171 lines
8.0 KiB
Python
171 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Untitled magazine entries ("Page 33"): propose merges into the previous article and titles.
|
|
|
|
Dry run by default: prints what would change and writes nothing. ``--apply`` first writes
|
|
``articles.backup-untitled-<timestamp>.json`` next to ``articles.json``, then merges every
|
|
suggestion at or above ``--min-confidence`` (page order, so chains of continuation pages collapse
|
|
into their article), marks merged articles ``audio_stale`` (audio files are kept), and writes
|
|
suggested titles into placeholder titles only. Titles someone set are never touched.
|
|
|
|
python3 scripts/magazine_untitled.py <doc-id | path/to/doc_dir> [--apply] [--no-llm] [--exclude ID,ID] [--json out.json]
|
|
|
|
See architecture/magazine-ingestion.md, "Untitled entries".
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as _dt
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
os.environ.setdefault("CONFIG_DIR", str(ROOT / "config"))
|
|
|
|
from core import magazine as mag # noqa: E402
|
|
from core import magazine_untitled as mu # noqa: E402
|
|
|
|
|
|
def resolve_doc_dir(arg: str) -> Path:
|
|
path = Path(arg)
|
|
if path.is_dir():
|
|
return path
|
|
if not re.fullmatch(r"[0-9a-f]{32}", arg):
|
|
raise SystemExit(f"Not a directory or document id: {arg}")
|
|
from core.constants import CONFIG_DIR
|
|
path = Path(CONFIG_DIR) / "reader_library" / arg
|
|
if not path.is_dir():
|
|
raise SystemExit(f"No magazine at {path}")
|
|
return path
|
|
|
|
|
|
def make_chat(timeout: float = 30.0):
|
|
"""A chat(messages) -> str callable on the app's configured LLM, or None when switched off."""
|
|
from core.config import _load_settings
|
|
settings = _load_settings()
|
|
if settings.get("magazine_llm_titles") is False:
|
|
return None
|
|
from routes.conversation import _post_llm_chat_completion
|
|
url = settings.get("llm_url") or "http://localhost:11434/v1"
|
|
headers = {"Authorization": f"Bearer {settings.get('llm_api_key') or 'sk-dummy-key'}"}
|
|
|
|
state = {"down": ""}
|
|
|
|
def chat(messages):
|
|
if state["down"]: # one failure is enough: fall back for the run
|
|
raise RuntimeError(state["down"])
|
|
try:
|
|
return _ask(messages)
|
|
except Exception as exc: # noqa: BLE001 - reported once, then deterministic
|
|
state["down"] = f"LLM unavailable ({str(exc)[:120]}); using deterministic titles and heuristics"
|
|
print(state["down"], file=sys.stderr)
|
|
raise
|
|
|
|
def _ask(messages):
|
|
payload = {"model": str(settings.get("llm_model") or "").strip(), "temperature": 0.0, "stream": False,
|
|
"max_tokens": 60, "messages": messages, "chat_template_kwargs": {"enable_thinking": False}}
|
|
resp = _post_llm_chat_completion(url, payload, headers, timeout)
|
|
if resp.status_code in (400, 422):
|
|
payload.pop("chat_template_kwargs")
|
|
resp = _post_llm_chat_completion(url, payload, headers, timeout)
|
|
resp.raise_for_status()
|
|
return resp.json()["choices"][0]["message"]["content"] or ""
|
|
return chat
|
|
|
|
|
|
def run(doc_dir: Path, *, apply: bool, min_confidence: float, chat=None, exclude=()) -> dict:
|
|
layout = mag.load_layout(doc_dir)
|
|
articles = mag.load_articles(doc_dir)
|
|
if not layout or not articles:
|
|
raise SystemExit(f"No layout/articles in {doc_dir}")
|
|
cache = mu.load_title_cache(doc_dir)
|
|
language = mu.doc_language(layout)
|
|
llm_check = mu.llm_continuation_check(chat) if chat else None
|
|
new_cache = {"entries": dict(cache.get("entries") or {})}
|
|
|
|
def titler(article_id: str, text: str) -> str | None:
|
|
if chat is None:
|
|
return None
|
|
return mu.llm_title(chat, text, language)
|
|
|
|
result = mu.plan(layout, articles, min_confidence=min_confidence, cache=cache,
|
|
llm_check=llm_check, titler=titler if chat else None, exclude=exclude)
|
|
index = mu._Index(layout, result["articles"])
|
|
for t in result["titles"]:
|
|
if t["source"] == "llm":
|
|
a = next(x for x in result["articles"] if x["id"] == t["id"])
|
|
new_cache["entries"][t["id"]] = {"title": t["after"], "hash": mu.text_hash(index, a),
|
|
"prompt": mu.PROMPT_VERSION}
|
|
summary = {
|
|
"doc_dir": str(doc_dir),
|
|
"mode": "apply" if apply else "dry-run",
|
|
"before": {"entries": len(articles), "untitled": result["before"]["count"],
|
|
"kinds": result["before"]["kinds"], "merge_suggestions": result["before"]["merge_suggestions"]},
|
|
"merges": result["merges"],
|
|
"titles": result["titles"],
|
|
"after": {"entries": len(result["articles"])},
|
|
"pending": result["pending"],
|
|
"audio_kept_orphaned": [m["source"] for m in result["merges"]
|
|
if (doc_dir / "audio" / f"{m['source']}.mp3").is_file()],
|
|
"audio_stale": sorted({m["target"] for m in result["merges"]}),
|
|
}
|
|
if apply:
|
|
stamp = _dt.datetime.now(_dt.timezone.utc).strftime("%Y%m%dT%H%M%S")
|
|
backup = doc_dir / f"articles.backup-untitled-{stamp}.json"
|
|
backup.write_text((doc_dir / "articles.json").read_text("utf-8"), "utf-8")
|
|
final = result["articles"]
|
|
summary["titles_applied"] = mu.apply_titles(final, result["titles"])
|
|
mag.save_articles(doc_dir, final)
|
|
mu.save_title_cache(doc_dir, new_cache)
|
|
summary["backup"] = str(backup)
|
|
return summary
|
|
|
|
|
|
def print_summary(s: dict) -> None:
|
|
b = s["before"]
|
|
print(f"{s['mode'].upper()} {s['doc_dir']}")
|
|
print(f"entries {b['entries']} -> {s['after']['entries']}; untitled/duplicate {b['untitled']}: {b['kinds']}")
|
|
print(f"merges: {len(s['merges'])} (audio marked stale on {len(s['audio_stale'])}, "
|
|
f"{len(s['audio_kept_orphaned'])} MP3s of removed entries kept on disk)")
|
|
for m in s["merges"]:
|
|
print(f" {m['source']} {m['title_before'][:28]!r:30} -> {m['target']} ({m['confidence']:.2f}) "
|
|
f"| {m['first_text'][:50]!r}")
|
|
print(f"suggestions left for review (below --min-confidence): {len(s['pending'])}")
|
|
for m in s["pending"]:
|
|
print(f" {m['id']} {m['title'][:28]!r:30} -> {m['into']} ({m['confidence']:.2f}) {m['page_label']}")
|
|
print(f"titles: {len(s['titles'])}")
|
|
for t in s["titles"]:
|
|
print(f" {t['id']} {t['before'][:12]!r:14} -> {t['after'][:60]!r} [{t['source']}/{t['kind']}] {t['page_label']}")
|
|
if s.get("backup"):
|
|
print(f"backup: {s['backup']}; titles applied: {s.get('titles_applied')}")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
p.add_argument("doc", help="magazine document id or its directory")
|
|
mode = p.add_mutually_exclusive_group()
|
|
mode.add_argument("--dry-run", action="store_true", default=True, help="report only (default)")
|
|
mode.add_argument("--apply", action="store_true", help="write changes (backup first)")
|
|
p.add_argument("--min-confidence", type=float, default=mu.HIGH_AT,
|
|
help=f"merge threshold (default {mu.HIGH_AT})")
|
|
p.add_argument("--exclude", action="append", default=[], metavar="ID[,ID...]",
|
|
help="leave these article ids as they are (not merged into another entry, no new title); repeatable")
|
|
p.add_argument("--no-llm", action="store_true", help="heuristics and deterministic titles only")
|
|
p.add_argument("--json", help="also write the full result as JSON to this path")
|
|
args = p.parse_args(argv)
|
|
doc_dir = resolve_doc_dir(args.doc)
|
|
chat = None if args.no_llm else make_chat()
|
|
summary = run(doc_dir, apply=args.apply, min_confidence=args.min_confidence, chat=chat,
|
|
exclude=[i.strip() for v in args.exclude for i in v.split(",") if i.strip()])
|
|
print_summary(summary)
|
|
if args.json:
|
|
Path(args.json).write_text(json.dumps(summary, ensure_ascii=False, indent=1), "utf-8")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|