feat(tools-pve/asr-pipeline): SQLite model config, task delete/rerefine, multi-recording review, lexicon CRUD

servers/tools-pve/asr-pipeline/asrservice/db.py

@@ -4,15 +4,19 @@ import sqlite3
 import time
 from typing import Any
 
-from .config import DB, DEFAULT_PROVIDER, ensure_dirs
+from asrpipe.config import CFG, bootstrap_default_settings, get_setting, load_db_settings, set_settings
+from .config import DB, ensure_dirs
 
 STATUSES = {
     "uploading", "queued", "transcribing", "refining", "review",
     "publishing", "complete", "failed",
 }
 
+_BOOTSTRAPPED_DB: str = ""
+
 
 def connect() -> sqlite3.Connection:
+    global _BOOTSTRAPPED_DB
     ensure_dirs()
     con = sqlite3.connect(DB, timeout=30)
     con.row_factory = sqlite3.Row
@@ -37,8 +41,14 @@ def connect() -> sqlite3.Connection:
         status TEXT NOT NULL, detail TEXT NOT NULL DEFAULT '', created_at REAL NOT NULL
       );
       CREATE INDEX IF NOT EXISTS ix_job_event_job ON job_event(job_id,id);
+      CREATE TABLE IF NOT EXISTS app_setting(
+        key TEXT PRIMARY KEY,
+        value TEXT NOT NULL DEFAULT '',
+        updated_at REAL NOT NULL
+      );
     """)
     cols = {r[1] for r in con.execute("PRAGMA table_info(intake_job)")}
+    default_prov = CFG.asr_provider or "qwen3cam"
     additions = {
         "draft": "TEXT NOT NULL DEFAULT ''",
         "result": "TEXT NOT NULL DEFAULT ''",
@@ -46,13 +56,19 @@ def connect() -> sqlite3.Connection:
         "sha256": "TEXT NOT NULL DEFAULT ''",
         "attempts": "INTEGER NOT NULL DEFAULT 0",
         "error_code": "TEXT NOT NULL DEFAULT ''",
-        "asr_provider": f"TEXT NOT NULL DEFAULT '{DEFAULT_PROVIDER}'",
+        "asr_provider": f"TEXT NOT NULL DEFAULT '{default_prov}'",
         "asr_model": "TEXT NOT NULL DEFAULT ''",
     }
     for name, ddl in additions.items():
         if name not in cols:
             con.execute(f"ALTER TABLE intake_job ADD COLUMN {name} {ddl}")
     con.commit()
+    if _BOOTSTRAPPED_DB != str(DB):
+        _BOOTSTRAPPED_DB = str(DB)
+        try:
+            bootstrap_default_settings()
+        except Exception:
+            pass
     return con
 
 
@@ -69,7 +85,7 @@ def create_job(
     asr_model: str = "",
 ) -> dict[str, Any]:
     now = time.time()
-    prov = (asr_provider or DEFAULT_PROVIDER or "qwen3cam").strip().lower()
+    prov = (asr_provider or CFG.asr_provider or "qwen3cam").strip().lower()
     with connect() as con:
         con.execute(
             "INSERT INTO intake_job(id,original_name,stored_name,title,context,size,"
@@ -88,7 +104,7 @@ def update_job(job_id: str, status: str, detail: str = "", **fields: Any) -> boo
         raise ValueError(f"invalid status: {status}")
     allowed = {
         "transcript", "draft", "result", "sha256", "attempts",
-        "error_code", "asr_provider", "asr_model",
+        "error_code", "asr_provider", "asr_model", "title", "context",
     }
     changes = {k: v for k, v in fields.items() if k in allowed and v is not None}
     sets = ["status=?", "updated_at=?", "detail=?"]
@@ -106,6 +122,44 @@ def update_job(job_id: str, status: str, detail: str = "", **fields: Any) -> boo
     return bool(cur.rowcount)
 
 
+def update_job_meta(
+    job_id: str,
+    *,
+    title: str | None = None,
+    context: str | None = None,
+    asr_provider: str | None = None,
+    asr_model: str | None = None,
+) -> bool:
+    sets = ["updated_at=?"]
+    values: list[Any] = [time.time()]
+    if title is not None:
+        sets.append("title=?")
+        values.append(title[:120])
+    if context is not None:
+        sets.append("context=?")
+        values.append(context[:2000])
+    if asr_provider is not None:
+        sets.append("asr_provider=?")
+        values.append(asr_provider[:40])
+    if asr_model is not None:
+        sets.append("asr_model=?")
+        values.append(asr_model[:80])
+    values.append(job_id)
+    with connect() as con:
+        cur = con.execute(f"UPDATE intake_job SET {','.join(sets)} WHERE id=?", values)
+    return bool(cur.rowcount)
+
+
+def delete_job(job_id: str) -> dict[str, Any] | None:
+    job = get_job(job_id)
+    if not job:
+        return None
+    with connect() as con:
+        con.execute("DELETE FROM job_event WHERE job_id=?", (job_id,))
+        con.execute("DELETE FROM intake_job WHERE id=?", (job_id,))
+    return job
+
+
 def get_job(job_id: str) -> dict[str, Any] | None:
     with connect() as con:
         row = con.execute("SELECT * FROM intake_job WHERE id=?", (job_id,)).fetchone()
@@ -120,13 +174,27 @@ def find_by_stored(stored_name: str) -> dict[str, Any] | None:
         return dict(row) if row else None
 
 
-def list_jobs(limit: int = 30, offset: int = 0, status: str = "") -> list[dict[str, Any]]:
+def list_jobs(
+    limit: int = 30,
+    offset: int = 0,
+    status: str = "",
+    q: str = "",
+    provider: str = "",
+) -> list[dict[str, Any]]:
     limit = min(max(limit, 1), 200)
     offset = max(offset, 0)
-    sql, args = "SELECT * FROM intake_job", []
+    sql = "SELECT * FROM intake_job WHERE 1=1"
+    args: list[Any] = []
     if status:
-        sql += " WHERE status=?"
+        sql += " AND status=?"
         args.append(status)
+    if provider:
+        sql += " AND asr_provider=?"
+        args.append(provider)
+    if q.strip():
+        pat = f"%{q.strip()}%"
+        sql += " AND (title LIKE ? OR original_name LIKE ? OR context LIKE ? OR detail LIKE ?)"
+        args.extend([pat, pat, pat, pat])
     sql += " ORDER BY created_at DESC LIMIT ? OFFSET ?"
     args.extend((limit, offset))
     with connect() as con:
@@ -145,3 +213,32 @@ def pending_recovery() -> list[dict[str, Any]]:
         return [dict(r) for r in con.execute(
             "SELECT * FROM intake_job WHERE status IN ('queued','transcribing','refining','publishing') "
             "ORDER BY created_at").fetchall()]
+
+
+def get_all_settings() -> dict[str, str]:
+    connect().close()
+    return load_db_settings(force=True)
+
+
+def save_settings(updates: dict[str, Any]) -> dict[str, str]:
+    connect().close()
+    set_settings(updates)
+    return load_db_settings(force=True)
+
+
+__all__ = [
+    "STATUSES",
+    "connect",
+    "create_job",
+    "update_job",
+    "update_job_meta",
+    "delete_job",
+    "get_job",
+    "find_by_stored",
+    "list_jobs",
+    "events",
+    "pending_recovery",
+    "get_all_settings",
+    "save_settings",
+    "get_setting",
+]

servers/tools-pve/asr-pipeline/asrservice/worker.py

@@ -4,6 +4,7 @@ import fcntl
 import hashlib
 import json
 import os
+import re
 import shutil
 import sqlite3
 import time
@@ -11,9 +12,10 @@ import uuid
 from pathlib import Path
 
 from asrpipe import graph, lexicon, providers, refine, review, revise, vault
+from asrpipe.config import CFG
 from . import db
 from .config import (
-    DATA, DEFAULT_PROVIDER, DRAFTS, FAILED, INBOX, LOGS, MAX_ATTEMPTS,
+    DATA, DRAFTS, FAILED, INBOX, LOGS, MAX_ATTEMPTS,
     POLL_SECONDS, PROCESSED, PROCESSING, PUBLIC_URL, REVISION_QUEUE,
     STABLE_SECONDS, STATE, SUPPORTED, TRANSCRIPTS, ensure_dirs,
 )
@@ -50,7 +52,7 @@ def transcribe(
 ) -> tuple[Path, providers.TranscribeResult]:
     res = providers.transcribe_audio(
         audio,
-        provider=provider or DEFAULT_PROVIDER,
+        provider=provider or CFG.asr_provider,
         model=model,
         title=title,
         user_context=user_context,
@@ -71,7 +73,7 @@ def transcribe(
 def process_audio(source: Path, job: dict) -> None:
     job_id = job["id"]
     attempts = int(job.get("attempts") or 0) + 1
-    prov = providers.normalize_provider_name(job.get("asr_provider") or DEFAULT_PROVIDER)
+    prov = providers.normalize_provider_name(job.get("asr_provider") or CFG.asr_provider)
     model = str(job.get("asr_model") or "")
     digest = sha256(source)
     prov_label = providers.PROVIDERS.get(prov, {}).get("label", prov)
@@ -120,6 +122,47 @@ def process_audio(source: Path, job: dict) -> None:
          f"{PUBLIC_URL}/jobs/{job_id}/review/")
 
 
+def rerefine_job(job: dict) -> None:
+    """Re-run LLM refinement on an existing raw transcript without re-running ASR."""
+    job_id = job["id"]
+    transcript = Path(str(job.get("transcript") or ""))
+    if not transcript.is_file():
+        raise FileNotFoundError(f"transcript file missing for job {job_id}: {transcript}")
+    raw = transcript.read_text(encoding="utf-8", errors="ignore")
+    intake = {
+        "job_id": job_id,
+        "title": job.get("title", ""),
+        "context": job.get("context", ""),
+        "source": job.get("source", ""),
+        "asr_provider": job.get("asr_provider") or CFG.asr_provider,
+        "asr_model": job.get("asr_model") or "",
+    }
+    intake_line = f"<!-- intake: {json.dumps(intake, ensure_ascii=False)} -->"
+    if re.match(r"^<!--\s*intake:\s*\{.*?\}\s*-->", raw):
+        raw = re.sub(r"^<!--\s*intake:\s*\{.*?\}\s*-->", intake_line, raw, count=1)
+    else:
+        raw = intake_line + "\n" + raw
+    transcript.write_text(raw, encoding="utf-8")
+
+    db.update_job(job_id, "refining", "正在基于现有转写重新生成精修草稿")
+    review.delete_for_job(
+        transcript_path=str(transcript),
+        draft_path=str(job.get("draft") or ""),
+        job_id=job_id,
+    )
+    result = refine.refine(
+        raw, date=time.strftime("%Y-%m-%d"),
+        transcript_path=str(transcript), queue_review=True,
+    )
+    draft = DRAFTS / f"{job_id}.md"
+    vault.write_draft_file(draft, result.draft)
+    questions = review.enqueue_draft(str(transcript), draft)
+    db.update_job(
+        job_id, "review", f"整理稿已重新生成，{questions} 个问题待确认",
+        transcript=str(transcript), draft=str(draft),
+    )
+
+
 def claim(path: Path) -> tuple[Path, dict]:
     job = db.find_by_stored(path.name)
     if not job:
@@ -127,7 +170,7 @@ def claim(path: Path) -> tuple[Path, dict]:
         job = db.create_job(
             job_id, path.name, path.name, title=path.stem,
             size=path.stat().st_size, source="ssh",
-            asr_provider=DEFAULT_PROVIDER,
+            asr_provider=CFG.asr_provider,
         )
     target = PROCESSING / path.name
     if target.exists():
@@ -164,8 +207,15 @@ def process_revisions() -> None:
     for marker in sorted(REVISION_QUEUE.glob("*.json")):
         try:
             payload = json.loads(marker.read_text(encoding="utf-8"))
-            draft = Path(payload["draft"])
+            action = str(payload.get("action") or "revise")
             job_id = str(payload.get("job_id") or "")
+            if action == "rerefine":
+                job = db.get_job(job_id)
+                if job:
+                    rerefine_job(job)
+                marker.unlink(missing_ok=True)
+                continue
+            draft = Path(payload["draft"])
             db.update_job(job_id, "publishing", "正在应用人工确认")
             revise.revise(draft)
             current = db.get_job(job_id) or {}
@@ -181,6 +231,14 @@ def process_revisions() -> None:
             marker.unlink(missing_ok=True)
         except Exception as exc:
             log(f"revision marker={marker.name} failed type={type(exc).__name__} err={exc}")
+            try:
+                payload = json.loads(marker.read_text(encoding="utf-8"))
+                jid = str(payload.get("job_id") or "")
+                if jid:
+                    db.update_job(jid, "failed", f"精修/修订失败：{exc}", error_code=type(exc).__name__)
+            except Exception:
+                pass
+            marker.unlink(missing_ok=True)
 
 
 def recover() -> None: