#!/usr/bin/env python3 """Persistent mobile control plane for recording intake and human review.""" from __future__ import annotations import hashlib import hmac import html import json import os import re import secrets import sqlite3 import time import urllib.parse from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from . import review, webauth from .config import CFG from .log import log STATE = CFG.state_dir / "review-web.json" DB = CFG.state_dir / "intake.db" RECORDINGS = Path(os.environ.get( "ASR_RECORDINGS_ROOT", str(CFG.obsidian / "recordings"))) INBOX = RECORDINGS / "inbox" META = RECORDINGS / "state" / "uploads" MAX_UPLOAD = int(os.environ.get("ASR_UPLOAD_MAX_BYTES", str(4 * 1024**3))) CHUNK = 1024 * 1024 SUPPORTED = {".m4a", ".mp3", ".wav", ".mp4", ".aac", ".ogg", ".opus", ".webm", ".flac"} STATUS_LABEL = { "uploading": "上传中", "queued": "等待转写", "transcribing": "正在转写", "refining": "正在整理", "review": "等待确认", "publishing": "正在发布", "complete": "已完成", "failed": "处理失败", } def _connect() -> sqlite3.Connection: CFG.state_dir.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(DB, timeout=20) con.row_factory = sqlite3.Row con.execute("PRAGMA journal_mode=WAL") con.execute(""" CREATE TABLE IF NOT EXISTS intake_job( id TEXT PRIMARY KEY, original_name TEXT NOT NULL, stored_name TEXT NOT NULL, title TEXT NOT NULL DEFAULT '', context TEXT NOT NULL DEFAULT '', size INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL, detail TEXT NOT NULL DEFAULT '', transcript TEXT NOT NULL DEFAULT '', created_at REAL NOT NULL, updated_at REAL NOT NULL ) """) columns = {row["name"] for row in con.execute("PRAGMA table_info(intake_job)")} for name in ("draft", "result"): if name not in columns: con.execute( f"ALTER TABLE intake_job ADD COLUMN {name} TEXT NOT NULL DEFAULT ''") return con def update_job(job_id: str, status: str, detail: str = "", transcript: str = "", draft: str = "", result: str = "") -> bool: if os.environ.get("ASR_ROLE"): from asrservice.db import update_job as service_update_job return service_update_job(job_id, status, detail, transcript=transcript, draft=draft, result=result) if not job_id: return False con = _connect() fields = ["status=?", "updated_at=?"] values: list[object] = [status, time.time()] if detail: fields.append("detail=?") values.append(detail[:1000]) if transcript: fields.append("transcript=?") values.append(transcript) if draft: fields.append("draft=?") values.append(draft) if result: fields.append("result=?") values.append(result) values.append(job_id) cur = con.execute( f"UPDATE intake_job SET {','.join(fields)} WHERE id=?", values) con.commit() con.close() return cur.rowcount > 0 def _jobs(limit: int = 30) -> list[dict]: con = _connect() rows = con.execute( "SELECT * FROM intake_job ORDER BY created_at DESC LIMIT ?", (limit,)).fetchall() con.close() return [dict(r) for r in rows] def _job(job_id: str) -> dict | None: con = _connect() row = con.execute("SELECT * FROM intake_job WHERE id=?", (job_id,)).fetchone() con.close() return dict(row) if row else None def _review_url(job_id: str = "") -> str: base = f"http://{CFG.review_public_host}:{CFG.review_port}" return f"{base}/jobs/{job_id}/review/" if job_id else f"{base}/review/" def _pending_for_job(job: dict, limit: int = 200) -> list[dict]: transcript = str(job.get("transcript") or "") draft = str(job.get("draft") or "") rows = [] for row in review.pending(limit): payload = json.loads(row["payload"] or "{}") if ((transcript and row["recording"] == transcript) or (draft and payload.get("draft_path") == draft) or payload.get("job_id") == job["id"]): rows.append(row) return rows def _safe_text(value: str, limit: int) -> str: value = re.sub(r"[\x00-\x1f\x7f]+", " ", value or "") return re.sub(r"\s+", " ", value).strip()[:limit] def _safe_stem(value: str) -> str: value = _safe_text(value, 80) value = re.sub(r"[^\w\u3400-\u9fff.-]+", "-", value, flags=re.UNICODE) return value.strip("-.") or "录音" def _write_state() -> None: CFG.state_dir.mkdir(parents=True, exist_ok=True) data = { "pid": os.getpid(), "url": _review_url(), "home": f"http://{CFG.review_public_host}:{CFG.review_port}/", "port": CFG.review_port, "started_at": time.time(), } tmp = STATE.with_suffix(".tmp") tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") os.chmod(tmp, 0o600) tmp.replace(STATE) def _shell(title: str, body: str, script: str = "") -> bytes: return f""" {html.escape(title)}
录音工作台
{body}
{script}""".encode("utf-8") def _home() -> bytes: body = """

上传录音

支持 mp3、m4a、wav、mp4、aac、ogg、opus、webm、flac。上传完成后自动进入现有 ASR 流程。

最近任务

正在读取…
""" script = """""" return _shell("录音工作台", body, script) def _review_page(action: str, job: dict | None = None) -> bytes: rows = _pending_for_job(job) if job else review.pending(100) if not rows: if job: label = STATUS_LABEL.get(job["status"], job["status"]) detail = html.escape(str(job.get("detail") or "")) return _shell("任务状态", f'
{html.escape(label)}
' f'

{html.escape(str(job.get("title") or job["original_name"]))}

' f'

{detail or "当前没有待确认问题。"}

' '

页面会随任务阶段更新，可以稍后刷新。

') return _shell("确认完成", '

当前没有待确认项目

' '

新草稿生成后，Telegram 会再次发送这里的链接。

') row = rows[0] payload = json.loads(row["payload"] or "{}") choices = payload.get("candidates") or [] buttons = "".join( f'' for v in choices) buttons += ('') source = html.escape(str(payload.get("sentence", ""))) source_html = f'
{source}
' if source else "" draft_html = "" if payload.get("draft_path"): try: path = Path(payload["draft_path"]).resolve() path.relative_to(CFG.drafts.resolve()) draft = path.read_text(encoding="utf-8", errors="ignore") draft_html = ('
查看完整草稿
'
                          + html.escape(draft) + "
") except (OSError, ValueError): pass placeholder = ("只输入姓名，例如：高瑶" if payload.get("kind") == "voiceprint" else ("背景不准时，写明正确背景并提交重跑" if payload.get("kind") == "background_check" else "输入正确名称或补充说明")) title = (str(job.get("title") or job["original_name"]) if job else "ASR 快速确认") body = f"""
{html.escape(title)} · 剩余 {len(rows)} 个 · 问题 #{row['id']}

{html.escape(row['question'])}

{source_html}{draft_html}
{buttons}
""" return _shell("ASR 快速确认", body) class Server(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True def __init__(self, addr, handler): super().__init__(addr, handler) class Handler(BaseHTTPRequestHandler): server_version = "ASRControl/1" def log_message(self, fmt, *args): status = args[1] if len(args) > 1 else "?" log(f"control web: {self.command} {urllib.parse.urlsplit(self.path).path} status={status}") def _headers(self, status=HTTPStatus.OK, content_type="text/html; charset=utf-8"): self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Cache-Control", "no-store, no-cache, must-revalidate") self.send_header("X-Content-Type-Options", "nosniff") def _send(self, data: bytes, status=HTTPStatus.OK, content_type="text/html; charset=utf-8"): self._headers(status, content_type) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _json(self, obj: object, status=HTTPStatus.OK): self._send(json.dumps(obj, ensure_ascii=False).encode("utf-8"), status, "application/json; charset=utf-8") def _review_job(self) -> tuple[bool, dict | None]: path = urllib.parse.urlsplit(self.path).path if path == "/review/": return True, None match = re.fullmatch(r"/jobs/([0-9a-z-]{8,64})/review/", path) if match: return True, _job(match.group(1)) return False, None def _browser_ok(self) -> bool: if webauth.allow_anon(): return True if not webauth.password(): return False return webauth.cookie_matches(self.headers.get("Cookie", "")) def _login_page(self, err: str = "") -> bytes: if not webauth.password() and not webauth.allow_anon(): err = err or "未配置 ASR_WEB_PASSWORD，控制面已关闭。" token = html.escape(webauth.csrf_token(), quote=True) alert = f"

{html.escape(err)}

" if err else "" body = ( f'

访问验证

{alert}' f'
' f'' f'' f'
' ) return _shell("登录", body) def do_GET(self): path = urllib.parse.urlsplit(self.path).path if path == "/health": return self._json({"ok": True, "pending": len(review.pending(100))}) if path in {"/login", "/logout"}: if path == "/logout": self.send_response(HTTPStatus.SEE_OTHER) self.send_header("Location", "/login") self.send_header("Set-Cookie", "asr_session=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0") self.end_headers() return return self._send(self._login_page()) if not self._browser_ok(): return self._send(self._login_page(), HTTPStatus.UNAUTHORIZED) if path == "/": return self._send(_home()) if path == "/api/jobs": jobs = _jobs() for j in jobs: j["status_label"] = STATUS_LABEL.get(j["status"], j["status"]) j["review_url"] = f"/jobs/{j['id']}/review/" j["has_draft"] = bool(j.get("draft")) j.pop("context", None) j.pop("transcript", None) j.pop("draft", None) j.pop("result", None) return self._json({"jobs": jobs}) is_review, job = self._review_job() if is_review: if path != "/review/" and not job: return self._send(b"Task not found", HTTPStatus.NOT_FOUND, "text/plain") return self._send(_review_page(path, job)) return self._send(b"Not found", HTTPStatus.NOT_FOUND, "text/plain") def _header(self, name: str, limit: int) -> str: return _safe_text(urllib.parse.unquote(self.headers.get(name, "")), limit) def _upload(self): try: length = int(self.headers.get("Content-Length", "0") or 0) except ValueError: return self._json({"error": "bad content-length"}, HTTPStatus.BAD_REQUEST) if length <= 0 or length > MAX_UPLOAD: return self._json({"error": "invalid upload size"}, HTTPStatus.REQUEST_ENTITY_TOO_LARGE) original = self._header("X-Filename", 240) title = self._header("X-Title", 80) context = self._header("X-Context", 2000) suffix = Path(original).suffix.lower() if suffix not in SUPPORTED: return self._json({"error": f"unsupported file type: {suffix or '(none)'}"}, HTTPStatus.UNSUPPORTED_MEDIA_TYPE) job_id = secrets.token_hex(12) stem = _safe_stem(title or Path(original).stem) stored = f"{time.strftime('%Y-%m-%d_%H_%M_%S')}-{stem}-{job_id[:8]}{suffix}" INBOX.mkdir(parents=True, exist_ok=True) staging = INBOX / ".uploading" staging.mkdir(parents=True, exist_ok=True) target = INBOX / stored part = staging / f"{job_id}.part" con = _connect() now = time.time() con.execute( "INSERT INTO intake_job(id,original_name,stored_name,title,context,size," "status,created_at,updated_at) VALUES(?,?,?,?,?,?,?, ?,?)", (job_id, original, stored, title, context, length, "uploading", now, now)) con.commit() con.close() remaining = length try: with part.open("xb") as f: while remaining: data = self.rfile.read(min(CHUNK, remaining)) if not data: raise EOFError("incomplete upload") f.write(data) remaining -= len(data) f.flush() os.fsync(f.fileno()) META.mkdir(parents=True, exist_ok=True) meta = { "job_id": job_id, "title": title, "context": context, "original_name": original, "stored_name": stored, } meta_tmp = META / f".{stored}.tmp" meta_tmp.write_text(json.dumps(meta, ensure_ascii=False), encoding="utf-8") meta_tmp.replace(META / f"{stored}.json") part.replace(target) update_job(job_id, "queued", "录音已落盘，等待 watcher 接收") return self._json({"ok": True, "job_id": job_id, "stored_name": stored}, HTTPStatus.CREATED) except EOFError: part.unlink(missing_ok=True) update_job(job_id, "failed", "上传未完成") return self._json({"error": "incomplete upload"}, HTTPStatus.BAD_REQUEST) except Exception as e: part.unlink(missing_ok=True) update_job(job_id, "failed", "上传失败") log(f"upload failed: {e}") return self._json({"error": "upload failed"}, HTTPStatus.INTERNAL_SERVER_ERROR) def _answer(self): try: try: length = int(self.headers.get("Content-Length", "0") or 0) except ValueError as exc: raise ValueError("bad content-length") from exc if length < 0 or length > 4096: raise ValueError("bad content-length") form = urllib.parse.parse_qs( self.rfile.read(length).decode("utf-8")) if not webauth.csrf_matches((form.get("csrf") or [""])[0]): raise ValueError("csrf") qid = int(form.get("id", ["0"])[0]) answer = (form.get("custom", [""])[0].strip() or form.get("answer", [""])[0].strip()) if not qid or not answer or len(answer) > 120: raise ValueError("invalid answer") is_review, job = self._review_job() if not is_review: raise ValueError("invalid review path") allowed = (_pending_for_job(job, 1000) if job else review.pending(1000)) if qid not in {row["id"] for row in allowed}: raise ValueError("问题不属于当前任务或已处理") review.answer(qid, answer) self.send_response(HTTPStatus.SEE_OTHER) self.send_header("Location", urllib.parse.urlsplit(self.path).path) self.send_header("Cache-Control", "no-store") self.end_headers() except Exception as e: self._send(_shell("提交失败", f'

提交失败

{html.escape(str(e))}

'), HTTPStatus.BAD_REQUEST) def do_POST(self): path = urllib.parse.urlsplit(self.path).path if path == "/login": try: length = int(self.headers.get("Content-Length", "0") or 0) except ValueError: return self._json({"error": "bad content-length"}, HTTPStatus.BAD_REQUEST) if length < 0: return self._json({"error": "bad content-length"}, HTTPStatus.BAD_REQUEST) length = min(length, 4096) form = urllib.parse.parse_qs(self.rfile.read(length).decode("utf-8", "ignore")) pw = form.get("password", [""])[0] if webauth.password() and hmac.compare_digest(pw, webauth.password()): self.send_response(HTTPStatus.SEE_OTHER) self.send_header("Location", "/") self.send_header( "Set-Cookie", f"asr_session={webauth.session_token()}; Path=/; HttpOnly; SameSite=Strict; Max-Age=86400", ) self.end_headers() return return self._send(self._login_page("口令错误或未配置 ASR_WEB_PASSWORD"), HTTPStatus.UNAUTHORIZED) if not self._browser_ok(): return self._json({"error": "unauthorized"}, HTTPStatus.UNAUTHORIZED) if path in ("/api/upload", "/hook/intake"): if not webauth.csrf_matches(self.headers.get("X-CSRF-Token", "")): return self._json({"error": "csrf"}, HTTPStatus.FORBIDDEN) return self._upload() if self._review_job()[0]: return self._answer() return self._send(b"Not found", HTTPStatus.NOT_FOUND, "text/plain") def serve() -> int: try: server = Server((CFG.review_host, CFG.review_port), Handler) except OSError as e: STATE.unlink(missing_ok=True) log(f"control web bind failed on {CFG.review_host}:{CFG.review_port}: {e}", level="ERROR") return 1 _write_state() log(f"control web listening on {CFG.review_public_host}:{CFG.review_port}") try: server.serve_forever(poll_interval=0.5) finally: server.server_close() STATE.unlink(missing_ok=True) return 0 if __name__ == "__main__": raise SystemExit(serve())