"""Untitled magazine entries: continuation detection, merging, suggested titles, CLI and routes.""" from __future__ import annotations import json import os import shutil import tempfile import unittest from pathlib import Path from unittest.mock import AsyncMock, Mock, patch _CONFIG = tempfile.mkdtemp(prefix="ttvc-magazine-untitled-test-") os.environ.setdefault("CONFIG_DIR", _CONFIG) from fastapi import HTTPException # noqa: E402 from core import magazine_untitled as mu # noqa: E402 def blk(bid, page, order, kind, text, x=40, w=240): return {"id": bid, "page": page, "reading_order": order, "type": kind, "text": text, "bbox": [x, 100 + order * 50, w, 40]} def art(aid, title, blocks, pages, status="approved", **extra): return {"id": aid, "title": title, "author": "", "pages": pages, "blocks": blocks, "images": [], "status": status, "confidence": 0.55, "section": "", "continued_to": [], "continued_from": [], "audio_status": "none", **extra} LONG = "Die Entwickler setzen inzwischen auf Agenten, die selbstständig Code schreiben und prüfen. " def sample(): """Page 10 opens a feature, 11 and 12 continue it as 'Page N' stand-ins, 13 is an advert, 14 is a new signed-off article, 15 holds only a picture credit.""" blocks = [ blk("b10-1", 10, 0, "title", "Per KI von der Idee zum Programm"), blk("b10-2", 10, 1, "byline", "Von Anna Beispiel"), blk("b10-3", 10, 2, "body", LONG * 3 + "Die Agenten arbeiten dabei ohne Pause und schreiben Hunder-"), blk("b11-1", 11, 0, "body", "te Zeilen am Tag. " + LONG * 3 + "Erst danach kommt der Mensch wieder ins"), blk("b12-1", 12, 0, "body", "Spiel und prüft die Ergebnisse. " + LONG * 2), blk("b12-2", 12, 1, "heading", "Blick nach vorn"), blk("b12-3", 12, 2, "body", LONG * 2 + "Das Projekt ist damit abgeschlossen. (hag) ct"), blk("b13-1", 13, 0, "ad", "Jetzt abonnieren! Drei Hefte gratis."), blk("b14-1", 14, 0, "body", "Recht an der eigenen Stimme. " + LONG * 3), blk("b15-1", 15, 0, "body", "Bild: Agentur"), ] layout = {"page_count": 15, "pages": [{"page": p, "width": 600, "height": 800} for p in range(10, 16)], "blocks": blocks, "images": [{"id": "img-13", "page": 13}]} articles = [ art("article-001", "Per KI von der Idee zum Programm", ["b10-1", "b10-2", "b10-3"], [10]), art("article-002", "Page 11", ["b11-1"], [11]), art("article-003", "Page 12", ["b12-1", "b12-2", "b12-3"], [12]), art("article-004", "Page 13", ["b13-1"], [13]), art("article-005", "Page 14", ["b14-1"], [14]), art("article-006", "Page 15", ["b15-1"], [15]), ] return layout, articles class PlaceholderTests(unittest.TestCase): def test_placeholder_titles(self): for t in ("", " ", "Page 33", "page 2", "Seite 7", "Untitled"): self.assertTrue(mu.is_placeholder_title(t), t) for t in ("Page 33 of the issue", "Inhalt", "Wer haftet, wenn KI einkauft?"): self.assertFalse(mu.is_placeholder_title(t), t) class ContinuationScoreTests(unittest.TestCase): def score(self, tail, head, **kw): args = dict(adjacent=True, similar_columns=True, starts_with_headline=False) args.update(kw) return mu.continuation_score(tail, {"type": "body", "text": head}, **args) def test_lowercase_start_after_open_sentence_is_high(self): s, ev = self.score("und schreiben Hunder-", "te Zeilen am Tag.") self.assertGreaterEqual(s, mu.HIGH_AT) self.assertIn("starts lower-case / mid-sentence", ev) self.assertIn("previous text ends in a hyphenated word break", ev) def test_signed_off_previous_article_blocks_a_merge(self): s, ev = self.score("Das Projekt ist abgeschlossen. (hag) ct", "Recht an der eigenen Stimme.") self.assertLess(s, mu.SUGGEST_AT) self.assertIn("previous article is signed off", ev) def test_own_headline_counts_against(self): base, _ = self.score("kommt der Mensch wieder ins", "Spiel und prüft.") with_head, ev = self.score("kommt der Mensch wieder ins", "Spiel und prüft.", starts_with_headline=True) self.assertLess(with_head, base) self.assertIn("starts with its own headline", ev) def test_finished_sentence_far_away_is_low(self): s, _ = self.score("Ende.", "Neuer Anfang.", adjacent=False, similar_columns=False) self.assertLess(s, mu.SUGGEST_AT) class ClassificationTests(unittest.TestCase): def setUp(self): self.layout, self.articles = sample() self.ann = mu.annotate(self.layout, self.articles, language="de") def test_kinds(self): kinds = {k: v["untitled_kind"] for k, v in self.ann.items()} self.assertIsNone(kinds["article-001"]) self.assertEqual(kinds["article-002"], "continuation") self.assertEqual(kinds["article-003"], "continuation") self.assertEqual(kinds["article-004"], "advert") self.assertEqual(kinds["article-006"], "junk") # previous article signed off, so page 14 is not a continuation self.assertNotEqual(kinds["article-005"], "continuation") def test_suggestion_points_at_the_previous_entry_with_evidence(self): s = self.ann["article-002"]["merge_suggestion"] self.assertEqual(s["into"], "article-001") self.assertEqual(s["level"], "high") self.assertTrue(s["evidence"]) self.assertEqual(self.ann["article-003"]["merge_suggestion"]["into"], "article-002") def test_display_fields(self): a1, a4, a2 = self.ann["article-001"], self.ann["article-004"], self.ann["article-002"] self.assertEqual(a1["display_title"], "Per KI von der Idee zum Programm") self.assertEqual(a1["title_source"], "title") self.assertEqual(a4["display_title"], "Anzeige") self.assertEqual(a4["title_source"], "suggested") self.assertEqual(a4["page_label"], "Seite 13") self.assertNotIn("Page", a2["display_title"]) def test_previous_entry_skips_advert_pages_but_not_long_gaps(self): layout, articles = sample() # page 14 follows page 12 across the advert on 13 -> previous entry is page 12's index = mu._Index(layout, articles) prev, _ = mu.previous_entry(index, articles[4]) self.assertEqual(prev["id"], "article-003") layout["blocks"].append(blk("b20-1", 20, 0, "body", "weiter geht es hier. " + LONG)) articles.append(art("article-007", "Page 20", ["b20-1"], [20])) prev, _ = mu.previous_entry(mu._Index(layout, articles), articles[-1]) self.assertIsNone(prev) def test_empty_stand_in(self): layout, articles = sample() articles.append(art("article-008", "Page 16", [], [16], status="detected")) layout["images"].append({"id": "img-16", "page": 16}) ann = mu.annotate(layout, articles, language="de") self.assertEqual(ann["article-008"]["untitled_kind"], "image_only") self.assertEqual(ann["article-008"]["display_title"], "Bildseite") def test_contents_heading(self): layout, articles = sample() layout["blocks"].append(blk("b16-1", 16, 0, "title", "Inhalt")) layout["blocks"].append(blk("b16-2", 16, 1, "body", "Seite 12 Agenten. Seite 30 Recht. " * 3)) articles.append(art("article-009", "", ["b16-1", "b16-2"], [16])) ann = mu.annotate(layout, articles, language="de") self.assertEqual(ann["article-009"]["untitled_kind"], "contents") self.assertEqual(ann["article-009"]["suggested_title"], "Inhalt") def test_article_head_gets_its_headline(self): layout, articles = sample() layout["blocks"].append(blk("b17-1", 17, 0, "title", "Wer haftet, wenn KI einkauft?")) layout["blocks"].append(blk("b17-2", 17, 1, "body", LONG * 3)) articles.append(art("article-010", "Page 17", ["b17-1", "b17-2"], [17])) ann = mu.annotate(layout, articles, language="de") self.assertEqual(ann["article-010"]["untitled_kind"], "article_head") self.assertEqual(ann["article-010"]["suggested_title"], "Wer haftet, wenn KI einkauft?") self.assertEqual(ann["article-010"]["suggested_title_source"], "heading") class LlmCheckTests(unittest.TestCase): def test_llm_only_consulted_in_the_ambiguous_band(self): layout, articles = sample() check = Mock(return_value=True) index = mu._Index(layout, articles) mu.merge_suggestion(index, articles[1], llm_check=check) # clear case: high already check.assert_not_called() # a page that starts a new capitalised sentence after a finished one: ambiguous layout["blocks"][3]["text"] = "Neue Zeilen am Tag. " + LONG * 3 layout["blocks"][2]["text"] = LONG * 3 index = mu._Index(layout, articles) without = mu.merge_suggestion(index, articles[1]) with_yes = mu.merge_suggestion(index, articles[1], llm_check=check) check.assert_called_once() self.assertIn("LLM: continues", with_yes["evidence"]) self.assertGreater(with_yes["confidence"], without["confidence"] if without else 0) no = mu.merge_suggestion(index, articles[1], llm_check=Mock(return_value=False)) self.assertIsNone(no) def test_llm_failure_is_ignored(self): layout, articles = sample() layout["blocks"][3]["text"] = "Neue Zeilen am Tag. " + LONG * 3 layout["blocks"][2]["text"] = LONG * 3 index = mu._Index(layout, articles) plain = mu.merge_suggestion(index, articles[1]) broken = mu.merge_suggestion(index, articles[1], llm_check=Mock(side_effect=RuntimeError("down"))) self.assertEqual(plain, broken) def test_continuation_check_parses_json(self): self.assertTrue(mu.llm_continuation_check(lambda m: '{"continues": true}')("a", "b")) self.assertFalse(mu.llm_continuation_check(lambda m: '```json\n{"continues": false}\n```')("a", "b")) self.assertIsNone(mu.llm_continuation_check(lambda m: "maybe")("a", "b")) class TitleTests(unittest.TestCase): def test_deterministic_sentence_skips_a_leading_fragment_and_trims(self): layout = {"blocks": [blk("x1", 5, 0, "body", "te Zeilen am Tag. Danach prüft der Mensch die " "Ergebnisse sehr gründlich und mit viel Geduld jeden Tag.")]} index = mu._Index(layout, []) title, source = mu.deterministic_title(index, art("a", "Page 5", ["x1"], [5]), "untitled_article") self.assertEqual(source, "sentence") self.assertTrue(title.startswith("Danach prüft der Mensch")) self.assertLessEqual(len(title.replace(" …", "").split()), mu.MAX_TITLE_WORDS) def test_type_labels_follow_the_language(self): index = mu._Index({"blocks": []}, []) self.assertEqual(mu.deterministic_title(index, art("a", "", [], []), "advert", "de")[0], "Anzeige") self.assertEqual(mu.deterministic_title(index, art("a", "", [], []), "advert", "en")[0], "Advertisement") self.assertEqual(mu.page_label(33, "en"), "page 33") def test_validate_llm_title(self): text = "Agenten schreiben Code, Menschen prüfen die Ergebnisse" self.assertEqual(mu.validate_llm_title('"Agenten schreiben Code"', text), "Agenten schreiben Code") self.assertEqual(mu.validate_llm_title("hm**Agenten prüfen Ergebnisse**", text), "Agenten prüfen Ergebnisse") for bad in ("", "Titel: Agenten schreiben Code", "Hier ist ein Titel", "Agenten", "Eins zwei drei vier fünf sechs sieben acht neun", "Zeile eins\nZeile zwei", "Quantencomputer erobern Weltraum", "A" * 80): self.assertIsNone(mu.validate_llm_title(bad, text), bad) def test_llm_title_falls_back_on_errors(self): text = "Agenten schreiben Code" self.assertEqual(mu.llm_title(lambda m: "Agenten schreiben Code", text, "de"), "Agenten schreiben Code") self.assertIsNone(mu.llm_title(Mock(side_effect=TimeoutError()), text, "de")) def test_prompt_is_strict_and_in_the_magazine_language(self): msgs = mu._title_messages("Text", "de") self.assertIn("ausschließlich", msgs[0]["content"]) self.assertIn("Deutsch", msgs[0]["content"]) self.assertIn("unvertrauenswürdige", msgs[0]["content"]) def test_cached_llm_title_is_used_only_while_the_text_matches(self): layout, articles = sample() index = mu._Index(layout, articles) cache = {"entries": {"article-005": {"title": "Recht an der eigenen Stimme", "hash": mu.text_hash(index, articles[4])}}} ann = mu.annotate(layout, articles, cache, language="de") self.assertEqual(ann["article-005"]["suggested_title"], "Recht an der eigenen Stimme") self.assertEqual(ann["article-005"]["suggested_title_source"], "llm") cache["entries"]["article-005"]["hash"] = "stale" ann = mu.annotate(layout, articles, cache, language="de") self.assertNotEqual(ann["article-005"]["suggested_title_source"], "llm") class MergeTests(unittest.TestCase): def test_merge_orders_blocks_and_marks_audio_stale(self): layout, articles = sample() articles[0]["blocks"] = ["b10-3", "b10-1", "b10-2"] rec = mu.merge_entries(layout, articles, "article-002", "article-001") merged = articles[0] self.assertEqual(merged["blocks"], ["b10-1", "b10-2", "b10-3", "b11-1"]) self.assertEqual(merged["pages"], [10, 11]) self.assertTrue(merged["audio_stale"]) self.assertEqual(merged["merged_from"], ["article-002"]) self.assertNotIn("article-002", [a["id"] for a in articles]) self.assertEqual(rec["status"], "approved") self.assertEqual(merged["title"], "Per KI von der Idee zum Programm") def test_merge_never_replaces_a_real_title_and_adopts_one_for_a_placeholder(self): layout, articles = sample() articles[1]["title"] = "Echter Titel" mu.merge_entries(layout, articles, "article-002", "article-001") self.assertEqual(articles[0]["title"], "Per KI von der Idee zum Programm") layout, articles = sample() articles[2]["title"] = "Durchstarten" mu.merge_entries(layout, articles, "article-003", "article-002") self.assertEqual(next(a for a in articles if a["id"] == "article-002")["title"], "Durchstarten") def test_unreviewed_source_makes_the_result_unreviewed(self): layout, articles = sample() articles[1]["status"] = "detected" mu.merge_entries(layout, articles, "article-002", "article-001") self.assertEqual(articles[0]["status"], "detected") def test_bad_ids(self): layout, articles = sample() with self.assertRaises(ValueError): mu.merge_entries(layout, articles, "article-001", "article-001") with self.assertRaises(KeyError): mu.merge_entries(layout, articles, "article-404", "article-001") class PlanTests(unittest.TestCase): def test_chain_collapses_into_the_root_and_input_is_untouched(self): layout, articles = sample() snapshot = json.dumps(articles) result = mu.plan(layout, articles) self.assertEqual(json.dumps(articles), snapshot) ids = [a["id"] for a in result["articles"]] self.assertNotIn("article-002", ids) self.assertNotIn("article-003", ids) root = result["articles"][0] self.assertEqual(root["pages"], [10, 11, 12]) self.assertEqual(root["merged_from"], ["article-002", "article-003"]) titled = {t["id"] for t in result["titles"]} self.assertNotIn("article-001", titled) self.assertIn("article-004", titled) def test_duplicate_title_tail_is_merged_and_named(self): layout, articles = sample() articles[0]["title"] = "Page 10" # the head is a stand-in ... articles[2]["title"] = "Per KI von der Idee zum Programm" # ... and the tail kept the title result = mu.plan(layout, articles) root = result["articles"][0] self.assertEqual(root["pages"], [10, 11, 12]) self.assertEqual(root["title"], "Per KI von der Idee zum Programm") def test_threshold_controls_merges(self): layout, articles = sample() self.assertEqual(mu.plan(layout, articles, min_confidence=1.01)["merges"], []) def test_excluded_entries_are_left_untouched(self): layout, articles = sample() base = mu.plan(layout, articles) self.assertTrue(base["merges"]) src = base["merges"][0]["source"] result = mu.plan(layout, articles, exclude=[src]) self.assertTrue(all(m["source"] != src for m in result["merges"])) self.assertNotIn(src, {t["id"] for t in result["titles"]}) kept = next(a for a in result["articles"] if a["id"] == src) orig = next(a for a in articles if a["id"] == src) self.assertEqual(kept["title"], orig["title"]) def test_apply_titles_only_touches_placeholders(self): _, articles = sample() n = mu.apply_titles(articles, [{"id": "article-001", "after": "X"}, {"id": "article-004", "after": "Anzeige", "source": "type"}]) self.assertEqual(n, 1) self.assertEqual(articles[0]["title"], "Per KI von der Idee zum Programm") self.assertEqual(articles[3]["title"], "Anzeige") self.assertEqual(articles[3]["title_placeholder"], "Page 13") class CliTests(unittest.TestCase): def setUp(self): self.dir = Path(tempfile.mkdtemp(prefix="untitled-cli-")) layout, articles = sample() (self.dir / "layout.json").write_text(json.dumps(layout), "utf-8") (self.dir / "articles.json").write_text(json.dumps({"articles": articles}), "utf-8") (self.dir / "audio").mkdir() (self.dir / "audio" / "article-002.mp3").write_bytes(b"ID3") def tearDown(self): shutil.rmtree(self.dir, ignore_errors=True) def test_dry_run_writes_nothing(self): from scripts import magazine_untitled as cli before = sorted(p.name for p in self.dir.rglob("*")) text = (self.dir / "articles.json").read_text("utf-8") s = cli.run(self.dir, apply=False, min_confidence=mu.HIGH_AT) self.assertEqual(sorted(p.name for p in self.dir.rglob("*")), before) self.assertEqual((self.dir / "articles.json").read_text("utf-8"), text) self.assertEqual(len(s["merges"]), 2) self.assertEqual(s["audio_kept_orphaned"], ["article-002"]) def test_apply_backs_up_merges_and_keeps_audio(self): from scripts import magazine_untitled as cli s = cli.run(self.dir, apply=True, min_confidence=mu.HIGH_AT) backups = list(self.dir.glob("articles.backup-untitled-*.json")) self.assertEqual(len(backups), 1) self.assertEqual(len(json.loads(backups[0].read_text("utf-8"))["articles"]), 6) after = json.loads((self.dir / "articles.json").read_text("utf-8"))["articles"] self.assertEqual(len(after), 4) self.assertTrue(after[0]["audio_stale"]) self.assertEqual(next(a for a in after if a["id"] == "article-004")["title"], "Anzeige") self.assertTrue((self.dir / "audio" / "article-002.mp3").is_file()) self.assertGreaterEqual(s["titles_applied"], 1) def test_llm_titles_with_mocked_chat(self): from scripts import magazine_untitled as cli chat = Mock(return_value="Recht an der eigenen Stimme") s = cli.run(self.dir, apply=False, min_confidence=mu.HIGH_AT, chat=chat) by_id = {t["id"]: t for t in s["titles"]} self.assertEqual(by_id["article-005"]["source"], "llm") self.assertEqual(by_id["article-005"]["after"], "Recht an der eigenen Stimme") class RouteTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.dir = Path(tempfile.mkdtemp(prefix="untitled-routes-")) layout, articles = sample() (self.dir / "layout.json").write_text(json.dumps(layout), "utf-8") (self.dir / "articles.json").write_text(json.dumps({"articles": articles}), "utf-8") def tearDown(self): shutil.rmtree(self.dir, ignore_errors=True) def articles(self): return json.loads((self.dir / "articles.json").read_text("utf-8"))["articles"] async def test_articles_listing_carries_display_fields(self): from routes.magazine import magazine_articles with patch("routes.magazine._require_magazine", return_value=(self.dir, {})): out = await magazine_articles("a" * 32) a4 = next(a for a in out["articles"] if a["id"] == "article-004") self.assertEqual(a4["display_title"], "Anzeige") self.assertEqual(a4["page_label"], "Seite 13") a2 = next(a for a in out["articles"] if a["id"] == "article-002") self.assertEqual(a2["merge_suggestion"]["into"], "article-001") self.assertEqual(self.articles()[1]["title"], "Page 11") # nothing saved async def test_merge_into_previous_checks_the_target_and_is_saved(self): from routes.magazine import magazine_merge_into_previous request = AsyncMock() request.body.return_value = b"{}" with patch("routes.magazine._require_magazine", return_value=(self.dir, {})): request.json.return_value = {"into": "article-005"} with self.assertRaises(HTTPException) as err: await magazine_merge_into_previous("a" * 32, "article-002", request) self.assertEqual(err.exception.status_code, 409) request.json.return_value = {"into": "article-001"} out = await magazine_merge_into_previous("a" * 32, "article-002", request) self.assertEqual(out["article"]["pages"], [10, 11]) with self.assertRaises(HTTPException) as err: # a titled, finished article: nothing proposed await magazine_merge_into_previous("a" * 32, "article-001", request) self.assertEqual(err.exception.status_code, 409) saved = self.articles() self.assertEqual(len(saved), 5) self.assertTrue(saved[0]["audio_stale"]) async def test_accept_title_never_overwrites_a_real_title(self): from routes.magazine import magazine_accept_title request = AsyncMock() request.body.return_value = b"{}" request.json.return_value = {} with patch("routes.magazine._require_magazine", return_value=(self.dir, {})): with self.assertRaises(HTTPException) as err: await magazine_accept_title("a" * 32, "article-001", request) self.assertEqual(err.exception.status_code, 409) out = await magazine_accept_title("a" * 32, "article-004", request) self.assertEqual(out["article"]["title"], "Anzeige") self.assertEqual(self.articles()[3]["title"], "Anzeige") self.assertEqual(self.articles()[3]["title_origin"], "type") async def test_untitled_report(self): from routes.magazine import magazine_untitled_report with patch("routes.magazine._require_magazine", return_value=(self.dir, {})): out = await magazine_untitled_report("a" * 32) self.assertEqual(out["kinds"]["continuation"], 2) self.assertEqual(out["merge_suggestions"], 2) def test_fresh_render_clears_audio_stale(self): from routes.magazine import _clear_audio_stale arts = self.articles() arts[0].update(audio_stale=True, audio_stale_reason="merged") (self.dir / "articles.json").write_text(json.dumps({"articles": arts}), "utf-8") _clear_audio_stale(self.dir, "article-001") self.assertNotIn("audio_stale", self.articles()[0]) def test_suggest_titles_job_uses_the_llm_and_caches(self): from routes import magazine as rm resp = Mock(status_code=200) resp.json.return_value = {"choices": [{"message": {"content": "Recht an der eigenen Stimme"}}]} with patch.object(rm, "_LIB", self.dir.parent), \ patch.object(rm, "_load_settings", return_value={}), \ patch.object(rm, "_llm_chat_fast", return_value=resp), \ patch.object(rm, "background_job_update") as upd: rm._suggest_titles_sync(self.dir.name, "job") self.assertEqual(upd.call_args.kwargs.get("stage"), "done") cache = mu.load_title_cache(self.dir) self.assertEqual(cache["entries"]["article-005"]["title"], "Recht an der eigenen Stimme") self.assertEqual(self.articles()[4]["title"], "Page 14") # titles themselves unchanged if __name__ == "__main__": unittest.main()