#!/usr/bin/env python3 """Ephemeral, capability-URL mobile UI for the human review queue.""" from __future__ import annotations import hmac import html import json import os import secrets import subprocess import sys import time import urllib.parse from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from . import review from .config import CFG from .log import log STATE = CFG.state_dir / "review-web.json" UNIT = "asr-review-web.service" def _url(token: str) -> str: return f"http://{CFG.review_public_host}:{CFG.review_port}/r/{token}/" def request_start(job_id: str = "") -> bool: """Ensure control service is running and send a task-scoped review URL.""" try: r = subprocess.run(["systemctl", "--user", "start", UNIT], capture_output=True, text=True, timeout=15) if r.returncode: log(f"review web start failed: {r.stderr.strip()[:240]}", level="WARN") return False from . import control_web job = control_web._job(job_id) if job_id else None if job_id and not job: log(f"review notification missing job: {job_id}", level="WARN") return False url = control_web._review_url(job_id) rows = (control_web._pending_for_job(job, 1000) if job else review.pending(1000)) from . import vault n = len(rows) title = str(job.get("title") or job["original_name"]) if job else "ASR" vault.notify("ASR 快速确认", f"{title}:{n} 个问题待确认\n{url}\n" f"录音上传入口:http://{CFG.review_public_host}:{CFG.review_port}/", redact_log=True) return True except Exception as e: log(f"review web start failed: {e}", level="WARN") return False def read_state() -> dict | None: try: return json.loads(STATE.read_text(encoding="utf-8")) except (OSError, ValueError): return None def _write_state(token: str, started: float) -> None: CFG.state_dir.mkdir(parents=True, exist_ok=True) data = { "pid": os.getpid(), "url": _url(token), "port": CFG.review_port, "started_at": started, "expires_at": started + CFG.review_max_seconds, } 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 _page(title: str, body: str) -> bytes: return f""" {html.escape(title)}
ASR 快速确认 确认结果会写入个人事实库,并用于后续录音
{body}
""".encode("utf-8") class Server(ThreadingHTTPServer): daemon_threads = True def __init__(self, addr, handler, token: str): super().__init__(addr, handler) self.token = token self.started_at = time.time() self.last_activity = self.started_at self.opened = False self.finish_at: float | None = None class Handler(BaseHTTPRequestHandler): server_version = "ASRReview/1" def log_message(self, fmt, *args): # BaseHTTPRequestHandler includes the full URL in its default log line. # The path is a bearer capability, so never persist it. status = args[1] if len(args) > 1 else "?" log(f"review web request: {self.command} 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("Pragma", "no-cache") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.send_header("Referrer-Policy", "no-referrer") self.send_header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; " "form-action 'self'; base-uri 'none'; frame-ancestors 'none'") def _valid_path(self) -> bool: parts = urllib.parse.urlsplit(self.path) want = f"/r/{self.server.token}/" return hmac.compare_digest(parts.path, want) def _send(self, data: bytes, status=HTTPStatus.OK): self._headers(status) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def do_GET(self): if self.path == "/health": body = json.dumps({"ok": True, "pending": len(review.pending(100))}).encode() self._headers(content_type="application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return if not self._valid_path(): self._send(b"Not found", HTTPStatus.NOT_FOUND) return self.server.last_activity = time.time() self.server.opened = True rows = review.pending(100) if not rows: if self.server.finish_at is None: self.server.finish_at = time.time() + 10 self._send(_page("完成", '

本轮确认完成

' '

结果已经写入个人事实库,本页面即将自动关闭。

')) return row = rows[0] payload = json.loads(row["payload"] or "{}") choices = payload.get("candidates") or [] buttons = [] for value in choices: v = html.escape(str(value), quote=True) buttons.append( f'') buttons.append( '') source = html.escape(str(payload.get("sentence", ""))) source_html = f'
{source}
' if source else "" draft_html = "" draft_path = payload.get("draft_path") if draft_path: try: path = Path(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): draft_html = "" placeholder = ("只输入姓名,例如:高瑶" if payload.get("kind") == "voiceprint" else ("背景不准时,在这里写明正确背景并提交重跑" if payload.get("kind") == "background_check" else "输入正确名称或补充说明")) body = f"""
剩余 {len(rows)} 个 · 问题 #{row['id']}

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

{source_html}{draft_html}
{''.join(buttons)}
""" self._send(_page("ASR 快速确认", body)) def do_POST(self): if not self._valid_path(): self._send(b"Not found", HTTPStatus.NOT_FOUND) return try: length = min(int(self.headers.get("Content-Length", "0")), 4096) form = urllib.parse.parse_qs(self.rfile.read(length).decode("utf-8")) qid = int(form.get("id", ["0"])[0]) custom = form.get("custom", [""])[0].strip() answer = custom or form.get("answer", [""])[0].strip() if not qid or not answer or len(answer) > 120: raise ValueError("invalid answer") pending_ids = {r["id"] for r in review.pending(1000)} if qid not in pending_ids: raise ValueError("问题已处理或不存在") review.answer(qid, answer) self.server.last_activity = time.time() self.send_response(HTTPStatus.SEE_OTHER) self.send_header("Location", f"/r/{self.server.token}/") self.send_header("Cache-Control", "no-store") self.end_headers() except Exception as e: self._send(_page("提交失败", f'

提交失败

{html.escape(str(e))}

'), HTTPStatus.BAD_REQUEST) def serve() -> int: rows = review.pending(100) if not rows: log("review web: no pending questions") return 0 token = secrets.token_urlsafe(32) started = time.time() try: server = Server((CFG.review_host, CFG.review_port), Handler, token) except OSError as e: STATE.unlink(missing_ok=True) log(f"review web bind failed on {CFG.review_host}:{CFG.review_port}: {e}", level="ERROR") return 1 _write_state(token, started) server.timeout = 1 url = _url(token) log(f"review web listening on {CFG.review_public_host}:{CFG.review_port} " f"({len(rows)} question(s))") try: from . import vault vault.notify("ASR 快速确认", f"{len(rows)} 个问题待确认:{url}\n" f"链接最多保留 {CFG.review_max_seconds // 60} 分钟,完成后自动销毁。", redact_log=True) except Exception as e: log(f"review web notification failed: {e}", level="WARN") try: while True: server.handle_request() now = time.time() if server.finish_at and now >= server.finish_at: # A new recording can enqueue work during the 10-second grace # period while systemd still considers this unit active. if review.pending(1): server.finish_at = None log("review web received new work during shutdown grace") else: break if now - started >= CFG.review_max_seconds: log("review web reached maximum lifetime") break if server.opened and now - server.last_activity >= CFG.review_idle_seconds: log("review web idle timeout") break finally: server.server_close() try: STATE.unlink(missing_ok=True) except OSError: pass return 0 if __name__ == "__main__": sys.exit(serve())