import pytest from gewu.adapters.text_extractor import TextExtractionError, TextMarkdownExtractor from gewu.domain.validation import extraction_digest def test_markdown_extraction_preserves_unicode_and_original_location_maps() -> None: source = "# Evidence\r\nCafe\u0301 has 400 visitors.\r\n" extraction = TextMarkdownExtractor().extract( "shared:source:r1", source.encode("utf-8"), "text/markdown" ) assert extraction.extraction_id == extraction_digest(extraction) block = extraction.blocks[0] assert "Café has 400 visitors." in block.normalized_text normalized_start = block.normalized_text.index("Café") normalized_end = normalized_start + len("Café") original_spans = block.normalization_map[normalized_start:normalized_end] original_quote = block.original_text[ min(span[0] for span in original_spans) : max(span[1] for span in original_spans) ] assert original_quote == "Cafe\u0301" def test_html_extraction_keeps_visible_text_and_excludes_active_content() -> None: source = ( "
Café has 400 visitors.
" "" "Second paragraph.
" ) extraction = TextMarkdownExtractor().extract( "shared:html:r1", source.encode("utf-8"), "text/html" ) assert extraction.parser_version == "0.3.0" text = "\n".join(block.original_text for block in extraction.blocks) assert "Evidence" in text assert "Café has 400 visitors." in "\n".join( block.normalized_text for block in extraction.blocks ) assert "ignoreThisSecret" not in text assert "fallback text" not in text assert "display:none" not in text block = extraction.blocks[0] normalized_start = block.normalized_text.index("Café") original_spans = block.normalization_map[normalized_start : normalized_start + 4] original_quote = block.original_text[ min(span[0] for span in original_spans) : max(span[1] for span in original_spans) ] assert original_quote == "Cafe\u0301" def test_extraction_splits_bounded_blocks_without_losing_text() -> None: source = ("a" * 80_000 + "\n" + "b" * 200).encode("utf-8") extraction = TextMarkdownExtractor().extract("shared:large:r1", source, "text/plain") assert len(extraction.blocks) == 2 assert "".join(block.original_text for block in extraction.blocks) == source.decode("utf-8") assert all(len(block.normalized_text) <= 80_000 for block in extraction.blocks) def test_webvtt_extraction_records_exact_cue_timestamps_and_text_offsets() -> None: source = ( "WEBVTT\n\n" "cue-1\n00:14:32.000 --> 00:14:36.500 align:start\n" "The audience entered the room.\n\n" "cue-2\n00:14:40.000 --> 00:14:42.000\n" "The host introduced the film.\n" ) extraction = TextMarkdownExtractor().extract( "shared:talk-transcript:r1", source.encode(), "text/vtt" ) cues = [block for block in extraction.blocks if block.time_start_ms is not None] assert [(cue.time_start_ms, cue.time_end_ms) for cue in cues] == [ (872_000, 876_500), (880_000, 882_000), ] quote = "The audience entered the room." assert quote in cues[0].normalized_text start = cues[0].normalized_text.index(quote) location = cues[0].normalization_map[start : start + len(quote)] original = cues[0].original_text[ min(span[0] for span in location) : max(span[1] for span in location) ] assert original == quote def test_subrip_transcript_extracts_comma_millisecond_timestamps() -> None: extraction = TextMarkdownExtractor().extract( "shared:interview:r1", b"1\n00:01:02,250 --> 00:01:03,750\nA precise quote.\n", "application/x-subrip", ) cue = next(block for block in extraction.blocks if block.time_start_ms is not None) assert (cue.time_start_ms, cue.time_end_ms) == (62_250, 63_750) assert "A precise quote." in cue.normalized_text @pytest.mark.parametrize( "content", [ b"WEBVTT\n\n00:00:05.000 --> 00:00:02.000\nReversed time.\n", b"WEBVTT\n\n00:00:01.000 --> invalid\nMalformed time.\n", b"WEBVTT\n\nNo timestamped cue.\n", ], ) def test_transcript_extraction_rejects_invalid_or_missing_timestamps(content: bytes) -> None: with pytest.raises(TextExtractionError): TextMarkdownExtractor().extract("shared:bad-transcript:r1", content, "text/vtt") @pytest.mark.parametrize( ("content", "media_type"), [(b"\xff", "text/plain"), (b"payload", "application/pdf"), (b"", "text/plain")], ) def test_extractor_rejects_invalid_or_unsupported_sources(content: bytes, media_type: str) -> None: with pytest.raises(TextExtractionError): TextMarkdownExtractor().extract("shared:source:r1", content, media_type)