from __future__ import annotations
import hashlib, hmac, html, json, os, re, time, urllib.parse, uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from asrpipe import review
from . import db
from .config import AGENT_TOKEN, DRAFTS, FAILED, INBOX, MAX_UPLOAD, PUBLIC_URL, RESULTS, STATE, SUPPORTED, TRANSCRIPTS, ensure_dirs
LABELS = {"queued":"等待处理","transcribing":"正在转写","refining":"正在整理","review":"等待确认","publishing":"正在定稿","complete":"已完成","failed":"处理失败"}
def safe(value, limit):
return re.sub(r"\s+", " ", re.sub(r"[\x00-\x1f\x7f]+", " ", value or "")).strip()[:limit]
def stem(value):
return re.sub(r"[^\w\u3400-\u9fff.-]+", "-", safe(value, 80)).strip("-.") or "录音"
def public(row, detailed=False):
out = {k: row.get(k) for k in ("id","original_name","title","size","status","detail","source","sha256","attempts","error_code","created_at","updated_at")}
out["status_label"] = LABELS.get(row["status"], row["status"])
out["web_url"] = f"{PUBLIC_URL}/jobs/{row['id']}/"
out["review_url"] = f"{PUBLIC_URL}/jobs/{row['id']}/review/"
out["artifacts"] = {name:f"/api/v1/jobs/{row['id']}/artifacts/{name}" for name in ("transcript","draft","result") if row.get(name)}
if detailed:
out["context"] = row.get("context", "")
out["events"] = db.events(row["id"])
return out
def pending(job):
found=[]
for row in review.pending(1000):
payload=json.loads(row["payload"] or "{}")
if ((job.get("transcript") and row["recording"]==job["transcript"])
or (job.get("draft") and payload.get("draft_path")==job["draft"])):
found.append(row)
return found
def shell(title, body, script=""):
return f'''{html.escape(title)}
" for e in db.events(job["id"]))
links=' · '.join(f'{label}' for n,label in (("transcript","转写"),("draft","草稿"),("result","正式稿")) if job.get(n))
retry=f'' if job["status"]=="failed" else ""
review_link=f'
")
row=rows[0]; payload=json.loads(row["payload"] or "{}"); choices=payload.get("candidates") or []
buttons=''.join(f'' for v in choices)+''
draft=""; path=Path(str(payload.get("draft_path") or ""))
try: path.resolve().relative_to(DRAFTS.resolve()); draft=path.read_text(encoding="utf-8")
except (OSError,ValueError): pass
return shell("人工确认",f"
剩余 {len(rows)} 个问题
{html.escape(row['question'])}
{html.escape(str(payload.get('sentence') or ''))}
查看草稿
{html.escape(draft)}
")
class Handler(BaseHTTPRequestHandler):
server_version="ASRService/1"
def log_message(self,*_): pass
def send_data(self,data,status=200,ctype="text/html; charset=utf-8",disposition=""):
self.send_response(status); self.send_header("Content-Type",ctype); self.send_header("Cache-Control","no-store"); self.send_header("X-Content-Type-Options","nosniff"); self.send_header("Content-Security-Policy","default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self' 'unsafe-inline'")
if disposition:self.send_header("Content-Disposition",disposition)
self.send_header("Content-Length",str(len(data)));self.end_headers();self.wfile.write(data)
def json(self,obj,status=200):self.send_data(json.dumps(obj,ensure_ascii=False).encode(),status,"application/json; charset=utf-8")
def auth(self):
value=self.headers.get("Authorization","")
return bool(AGENT_TOKEN and value.startswith("Bearer ") and hmac.compare_digest(value[7:],AGENT_TOKEN))
def need_auth(self):
if self.auth():return True
self.json({"error":"unauthorized"},401);return False
def header(self,name,limit):return safe(urllib.parse.unquote(self.headers.get(name,"")),limit)
def upload(self,source,protected):
if protected and not self.auth():return self.json({"error":"unauthorized"},401)
try:length=int(self.headers.get("Content-Length","0"))
except ValueError:length=0
name=self.header("X-Filename",240);suffix=Path(name).suffix.lower()
if length<=0 or length>MAX_UPLOAD:return self.json({"error":"invalid size"},413)
if suffix not in SUPPORTED:return self.json({"error":"unsupported type"},415)
jid=uuid.uuid4().hex[:24]; title=self.header("X-Title",80); stored=f"{time.strftime('%Y-%m-%d_%H_%M_%S')}-{stem(title or Path(name).stem)}-{jid[:8]}{suffix}"; staging=INBOX/".uploading";staging.mkdir(parents=True,exist_ok=True);part=staging/f"{jid}.part";remaining=length;digest=hashlib.sha256()
try:
with part.open("xb") as out:
while remaining:
chunk=self.rfile.read(min(1024*1024,remaining))
if not chunk:raise EOFError
out.write(chunk);digest.update(chunk);remaining-=len(chunk)
out.flush();os.fsync(out.fileno())
job=db.create_job(jid,name,stored,title=title,context=self.header("X-Context",2000),size=length,source=source);db.update_job(jid,"queued","录音已落盘，等待 worker",sha256=digest.hexdigest());part.replace(INBOX/stored);self.json(public(job),201)
except Exception:part.unlink(missing_ok=True);self.json({"error":"upload failed"},500)
def artifact(self,job,name,api=False):
if not job:return self.json({"error":"not found"},404)
roots={"transcript":TRANSCRIPTS,"draft":DRAFTS,"result":RESULTS};path=Path(job.get(name) or "")
try:path.resolve().relative_to(roots[name].resolve());data=path.read_bytes()
except (OSError,ValueError):return self.json({"error":"not found"},404)
self.send_data(data,200,"text/markdown; charset=utf-8",f'inline; filename="{path.name}"')
def do_GET(self):
path=urllib.parse.urlsplit(self.path).path
if path in ("/health","/healthz"):return self.json({"ok":True,"service":"asr-web"})
if path=="/readyz":
try:db.connect().close();ok=(STATE/"worker.heartbeat").exists() and time.time()-(STATE/"worker.heartbeat").stat().st_mtime<90
except Exception:ok=False
return self.json({"ready":ok},200 if ok else 503)
if path=="/":return self.send_data(home())
if path=="/docs":return self.send_data(shell("Agent API","
"))
if path=="/ui/jobs":return self.json({"jobs":[public(x) for x in db.list_jobs()]})
m=re.fullmatch(r"/jobs/([0-9a-f]{24})/",path)
if m:
job=db.get_job(m.group(1));return self.send_data(job_page(job)) if job else self.send_data(b"Not found",404,"text/plain")
m=re.fullmatch(r"/jobs/([0-9a-f]{24})/review/",path)
if m:
job=db.get_job(m.group(1));return self.send_data(review_page(job)) if job else self.send_data(b"Not found",404,"text/plain")
m=re.fullmatch(r"/jobs/([0-9a-f]{24})/artifact/(transcript|draft|result)",path)
if m:return self.artifact(db.get_job(m.group(1)),m.group(2))
if path=="/api/v1/jobs":
if not self.need_auth():return
q=urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query);return self.json({"jobs":[public(x) for x in db.list_jobs(int(q.get('limit',[30])[0]),int(q.get('offset',[0])[0]),q.get('status',[''])[0])]})
m=re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})",path)
if m:
if not self.need_auth():return
job=db.get_job(m.group(1));return self.json(public(job,True)) if job else self.json({"error":"not found"},404)
m=re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})/artifacts/(transcript|draft|result)",path)
if m:
if not self.need_auth():return
return self.artifact(db.get_job(m.group(1)),m.group(2),True)
m=re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})/reviews",path)
if m:
if not self.need_auth():return
job=db.get_job(m.group(1))
if not job:return self.json({"error":"not found"},404)
rows=[]
for row in pending(job):
payload=json.loads(row["payload"] or "{}")
rows.append({"id":row["id"],"question":row["question"],
"kind":payload.get("kind",""),
"sentence":payload.get("sentence",""),
"candidates":payload.get("candidates") or []})
return self.json({"reviews":rows})
self.send_data(b"Not found",404,"text/plain")
def do_POST(self):
path=urllib.parse.urlsplit(self.path).path
if path=="/ui/upload":return self.upload("web",False)
if path=="/api/v1/jobs":return self.upload("agent",True)
m=re.fullmatch(r"/jobs/([0-9a-f]{24})/review/",path)
if m:
job=db.get_job(m.group(1));length=min(int(self.headers.get("Content-Length","0")),4096);form=urllib.parse.parse_qs(self.rfile.read(length).decode());qid=int(form.get("id",["0"])[0]);answer=(form.get("custom",[""])[0].strip() or form.get("answer",[""])[0].strip())
if not job or qid not in {x["id"] for x in pending(job)} or not answer:return self.send_data(b"Invalid answer",400,"text/plain")
review.answer(qid,answer[:120]);self.send_response(303);self.send_header("Location",path);self.end_headers();return
m=re.fullmatch(r"/(?:api/v1/)?jobs/([0-9a-f]{24})/retry",path)
if m:
api=path.startswith("/api/")
if api and not self.need_auth():return
job=db.get_job(m.group(1));source=next((p for p in FAILED.iterdir() if job and p.name==job["stored_name"]),None)
if not job or job["status"]!="failed" or not source:return self.json({"error":"cannot retry"},409)
source.replace(INBOX/source.name);db.update_job(job["id"],"queued","重新排队",error_code="")
if api:return self.json({"ok":True},202)
self.send_response(303);self.send_header("Location",f"/jobs/{job['id']}/");self.end_headers();return
m=re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})/reviews/(\d+)",path)
if m:
if not self.need_auth():return
job=db.get_job(m.group(1));length=min(int(self.headers.get("Content-Length","0")),4096)
try:answer=str(json.loads(self.rfile.read(length)).get("answer") or "").strip()
except (ValueError,AttributeError):answer=""
qid=int(m.group(2))
if not job or qid not in {x["id"] for x in pending(job)} or not answer:
return self.json({"error":"invalid review answer"},400)
return self.json(review.answer(qid,answer[:120]))
self.send_data(b"Not found",404,"text/plain")
def main():
ensure_dirs();db.connect().close();server=ThreadingHTTPServer(("0.0.0.0",8788),Handler);server.daemon_threads=True;server.serve_forever()
if __name__=="__main__":main()