mARTin-B78
736aeaaeac
Magazines break the audiobook pipeline's assumptions: several unrelated articles share a page, columns run in parallel, adverts and sidebars interrupt body copy, and a feature routinely jumps twenty pages to its tail. Narrating page-by-page walks out of an article mid-sentence and into an advert, so the canonical object is the article, not the page. core/magazine.py reconstructs articles from a flat PDF: PyMuPDF extracts blocks with bboxes and font metrics, tesseract supplies a TSV fallback so scanned pages join the same column grouping, blocks are classified and ordered column-major, and articles are anchored by the parsed table of contents with "continued on page N" tails linked back. routes/magazine.py adds 13 endpoints. Ingestion is a restart-safe background job. Nothing is narrated until a human approves the boundaries; podcast generation refuses unapproved articles. Podcast mode renders an approved article as a 2-3 host discussion using the configured LLM and voices the user picks, alongside narration rather than instead of it. Verified end-to-end over HTTP on a synthetic 8-page issue: the feature came back as pages [3, 8] with its jump tail absorbed and the advert page excluded, and an image-only rendering recovered all 47 blocks via OCR. 44 new tests; full suite 90 passing. The Studio UI is not built — this is reachable only via the API. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
"""Magazine route-layer regressions: path safety, review gating, script parsing."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
|
|
_CONFIG = tempfile.mkdtemp(prefix="ttvc-magazine-routes-test-")
|
|
os.environ.setdefault("CONFIG_DIR", _CONFIG)
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from routes.magazine import _article_id, _doc_dir, _parse_script
|
|
|
|
|
|
class PathSafetyTests(unittest.TestCase):
|
|
def test_traversal_and_junk_ids_are_rejected(self):
|
|
for bad in ("../../etc", "a/b", "", "..", "x" * 41, "id-with-dash", "a.b"):
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
_doc_dir(bad)
|
|
self.assertEqual(ctx.exception.status_code, 400, bad)
|
|
|
|
def test_wellformed_but_missing_id_is_a_404_not_a_400(self):
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
_doc_dir("0123456789abcdef0123456789abcdef")
|
|
self.assertEqual(ctx.exception.status_code, 404)
|
|
|
|
def test_article_ids_reject_separators(self):
|
|
self.assertEqual(_article_id("article-001"), "article-001")
|
|
for bad in ("../x", "a/b", "", "a" * 65, "a b"):
|
|
with self.assertRaises(HTTPException):
|
|
_article_id(bad)
|
|
|
|
|
|
class PodcastScriptParsingTests(unittest.TestCase):
|
|
def test_plain_json_parses(self):
|
|
raw = json.dumps({
|
|
"title": "Batteries",
|
|
"turns": [{"host": "A", "text": "Hello."},
|
|
{"host": "B", "text": "Hi."}],
|
|
})
|
|
got = _parse_script(raw, ["A", "B"])
|
|
self.assertEqual(got["title"], "Batteries")
|
|
self.assertEqual([t["host"] for t in got["turns"]], ["A", "B"])
|
|
|
|
def test_markdown_fence_is_tolerated(self):
|
|
raw = '```json\n{"title":"T","turns":[{"host":"A","text":"Hi."}]}\n```'
|
|
self.assertEqual(_parse_script(raw, ["A", "B"])["turns"][0]["text"], "Hi.")
|
|
|
|
def test_prose_around_the_json_is_stripped(self):
|
|
raw = 'Sure! Here is the script:\n{"title":"T","turns":[{"host":"A","text":"Hi."}]}\nHope that helps.'
|
|
self.assertEqual(_parse_script(raw, ["A", "B"])["title"], "T")
|
|
|
|
def test_unknown_host_label_is_reassigned_not_dropped(self):
|
|
"""An unrecognised label would otherwise leave a turn with no voice."""
|
|
raw = json.dumps({"turns": [
|
|
{"host": "NARRATOR", "text": "One."},
|
|
{"host": "B", "text": "Two."},
|
|
]})
|
|
got = _parse_script(raw, ["A", "B"])
|
|
self.assertEqual(len(got["turns"]), 2)
|
|
self.assertTrue(all(t["host"] in ("A", "B") for t in got["turns"]))
|
|
|
|
def test_empty_turns_are_skipped(self):
|
|
raw = json.dumps({"turns": [
|
|
{"host": "A", "text": " "},
|
|
{"host": "B", "text": "Real line."},
|
|
]})
|
|
got = _parse_script(raw, ["A", "B"])
|
|
self.assertEqual(len(got["turns"]), 1)
|
|
|
|
def test_unusable_output_raises_502_not_500(self):
|
|
for raw in ("not json at all", "{}", json.dumps({"turns": []}),
|
|
json.dumps({"turns": [{"host": "A", "text": ""}]})):
|
|
with self.assertRaises(HTTPException) as ctx:
|
|
_parse_script(raw, ["A", "B"])
|
|
self.assertEqual(ctx.exception.status_code, 502, raw)
|
|
|
|
def test_overlong_turn_text_is_capped(self):
|
|
raw = json.dumps({"turns": [{"host": "A", "text": "x" * 5000}]})
|
|
self.assertEqual(len(_parse_script(raw, ["A"])["turns"][0]["text"]), 2000)
|
|
|
|
|
|
def tearDownModule():
|
|
shutil.rmtree(_CONFIG, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|