tts-voice-creator-clone-and-design-2/tests/test_magazine_untitled.py

"""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("<think>hm</think>**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_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()