From d0fd37c1b56766b39b21e7e80de5353af35b59b5 Mon Sep 17 00:00:00 2001 From: Michele Bigi Date: Sun, 20 Sep 2026 19:31:41 +0200 Subject: [PATCH] Rewrite leftover HTTP routers except auth. Project ingest, pipeline enqueue/SSE, reports, admin, contact, survey, and feedback now resolve from src. Auth router is unchanged. --- periscope/src/backend/routers/admin.py | 499 +++++++++++ periscope/src/backend/routers/contact.py | 118 +++ periscope/src/backend/routers/deps.py | 33 + periscope/src/backend/routers/feedback.py | 250 ++++++ periscope/src/backend/routers/pipeline.py | 783 +++++++++++++++++ periscope/src/backend/routers/projects.py | 990 ++++++++++++++++++++++ periscope/src/backend/routers/reports.py | 378 +++++++++ periscope/src/backend/routers/survey.py | 69 ++ tests/test_periscope_routers_rewrite.py | 49 ++ 9 files changed, 3169 insertions(+) create mode 100644 periscope/src/backend/routers/admin.py create mode 100644 periscope/src/backend/routers/contact.py create mode 100644 periscope/src/backend/routers/deps.py create mode 100644 periscope/src/backend/routers/feedback.py create mode 100644 periscope/src/backend/routers/pipeline.py create mode 100644 periscope/src/backend/routers/projects.py create mode 100644 periscope/src/backend/routers/reports.py create mode 100644 periscope/src/backend/routers/survey.py create mode 100644 tests/test_periscope_routers_rewrite.py diff --git a/periscope/src/backend/routers/admin.py b/periscope/src/backend/routers/admin.py new file mode 100644 index 0000000..b51790a --- /dev/null +++ b/periscope/src/backend/routers/admin.py @@ -0,0 +1,499 @@ +"""Admin library, users, usage, sweeper, and report overrides.""" +from __future__ import annotations + +import asyncio +import re +from datetime import datetime, timezone + +import httpx +from fastapi import APIRouter, HTTPException, Request +from fastapi.responses import JSONResponse +from pydantic import BaseModel + +from backend.config import settings +from backend.periscopex.utils import safe_mpn +from backend.routers.deps import get_storage, resolve_or_404 +from backend.services import admin_settings as settings_svc +from backend.services import projects as proj_svc +from backend.services.billing_hook import get_billing + +router = APIRouter(prefix="/admin", tags=["admin"]) +_CLERK_USERS = "https://api.clerk.com/v1/users" + + +async def is_admin(request: Request) -> bool: + cached = getattr(request.state, "_is_admin", None) + if cached is not None: + return cached + + user_id: str = request.state.user_id + if not settings.use_auth: + request.state._is_admin = True + return True + + if settings.use_local_auth: + from backend.services import local_users + + user = local_users.get_user(user_id) + ok = bool(user and user.is_admin) + request.state._is_admin = ok + return ok + + ok = False + try: + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{_CLERK_USERS}/{user_id}", + headers={"Authorization": f"Bearer {settings.clerk_secret_key}"}, + ) + if resp.status_code == 200: + ok = resp.json().get("public_metadata", {}).get("role") == "admin" + except Exception: + ok = False + request.state._is_admin = ok + return ok + + +async def _require_admin(request: Request) -> str: + if not await is_admin(request): + raise HTTPException(403, "Admin access required") + return request.state.user_id + + +def _safe_name(name: str) -> str: + safe = safe_mpn(name) + if ".." in safe or not re.match(r"^[A-Za-z0-9]", safe): + raise HTTPException(400, "Invalid component name") + return safe + + +def _component_key(storage, component_type: str, safe: str) -> str: + if component_type == "ic": + return f"library/extracted/{safe}.json" + if component_type == "passive": + return f"library/patterns/{safe}.json" + if component_type == "simple": + key = f"library/passives/{safe}.json" + if storage.exists(key): + return key + return f"library/models/{safe}.json" + raise HTTPException(400, f"Unknown component type: {component_type}") + + +@router.get("/components") +async def list_components(request: Request): + await _require_admin(request) + catalog = proj_svc.list_library_catalog(get_storage(request)) + return JSONResponse( + content={ + "ics": catalog["ics"], + "passives": catalog["passives"], + "simple": catalog["simple"], + }, + headers={"Cache-Control": "no-store"}, + ) + + +@router.get("/components/{component_type}/{name:path}") +async def get_component(component_type: str, name: str, request: Request): + await _require_admin(request) + storage = get_storage(request) + key = _component_key(storage, component_type, _safe_name(name)) + if not storage.exists(key): + raise HTTPException(404, f"Component not found: {name}") + return JSONResponse(content=storage.read_json(key)) + + +@router.delete("/components/{component_type}/{name:path}") +async def delete_component(component_type: str, name: str, request: Request): + await _require_admin(request) + storage = get_storage(request) + key = _component_key(storage, component_type, _safe_name(name)) + if not storage.exists(key): + raise HTTPException(404, f"Component not found: {name}") + storage.delete_key(key) + from backend.services.datasheet_store import delete_datasheet_ref + + deleted_datasheets = 0 + if delete_datasheet_ref(storage, name): + deleted_datasheets += 1 + ds_key = f"library/datasheets/{_safe_name(name)}.pdf" + if storage.exists(ds_key): + storage.delete_key(ds_key) + deleted_datasheets += 1 + return {"deleted": key, "deleted_datasheets": deleted_datasheets} + + +def _clerk_profile_fields(clerk: dict) -> dict: + first = clerk.get("first_name") or "" + last = clerk.get("last_name") or "" + emails = clerk.get("email_addresses") or [] + return { + "name": f"{first} {last}".strip() or None, + "email": emails[0].get("email_address") if emails else None, + "image_url": clerk.get("image_url"), + } + + +def _base_admin_user(storage, uid: str) -> dict: + try: + project_count = len(proj_svc.list_projects(storage, uid)) + except Exception: + project_count = 0 + try: + balance = get_billing().get_balance(storage, uid) + except Exception: + balance = 0.0 + return { + "user_id": uid, + "project_count": project_count, + "balance": round(balance, 4), + "name": None, + "email": None, + "image_url": None, + } + + +def _user_ids_from_prefix(storage) -> set[str]: + ids: set[str] = set() + for entry in storage.list_prefix("users/"): + parts = entry.split("/") + if len(parts) >= 2 and parts[1]: + ids.add(parts[1]) + return ids + + +async def _enrich_clerk_profiles(users: dict[str, dict]) -> None: + sem = asyncio.Semaphore(10) + + async with httpx.AsyncClient(timeout=10.0) as client: + + async def _one(uid: str) -> None: + async with sem: + try: + resp = await client.get( + f"{_CLERK_USERS}/{uid}", + headers={"Authorization": f"Bearer {settings.clerk_secret_key}"}, + ) + if resp.status_code == 200: + users[uid].update(_clerk_profile_fields(resp.json())) + except Exception: + return + + await asyncio.gather(*(_one(uid) for uid in users)) + + +@router.get("/users") +async def list_users(request: Request): + await _require_admin(request) + storage = get_storage(request) + user_ids = _user_ids_from_prefix(storage) + user_ids.update(get_billing().list_user_ids(storage)) + users = {uid: _base_admin_user(storage, uid) for uid in user_ids} + if settings.use_auth and users: + await _enrich_clerk_profiles(users) + return list(users.values()) + + +@router.get("/users/search") +async def search_users(request: Request, email: str): + await _require_admin(request) + storage = get_storage(request) + email = email.strip() + if not email: + return [] + if not settings.use_auth: + raise HTTPException(400, "User search requires authentication to be enabled") + async with httpx.AsyncClient(timeout=10.0) as client: + try: + resp = await client.get( + _CLERK_USERS, + params={"email_address": [email]}, + headers={"Authorization": f"Bearer {settings.clerk_secret_key}"}, + ) + except Exception as exc: + raise HTTPException(502, "Failed to look up user") from exc + if resp.status_code != 200: + raise HTTPException(502, "Failed to look up user") + results: list[dict] = [] + for clerk in resp.json(): + uid = clerk.get("id") + if not uid: + continue + entry = _base_admin_user(storage, uid) + entry.update(_clerk_profile_fields(clerk)) + results.append(entry) + return results + + +@router.get("/usage") +async def get_usage(request: Request): + await _require_admin(request) + storage = get_storage(request) + seen: set[str] = set() + user_rows: list[dict] = [] + grand_total = 0.0 + for uid in _user_ids_from_prefix(storage): + if uid in seen: + continue + seen.add(uid) + projects = proj_svc.list_projects(storage, uid) + user_cost = 0.0 + details = [] + for p in projects: + cost = p.total_cost_usd or 0.0 + user_cost += cost + details.append( + { + "id": p.id, + "name": p.name, + "status": p.status, + "cost_usd": cost, + "created": p.created, + } + ) + user_rows.append( + { + "user_id": uid, + "project_count": len(projects), + "total_cost_usd": round(user_cost, 4), + "projects": details, + "name": None, + "email": None, + } + ) + grand_total += user_cost + if settings.use_auth and user_rows: + async with httpx.AsyncClient() as client: + for row in user_rows: + try: + resp = await client.get( + f"{_CLERK_USERS}/{row['user_id']}", + headers={"Authorization": f"Bearer {settings.clerk_secret_key}"}, + ) + if resp.status_code == 200: + clerk = resp.json() + first = clerk.get("first_name") or "" + last = clerk.get("last_name") or "" + row["name"] = f"{first} {last}".strip() or None + emails = clerk.get("email_addresses") or [] + row["email"] = emails[0].get("email_address") if emails else None + except Exception: + continue + return {"grand_total_usd": round(grand_total, 4), "users": user_rows} + + +async def _enrich_with_clerk_info( + items: list[dict], + uid_key: str = "user_id", + name_key: str = "owner_name", + email_key: str = "owner_email", +) -> None: + if not settings.use_auth or not items: + return + cache: dict[str, dict] = {} + async with httpx.AsyncClient() as client: + for item in items: + uid = item[uid_key] + if uid not in cache: + try: + resp = await client.get( + f"{_CLERK_USERS}/{uid}", + headers={"Authorization": f"Bearer {settings.clerk_secret_key}"}, + ) + if resp.status_code == 200: + clerk = resp.json() + first = clerk.get("first_name") or "" + last = clerk.get("last_name") or "" + emails = clerk.get("email_addresses") or [] + cache[uid] = { + name_key: f"{first} {last}".strip() or None, + email_key: emails[0].get("email_address") if emails else None, + } + else: + cache[uid] = {name_key: None, email_key: None} + except Exception: + cache[uid] = {name_key: None, email_key: None} + item.update(cache[uid]) + + +@router.get("/projects") +async def list_all_projects(request: Request): + await _require_admin(request) + storage = get_storage(request) + all_projects: list[dict] = [] + for uid in _user_ids_from_prefix(storage): + for p in proj_svc.list_projects(storage, uid): + all_projects.append( + { + "id": p.id, + "name": p.name, + "user_id": p.user_id, + "status": p.status, + "created": p.created, + "updated": p.updated, + "has_bom": p.has_bom, + "has_netlist": p.has_netlist, + "datasheet_count": p.datasheet_count, + "total_cost_usd": p.total_cost_usd, + "pipeline_state": p.pipeline_state, + "summary": p.summary, + "owner_name": None, + "owner_email": None, + } + ) + await _enrich_with_clerk_info(all_projects) + return all_projects + + +@router.get("/runs") +async def list_running_pipelines(request: Request): + from backend.services import job_runner + + await _require_admin(request) + storage = get_storage(request) + now = datetime.now(timezone.utc) + runs: list[dict] = [] + for uid in _user_ids_from_prefix(storage): + prefix = f"users/{uid}/projects/" + for proj_entry in storage.list_prefix(prefix): + meta_key = ( + proj_entry + if proj_entry.endswith("/project.json") + else f"{proj_entry}/project.json" + ) + if not storage.exists(meta_key): + continue + try: + meta = proj_svc.ProjectMeta.model_validate(storage.read_json(meta_key)) + except Exception: + continue + if meta.status not in (proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING): + continue + if proj_svc.heal_if_pipeline_finished(storage, uid, meta.id) is not None: + continue + exec_state = "unknown" + if meta.execution_name: + exec_state = job_runner.get_execution_state(meta.execution_name) + elif not job_runner.use_cloud_run_jobs(): + exec_state = job_runner.get_execution_state(f"local/projects/{meta.id}") + if exec_state in ("succeeded", "failed", "cancelled"): + try: + last_update = datetime.fromisoformat(meta.updated) + age = (now - last_update).total_seconds() + except Exception: + age = settings.pipeline_sweeper_stale_seconds + 1 + if age >= settings.pipeline_sweeper_stale_seconds: + proj_svc.mark_stale_running( + storage, + uid, + meta.id, + f"Worker terminated (execution state={exec_state}); please restart.", + ) + continue + try: + started_at = datetime.fromisoformat(meta.updated) + except Exception: + started_at = now + runs.append( + { + "project_id": meta.id, + "project_name": meta.name, + "user_id": uid, + "status": meta.status, + "execution_name": meta.execution_name, + "execution_state": exec_state, + "started_at": started_at.isoformat(), + "duration_seconds": int((now - started_at).total_seconds()), + "owner_name": None, + "owner_email": None, + } + ) + await _enrich_with_clerk_info(runs) + return runs + + +class UpdateMinVersionRequest(BaseModel): + min_model_version: str + + +@router.get("/settings") +async def get_settings(request: Request): + await _require_admin(request) + storage = get_storage(request) + data = settings_svc.get_admin_settings(storage) + data["default_model_version"] = settings.get_default_model_version() + return data + + +@router.put("/settings/min-model-version") +async def set_min_model_version(req: UpdateMinVersionRequest, request: Request): + await _require_admin(request) + storage = get_storage(request) + try: + settings_svc.set_min_model_version(storage, req.min_model_version) + except Exception as e: + raise HTTPException(400, f"Invalid version: {e}") + return {"min_model_version": req.min_model_version} + + +class TestEmailRequest(BaseModel): + to_email: str + + +@router.post("/test-email") +async def test_email(req: TestEmailRequest, request: Request): + await _require_admin(request) + from backend.services.email import send_test_email + + return await send_test_email(req.to_email) + + +@router.post("/projects/{project_id}/mark-complete") +async def mark_project_complete(project_id: str, request: Request): + await _require_admin(request) + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + if meta.status in (proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED): + raise HTTPException(409, "Cannot mark a running pipeline complete; cancel it first") + if meta.status == "complete": + return {"status": "complete", "project_id": project_id} + proj_svc.update_project( + storage, + owner, + project_id, + status="complete", + pause_checkpoint=None, + pause_reason=None, + ) + return {"status": "complete", "project_id": project_id} + + +@router.delete("/projects/{project_id}/findings/{finding_id}") +async def delete_finding(project_id: str, finding_id: str, request: Request): + await _require_admin(request) + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + key = f"{proj_svc.project_prefix(owner, project_id)}/report.json" + if not storage.exists(key): + raise HTTPException(404, "Report not found") + report = storage.read_json(key) + findings = report.get("findings", []) or [] + remaining = [f for f in findings if f.get("finding_id") != finding_id] + if len(remaining) == len(findings): + raise HTTPException(404, f"Finding not found: {finding_id}") + summary = {"total": len(remaining), "ERROR": 0, "WARNING": 0, "INFO": 0} + for f in remaining: + status = f.get("status") + if status in summary: + summary[status] += 1 + report["findings"] = remaining + report["summary"] = summary + storage.write_json(key, report) + proj_svc.update_project(storage, owner, project_id, summary=summary) + return { + "deleted": finding_id, + "project_id": project_id, + "remaining": len(remaining), + "summary": summary, + } diff --git a/periscope/src/backend/routers/contact.py b/periscope/src/backend/routers/contact.py new file mode 100644 index 0000000..a461634 --- /dev/null +++ b/periscope/src/backend/routers/contact.py @@ -0,0 +1,118 @@ +"""Unauthenticated contact mailbox. Honeypot and per-IP delay live here.""" +from __future__ import annotations + +import html +import logging +import time +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText + +from fastapi import APIRouter, Request +from pydantic import BaseModel, EmailStr, Field + +from backend.config import settings +from backend.services.email import _send_raw + +logger = logging.getLogger(__name__) +router = APIRouter() + +_COOLDOWN_S = 60 +_seen: dict[str, float] = {} + + +class ContactRequest(BaseModel): + name: str = Field(..., min_length=1, max_length=200) + email: EmailStr = Field(..., max_length=254) + message: str = Field(..., min_length=1, max_length=5000) + company: str = Field("", max_length=200) + subject: str = Field("", max_length=200) + honeypot: str = Field("", alias="_honey") + + +class ContactResponse(BaseModel): + success: bool + message: str + + +def _client_ip(request: Request) -> str: + forwarded = request.headers.get("x-forwarded-for", "").split(",")[0].strip() + if forwarded: + return forwarded + return request.client.host if request.client else "unknown" + + +def _compose(data: ContactRequest) -> MIMEMultipart: + msg = MIMEMultipart("alternative") + msg["From"] = f"Periscope <{settings.email_sender}>" + msg["To"] = settings.contact_recipient + msg["Reply-To"] = data.email + title = data.subject or "New message" + msg["Subject"] = f"[Periscope Contact] {title} from {data.name}" + + plain = [f"Name: {data.name}", f"Email: {data.email}"] + if data.company: + plain.append(f"Company: {data.company}") + if data.subject: + plain.append(f"Subject: {data.subject}") + plain.extend(["", data.message, "", "— Sent from the Periscope contact form"]) + msg.attach(MIMEText("\n".join(plain), "plain")) + + def cell(label: str, value: str) -> str: + return ( + "" + f'{html.escape(label)}' + f'{value}' + "" + ) + + rows = cell("Name", html.escape(data.name)) + email = html.escape(data.email) + rows += ( + "" + 'Email' + f'{email}' + "" + ) + if data.company: + rows += cell("Company", html.escape(data.company)) + if data.subject: + rows += cell("Subject", html.escape(data.subject)) + body = html.escape(data.message) + html_body = f"""\ +
+

New contact form submission

+ {rows}
+
{body}
+

Sent from the Periscope contact form

+
""" + msg.attach(MIMEText(html_body, "html")) + return msg + + +@router.post("/contact", response_model=ContactResponse) +async def submit_contact(data: ContactRequest, request: Request): + ok = "Message sent! We'll get back to you soon." + if data.honeypot: + return ContactResponse(success=True, message=ok) + + ip = _client_ip(request) + now = time.time() + last = _seen.get(ip) + if last is not None and now - last < _COOLDOWN_S: + return ContactResponse( + success=False, message="Please wait a minute before submitting again." + ) + _seen[ip] = now + if len(_seen) > 1000: + stale = now - _COOLDOWN_S + for key in [k for k, ts in _seen.items() if ts < stale]: + del _seen[key] + + if not settings.use_email or not settings.contact_recipient: + logger.warning("Contact form submitted but email is not configured") + return ContactResponse( + success=False, message="Email is not configured on this server." + ) + + await _send_raw(settings.contact_recipient, _compose(data), "Contact form") + return ContactResponse(success=True, message=ok) diff --git a/periscope/src/backend/routers/deps.py b/periscope/src/backend/routers/deps.py new file mode 100644 index 0000000..074831f --- /dev/null +++ b/periscope/src/backend/routers/deps.py @@ -0,0 +1,33 @@ +"""Request helpers shared by HTTP routers.""" +from __future__ import annotations + +from fastapi import HTTPException, Request + +from backend.services import projects as proj_svc +from backend.services.storage import StorageBackend + + +def get_storage(request: Request) -> StorageBackend: + return request.app.state.storage + + +def get_user_id(request: Request) -> str: + return request.state.user_id + + +async def resolve_or_404(request: Request, project_id: str) -> tuple[str, proj_svc.ProjectMeta]: + """Owner, collaborator, or admin; otherwise 404 (never 403 for missing projects).""" + storage = get_storage(request) + user_id = get_user_id(request) + hit = proj_svc.resolve_project_access(storage, user_id, project_id) + if hit: + return hit + + from backend.routers.admin import is_admin + + if await is_admin(request): + found = proj_svc.find_project_any_user(storage, project_id) + if found: + return found + + raise HTTPException(404, "Project not found") diff --git a/periscope/src/backend/routers/feedback.py b/periscope/src/backend/routers/feedback.py new file mode 100644 index 0000000..faf0465 --- /dev/null +++ b/periscope/src/backend/routers/feedback.py @@ -0,0 +1,250 @@ +"""Feedback tickets: per-user JSON plus JSONL indexes for listing.""" +from __future__ import annotations + +import json +import logging +import uuid +from datetime import datetime, timezone +from typing import Literal + +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel, Field + +from backend.routers.deps import get_storage, get_user_id +from backend.services.storage import StorageBackend + +logger = logging.getLogger(__name__) +router = APIRouter() + +_TICKETS = "admin/feedback/tickets/" +_BY_USER = "admin/feedback/index/by_user/" +_BY_PROJECT = "admin/feedback/index/by_project/" +_ALL = "admin/feedback/index/all.jsonl" + +FeedbackType = Literal["bug", "rule_feedback", "feature_request"] +FeedbackStatus = Literal["open", "acknowledged", "resolved"] + + +def _ticket_key(ticket_id: str) -> str: + return f"{_TICKETS}{ticket_id}.json" + + +def _user_index_key(user_id: str) -> str: + return f"{_BY_USER}{user_id}.jsonl" + + +def _project_index_key(project_id: str) -> str: + return f"{_BY_PROJECT}{project_id}.jsonl" + + +class FeedbackTicket(BaseModel): + ticket_id: str + user_id: str + user_name: str | None = None + user_email: str | None = None + project_id: str | None = None + project_name: str | None = None + type: FeedbackType + status: FeedbackStatus = "open" + finding_id: str | None = None + finding_text: str | None = None + finding_designator: str | None = None + finding_mpn: str | None = None + finding_status: str | None = None + message: str + admin_notes: str | None = None + created_at: str + updated_at: str + + +class CreateFeedbackRequest(BaseModel): + type: FeedbackType + message: str = Field(..., min_length=1, max_length=5000) + project_id: str | None = None + project_name: str | None = None + user_name: str | None = None + user_email: str | None = None + finding_id: str | None = None + finding_text: str | None = None + finding_designator: str | None = None + finding_mpn: str | None = None + finding_status: str | None = None + + +class UpdateFeedbackRequest(BaseModel): + status: FeedbackStatus | None = None + admin_notes: str | None = None + + +def _append_index(storage: StorageBackend, key: str, entry: dict) -> None: + prior = storage.read_text(key) if storage.exists(key) else "" + storage.write_text(key, prior + json.dumps(entry) + "\n") + + +def _read_index(storage: StorageBackend, key: str) -> list[dict]: + if not storage.exists(key): + return [] + rows: list[dict] = [] + for line in storage.read_text(key).splitlines(): + line = line.strip() + if not line: + continue + try: + rows.append(json.loads(line)) + except json.JSONDecodeError: + logger.warning("Skipping malformed feedback index line in %s", key) + return rows + + +def _read_ticket(storage: StorageBackend, ticket_id: str) -> FeedbackTicket | None: + key = _ticket_key(ticket_id) + if not storage.exists(key): + return None + try: + return FeedbackTicket(**storage.read_json(key)) + except Exception: + logger.warning("Failed to read ticket %s", ticket_id) + return None + + +def _tickets_from_index( + storage: StorageBackend, + index_key: str, + *, + status: str | None = None, + ticket_type: str | None = None, + project_id: str | None = None, +) -> list[FeedbackTicket]: + out: list[FeedbackTicket] = [] + for entry in reversed(_read_index(storage, index_key)): + tid = entry.get("ticket_id") + if not tid: + continue + ticket = _read_ticket(storage, tid) + if ticket is None: + continue + if status and ticket.status != status: + continue + if ticket_type and ticket.type != ticket_type: + continue + if project_id and ticket.project_id != project_id: + continue + out.append(ticket) + return out + + +@router.post("/feedback", response_model=FeedbackTicket) +async def create_feedback(body: CreateFeedbackRequest, request: Request): + storage = get_storage(request) + user_id = get_user_id(request) + now = datetime.now(timezone.utc).isoformat() + ticket_id = uuid.uuid4().hex[:12] + ticket = FeedbackTicket( + ticket_id=ticket_id, + user_id=user_id, + user_name=body.user_name, + user_email=body.user_email, + project_id=body.project_id, + project_name=body.project_name, + type=body.type, + status="open", + finding_id=body.finding_id, + finding_text=body.finding_text, + finding_designator=body.finding_designator, + finding_mpn=body.finding_mpn, + finding_status=body.finding_status, + message=body.message, + admin_notes=None, + created_at=now, + updated_at=now, + ) + storage.write_json(_ticket_key(ticket_id), ticket.model_dump()) + index = {"ticket_id": ticket_id, "created_at": now} + _append_index(storage, _user_index_key(user_id), index) + _append_index(storage, _ALL, index) + if body.project_id: + _append_index(storage, _project_index_key(body.project_id), index) + try: + from backend.services.email import send_feedback_received_email + + await send_feedback_received_email( + ticket_id=ticket_id, + user_id=user_id, + feedback_type=body.type, + message=body.message, + submitter_name=body.user_name, + submitter_email=body.user_email, + project_name=body.project_name, + project_id=body.project_id, + finding_designator=body.finding_designator, + finding_mpn=body.finding_mpn, + finding_status=body.finding_status, + finding_text=body.finding_text, + ) + except Exception: + logger.exception("Failed to enqueue feedback-received email for %s", ticket_id) + return ticket + + +@router.get("/feedback", response_model=list[FeedbackTicket]) +async def list_my_feedback(request: Request, status: str | None = None): + return _tickets_from_index( + get_storage(request), _user_index_key(get_user_id(request)), status=status + ) + + +@router.get("/admin/feedback", response_model=list[FeedbackTicket]) +async def list_all_feedback( + request: Request, + status: str | None = None, + type: str | None = None, + project_id: str | None = None, +): + from backend.routers.admin import _require_admin + + await _require_admin(request) + return _tickets_from_index( + get_storage(request), + _ALL, + status=status, + ticket_type=type, + project_id=project_id, + ) + + +@router.put("/admin/feedback/{ticket_id}", response_model=FeedbackTicket) +async def update_feedback(ticket_id: str, body: UpdateFeedbackRequest, request: Request): + from backend.routers.admin import _require_admin + + await _require_admin(request) + storage = get_storage(request) + ticket = _read_ticket(storage, ticket_id) + if ticket is None: + raise HTTPException(404, "Ticket not found") + previous = (ticket.admin_notes or "").strip() + if body.status is not None: + ticket.status = body.status + if body.admin_notes is not None: + ticket.admin_notes = body.admin_notes + ticket.updated_at = datetime.now(timezone.utc).isoformat() + storage.write_json(_ticket_key(ticket_id), ticket.model_dump()) + notes = (ticket.admin_notes or "").strip() + if notes and notes != previous: + try: + from backend.services.email import send_feedback_reply_email + + await send_feedback_reply_email( + user_id=ticket.user_id, + reply_text=notes, + original_message=ticket.message, + recipient_name=ticket.user_name, + recipient_email=ticket.user_email, + project_name=ticket.project_name, + finding_designator=ticket.finding_designator, + finding_mpn=ticket.finding_mpn, + ) + except Exception: + logger.exception( + "Failed to enqueue feedback-reply email for ticket %s", ticket_id + ) + return ticket diff --git a/periscope/src/backend/routers/pipeline.py b/periscope/src/backend/routers/pipeline.py new file mode 100644 index 0000000..10899f1 --- /dev/null +++ b/periscope/src/backend/routers/pipeline.py @@ -0,0 +1,783 @@ +"""HTTP surface for analysis, placement, and PCB jobs. + +The API process never runs extract/review itself: it CAS-transitions +project meta, asks ``job_runner`` to spawn a worker, and SSE-tails the +storage event log (plus the in-memory ``job_workspace`` broker). +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import AsyncIterator, Callable +from typing import Any, Literal + +from fastapi import APIRouter, HTTPException, Request +from pydantic import BaseModel +from sse_starlette.sse import EventSourceResponse + +from backend.routers.deps import get_storage, get_user_id, resolve_or_404 +from backend.services import event_bridge, job_runner, job_workspace +from backend.services import projects as proj_svc +from backend.services.billing_hook import get_billing +from backend.services.cost_estimator import estimate_pipeline_cost +from backend.services.pcb_pipeline import ( + analysis_busy as pcb_analysis_busy, + pcb_busy, + pcb_sse_terminal_from_status, + placement_busy as pcb_placement_busy, +) +from backend.services.placement_pipeline import ( + analysis_busy as placement_analysis_busy, + placement_busy, +) + +log = logging.getLogger(__name__) + +router = APIRouter(tags=["pipeline"]) + +_REGEN_STAGES = {"derating"} +_ANALYSIS_IDLE = frozenset({ + proj_svc.STATUS_DRAFT, + proj_svc.STATUS_COMPLETE, + proj_svc.STATUS_ERROR, + proj_svc.STATUS_CANCELLED, +}) +_REPROCESS_IDLE = frozenset({ + proj_svc.STATUS_COMPLETE, + proj_svc.STATUS_ERROR, + proj_svc.STATUS_CANCELLED, +}) +_WORKER_DONE = frozenset({"succeeded", "failed", "cancelled"}) +_ANALYSIS_SSE_DONE = frozenset({ + "pipeline_complete", + "pipeline_error", + "pipeline_cancelled", + "pipeline_paused", +}) +_PLACEMENT_IDLE = frozenset({"draft", "complete", "error", "cancelled"}) +_PLACEMENT_SSE_DONE = frozenset({ + "placement_complete", + "placement_error", + "placement_cancelled", +}) +_PCB_IDLE = frozenset({"draft", "complete", "error", "cancelled"}) +_PCB_SSE_DONE = frozenset({ + "pcb_complete", + "pcb_error", + "pcb_cancelled", +}) +_SSE_HEADERS = { + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + "Connection": "keep-alive", +} +_ENQUEUE_FAIL = "Failed to enqueue pipeline worker; please retry" + + +class RegenRequest(BaseModel): + stages: list[str] + + +class ReprocessRequest(BaseModel): + mode: Literal["failed", "all"] = "failed" + + +def _project_active(meta: proj_svc.ProjectMeta) -> bool: + """True while analysis is queued or running (meta only; no Cloud Run poll).""" + return meta.status in (proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING) + + +def _need_bom_netlist(meta: proj_svc.ProjectMeta, detail: str) -> None: + if not meta.has_bom or not meta.has_netlist: + raise HTTPException(400, detail) + + +def _reset_event_history(storage, owner: str, project_id: str) -> None: + """Drop GCS log objects and any in-process broker history for this project.""" + try: + event_bridge.GCSEventBroker(storage, owner).clear_history(project_id) + except Exception: + log.exception("event log clear failed for %s", project_id) + job_workspace.broker.clear_history(project_id) + + +def _record_exec(storage, owner: str, project_id: str, execution_name: str, **extra: Any) -> None: + proj_svc.update_project( + storage, owner, project_id, execution_name=execution_name, **extra, + ) + + +def _enqueue_or_503( + storage, + owner: str, + project_id: str, + *, + enqueue: Callable[[], str], + fail_status: dict[str, Any], + http_detail: str, +) -> str: + try: + return enqueue() + except Exception: + log.exception("worker enqueue failed for %s", project_id) + proj_svc.update_project(storage, owner, project_id, **fail_status) + raise HTTPException(503, http_detail) + + +def _cas_queued( + storage, + owner: str, + project_id: str, + *, + from_status: frozenset[str], + conflict: str, + **fields: Any, +) -> None: + try: + proj_svc.transition_status( + storage, owner, project_id, + from_status=from_status, + to_status=proj_svc.STATUS_QUEUED, + **fields, + ) + except proj_svc.StatusConflict: + raise HTTPException(409, conflict) + + +async def _await_terminal( + storage, user_id: str, project_id: str, *, timeout_s: float, +) -> None: + """Poll until analysis meta is terminal, or ``timeout_s`` elapses.""" + step = 0.5 + spent = 0.0 + while spent < timeout_s: + try: + meta = proj_svc.get_project(storage, user_id, project_id) + except Exception: + meta = None + if meta is None or meta.status in proj_svc.TERMINAL_STATUSES: + return + await asyncio.sleep(step) + spent += step + + +async def _interrupt_active_pipeline(storage, user_id: str, project_id: str) -> None: + """Soft-cancel, then hard-cancel, then force-cancel leftover running meta.""" + proj_svc.request_cancel(storage, user_id, project_id) + await _await_terminal(storage, user_id, project_id, timeout_s=10.0) + meta = proj_svc.get_project(storage, user_id, project_id) + if meta is None: + return + if _project_active(meta) and meta.execution_name: + job_runner.cancel_execution(meta.execution_name) + await _await_terminal(storage, user_id, project_id, timeout_s=5.0) + meta = proj_svc.get_project(storage, user_id, project_id) or meta + if _project_active(meta): + try: + proj_svc.transition_status( + storage, user_id, project_id, + from_status={proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED}, + to_status=proj_svc.STATUS_CANCELLED, + pipeline_state={"error": "Superseded by reprocess"}, + cancel_requested=False, + ) + except proj_svc.StatusConflict: + pass + + +async def _stop_then_wait(storage, owner: str, project_id: str, meta: proj_svc.ProjectMeta) -> None: + if not _project_active(meta): + return + proj_svc.request_cancel(storage, owner, project_id) + await _await_terminal(storage, owner, project_id, timeout_s=10.0) + latest = proj_svc.get_project(storage, owner, project_id) or meta + if _project_active(latest) and latest.execution_name: + job_runner.cancel_execution(latest.execution_name) + await _await_terminal(storage, owner, project_id, timeout_s=5.0) + + +def _sse(gen: AsyncIterator[dict]) -> EventSourceResponse: + return EventSourceResponse(gen, ping=15, headers=_SSE_HEADERS) + + +def _yield_sse(ev: str, data: dict) -> dict: + return {"event": ev, "data": json.dumps(data)} + + +async def _poll_until_crash( + storage, + owner: str, + project_id: str, + *, + execution_name: str | None, + live: Callable[[proj_svc.ProjectMeta], bool], + finished: Callable[[proj_svc.ProjectMeta], bool], + reason_of: Callable[[proj_svc.ProjectMeta], str], + crash: dict[str, str | None], + initially_live: bool, +) -> None: + saw = initially_live + while True: + await asyncio.sleep(2.0) + try: + cur = proj_svc.get_project(storage, owner, project_id) + except Exception: + continue + if cur is None: + continue + if live(cur): + saw = True + elif saw and finished(cur): + crash["reason"] = reason_of(cur) + return + if execution_name and (saw or live(cur)): + try: + state = job_runner.get_execution_state(execution_name) + except Exception: + state = "unknown" + if state in _WORKER_DONE: + crash["reason"] = f"execution state={state}" + return + + +# --------------------------------------------------------------------------- +# Analysis +# --------------------------------------------------------------------------- + + +@router.post("/pipeline/{project_id}/start", status_code=202) +async def start(project_id: str, request: Request): + from backend._version import PERISCOPE_VERSION + + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + _need_bom_netlist(meta, "Upload BOM and netlist before starting pipeline") + get_billing().ensure_trial_grant(storage, get_user_id(request)) + _cas_queued( + storage, owner, project_id, + from_status=_ANALYSIS_IDLE, + conflict="Pipeline already running or queued", + cancel_requested=False, + execution_name=None, + periscope_version=PERISCOPE_VERSION, + ) + name = _enqueue_or_503( + storage, owner, project_id, + enqueue=lambda: job_runner.enqueue_pipeline( + project_id, owner, resume=False, free=False, + ), + fail_status={ + "status": proj_svc.STATUS_ERROR, + "pipeline_state": {"error": "Failed to enqueue worker"}, + }, + http_detail=_ENQUEUE_FAIL, + ) + _record_exec(storage, owner, project_id, name) + return {"status": "started", "project_id": project_id} + + +@router.post("/pipeline/{project_id}/cancel") +async def cancel(project_id: str, request: Request): + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + if not _project_active(meta): + raise HTTPException(409, f"Pipeline is not running (status={meta.status})") + proj_svc.request_cancel(storage, owner, project_id) + return {"status": "cancel_requested", "project_id": project_id} + + +@router.post("/pipeline/{project_id}/estimate") +async def estimate(project_id: str, request: Request): + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + if not meta.has_bom: + raise HTTPException(400, "Upload a BOM before requesting an estimate") + try: + est = estimate_pipeline_cost(storage, owner, project_id) + except FileNotFoundError as exc: + raise HTTPException(400, str(exc)) from exc + return est.model_dump() + + +@router.post("/pipeline/{project_id}/resume", status_code=202) +async def resume(project_id: str, request: Request): + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + if meta.status != proj_svc.STATUS_PAUSED: + raise HTTPException( + 400, + f"Project is not paused (status={meta.status}); nothing to resume.", + ) + _need_bom_netlist(meta, "Project is missing BOM or netlist") + _cas_queued( + storage, owner, project_id, + from_status=frozenset({proj_svc.STATUS_PAUSED}), + conflict="Project state changed; refresh and retry", + cancel_requested=False, + ) + name = _enqueue_or_503( + storage, owner, project_id, + enqueue=lambda: job_runner.enqueue_pipeline( + project_id, owner, resume=True, free=False, + ), + fail_status={ + "status": proj_svc.STATUS_ERROR, + "pipeline_state": {"error": "Failed to enqueue worker"}, + }, + http_detail=_ENQUEUE_FAIL, + ) + _record_exec(storage, owner, project_id, name) + return {"status": "resumed", "project_id": project_id} + + +@router.post("/pipeline/{project_id}/reprocess", status_code=202) +async def reprocess( + project_id: str, request: Request, req: ReprocessRequest | None = None, +): + from backend._version import PERISCOPE_VERSION + + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + _need_bom_netlist(meta, "Upload BOM and netlist before reprocessing") + if _project_active(meta): + await _interrupt_active_pipeline(storage, owner, project_id) + meta = proj_svc.get_project(storage, owner, project_id) or meta + + allowed = _REPROCESS_IDLE | {proj_svc.STATUS_PAUSED} if meta.status == proj_svc.STATUS_PAUSED else _REPROCESS_IDLE + if meta.status not in allowed and not _project_active(meta): + raise HTTPException(409, f"Cannot reprocess from status={meta.status}.") + + body = req or ReprocessRequest() + retry_failed = body.mode == "failed" + keep_refs = ( + proj_svc.completed_review_refs_for_retry(storage, owner, project_id) + if retry_failed else [] + ) + _cas_queued( + storage, owner, project_id, + from_status=allowed | {proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING}, + conflict="Pipeline already running or queued", + cancel_requested=False, + execution_name=None, + pipeline_state=None, + pause_checkpoint=None, + pause_reason=None, + completed_review_refs=keep_refs, + periscope_version=PERISCOPE_VERSION, + ) + name = _enqueue_or_503( + storage, owner, project_id, + enqueue=lambda: job_runner.enqueue_pipeline( + project_id, owner, resume=retry_failed, free=False, + ), + fail_status={ + "status": proj_svc.STATUS_ERROR, + "pipeline_state": {"error": "Failed to enqueue worker"}, + }, + http_detail=_ENQUEUE_FAIL, + ) + _record_exec(storage, owner, project_id, name) + return { + "status": "reprocess_started", + "project_id": project_id, + "mode": body.mode, + "resume": retry_failed, + "kept_review_refs": keep_refs, + } + + +@router.post("/pipeline/{project_id}/restart", status_code=202) +async def restart(project_id: str, request: Request): + from backend.routers.admin import _require_admin + + await _require_admin(request) + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + _need_bom_netlist(meta, "Upload BOM and netlist before starting pipeline") + await _stop_then_wait(storage, owner, project_id, meta) + proj_svc.clear_project_extractions(storage, owner, project_id) + _cas_queued( + storage, owner, project_id, + from_status=_ANALYSIS_IDLE | {proj_svc.STATUS_PAUSED}, + conflict="Pipeline is busy; cancel first then retry", + cancel_requested=False, + execution_name=None, + ) + name = _enqueue_or_503( + storage, owner, project_id, + enqueue=lambda: job_runner.enqueue_pipeline( + project_id, owner, resume=False, free=True, + ), + fail_status={ + "status": proj_svc.STATUS_ERROR, + "pipeline_state": {"error": "Failed to enqueue worker"}, + }, + http_detail=_ENQUEUE_FAIL, + ) + _record_exec(storage, owner, project_id, name) + return {"status": "restarted", "project_id": project_id} + + +@router.post("/pipeline/{project_id}/regen", status_code=202) +async def regen(project_id: str, req: RegenRequest, request: Request): + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + _need_bom_netlist(meta, "Upload BOM and netlist before running regen") + unknown = set(req.stages) - _REGEN_STAGES + if unknown: + raise HTTPException( + 400, + f"Invalid regen stages: {sorted(unknown)}. Valid: {sorted(_REGEN_STAGES)}", + ) + if not req.stages: + raise HTTPException(400, "At least one stage is required") + await _stop_then_wait(storage, owner, project_id, meta) + _cas_queued( + storage, owner, project_id, + from_status=_ANALYSIS_IDLE | {proj_svc.STATUS_PAUSED}, + conflict="Pipeline is busy; cancel first then retry", + cancel_requested=False, + execution_name=None, + ) + name = _enqueue_or_503( + storage, owner, project_id, + enqueue=lambda: job_runner.enqueue_pipeline_regen( + project_id, owner, stages=req.stages, + ), + fail_status={ + "status": proj_svc.STATUS_ERROR, + "pipeline_state": {"error": "Failed to enqueue worker"}, + }, + http_detail=_ENQUEUE_FAIL, + ) + _record_exec(storage, owner, project_id, name) + return {"status": "regen_started", "project_id": project_id, "stages": req.stages} + + +@router.get("/pipeline/{project_id}/events") +async def events(project_id: str, request: Request): + owner, meta = await resolve_or_404(request, project_id) + storage = get_storage(request) + + async def gen(): + crash: dict[str, str | None] = {"reason": None} + watcher = asyncio.create_task(_poll_until_crash( + storage, owner, project_id, + execution_name=meta.execution_name, + live=lambda m: m.status in (proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING), + finished=lambda m: m.status in proj_svc.TERMINAL_STATUSES, + reason_of=lambda m: f"project status={m.status} (terminal)", + crash=crash, + initially_live=meta.status in ( + proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING, + ), + )) + try: + async for msg in event_bridge.tail_events( + storage, owner, project_id, + terminal_events=_ANALYSIS_SSE_DONE, + ): + if crash["reason"] is not None: + break + ev = msg["event"] + if ev.startswith("placement_") or ev.startswith("pcb_"): + continue + yield _yield_sse(ev, msg.get("data") or {}) + if ev in _ANALYSIS_SSE_DONE: + return + if crash["reason"] is not None: + cur = proj_svc.get_project(storage, owner, project_id) + err = ( + (cur.pipeline_state or {}).get("error") + if cur and cur.pipeline_state + else crash["reason"] + ) + yield _yield_sse("pipeline_error", { + "error": err or "worker terminated without writing a terminal event", + "synthetic": True, + }) + finally: + watcher.cancel() + try: + await watcher + except (asyncio.CancelledError, Exception): + pass + + return _sse(gen()) + + +@router.get("/pipeline/{project_id}/status") +async def status(project_id: str, request: Request): + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + healed = proj_svc.heal_if_pipeline_finished(storage, owner, project_id) + if healed is not None: + meta = healed + place = meta.placement_status or "draft" + pcb = meta.pcb_status or "draft" + return { + "status": meta.status, + "summary": meta.summary, + "pipeline_state": meta.pipeline_state, + "running": meta.status in (proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED), + "placement_status": meta.placement_status, + "placement_state": meta.placement_state, + "placement_running": place in ("queued", "running"), + "pcb_status": meta.pcb_status, + "pcb_state": meta.pcb_state, + "pcb_running": pcb in ("queued", "running"), + "healed": healed is not None, + } + + +# --------------------------------------------------------------------------- +# Placement (topology; no credits) +# --------------------------------------------------------------------------- + + +@router.post("/pipeline/{project_id}/placement/start", status_code=202) +async def start_placement(project_id: str, request: Request): + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + _need_bom_netlist(meta, "Upload BOM and netlist before starting placement") + if placement_analysis_busy(meta): + raise HTTPException(409, "Analysis pipeline is running; wait or cancel it first") + if placement_busy(meta): + raise HTTPException(409, "Placement pipeline already running or queued") + if pcb_busy(meta): + raise HTTPException(409, "PCB review is running; wait or cancel it first") + if (meta.placement_status or "draft") not in _PLACEMENT_IDLE: + raise HTTPException( + 409, + f"Cannot start placement from placement_status={meta.placement_status}", + ) + proj_svc.update_project( + storage, owner, project_id, + placement_status="queued", + placement_cancel_requested=False, + placement_state=None, + placement_execution_name=None, + ) + _reset_event_history(storage, owner, project_id) + name = _enqueue_or_503( + storage, owner, project_id, + enqueue=lambda: job_runner.enqueue_placement_pipeline(project_id, owner), + fail_status={ + "placement_status": "error", + "placement_state": {"error": "Failed to enqueue placement worker"}, + }, + http_detail="Failed to enqueue placement worker; please retry", + ) + proj_svc.update_project( + storage, owner, project_id, placement_execution_name=name, + ) + return {"status": "started", "project_id": project_id} + + +@router.post("/pipeline/{project_id}/placement/cancel") +async def cancel_placement(project_id: str, request: Request): + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + if not placement_busy(meta): + raise HTTPException( + 409, + f"Placement is not running (placement_status={meta.placement_status})", + ) + proj_svc.update_project( + storage, owner, project_id, placement_cancel_requested=True, + ) + return {"status": "cancel_requested", "project_id": project_id} + + +@router.get("/pipeline/{project_id}/placement/plan") +async def get_placement_plan(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + prefix = proj_svc.project_prefix(owner, project_id) + for name in ("placement_plan.json", "functional_groups.json"): + key = f"{prefix}/{name}" + if storage.exists(key): + return storage.read_json(key) + raise HTTPException(404, "Placement plan not found — run placement first") + + +@router.get("/pipeline/{project_id}/placement/pack") +async def get_placement_pack(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + key = f"{proj_svc.project_prefix(owner, project_id)}/placement_pack.json" + if not storage.exists(key): + raise HTTPException(404, "Placement pack not found — run placement first") + return storage.read_json(key) + + +@router.get("/pipeline/{project_id}/placement/events") +async def placement_events(project_id: str, request: Request): + owner, meta = await resolve_or_404(request, project_id) + storage = get_storage(request) + + async def gen(): + crash: dict[str, str | None] = {"reason": None} + watcher = asyncio.create_task(_poll_until_crash( + storage, owner, project_id, + execution_name=meta.placement_execution_name, + live=lambda m: (m.placement_status or "draft") in ("queued", "running"), + finished=lambda m: (m.placement_status or "draft") in ( + "complete", "error", "cancelled", + ), + reason_of=lambda m: f"placement_status={m.placement_status or 'draft'} (terminal)", + crash=crash, + initially_live=(meta.placement_status or "draft") in ("queued", "running"), + )) + try: + async for msg in event_bridge.tail_events( + storage, owner, project_id, + terminal_events=_PLACEMENT_SSE_DONE, + ): + if crash["reason"] is not None: + break + ev = msg["event"] + keep = ev.startswith("placement_") or ev == "heartbeat" + if not keep or ev.startswith("pcb_"): + continue + yield _yield_sse(ev, msg.get("data") or {}) + if ev in _PLACEMENT_SSE_DONE: + return + if crash["reason"] is not None: + cur = proj_svc.get_project(storage, owner, project_id) + err = cur.placement_state.get("error") if cur and cur.placement_state else None + yield _yield_sse("placement_error", { + "error": err or crash["reason"] + or "placement worker terminated without a terminal event", + "synthetic": True, + }) + finally: + watcher.cancel() + try: + await watcher + except (asyncio.CancelledError, Exception): + pass + + return _sse(gen()) + + +# --------------------------------------------------------------------------- +# PCB review +# --------------------------------------------------------------------------- + + +@router.post("/pipeline/{project_id}/pcb/start", status_code=202) +async def start_pcb(project_id: str, request: Request): + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + if not meta.has_pcb: + raise HTTPException(400, "Upload a .kicad_pcb before starting PCB review") + _need_bom_netlist(meta, "Upload BOM and netlist before starting PCB review") + if pcb_analysis_busy(meta): + raise HTTPException(409, "Analysis pipeline is running; wait or cancel it first") + if pcb_placement_busy(meta): + raise HTTPException(409, "Placement pipeline is running; wait or cancel it first") + if pcb_busy(meta): + raise HTTPException(409, "PCB review already running or queued") + if (meta.pcb_status or "draft") not in _PCB_IDLE: + raise HTTPException( + 409, + f"Cannot start PCB review from pcb_status={meta.pcb_status}", + ) + proj_svc.update_project( + storage, owner, project_id, + pcb_status="queued", + pcb_cancel_requested=False, + pcb_state=None, + pcb_execution_name=None, + ) + _reset_event_history(storage, owner, project_id) + name = _enqueue_or_503( + storage, owner, project_id, + enqueue=lambda: job_runner.enqueue_pcb_pipeline(project_id, owner), + fail_status={ + "pcb_status": "error", + "pcb_state": {"error": "Failed to enqueue PCB worker"}, + }, + http_detail="Failed to enqueue PCB worker; please retry", + ) + proj_svc.update_project( + storage, owner, project_id, pcb_execution_name=name, + ) + return {"status": "started", "project_id": project_id} + + +@router.post("/pipeline/{project_id}/pcb/cancel") +async def cancel_pcb(project_id: str, request: Request): + storage = get_storage(request) + owner, meta = await resolve_or_404(request, project_id) + if not pcb_busy(meta): + raise HTTPException( + 409, + f"PCB review is not running (pcb_status={meta.pcb_status})", + ) + proj_svc.update_project( + storage, owner, project_id, pcb_cancel_requested=True, + ) + return {"status": "cancel_requested", "project_id": project_id} + + +@router.get("/pipeline/{project_id}/pcb/inventory") +async def get_pcb_inventory(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + key = f"{proj_svc.project_prefix(owner, project_id)}/pcb_inventory.json" + if not storage.exists(key): + raise HTTPException(404, "PCB inventory not found — run PCB review first") + return storage.read_json(key) + + +@router.get("/pipeline/{project_id}/pcb/events") +async def pcb_events(project_id: str, request: Request): + owner, meta = await resolve_or_404(request, project_id) + storage = get_storage(request) + + async def gen(): + crash: dict[str, str | None] = {"reason": None} + watcher = asyncio.create_task(_poll_until_crash( + storage, owner, project_id, + execution_name=meta.pcb_execution_name, + live=lambda m: (m.pcb_status or "draft") in ("queued", "running"), + finished=lambda m: (m.pcb_status or "draft") in ( + "complete", "error", "cancelled", + ), + reason_of=lambda m: f"pcb_status={m.pcb_status or 'draft'} (terminal)", + crash=crash, + initially_live=(meta.pcb_status or "draft") in ("queued", "running"), + )) + try: + async for msg in event_bridge.tail_events( + storage, owner, project_id, + terminal_events=_PCB_SSE_DONE, + ): + if crash["reason"] is not None: + break + ev = msg["event"] + if not (ev.startswith("pcb_") or ev == "heartbeat"): + continue + yield _yield_sse(ev, msg.get("data") or {}) + if ev in _PCB_SSE_DONE: + return + if crash["reason"] is not None: + cur = proj_svc.get_project(storage, owner, project_id) + ev, payload = pcb_sse_terminal_from_status( + cur.pcb_status if cur else None, + cur.pcb_state if cur else None, + crash["reason"], + ) + yield _yield_sse(ev, payload) + finally: + watcher.cancel() + try: + await watcher + except (asyncio.CancelledError, Exception): + pass + + return _sse(gen()) diff --git a/periscope/src/backend/routers/projects.py b/periscope/src/backend/routers/projects.py new file mode 100644 index 0000000..426ba56 --- /dev/null +++ b/periscope/src/backend/routers/projects.py @@ -0,0 +1,990 @@ +"""Project CRUD, library lookup, and file ingestion.""" + +from __future__ import annotations + +import csv +import io +import json +import logging +import os +import shutil +import tempfile +from pathlib import Path + +from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile +from fastapi.responses import JSONResponse, Response +from pydantic import BaseModel + +from backend.config import settings +from backend.periscopex.utils import natural_sort_key, safe_mpn +from backend.routers.deps import get_storage, get_user_id, resolve_or_404 +from backend.services import projects as proj_svc + +log = logging.getLogger(__name__) + +MAX_UPLOAD_BYTES = 30 * 1024 * 1024 +_MAX_MB = MAX_UPLOAD_BYTES // 1024 // 1024 + +router = APIRouter(tags=["projects"]) + + +def _xlsx_to_csv(raw: bytes) -> bytes: + import openpyxl + + wb = openpyxl.load_workbook(io.BytesIO(raw), read_only=True, data_only=True) + try: + sheet = wb.active + buf = io.StringIO() + writer = csv.writer(buf) + for row in sheet.iter_rows(values_only=True): + writer.writerow(["" if cell is None else str(cell) for cell in row]) + return buf.getvalue().encode("utf-8") + finally: + wb.close() + + +def _bom_file_to_csv_bytes(path: Path) -> bytes | None: + """CSV as-is; xlsx flattened to CSV; anything else ignored (KiCad zip).""" + suffix = path.suffix.lower() + payload = path.read_bytes() + if suffix == ".csv": + return payload + if suffix == ".xlsx": + try: + return _xlsx_to_csv(payload) + except Exception: + return None + return None + + +def _reject_oversize(nbytes: int, *, label: str | None = None) -> None: + if nbytes <= MAX_UPLOAD_BYTES: + return + if label: + mb = nbytes / 1024 / 1024 + raise HTTPException(413, f"{label} is {mb:.1f} MB — exceeds {_MAX_MB} MB limit") + raise HTTPException(413, f"File too large (max {_MAX_MB} MB)") + + +def _access_or_404(request: Request, project_id: str): + hit = proj_svc.resolve_project_access( + get_storage(request), get_user_id(request), project_id, + ) + if not hit: + raise HTTPException(404, "Project not found") + return hit + + +def _parse_bom_csv( + data: bytes, *, reference_column: str, mpn_column: str, +) -> dict: + from backend.periscopex.parsers import parse_bom + + with tempfile.NamedTemporaryFile(delete=False, suffix=".csv") as tmp: + tmp.write(data) + tmp_path = tmp.name + try: + return parse_bom(tmp_path, reference_col=reference_column, mpn_col=mpn_column) + finally: + os.unlink(tmp_path) + + +def _bucket_mpns(bom: dict) -> tuple[list[str], list[str], list[str]]: + from backend.periscopex.taxonomy import SIMPLE_TYPES, type_for_ref + + ic: list[str] = [] + passive: list[str] = [] + simple: list[str] = [] + seen_ic: set[str] = set() + seen_passive: set[str] = set() + seen_simple: set[str] = set() + for ref, info in sorted(bom.items()): + mpn = info.get("mpn") + if not mpn: + continue + kind = type_for_ref(ref) + if kind == "ic" and mpn not in seen_ic: + seen_ic.add(mpn) + ic.append(mpn) + elif kind == "passive" and mpn not in seen_passive: + seen_passive.add(mpn) + passive.append(mpn) + elif kind and kind in SIMPLE_TYPES and mpn not in seen_simple: + seen_simple.add(mpn) + simple.append(mpn) + return ic, passive, simple + + +def _pins_by_ref( + parts: dict[str, str], + nets: dict[str, list[tuple[str, str]]], +) -> list[dict]: + by_ref: dict[str, dict[str, str]] = {ref: {} for ref in parts} + for net_name, pins in nets.items(): + for ref, pin in pins: + by_ref.setdefault(ref, {}).setdefault(pin, net_name) + rows: list[dict] = [] + for ref in sorted(by_ref, key=natural_sort_key): + pin_map = by_ref[ref] + rows.append({ + "ref": ref, + "pins": [ + {"number": num, "net_name": pin_map[num]} + for num in sorted(pin_map, key=natural_sort_key) + ], + }) + return rows + + +def _netlist_ext(key: str) -> str: + leaf = key.rsplit("/", 1)[-1] + return leaf.rsplit(".", 1)[-1] if "." in leaf else "asc" + + +async def _gather_netlist_blobs( + file: UploadFile | None, + files: list[UploadFile] | None, + paths: str | None, +) -> list[tuple[str, bytes]]: + uploads = list(files or []) if files else ([file] if file is not None else []) + rels: list[str] | None = None + if paths: + try: + parsed = json.loads(paths) + except json.JSONDecodeError: + parsed = None + if isinstance(parsed, list) and all(isinstance(x, str) for x in parsed): + rels = parsed + blobs: list[tuple[str, bytes]] = [] + seen: set[tuple[str, int]] = set() + for i, uf in enumerate(uploads): + data = await uf.read() + _reject_oversize(len(data)) + name = rels[i] if rels is not None and i < len(rels) else (uf.filename or "netlist") + mark = (name, len(data)) + if mark in seen: + continue + seen.add(mark) + blobs.append((name, data)) + if not blobs: + raise HTTPException(400, "No netlist file uploaded") + return blobs + + +def _passive_hit_json(mpn: str, safe: str, model_data: dict, *, cached: bool, lcsc_id: str) -> dict: + return { + "mpn": mpn, + "safe_mpn": safe, + "model": model_data, + "cached": cached, + "lcsc_id": lcsc_id, + } + + +# --- Library --- + + +class LibraryCheckRequest(BaseModel): + ic_mpns: list[str] = [] + passive_mpns: list[str] = [] + simple_mpns: list[str] = [] + + +@router.post("/library/check") +async def check_library(body: LibraryCheckRequest, request: Request): + storage = get_storage(request) + ics = [mpn for mpn in body.ic_mpns if proj_svc.library_has_extraction(storage, mpn)] + patterns = proj_svc.load_library_patterns(storage) if body.passive_mpns else [] + passives: list[str] = [] + if body.passive_mpns: + from backend.periscopex.resolve_passives import resolve_mpn + + passives = [ + mpn for mpn in body.passive_mpns + if resolve_mpn(mpn, patterns) is not None + or proj_svc.library_has_passive_model(storage, mpn) is not None + ] + simples = [mpn for mpn in body.simple_mpns if proj_svc.library_has_model(storage, mpn)] + wanted = set(body.ic_mpns + body.passive_mpns + body.simple_mpns) + sheets = [ + mpn for mpn in wanted + if proj_svc.library_has_datasheet(storage, mpn, patterns=patterns) + ] + return { + "ic_resolved": ics, + "passive_resolved": passives, + "simple_resolved": simples, + "datasheets_available": sheets, + } + + +@router.get("/library") +async def get_library(request: Request): + storage = get_storage(request) + return JSONResponse( + content=proj_svc.list_library_catalog(storage), + headers={"Cache-Control": "no-store"}, + ) + + +@router.get("/library/datasheet/{mpn:path}") +async def get_library_datasheet(mpn: str, request: Request): + storage = get_storage(request) + key = proj_svc.library_has_datasheet(storage, mpn) + if not key: + raise HTTPException(404, f"Datasheet not in library: {mpn}") + return Response( + content=storage.read_bytes(key), + media_type="application/pdf", + headers={"Content-Disposition": f'inline; filename="{mpn}.pdf"'}, + ) + + +# --- CRUD --- + + +class CreateProjectRequest(BaseModel): + name: str + + +class RenameRequest(BaseModel): + name: str + + +@router.post("/projects") +async def create_project(body: CreateProjectRequest, request: Request): + meta = proj_svc.create_project(get_storage(request), get_user_id(request), body.name) + return meta.model_dump() + + +@router.get("/projects") +async def list_projects(request: Request): + storage = get_storage(request) + uid = get_user_id(request) + owned = proj_svc.list_projects(storage, uid) + shared = proj_svc.list_shared_projects(storage, uid) + return [m.model_dump() for m in owned + shared] + + +@router.get("/projects/{project_id}") +async def get_project(project_id: str, request: Request): + storage = get_storage(request) + owner_id, meta = await resolve_or_404(request, project_id) + for healer in ( + proj_svc.heal_if_pipeline_finished, + proj_svc.heal_if_placement_stuck, + proj_svc.heal_if_pcb_stuck, + ): + healed = healer(storage, owner_id, project_id) + if healed is not None: + meta = healed + return meta.model_dump() + + +@router.delete("/projects/{project_id}") +async def delete_project(project_id: str, request: Request): + storage = get_storage(request) + uid = get_user_id(request) + if proj_svc.delete_project(storage, uid, project_id): + return {"ok": True} + hit = proj_svc.resolve_project_access(storage, uid, project_id) + if hit: + owner_id, _ = hit + proj_svc.remove_collaborator(storage, owner_id, project_id, uid) + return {"ok": True, "removed_self": True} + raise HTTPException(404, "Project not found") + + +@router.patch("/projects/{project_id}") +async def rename_project(project_id: str, body: RenameRequest, request: Request): + storage = get_storage(request) + owner_id, _ = _access_or_404(request, project_id) + name = body.name.strip() + if not name: + raise HTTPException(400, "Name must be non-empty") + meta = proj_svc.update_project(storage, owner_id, project_id, name=name) + return meta.model_dump() + + +@router.post("/projects/{project_id}/reopen") +async def reopen_project(project_id: str, request: Request): + storage = get_storage(request) + owner_id, meta = _access_or_404(request, project_id) + if meta.status in (proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED): + raise HTTPException(409, "Pipeline is running; cancel it before reopening") + meta = proj_svc.reopen_project(storage, owner_id, project_id) + return meta.model_dump() + + +# --- Downloads --- + + +@router.get("/projects/{project_id}/files/bom") +async def download_bom(project_id: str, request: Request): + storage = get_storage(request) + owner_id, _ = await resolve_or_404(request, project_id) + key = proj_svc.get_bom_key(storage, owner_id, project_id) + if not key: + raise HTTPException(404, "BOM not uploaded") + return Response( + content=storage.read_bytes(key), + media_type="text/csv", + headers={"Content-Disposition": 'attachment; filename="bom.csv"'}, + ) + + +@router.get("/projects/{project_id}/files/netlist") +async def download_netlist(project_id: str, request: Request): + storage = get_storage(request) + owner_id, _ = await resolve_or_404(request, project_id) + key = proj_svc.get_netlist_key(storage, owner_id, project_id) + if not key: + raise HTTPException(404, "Netlist not uploaded") + ext = _netlist_ext(key) + return Response( + content=storage.read_bytes(key), + media_type="text/plain", + headers={"Content-Disposition": f'attachment; filename="netlist.{ext}"'}, + ) + + +@router.get("/projects/{project_id}/netlist/subdesigns") +async def get_netlist_subdesigns(project_id: str, request: Request): + from backend.periscopex.parsers_edif import list_edif_subdesigns + + storage = get_storage(request) + owner_id, meta = await resolve_or_404(request, project_id) + empty = {"sub_designs": [], "selected": None} + if meta.netlist_format != "edif": + return empty + key = proj_svc.get_netlist_key(storage, owner_id, project_id) + if not key: + return empty + data = storage.read_bytes(key) + with tempfile.NamedTemporaryFile(delete=False, suffix=".edn") as tmp: + tmp.write(data) + tmp_path = tmp.name + try: + subs = list_edif_subdesigns(tmp_path) + finally: + os.unlink(tmp_path) + return {"sub_designs": subs, "selected": meta.netlist_subdesigns} + + +class NetlistSubdesignsUpdate(BaseModel): + selected: list[str] | None + + +@router.put("/projects/{project_id}/netlist/subdesigns") +async def set_netlist_subdesigns( + project_id: str, payload: NetlistSubdesignsUpdate, request: Request, +): + storage = get_storage(request) + owner_id, _ = _access_or_404(request, project_id) + cleaned = [s.strip() for s in (payload.selected or []) if s and s.strip()] + meta = proj_svc.update_project( + storage, owner_id, project_id, + netlist_subdesigns=cleaned if payload.selected is not None else None, + ) + return meta.model_dump() + + +@router.get("/projects/{project_id}/files/datasheets") +async def list_datasheets(project_id: str, request: Request): + storage = get_storage(request) + owner_id, _ = await resolve_or_404(request, project_id) + stems = proj_svc.list_project_datasheets(storage, owner_id, project_id) + return {"stems": sorted(stems)} + + +# --- Uploads --- + + +@router.post("/projects/{project_id}/upload/bom") +async def upload_bom( + project_id: str, + file: UploadFile, + request: Request, + reference_column: str = "Reference", + mpn_column: str = "Manufacturer Part Number", + column_is_lcsc: bool | None = None, +): + storage = get_storage(request) + owner_id, _ = _access_or_404(request, project_id) + data = await file.read() + _reject_oversize(len(data)) + filename = file.filename or "" + if filename.lower().endswith(".xlsx"): + try: + data = _xlsx_to_csv(data) + except Exception as exc: + raise HTTPException(400, f"Invalid Excel file: {exc}") from exc + + try: + bom = _parse_bom_csv( + data, reference_column=reference_column, mpn_column=mpn_column, + ) + except Exception as exc: + raise HTTPException(400, f"Invalid BOM file: {exc}") from exc + + lcsc_resolved = 0 + lcsc_detected = False + lcsc_map: dict[str, str] = {} + lcsc_payloads: dict[str, dict] = {} + try: + from backend.services.purple_parts import ( + detect_lcsc_column, + resolve_lcsc_column_bytes, + ) + + lcsc_detected = detect_lcsc_column(data, mpn_column) + if column_is_lcsc or lcsc_detected: + data, lcsc_resolved, lcsc_map, lcsc_payloads = await resolve_lcsc_column_bytes( + data, mpn_col=mpn_column, + ) + except Exception: + log.warning("purple-parts BOM resolve failed", exc_info=True) + + if lcsc_resolved: + try: + bom = _parse_bom_csv( + data, reference_column=reference_column, mpn_column=mpn_column, + ) + except Exception: + log.warning( + "Re-parse after LCSC rewrite failed; using pre-rewrite BOM for classification", + exc_info=True, + ) + + ic_mpns, passive_mpns, simple_mpns = _bucket_mpns(bom) + + if ( + settings.use_purple_parts + and not lcsc_detected + and not column_is_lcsc + and passive_mpns + ): + try: + from backend.services.purple_parts import lookup_mpn_batch + + parts = await lookup_mpn_batch(passive_mpns) + for pmpn, part in parts.items(): + if part and part.get("lcsc") and part.get("description"): + lcsc_map[part["lcsc"]] = pmpn + lcsc_payloads[part["lcsc"]] = dict(part) + except Exception: + log.warning("purple-parts by-mpn passive enrich failed", exc_info=True) + + key = proj_svc.save_bom(storage, owner_id, project_id, data) + fields: dict = { + "bom_columns": {"reference": reference_column, "mpn": mpn_column}, + "component_mpns": { + "ic": ic_mpns, + "passive": passive_mpns, + "simple": simple_mpns, + }, + } + if lcsc_map: + fields["lcsc_to_mpn"] = lcsc_map + if lcsc_payloads: + fields["lcsc_payloads"] = lcsc_payloads + proj_svc.update_project(storage, owner_id, project_id, **fields) + return { + "path": key, + "components": len(bom), + "lcsc_resolved": lcsc_resolved, + "lcsc_detected": lcsc_detected, + "lcsc_to_mpn": lcsc_map, + } + + +@router.post("/projects/{project_id}/upload/netlist") +async def upload_netlist( + project_id: str, + request: Request, + file: UploadFile | None = File(default=None), + files: list[UploadFile] | None = File(default=None), + paths: str | None = Form(default=None), +): + from backend.periscopex.netlist_bundle import materialize_netlist_upload + from backend.periscopex.parsers import parse_netlist_any, validate_netlist + from backend.periscopex.parsers_edif import list_edif_subdesigns + + storage = get_storage(request) + owner_id, _ = _access_or_404(request, project_id) + blobs = await _gather_netlist_blobs(file, files, paths) + + sub_designs: list[dict] = [] + bom_saved = False + pcb_saved = False + sheets = 1 + try: + with tempfile.TemporaryDirectory() as tmp: + parsed = materialize_netlist_upload(blobs, Path(tmp) / "work") + parts, nets, fmt = parse_netlist_any(parsed.root) + if fmt == "edif": + sub_designs = list_edif_subdesigns(parsed.root) + issues = validate_netlist(parts, nets) + if issues: + raise ValueError("; ".join(issues)) + root_bytes = parsed.root.read_bytes() + key = proj_svc.save_netlist(storage, owner_id, project_id, root_bytes, fmt=fmt) + if fmt == "kicad_sch": + proj_svc.save_companion_sheets( + storage, owner_id, project_id, parsed.root, parsed.extra_sch, + ) + sheets = 1 + len(parsed.extra_sch) + else: + proj_svc.clear_companion_sheets(storage, owner_id, project_id) + if parsed.pcb is not None: + proj_svc.save_pcb(storage, owner_id, project_id, parsed.pcb.read_bytes()) + pcb_saved = True + if parsed.bom is not None: + bom_bytes = _bom_file_to_csv_bytes(parsed.bom) + if bom_bytes: + proj_svc.save_bom(storage, owner_id, project_id, bom_bytes) + proj_svc.update_project( + storage, owner_id, project_id, + bom_columns={ + "reference": "Reference", + "mpn": "Manufacturer Part Number", + }, + ) + bom_saved = True + except HTTPException: + raise + except Exception as exc: + raise HTTPException(400, f"Netlist failed sanity check: {exc}") from exc + + designator_pins: list[dict] = [] + if fmt != "pads": + designator_pins = _pins_by_ref(parts, nets) + return { + "path": key, + "parts": len(parts), + "nets": len(nets), + "format": fmt, + "sub_designs": sub_designs, + "designator_pins": designator_pins, + "pcb_saved": pcb_saved, + "bom_saved": bom_saved, + "sheets": sheets, + } + + +@router.post("/projects/{project_id}/upload/pcb") +async def upload_pcb(project_id: str, file: UploadFile, request: Request): + from backend.periscopex.parsers_kicad_pcb import parse_kicad_pcb + + storage = get_storage(request) + owner_id, _ = _access_or_404(request, project_id) + data = await file.read() + _reject_oversize(len(data)) + with tempfile.NamedTemporaryFile(delete=False, suffix=".kicad_pcb") as tmp: + tmp.write(data) + tmp_path = tmp.name + try: + layout = parse_kicad_pcb(tmp_path) + except Exception as exc: + raise HTTPException(400, f"Invalid KiCad PCB: {exc}") from exc + finally: + os.unlink(tmp_path) + key = proj_svc.save_pcb(storage, owner_id, project_id, data) + return { + "path": key, + "footprints": len(layout.footprints), + "nets": len(layout.nets), + "segments": len(layout.segments), + } + + +@router.post("/projects/{project_id}/upload/datasheets") +async def upload_datasheets( + project_id: str, file: UploadFile, mpn: str, + request: Request, also_for: str | None = None, +): + storage = get_storage(request) + owner_id, _ = _access_or_404(request, project_id) + if not file.filename or not file.filename.lower().endswith(".pdf"): + raise HTTPException(400, "File must be a PDF") + data = await file.read() + _reject_oversize(len(data), label=file.filename or mpn) + key = proj_svc.save_datasheet(storage, owner_id, project_id, mpn, data) + extra_mpns: list[str] = [] + if also_for: + for extra in also_for.split(","): + extra = extra.strip() + if extra: + proj_svc.save_datasheet(storage, owner_id, project_id, extra, data) + extra_mpns.append(extra) + return {"path": key, "mpn": mpn, "also_for": extra_mpns} + + +# --- Collaborators --- + + +class AddCollaboratorRequest(BaseModel): + email: str + + +def _local_member(uid: str, owner_id: str) -> dict: + return { + "user_id": uid, + "name": None, + "email": None, + "image_url": None, + "role": "owner" if uid == owner_id else "collaborator", + } + + +@router.get("/projects/{project_id}/collaborators") +async def list_collaborators(project_id: str, request: Request): + storage = get_storage(request) + owner_id, meta = _access_or_404(request, project_id) + ids = [owner_id] + [c for c in meta.collaborators if c != owner_id] + if settings.use_auth: + from backend.services.user_directory import get_user_profile + + members = [] + for uid in ids: + profile = await get_user_profile(uid) + members.append({ + "user_id": uid, + "name": profile.get("name"), + "email": profile.get("email"), + "image_url": profile.get("image_url"), + "role": "owner" if uid == owner_id else "collaborator", + }) + else: + members = [_local_member(uid, owner_id) for uid in ids] + return {"owner_user_id": owner_id, "collaborators": members} + + +@router.post("/projects/{project_id}/collaborators") +async def add_collaborator(project_id: str, body: AddCollaboratorRequest, request: Request): + storage = get_storage(request) + uid = get_user_id(request) + meta = proj_svc.get_project(storage, uid, project_id) + if not meta: + raise HTTPException(404, "Project not found") + if not settings.use_auth: + raise HTTPException(400, "Collaboration requires authentication to be enabled") + + from backend.services.user_directory import find_user_id_by_email, get_user_profile + + collab_id = await find_user_id_by_email(body.email) + if not collab_id: + raise HTTPException(404, "No user found with that email") + if collab_id == uid: + raise HTTPException(400, "Cannot add yourself as a collaborator") + if collab_id in meta.collaborators: + raise HTTPException(409, "User is already a collaborator") + proj_svc.add_collaborator(storage, uid, project_id, collab_id) + profile = await get_user_profile(collab_id) + return { + "user_id": collab_id, + "name": profile.get("name"), + "email": profile.get("email"), + "image_url": profile.get("image_url"), + } + + +@router.delete("/projects/{project_id}/collaborators/{collaborator_user_id}") +async def remove_collaborator(project_id: str, collaborator_user_id: str, request: Request): + from backend.routers.admin import is_admin + + storage = get_storage(request) + uid = get_user_id(request) + meta = proj_svc.get_project(storage, uid, project_id) + owner_id = uid + if not meta: + if not await is_admin(request): + raise HTTPException(404, "Project not found") + found = proj_svc.find_project_any_user(storage, project_id) + if not found: + raise HTTPException(404, "Project not found") + owner_id, meta = found + if collaborator_user_id not in meta.collaborators: + raise HTTPException(404, "User is not a collaborator") + proj_svc.remove_collaborator(storage, owner_id, project_id, collaborator_user_id) + return {"ok": True} + + +@router.post("/projects/{project_id}/collaborators/{collaborator_user_id}/make-owner") +async def make_collaborator_owner( + project_id: str, collaborator_user_id: str, request: Request, +): + from backend.routers.admin import is_admin + + if not await is_admin(request): + raise HTTPException(403, "Admin access required") + storage = get_storage(request) + found = proj_svc.find_project_any_user(storage, project_id) + if not found: + raise HTTPException(404, "Project not found") + current_owner, _ = found + try: + proj_svc.transfer_ownership( + storage, current_owner, project_id, collaborator_user_id, + ) + except ValueError as exc: + raise HTTPException(400, str(exc)) from exc + return {"ok": True, "owner_user_id": collaborator_user_id} + + +# --- Datasheet fetch + DigiKey --- + + +@router.get("/digikey/datasheet") +@router.get("/datasheets/fetch") +async def fetch_auto_datasheet(mpn: str, request: Request, lcsc: str | None = None): + from backend.services.datasheet_finder import find_datasheet + + result = await find_datasheet(mpn, lcsc_id=lcsc) + if not result.ok: + return JSONResponse( + status_code=404, + content={ + "detail": result.error or "Failed to fetch datasheet", + "url": result.url, + "urls": result.suggested_urls or ([result.url] if result.url else []), + "source": result.source, + }, + ) + headers = {"Content-Disposition": f'attachment; filename="{mpn}.pdf"'} + if result.url: + headers["X-Datasheet-Url"] = result.url + if result.source: + headers["X-Datasheet-Source"] = result.source + try: + proj_svc.remember_datasheet( + get_storage(request), mpn, result.pdf_bytes, extra_mpns=result.alias_mpns, + ) + except Exception: + pass + return Response(content=result.pdf_bytes, media_type="application/pdf", headers=headers) + + +class AutoResolveItem(BaseModel): + mpn: str + component_type: str + + +class AutoResolveRequest(BaseModel): + items: list[AutoResolveItem] + + +@router.post("/digikey/auto-resolve") +async def auto_resolve(body: AutoResolveRequest, request: Request): + import asyncio + + from backend.services.datasheet_extract import CatalogResolveMiss, auto_resolve_specs + from backend.services.digikey import fetch_params + + if not settings.use_digikey: + raise HTTPException(400, "DigiKey API not configured") + + storage = get_storage(request) + sem = asyncio.Semaphore(10) + + async def resolve_one(item: AutoResolveItem) -> dict: + async with sem: + try: + safe = safe_mpn(item.mpn) + if item.component_type == "passive": + lib_key = f"library/passives/{safe}.json" + if not storage.exists(lib_key): + legacy = f"library/models/{safe}.json" + if storage.exists(legacy): + return {"mpn": item.mpn, "status": "resolved"} + else: + lib_key = f"library/models/{safe}.json" + if storage.exists(lib_key): + return {"mpn": item.mpn, "status": "resolved"} + + result = await fetch_params(item.mpn) + if not result.ok or not result.params: + return { + "mpn": item.mpn, + "status": "failed", + "error": result.error or "No parameters", + } + try: + model = await auto_resolve_specs( + mpn=item.mpn, + digikey_params=result.params.parameters, + digikey_category=result.params.category, + digikey_description=result.params.description, + component_type=item.component_type, + use_llm=settings.has_llm_credentials(), + ) + except CatalogResolveMiss as exc: + return {"mpn": item.mpn, "status": "failed", "error": str(exc)} + storage.write_json(lib_key, model.model_dump()) + return {"mpn": item.mpn, "status": "resolved"} + except Exception as exc: + log.warning("Auto-resolve failed for %s: %s", item.mpn, exc, exc_info=True) + return { + "mpn": item.mpn, + "status": "failed", + "error": str(exc) or type(exc).__name__, + } + + results = await asyncio.gather(*(resolve_one(item) for item in body.items)) + return {"results": results} + + +# --- LCSC wizard resolve --- + + +class LcscResolvePassiveRequest(BaseModel): + lcsc_id: str + + +def _seed_taxonomy_dir(storage, tax_dir: Path) -> None: + for key in storage.list_prefix("taxonomy/"): + if key.endswith(".json"): + storage.download_to_local(key, tax_dir / key.rsplit("/", 1)[-1]) + if any(tax_dir.glob("*.json")): + return + repo_tax = settings.taxonomy_dir + if repo_tax.is_dir(): + for path in repo_tax.glob("*.json"): + shutil.copy2(path, tax_dir / path.name) + + +@router.post("/projects/{project_id}/lcsc/resolve-passive") +async def lcsc_resolve_passive( + project_id: str, body: LcscResolvePassiveRequest, request: Request, +): + from backend.services.api_logs import ApiLogger + from backend.services.billing_hook import InsufficientCredits, get_billing + from backend.services.datasheet_extract import auto_resolve_specs + from backend.services.passive_from_distributor import specs_from_lcsc_payload + + storage = get_storage(request) + owner_id, meta = _access_or_404(request, project_id) + payload = (meta.lcsc_payloads or {}).get(body.lcsc_id) + if not payload: + raise HTTPException(404, f"No cached payload for LCSC id {body.lcsc_id!r}") + mpn = (payload.get("mpn") or "").strip() + if not mpn: + raise HTTPException(404, f"Cached payload for {body.lcsc_id!r} has no MPN") + + safe = safe_mpn(mpn) + project_model_key = ( + f"{proj_svc.project_prefix(owner_id, project_id)}/models/{safe}.json" + ) + if storage.exists(project_model_key): + return _passive_hit_json( + mpn, safe, storage.read_json(project_model_key), + cached=True, lcsc_id=body.lcsc_id, + ) + lib_key = proj_svc.library_has_passive_model(storage, mpn) + if lib_key: + storage.copy_object(lib_key, project_model_key) + return _passive_hit_json( + mpn, safe, storage.read_json(project_model_key), + cached=True, lcsc_id=body.lcsc_id, + ) + + synth_category = " / ".join( + p for p in (payload.get("category"), payload.get("subcategory")) if p + ) or None + synth_params: list[dict[str, str]] = [] + if payload.get("package"): + synth_params.append({"name": "Package / Case", "value": payload["package"]}) + if payload.get("manufacturer"): + synth_params.append({"name": "Manufacturer", "value": payload["manufacturer"]}) + description = payload.get("description") or "" + if not description: + raise HTTPException( + 502, + f"Cached LCSC payload for {body.lcsc_id!r} has no description — " + "cannot auto-resolve", + ) + + catalog_model = specs_from_lcsc_payload(mpn, payload) + if catalog_model is not None: + storage.write_json(project_model_key, catalog_model.model_dump()) + proj_svc.save_to_library( + storage, project_model_key, "passives", f"{safe}.json", + ) + return _passive_hit_json( + mpn, safe, catalog_model.model_dump(), + cached=False, lcsc_id=body.lcsc_id, + ) + + api_logger = ApiLogger() + with tempfile.TemporaryDirectory() as tmpdir: + tax_dir = Path(tmpdir) / "taxonomy" + tax_dir.mkdir() + _seed_taxonomy_dir(storage, tax_dir) + try: + model = await auto_resolve_specs( + mpn=mpn, + digikey_params=synth_params, + digikey_category=synth_category or "", + digikey_description=description, + component_type="passive", + taxonomy_dir=tax_dir, + api_logger=api_logger, + ) + except Exception as exc: + log.warning( + "lcsc_resolve_passive: auto_resolve_specs failed for %s (lcsc=%s)", + mpn, body.lcsc_id, exc_info=True, + ) + raise HTTPException(502, f"Auto-resolve failed: {exc}") from exc + + total_credits = sum(float(e.get("credits_charged") or 0) for e in api_logger.entries) + if total_credits > 0: + try: + get_billing().charge( + storage, owner_id, round(total_credits, 4), + reason="pipeline_charge", + run_id=project_id, + unit_id=f"lcsc_resolve_passive:{mpn}", + allow_overdraft=False, + ) + except InsufficientCredits as short: + raise HTTPException( + 402, + detail={ + "reason": "insufficient_credits", + "required": short.required, + "available": short.available, + }, + ) from short + + storage.write_json(project_model_key, model.model_dump()) + proj_svc.save_to_library( + storage, project_model_key, "passives", f"{safe}.json", + ) + try: + logs_key = f"{proj_svc.project_prefix(owner_id, project_id)}/api_logs.jsonl" + existing = storage.read_text(logs_key) if storage.exists(logs_key) else "" + appended = existing + api_logger.to_jsonl() + if appended: + storage.write_text(logs_key, appended) + except Exception: + log.warning("lcsc_resolve_passive: failed to append api_logs.jsonl", exc_info=True) + + try: + from backend.services.api_logs import total_cost as _total_cost + + added_cost = _total_cost(api_logger.entries) + if added_cost > 0: + current_total = float(meta.total_cost_usd or 0) + proj_svc.update_project( + storage, owner_id, project_id, + total_cost_usd=round(current_total + added_cost, 6), + credits_spent=round(float(meta.credits_spent or 0) + total_credits, 4), + ) + except Exception: + log.warning("lcsc_resolve_passive: failed to update total_cost_usd", exc_info=True) + + return _passive_hit_json( + mpn, safe, model.model_dump(), + cached=False, lcsc_id=body.lcsc_id, + ) diff --git a/periscope/src/backend/routers/reports.py b/periscope/src/backend/routers/reports.py new file mode 100644 index 0000000..c95a4b7 --- /dev/null +++ b/periscope/src/backend/routers/reports.py @@ -0,0 +1,378 @@ +"""Serve reports, graphs, datasheets, and finding review mutations.""" +from __future__ import annotations + +import json +import logging +import re +import uuid +from datetime import datetime, timezone + +from fastapi import APIRouter, HTTPException, Request +from fastapi.responses import JSONResponse, Response +from pydantic import BaseModel + +from backend.periscopex.finding_engine import ( + apply_decisions, + complete_findings, + decision_from_review, + sort_findings, + upsert_decision, +) +from backend.periscopex.models import Finding +from backend.periscopex.pcb_checks import merge_schema_pcb_reports +from backend.periscopex.review_workflow import ( + ReviewError, + apply_review_state, + build_eco, + eco_csv, + sign_report, +) +from backend.periscopex.utils import safe_mpn +from backend.routers.deps import get_storage, get_user_id, resolve_or_404 +from backend.services import projects as proj_svc + +log = logging.getLogger(__name__) +router = APIRouter(tags=["reports"]) + +_MPN_OK = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_\-\.:/ +#,()]*$") + + +def _require_mpn(mpn: str) -> None: + if not _MPN_OK.match(mpn) or ".." in mpn: + raise HTTPException(400, "Invalid MPN format") + + +def _prefix(owner: str, project_id: str) -> str: + return proj_svc.project_prefix(owner, project_id) + + +def _load_report(storage, owner: str, project_id: str) -> tuple[str, dict]: + key = f"{_prefix(owner, project_id)}/report.json" + if not storage.exists(key): + raise HTTPException(404, "Report not found") + return key, storage.read_json(key) + + +def _findings_from_report(report_data: dict) -> list[Finding]: + out: list[Finding] = [] + for raw in report_data.get("findings") or []: + try: + out.append(Finding.model_validate(raw)) + except Exception: + log.warning("Skipping malformed finding in report", exc_info=True) + return out + + +@router.get("/report/{project_id}") +async def get_report(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + prefix = _prefix(owner, project_id) + schema_key = f"{prefix}/report.json" + pcb_key = f"{prefix}/pcb_report.json" + schema = storage.read_json(schema_key) if storage.exists(schema_key) else None + pcb = storage.read_json(pcb_key) if storage.exists(pcb_key) else None + merged = merge_schema_pcb_reports(schema, pcb) + if merged is None: + raise HTTPException(404, "Report not found — run the pipeline first") + findings = _findings_from_report(merged) + try: + complete_findings(findings) + except Exception: + log.exception("complete_findings failed while serving report %s", project_id) + sort_findings(findings) + dec_key = f"{prefix}/decisions.json" + if storage.exists(dec_key): + try: + apply_decisions(findings, storage.read_json(dec_key) or []) + except Exception: + log.exception("apply_decisions failed for report %s", project_id) + merged["findings"] = [json.loads(f.model_dump_json()) for f in findings] + summary = {"ERROR": 0, "WARNING": 0, "INFO": 0, "total": len(findings)} + for finding in findings: + if finding.status in summary: + summary[finding.status] += 1 + merged["summary"] = summary + return JSONResponse(merged) + + +@router.get("/report/{project_id}/cad-bridge") +async def get_cad_bridge(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + key = f"{_prefix(owner, project_id)}/periscope-findings.json" + if not storage.exists(key): + raise HTTPException(404, "CAD bridge not found — run the pipeline first") + return JSONResponse(storage.read_json(key)) + + +class AddCommentBody(BaseModel): + finding_id: str + text: str + user_name: str + mentions: list[str] = [] + + +@router.post("/report/{project_id}/comments") +async def add_comment(project_id: str, body: AddCommentBody, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + user_id = get_user_id(request) + key = f"{_prefix(owner, project_id)}/report.json" + if not storage.exists(key): + raise HTTPException(404, "Report not found") + report = storage.read_json(key) + comment = { + "comment_id": str(uuid.uuid4()), + "finding_id": body.finding_id, + "user_id": user_id, + "user_name": body.user_name, + "text": body.text, + "mentions": body.mentions, + "created_at": datetime.now(timezone.utc).isoformat(), + } + bucket = report.setdefault("comments", {}) + bucket.setdefault(body.finding_id, []).append(comment) + storage.write_json(key, report) + return JSONResponse(comment, status_code=201) + + +@router.delete("/report/{project_id}/comments/{comment_id}") +async def delete_comment(project_id: str, comment_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + user_id = get_user_id(request) + key = f"{_prefix(owner, project_id)}/report.json" + if not storage.exists(key): + raise HTTPException(404, "Report not found") + report = storage.read_json(key) + comments = report.get("comments", {}) + found = False + for finding_id, rows in list(comments.items()): + for i, row in enumerate(rows): + if row.get("comment_id") != comment_id: + continue + found = True + if row.get("user_id") != user_id and user_id != owner: + raise HTTPException(403, "Cannot delete another user's comment") + rows.pop(i) + if not rows: + del comments[finding_id] + storage.write_json(key, report) + return JSONResponse({"ok": True}) + if not found: + raise HTTPException(404, "Comment not found") + return JSONResponse({"ok": True}) + + +class ReviewBody(BaseModel): + state: str + reason: str = "" + user_name: str = "" + + +@router.put("/report/{project_id}/findings/{finding_id}/review") +async def put_finding_review(project_id: str, finding_id: str, body: ReviewBody, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + user_id = get_user_id(request) + key, report = _load_report(storage, owner, project_id) + prefix = _prefix(owner, project_id) + findings = _findings_from_report(report) + ids = {f.finding_id for f in findings if f.finding_id} + if finding_id not in ids: + pcb_key = f"{prefix}/pcb_report.json" + if storage.exists(pcb_key): + pcb = storage.read_json(pcb_key) + pcb_findings = _findings_from_report(pcb) + if finding_id in {f.finding_id for f in pcb_findings if f.finding_id}: + key, report, findings = pcb_key, pcb, pcb_findings + ids = {f.finding_id for f in findings if f.finding_id} + if finding_id not in ids: + raise HTTPException(404, "Finding not found") + try: + states = apply_review_state( + report.get("review_states") or {}, + finding_id, + state=body.state, + reason=body.reason, + user_id=user_id, + user_name=body.user_name, + ) + except ReviewError as exc: + raise HTTPException(400, str(exc)) from exc + report["review_states"] = states + storage.write_json(key, report) + if body.state in {"wontfix", "false_positive"}: + found = next( + (f for f in _findings_from_report(report) if f.finding_id == finding_id), + None, + ) + if found is not None: + dec = decision_from_review( + found, state=body.state, reason=body.reason, user_id=user_id + ) + if dec is not None: + dkey = f"{prefix}/decisions.json" + existing = storage.read_json(dkey) if storage.exists(dkey) else [] + if not isinstance(existing, list): + existing = [] + storage.write_json(dkey, upsert_decision(existing, dec)) + return JSONResponse(states.get(finding_id) or {"state": "open", "reason": ""}) + + +@router.get("/report/{project_id}/eco.json") +async def get_eco_json(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + _, report = _load_report(storage, owner, project_id) + return JSONResponse( + {"items": build_eco(_findings_from_report(report), report.get("review_states") or {})} + ) + + +@router.get("/report/{project_id}/eco.csv") +async def get_eco_csv(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + _, report = _load_report(storage, owner, project_id) + items = build_eco(_findings_from_report(report), report.get("review_states") or {}) + return Response(eco_csv(items), media_type="text/csv") + + +@router.post("/report/{project_id}/sign") +async def post_sign_report(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + user_id = get_user_id(request) + key, report = _load_report(storage, owner, project_id) + release = sign_report(report, user_id=user_id) + report["release"] = release + storage.write_json(key, report) + return JSONResponse(release) + + +@router.get("/bom/{project_id}") +async def get_bom_summary(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + key = f"{_prefix(owner, project_id)}/bom_summary.json" + if not storage.exists(key): + raise HTTPException(404, "BOM summary not found — run the pipeline first") + return JSONResponse(storage.read_json(key)) + + +@router.get("/derating/{project_id}") +async def get_derating(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + key = f"{_prefix(owner, project_id)}/derating.json" + if not storage.exists(key): + raise HTTPException(404, "Derating data not found — run the pipeline first") + return JSONResponse(storage.read_json(key)) + + +@router.get("/graph/{project_id}") +async def get_graph(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + key = f"{_prefix(owner, project_id)}/design_graph.json" + if not storage.exists(key): + raise HTTPException(404, "Design graph not found — run the pipeline first") + return JSONResponse(storage.read_json(key)) + + +@router.get("/projects/{project_id}/logs") +async def get_project_logs(project_id: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + key = f"{_prefix(owner, project_id)}/api_logs.jsonl" + if not storage.exists(key): + return JSONResponse([]) + text = storage.read_text(key) + entries = [json.loads(line) for line in text.strip().split("\n") if line.strip()] + from backend.services.llm.pricing import cost_for_entry + + for entry in entries: + if any( + entry.get(k) + for k in ( + "input_tokens", + "output_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + ) + ): + entry["cost_usd"] = round(cost_for_entry(entry), 6) + return JSONResponse(entries) + + +def _find_datasheet_key(storage, owner: str, project_id: str, safe: str, mpn: str | None = None) -> str | None: + from backend.services.datasheet_store import resolve_datasheet + + uploaded = f"{_prefix(owner, project_id)}/uploads/datasheets/{safe}.pdf" + if storage.exists(uploaded): + return uploaded + resolved = resolve_datasheet(storage, safe) + if resolved: + return resolved + legacy = f"library/datasheets/{safe}.pdf" + if storage.exists(legacy): + return legacy + if mpn: + return proj_svc.library_has_datasheet(storage, mpn) + return None + + +@router.get("/projects/{project_id}/datasheet-url/{mpn:path}") +async def get_datasheet_url(project_id: str, mpn: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + _require_mpn(mpn) + key = _find_datasheet_key(storage, owner, project_id, safe_mpn(mpn), mpn=mpn) + if key is None: + raise HTTPException(404, f"Datasheet not found for MPN: {mpn}") + proxy = f"/api/projects/{project_id}/datasheet/{mpn}" + return {"url": f"{str(request.base_url).rstrip('/')}{proxy}"} + + +@router.get("/projects/{project_id}/datasheet/{mpn:path}") +async def get_datasheet_proxy(project_id: str, mpn: str, request: Request): + storage = get_storage(request) + owner, _ = await resolve_or_404(request, project_id) + _require_mpn(mpn) + safe = safe_mpn(mpn) + key = _find_datasheet_key(storage, owner, project_id, safe, mpn=mpn) + if key is None: + raise HTTPException(404, f"Datasheet not found for MPN: {mpn}") + return Response( + content=storage.read_bytes(key), + media_type="application/pdf", + headers={"Content-Disposition": f'inline; filename="{safe}.pdf"'}, + ) + + +@router.get("/datasheets/{mpn}") +async def get_datasheet(mpn: str, request: Request): + from fastapi.responses import FileResponse + + from backend.services.storage import LocalStorageBackend + + storage = get_storage(request) + user_id = get_user_id(request) + _require_mpn(mpn) + safe = safe_mpn(mpn) + if not isinstance(storage, LocalStorageBackend): + raise HTTPException( + 400, + "Use GET /projects/{project_id}/datasheet-url/{mpn} for cloud storage", + ) + for entry in storage.list_prefix(f"users/{user_id}/projects/"): + pdf_key = f"{entry}/uploads/datasheets/{safe}.pdf" + if storage.exists(pdf_key): + return FileResponse( + storage._path(pdf_key), + media_type="application/pdf", + filename=f"{safe}.pdf", + ) + raise HTTPException(404, f"Datasheet not found for MPN: {mpn}") diff --git a/periscope/src/backend/routers/survey.py b/periscope/src/backend/routers/survey.py new file mode 100644 index 0000000..b1378ed --- /dev/null +++ b/periscope/src/backend/routers/survey.py @@ -0,0 +1,69 @@ +"""Onboarding survey: status flag plus one-shot Google Sheet append.""" +from __future__ import annotations + +import logging + +from fastapi import APIRouter, Request +from pydantic import BaseModel + +from backend.config import settings +from backend.routers.deps import get_storage, get_user_id +from backend.services import survey as survey_svc + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/survey", tags=["survey"]) + + +class SurveySubmission(BaseModel): + referral_source: str + user_profile: str + + +@router.get("/status") +async def survey_status(request: Request): + return { + "completed": survey_svc.is_completed(get_storage(request), get_user_id(request)) + } + + +async def _identity(user_id: str) -> tuple[str, str]: + email, name = "unknown", "unknown" + if not settings.use_auth: + return email, name + try: + from backend.services.email import _resolve_clerk_user + + clerk_user = await _resolve_clerk_user(user_id) + except Exception: + logger.warning("Failed to resolve Clerk user %s for survey", user_id) + return email, name + if not clerk_user: + return email, name + emails = clerk_user.get("email_addresses") or [] + if emails: + email = emails[0].get("email_address") or "unknown" + first = clerk_user.get("first_name") or "" + last = clerk_user.get("last_name") or "" + name = f"{first} {last}".strip() or "unknown" + return email, name + + +@router.post("") +async def submit_survey(request: Request, body: SurveySubmission): + storage = get_storage(request) + user_id = get_user_id(request) + if survey_svc.is_completed(storage, user_id): + return {"ok": True, "detail": "already_submitted"} + + email, name = await _identity(user_id) + sheet_ok = await survey_svc.append_to_sheet( + user_id=user_id, + email=email, + name=name, + referral_source=body.referral_source, + user_profile=body.user_profile, + ) + if sheet_ok or not settings.survey_sheet_id: + survey_svc._mark_completed(storage, user_id) + return {"ok": True} + return {"ok": False, "detail": "sheet_write_failed"} diff --git a/tests/test_periscope_routers_rewrite.py b/tests/test_periscope_routers_rewrite.py new file mode 100644 index 0000000..907bd38 --- /dev/null +++ b/tests/test_periscope_routers_rewrite.py @@ -0,0 +1,49 @@ +"""HTTP routers resolve from periscope/src (auth stays leftover).""" + +from __future__ import annotations + +from pathlib import Path + +import backend.routers.admin as admin +import backend.routers.contact as contact +import backend.routers.deps as deps +import backend.routers.feedback as feedback +import backend.routers.pipeline as pipeline +import backend.routers.projects as projects +import backend.routers.reports as reports +import backend.routers.survey as survey + + +def _src(mod, name: str) -> None: + path = Path(mod.__file__).resolve() + assert path.name == name + assert "src" in path.parts + assert "Native Periscope overlay" not in path.read_text(encoding="utf-8")[:400] + + +def test_inherited_routers_are_src(): + for mod, name in ( + (admin, "admin.py"), + (contact, "contact.py"), + (deps, "deps.py"), + (feedback, "feedback.py"), + (pipeline, "pipeline.py"), + (projects, "projects.py"), + (reports, "reports.py"), + (survey, "survey.py"), + ): + _src(mod, name) + + +def test_auth_router_is_not_rewritten_this_slice(): + import backend.routers.auth as auth + + path = Path(auth.__file__).resolve() + assert path.name == "auth.py" + assert "src" in path.parts + + +def test_reprocess_helper_names(): + assert callable(pipeline._await_terminal) + assert callable(pipeline._project_active) + assert callable(projects._bom_file_to_csv_bytes)