"""workshop — the project notebook behind the workshop/office assistant, from docs/workshop-assistant.md. STEP 1 OF THAT DOC, AND DELIBERATELY ONLY STEP 1. The feasibility note ranks four capabilities and says to build this one first, because it is the only one with no perception problem in it and because it is where everything else lands: an identified mainboard is worth nothing until there is a project to attach it to. Camera identification, spec research and guide lookup all arrive later as *writers into this database*, not as separate stores. THE SPLIT THAT SHAPES THIS FILE -------------------------------- **SQLite is the index; the SMB share is the filing cabinet.** State that has to be queried — projects, decisions, what step you are on — lives in the database. (What you *own* is knowledge, not work product: it lives in knowledge.py's separate, never-pruned database, because there must be exactly one inventory.) Artefacts — datasheets, generated diagrams, photos of the board, notes you want to open from a laptop — live as ordinary files under WORKSPACE_DIR, which is exported read-write over SMB. Mixing the two gets you the worst of both: a database you cannot browse and files nothing can query. That is also why this service never returns file *contents* for anything in the workspace. It lists what is there and it hands back paths; opening the file is the share's job, and a second copy of the bytes over HTTP would be a second place for them to be stale. `GET /projects//files` is a directory listing, not a file server. WHY DECISIONS ARE THEIR OWN TABLE ---------------------------------- "What did I decide last Tuesday, and why" is the single question a project notebook exists to answer, and it is the one a pile of notes answers worst. A decision has a question, an answer, and a reason, and it is worth the four columns to keep those apart — the same argument this project already makes for every "flag the reasoning, not just the outcome" README in the repo. SECURITY BOUNDARY, same as every other service here: one bearer token, checked on every request including the GETs, failing closed when unset. This one holds no credentials and no personal data, but it *writes files into a share*, so see _safe_slug() for the one rule that keeps that from becoming a path-traversal question. """ from __future__ import annotations import json import logging import os import re import sqlite3 import sys from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlsplit import fleet import git_ops import gitea import health import knowledge LOG = logging.getLogger("workshop") TOKEN = os.environ.get("WORKSHOP_TOKEN", "") DB_PATH = Path(os.environ.get("WORKSHOP_DB_PATH", "/data/workshop.db")) # The SMB-exported workspace. Every project gets a directory under here; see # docs/workshop-assistant.md for the layout and for why it is a share at all. WORKSPACE_DIR = Path(os.environ.get("WORKSHOP_WORKSPACE_DIR", "/workspace")) PROJECT_SUBDIRS = ("notes", "datasheets", "diagrams", "photos", "scratch") MAX_JSON_BYTES = 256 * 1024 # A slug is a directory name on a writable share, which makes it the one piece of # user input in this service with a filesystem consequence. Anchored, no dots, no # separators — "../../etc" cannot survive this, and neither can a name that only # differs from another by case on a case-insensitive mount. SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$") def _now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def _safe_slug(value: str) -> str | None: """A slug, or None. Never raises, never sanitises-and-continues. Sanitising is the tempting option and the wrong one: silently turning `../x` into `x` means two different requests write to one directory, and the person who typed the first one never finds out. A bad slug is a 400. """ slug = str(value or "").strip().lower() return slug if SLUG_RE.match(slug) else None def _slugify(name: str) -> str: """A first-guess slug from a project name, for the common case where nobody supplied one. Not authoritative — the result still goes through _safe_slug().""" slug = re.sub(r"[^a-z0-9]+", "-", str(name or "").strip().lower()).strip("-") return slug[:63] or "project" # --- storage ---------------------------------------------------------------------- def _db() -> sqlite3.Connection: DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(DB_PATH, timeout=10) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") return conn def init_db() -> None: with _db() as conn: conn.executescript( """ CREATE TABLE IF NOT EXISTS projects ( id INTEGER PRIMARY KEY, slug TEXT NOT NULL UNIQUE, name TEXT NOT NULL, -- The HA area this project belongs to, so a room-scoped assistant can -- answer "what am I working on" with the projects of the room it was -- asked in. An area_id, the same vocabulary as everything else in this -- project — see docs/rooms-and-endpoints.md. room TEXT, -- active | parked | done. Parked is not done: a shelved project whose -- parts are still allocated to it is exactly what you want to find -- before buying those parts again. status TEXT NOT NULL DEFAULT 'active', summary TEXT, -- The project's Gitea repository, once one exists. A URL, not a flag: -- what anybody actually wants from this column is something to click -- or clone, and "true" would send every reader to Gitea to search. repo_url TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS notes ( id INTEGER PRIMARY KEY, project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, body TEXT NOT NULL, -- Who or what wrote this. 'human' or an assistant identifier — kept so -- a note the model wrote is never mistaken for something you said, the -- same distinction digest-engine draws between a source's claim and a -- fact. author TEXT NOT NULL DEFAULT 'human', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS decisions ( id INTEGER PRIMARY KEY, project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, question TEXT NOT NULL, answer TEXT NOT NULL, -- The column this table exists for. An answer without its reasoning is -- a thing you will re-litigate in three weeks. because TEXT, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS notes_by_project ON notes (project_id, created_at); CREATE INDEX IF NOT EXISTS decisions_by_project ON decisions (project_id, created_at); """ ) def _project_dir(slug: str) -> Path: return WORKSPACE_DIR / "projects" / slug def _ensure_project_dir(slug: str) -> list[str]: """Create the project's workspace directories. Best-effort by design. An unmounted or read-only share must not stop a project from being created — the notebook is in SQLite and is the part that matters, and a share that comes back later can have its directories made then (every write path calls this). Returns the subdirectories that exist, so the API can say plainly whether the share is working rather than implying it is. """ made = [] for sub in PROJECT_SUBDIRS: try: (_project_dir(slug) / sub).mkdir(parents=True, exist_ok=True) made.append(sub) except OSError: LOG.warning("workshop: could not create %s/%s in the workspace", slug, sub, exc_info=True) return made def _row(row) -> dict: return {k: row[k] for k in row.keys()} def list_projects(room: str = "") -> dict: with _db() as conn: if room: rows = conn.execute( "SELECT * FROM projects WHERE room = ? ORDER BY updated_at DESC", (room,) ).fetchall() else: rows = conn.execute("SELECT * FROM projects ORDER BY updated_at DESC").fetchall() return {"projects": [_row(r) for r in rows]} def create_project(payload: dict) -> dict: name = str(payload.get("name") or "").strip() if not name: return {"ok": False, "reason": "bad_field", "message": "'name' is required."} slug = _safe_slug(payload.get("slug") or _slugify(name)) if slug is None: return { "ok": False, "reason": "bad_slug", "message": "A slug is lowercase letters, digits and hyphens — it is also a " "directory name on the share, so it can't contain dots or slashes.", } room = str(payload.get("room") or "").strip() with _db() as conn: if conn.execute("SELECT 1 FROM projects WHERE slug = ?", (slug,)).fetchone(): return {"ok": False, "reason": "duplicate", "message": f"A project called {slug!r} already exists."} conn.execute( "INSERT INTO projects (slug, name, room, status, summary, created_at, updated_at) " "VALUES (?, ?, ?, 'active', ?, ?, ?)", (slug, name, room or None, str(payload.get("summary") or "").strip() or None, _now(), _now()), ) row = conn.execute("SELECT * FROM projects WHERE slug = ?", (slug,)).fetchone() made = _ensure_project_dir(slug) LOG.info("workshop: created project %s (workspace dirs: %s)", slug, ", ".join(made) or "none") return {"ok": True, "project": _row(row), "workspace": _workspace_status(slug, made)} def _workspace_status(slug: str, made: list[str]) -> dict: """Says plainly whether the share is usable, rather than implying it is by silence.""" return { "path": str(_project_dir(slug)), "subdirs": made, "available": len(made) == len(PROJECT_SUBDIRS), } def get_project(slug: str) -> dict | None: with _db() as conn: row = conn.execute("SELECT * FROM projects WHERE slug = ?", (slug,)).fetchone() if row is None: return None project = _row(row) pid = row["id"] project["notes"] = [_row(r) for r in conn.execute( "SELECT * FROM notes WHERE project_id = ? ORDER BY created_at DESC", (pid,))] project["decisions"] = [_row(r) for r in conn.execute( "SELECT * FROM decisions WHERE project_id = ? ORDER BY created_at DESC", (pid,))] # Hardware lives in the knowledge database, not here — see knowledge.py's docstring # on why there is exactly one inventory. A project reads its claims from it. project["hardware"] = knowledge.list_hardware(project_slug=slug)["hardware"] project["knowledge"] = knowledge.list_project_knowledge(slug)["project_knowledge"] project["workspace"] = _workspace_status(slug, [ s for s in PROJECT_SUBDIRS if (_project_dir(slug) / s).is_dir() ]) return project def update_project(slug: str, payload: dict) -> dict: updates = [] if "name" in payload: name = str(payload["name"] or "").strip() if not name: return {"ok": False, "reason": "bad_field", "message": "'name' cannot be empty."} updates.append(("name", name)) if "status" in payload: status = str(payload["status"] or "").strip().lower() if status not in ("active", "parked", "done"): return {"ok": False, "reason": "bad_field", "message": "status is active, parked or done."} updates.append(("status", status)) for field in ("summary", "room"): if field in payload: updates.append((field, str(payload[field] or "").strip() or None)) if not updates: return {"ok": False, "reason": "bad_field", "message": "Nothing to update."} with _db() as conn: if conn.execute("SELECT 1 FROM projects WHERE slug = ?", (slug,)).fetchone() is None: return {"ok": False, "reason": "no_such_project", "message": "No project with that slug."} for column, value in updates: conn.execute(f"UPDATE projects SET {column} = ? WHERE slug = ?", (value, slug)) conn.execute("UPDATE projects SET updated_at = ? WHERE slug = ?", (_now(), slug)) row = conn.execute("SELECT * FROM projects WHERE slug = ?", (slug,)).fetchone() return {"ok": True, "project": _row(row)} def _project_id(conn, slug: str) -> int | None: row = conn.execute("SELECT id FROM projects WHERE slug = ?", (slug,)).fetchone() return int(row["id"]) if row else None def add_note(slug: str, payload: dict) -> dict: body = str(payload.get("body") or "").strip() if not body: return {"ok": False, "reason": "bad_field", "message": "'body' is required."} author = str(payload.get("author") or "human").strip() or "human" with _db() as conn: pid = _project_id(conn, slug) if pid is None: return {"ok": False, "reason": "no_such_project", "message": "No project with that slug."} conn.execute( "INSERT INTO notes (project_id, body, author, created_at) VALUES (?, ?, ?, ?)", (pid, body, author, _now()), ) conn.execute("UPDATE projects SET updated_at = ? WHERE id = ?", (_now(), pid)) return {"ok": True} def add_decision(slug: str, payload: dict) -> dict: question = str(payload.get("question") or "").strip() answer = str(payload.get("answer") or "").strip() if not question or not answer: return {"ok": False, "reason": "bad_field", "message": "'question' and 'answer' are both required."} with _db() as conn: pid = _project_id(conn, slug) if pid is None: return {"ok": False, "reason": "no_such_project", "message": "No project with that slug."} conn.execute( "INSERT INTO decisions (project_id, question, answer, because, created_at) VALUES (?, ?, ?, ?, ?)", (pid, question, answer, str(payload.get("because") or "").strip() or None, _now()), ) conn.execute("UPDATE projects SET updated_at = ? WHERE id = ?", (_now(), pid)) return {"ok": True} def create_project_repo(slug: str, payload: dict) -> dict: """Give a project a Gitea repository and remember where it is. The repo name defaults to the project slug — the slug is already the project's stable, filesystem-safe identity, and having the directory on the share, the row in the database and the repository all answer to one name is worth more than letting each pick its own. """ with _db() as conn: row = conn.execute("SELECT * FROM projects WHERE slug = ?", (slug,)).fetchone() if row is None: return {"ok": False, "reason": "no_such_project", "message": "No project with that slug."} name = str(payload.get("name") or slug).strip() description = str(payload.get("description") or row["summary"] or f"Workshop project: {row['name']}") result = gitea.create_repo(name, description, payload.get("private")) if not result.get("ok"): return result url = result["repo"].get("html_url") # Set up code/ as a working tree pointed at the new remote, so the first commit # does not need a second call. Best-effort: a repo with no local tree yet is fine. result["local"] = git_ops.ensure_repo(_project_dir(slug), result["repo"].get("clone_url") or "") with _db() as conn: conn.execute("UPDATE projects SET repo_url = ?, updated_at = ? WHERE slug = ?", (url, _now(), slug)) LOG.info("workshop: project %s -> %s", slug, url) return result def cameras() -> dict: """The network cameras the workshop display may show. Network, not USB: every USB camera in this project is aimed at one fixed thing (an item held up to the kitchen display, an appliance door) at an angle that is useless for anything else. The cameras worth putting on a workshop screen are the ones already on the network and already ingested by go2rtc/Frigate. """ raw = os.environ.get("WORKSHOP_CAMERAS", "").strip() entries = [] for item in raw.split(","): item = item.strip() if not item or ":" not in item: continue name, _, stream = item.partition(":") if name.strip() and stream.strip(): entries.append({"name": name.strip(), "stream": stream.strip()}) return {"cameras": entries, "go2rtc_url": os.environ.get("GO2RTC_URL", "").rstrip("/")} def list_files(slug: str) -> dict: """A directory listing of the project's workspace — names and sizes, never bytes. The share serves the files; this says what is there. Symlinks are reported but never followed, and nothing outside the project directory is ever listed, because the whole point of the slug rule is that this path is not user-controlled beyond one directory name. """ base = _project_dir(slug) result: dict[str, list] = {} for sub in PROJECT_SUBDIRS: entries = [] directory = base / sub try: for entry in sorted(directory.iterdir(), key=lambda p: p.name): try: stat = entry.stat() entries.append({ "name": entry.name, "bytes": stat.st_size, "modified_at": datetime.fromtimestamp(stat.st_mtime, timezone.utc) .isoformat().replace("+00:00", "Z"), "is_dir": entry.is_dir(), }) except OSError: continue except OSError: # Missing or unreadable: an empty list plus `available: false` below, not # an error — a share that is temporarily not mounted must not look like a # project that has no files. pass result[sub] = entries return { "path": str(base), "available": base.is_dir(), "files": result, } # --- HTTP ------------------------------------------------------------------------- class Handler(BaseHTTPRequestHandler): server_version = "workshop/1" def log_message(self, format, *args): # noqa: A002 LOG.info("%s - %s", self.address_string(), format % args) def _authorized(self) -> bool: return bool(TOKEN) and self.headers.get("Authorization", "") == f"Bearer {TOKEN}" def _respond(self, status: HTTPStatus, payload) -> None: body = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(body) def _json_body(self) -> dict | None: try: length = int(self.headers.get("Content-Length") or 0) if length > MAX_JSON_BYTES: raise ValueError(f"body too large ({length} bytes)") payload = json.loads(self.rfile.read(length) or b"{}") except (ValueError, json.JSONDecodeError) as exc: self._respond(HTTPStatus.BAD_REQUEST, {"error": f"bad request body: {exc}"}) return None if not isinstance(payload, dict): self._respond(HTTPStatus.BAD_REQUEST, {"error": "body must be a JSON object"}) return None return payload def _slug_or_400(self, raw: str) -> str | None: slug = _safe_slug(raw) if slug is None: self._respond(HTTPStatus.BAD_REQUEST, {"error": f"{raw!r} is not a valid project slug"}) return slug def _result(self, result: dict) -> None: if result.get("ok"): self._respond(HTTPStatus.OK, result) return status = { "no_such_project": HTTPStatus.NOT_FOUND, "no_such_item": HTTPStatus.NOT_FOUND, "no_such_version": HTTPStatus.NOT_FOUND, "duplicate": HTTPStatus.CONFLICT, }.get(str(result.get("reason")), HTTPStatus.BAD_REQUEST) self._respond(status, result) def do_OPTIONS(self): # noqa: N802 self.send_response(HTTPStatus.NO_CONTENT) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type") self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS") self.end_headers() def do_GET(self): # noqa: N802 if not self._authorized(): self._respond(HTTPStatus.UNAUTHORIZED, {"error": "missing or invalid bearer token"}) return split = urlsplit(self.path) path, query = split.path, parse_qs(split.query) project_match = re.match(r"^/projects/([^/]+)$", path) files_match = re.match(r"^/projects/([^/]+)/files$", path) git_match = re.match(r"^/projects/([^/]+)/git$", path) fleet_script_match = re.match(r"^/fleet/script/([a-z0-9-]+)$", path) if path == "/projects": self._respond(HTTPStatus.OK, list_projects((query.get("room") or [""])[0].strip())) elif path == "/hardware": self._respond(HTTPStatus.OK, knowledge.list_hardware( (query.get("q") or [""])[0].strip(), (query.get("status") or [""])[0].strip(), (query.get("project") or [""])[0].strip(), )) elif path == "/knowledge/workflow": self._respond(HTTPStatus.OK, knowledge.list_workflow_notes( (query.get("activity") or [""])[0].strip())) elif path == "/knowledge/facts": self._respond(HTTPStatus.OK, knowledge.list_facts((query.get("q") or [""])[0].strip())) elif path == "/knowledge/projects": self._respond(HTTPStatus.OK, knowledge.list_project_knowledge( (query.get("project") or [""])[0].strip(), (query.get("q") or [""])[0].strip())) elif path == "/fleet": self._respond(HTTPStatus.OK, fleet.overview()) elif fleet_script_match: # What an endpoint of this platform should run. Serves the PUBLISHED version # only — a draft is invisible here by design. entry = fleet.published(fleet_script_match.group(1)) if entry is None: self._respond(HTTPStatus.NOT_FOUND, {"error": "no published script for that platform"}) else: self._respond(HTTPStatus.OK, entry) elif path == "/cameras": # Names and stream ids only — this service never proxies video. Putting a # Python HTTP server in the path of an H.264 stream is how a working camera # becomes a stuttering one; the kiosk talks to go2rtc directly. self._respond(HTTPStatus.OK, cameras()) elif path == "/health": # The workshop display's overlay reads this. `overall` is one word so the # display never re-derives "is anything wrong" from a list. self._respond(HTTPStatus.OK, health.latest()) elif path == "/context": # The payoff: everything the assistant should already know, in one call. self._respond(HTTPStatus.OK, knowledge.context( (query.get("activity") or [""])[0].strip(), (query.get("q") or [""])[0].strip(), (query.get("project") or [""])[0].strip(), )) elif project_match: slug = self._slug_or_400(project_match.group(1)) if slug is None: return project = get_project(slug) if project is None: self._respond(HTTPStatus.NOT_FOUND, {"error": "no such project"}) else: self._respond(HTTPStatus.OK, project) elif git_match: slug = self._slug_or_400(git_match.group(1)) if slug is not None: self._respond(HTTPStatus.OK, git_ops.status(_project_dir(slug))) elif files_match: slug = self._slug_or_400(files_match.group(1)) if slug is not None: self._respond(HTTPStatus.OK, list_files(slug)) else: self._respond(HTTPStatus.NOT_FOUND, {"error": "no such endpoint"}) def do_DELETE(self): # noqa: N802 """Corrections. "No retention period" means nothing expires by itself — it does not mean nothing is ever wrong, and a knowledge base you cannot correct is one people route around.""" if not self._authorized(): self._respond(HTTPStatus.UNAUTHORIZED, {"error": "missing or invalid bearer token"}) return path = urlsplit(self.path).path match = re.match(r"^/(hardware|knowledge/workflow|knowledge/facts|knowledge/projects)/(\d+)$", path) if not match: self._respond(HTTPStatus.NOT_FOUND, {"error": "no such endpoint"}) return table = { "hardware": "hardware", "knowledge/workflow": "workflow_notes", "knowledge/facts": "facts", "knowledge/projects": "project_knowledge", }[match.group(1)] self._result(knowledge.delete_row(table, int(match.group(2)))) def do_POST(self): # noqa: N802 if not self._authorized(): self._respond(HTTPStatus.UNAUTHORIZED, {"error": "missing or invalid bearer token"}) return path = urlsplit(self.path).path note_match = re.match(r"^/projects/([^/]+)/notes$", path) knowledge_match = re.match(r"^/projects/([^/]+)/knowledge$", path) repo_match = re.match(r"^/projects/([^/]+)/repo$", path) commit_match = re.match(r"^/projects/([^/]+)/commit$", path) branch_match = re.match(r"^/projects/([^/]+)/branch$", path) scrub_match = re.match(r"^/projects/([^/]+)/scrub-request$", path) hardware_match = re.match(r"^/hardware/(\d+)$", path) decision_match = re.match(r"^/projects/([^/]+)/decisions$", path) project_match = re.match(r"^/projects/([^/]+)$", path) payload = self._json_body() if payload is None: return if path == "/projects": self._result(create_project(payload)) elif path == "/hardware": self._result(knowledge.add_hardware(payload)) elif hardware_match: self._result(knowledge.update_hardware(int(hardware_match.group(1)), payload)) elif path == "/fleet/upload": self._result(fleet.upload(payload.get("platform", ""), payload.get("body", ""), payload.get("note", ""))) elif path == "/fleet/publish": # The second, deliberate action. Upload is not deploy — see fleet.py. try: version = int(payload.get("version")) except (TypeError, ValueError): self._respond(HTTPStatus.BAD_REQUEST, {"error": "'version' is required"}) return self._result(fleet.publish(payload.get("platform", ""), version)) elif path == "/fleet/report": self._result(fleet.report(payload)) elif path == "/knowledge/workflow": self._result(knowledge.add_workflow_note(payload)) elif path == "/knowledge/facts": self._result(knowledge.add_fact(payload)) elif note_match: slug = self._slug_or_400(note_match.group(1)) if slug is not None: self._result(add_note(slug, payload)) elif repo_match: slug = self._slug_or_400(repo_match.group(1)) if slug is not None: self._result(create_project_repo(slug, payload)) elif commit_match: slug = self._slug_or_400(commit_match.group(1)) if slug is not None: self._result(git_ops.commit( _project_dir(slug), payload.get("message", ""), bool(payload.get("push", True)))) elif branch_match: slug = self._slug_or_400(branch_match.group(1)) if slug is not None: self._result(git_ops.branch(_project_dir(slug), payload.get("name", ""))) elif scrub_match: # Never executes anything. Returns the commands for a human to run from the # repository on the share — see git_ops.scrub_instructions(). slug = self._slug_or_400(scrub_match.group(1)) if slug is not None: self._result(git_ops.scrub_instructions( _project_dir(slug), payload.get("path", ""), payload.get("note", ""))) elif knowledge_match: slug = self._slug_or_400(knowledge_match.group(1)) if slug is not None: self._result(knowledge.add_project_knowledge(slug, payload)) elif decision_match: slug = self._slug_or_400(decision_match.group(1)) if slug is not None: self._result(add_decision(slug, payload)) elif project_match: slug = self._slug_or_400(project_match.group(1)) if slug is not None: self._result(update_project(slug, payload)) else: self._respond(HTTPStatus.NOT_FOUND, {"error": "no such endpoint"}) def main() -> int: logging.basicConfig( level=os.environ.get("LOG_LEVEL", "INFO").upper(), format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", ) if not TOKEN: LOG.error("WORKSHOP_TOKEN is not set — every request will be rejected until it is.") init_db() knowledge.init_db() health.init_db() fleet.init_db() health.start_poller() try: (WORKSPACE_DIR / "projects").mkdir(parents=True, exist_ok=True) except OSError: LOG.warning( "workshop: %s is not writable — the notebook still works, but nothing can be " "filed on the share until it is", WORKSPACE_DIR, exc_info=True, ) port = int(os.environ.get("WORKSHOP_PORT", "8102")) server = ThreadingHTTPServer(("0.0.0.0", port), Handler) LOG.info("workshop listening on :%d (db: %s, workspace: %s)", port, DB_PATH, WORKSPACE_DIR) try: server.serve_forever() except KeyboardInterrupt: pass return 0 if __name__ == "__main__": sys.exit(main())