Files
infra-docker-configs/servers/tools-pve/asr-pipeline/asrpipe/review_web.py
T

306 lines
13 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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"""<!doctype html>
<html lang="zh-CN"><head>
<meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<meta name="color-scheme" content="light">
<meta http-equiv="Cache-Control" content="no-store">
<title>{html.escape(title)}</title>
<style>
:root{{--ink:#17202a;--muted:#637083;--line:#d9dee5;--paper:#f7f8fa;
--accent:#176b5b;--accent2:#c84b31;--white:#fff}}
*{{box-sizing:border-box}} body{{margin:0;background:var(--paper);color:var(--ink);
font-family:system-ui,-apple-system,"PingFang SC","Microsoft YaHei",sans-serif;
font-size:17px;line-height:1.55;letter-spacing:0}}
header{{background:var(--white);border-bottom:1px solid var(--line);padding:16px 18px}}
header strong{{font-size:18px}} header span{{display:block;color:var(--muted);font-size:14px}}
main{{max-width:680px;margin:0 auto;padding:18px 14px 40px}}
.question{{background:var(--white);border:1px solid var(--line);border-radius:8px;padding:18px}}
h1{{font-size:21px;line-height:1.35;margin:0 0 12px}} .source{{margin:14px 0;
padding:12px;border-left:4px solid #d4a72c;background:#fff9e8;color:#39424e}}
form{{display:grid;gap:10px}} button{{width:100%;min-height:50px;border:0;border-radius:6px;
padding:12px 14px;background:var(--accent);color:white;font:inherit;font-weight:650}}
button.secondary{{background:#e9edf1;color:var(--ink);border:1px solid var(--line)}}
input{{width:100%;min-height:50px;border:1px solid #9aa5b1;border-radius:6px;
padding:10px 12px;font:inherit;background:white}} .custom{{display:flex;gap:8px}}
.custom input{{min-width:0;flex:1}} .custom button{{width:auto;min-width:82px}}
.meta{{color:var(--muted);font-size:14px;margin-bottom:10px}} .done{{text-align:center;
background:var(--white);border:1px solid var(--line);border-radius:8px;padding:36px 18px}}
.draft{{margin:14px 0;border:1px solid var(--line);border-radius:6px;background:#fbfcfd}}
.draft summary{{padding:12px 14px;font-weight:650;cursor:pointer}}
.draft pre{{margin:0;padding:14px;border-top:1px solid var(--line);max-height:58vh;
overflow:auto;white-space:pre-wrap;overflow-wrap:anywhere;font:14px/1.55 ui-monospace,
SFMono-Regular,Consolas,monospace}}
.done h1{{color:var(--accent)}} @media(max-width:420px){{main{{padding:12px 10px 28px}}
.question{{padding:15px}} .custom{{display:grid}} .custom button{{width:100%}}}}
</style></head><body><header><strong>ASR 快速确认</strong>
<span>确认结果会写入个人事实库,并用于后续录音</span></header>
<main>{body}</main></body></html>""".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("完成",
'<section class="done"><h1>本轮确认完成</h1>'
'<p>结果已经写入个人事实库,本页面即将自动关闭。</p></section>'))
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'<button type="submit" name="answer" value="{v}">{v}</button>')
buttons.append(
'<button class="secondary" type="submit" name="answer" '
'value="无法确认,保留原文">都不是 / 无法确认</button>')
source = html.escape(str(payload.get("sentence", "")))
source_html = f'<div class="source">{source}</div>' 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 = (
'<details class="draft"><summary>查看完整草稿</summary><pre>'
+ html.escape(draft) + "</pre></details>")
except (OSError, ValueError):
draft_html = ""
placeholder = ("只输入姓名,例如:高瑶"
if payload.get("kind") == "voiceprint"
else ("背景不准时,在这里写明正确背景并提交重跑"
if payload.get("kind") == "background_check"
else "输入正确名称或补充说明"))
body = f"""<section class="question">
<div class="meta">剩余 {len(rows)} 个 · 问题 #{row['id']}</div>
<h1>{html.escape(row['question'])}</h1>{source_html}{draft_html}
<form method="post" action="/r/{self.server.token}/">
<input type="hidden" name="id" value="{row['id']}">
{''.join(buttons)}
<div class="custom"><input name="custom" maxlength="120" autocomplete="off"
placeholder="{html.escape(placeholder, quote=True)}"><button type="submit">提交</button></div>
</form></section>"""
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'<section class="done"><h1>提交失败</h1><p>{html.escape(str(e))}</p></section>'),
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())