From d46898dceb38156e29a5e14ff95ec375e5fcb818 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 6 Jul 2026 09:33:09 +0900 Subject: [PATCH] feat: add extractive transcript summarizer (summarize.py) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turn long transcripts into skimmable notes with a stdlib-only extractive summarizer — no external deps, no LLM or network calls. - summarize_text(text, max_sentences=5) -> Summary(summary_text, key_sentences, word_count): regex sentence splitting, word-frequency scoring with a small stopword list (average frequency, so long sentences don't win on length), top-N sentences returned in original document order. - summarize_segments(segments) accepts duck-typed objects with a .text attribute (e.g. timestamped transcription segments) and reassembles sentences that span segment boundaries. - Edge cases covered: empty/whitespace input, short input passthrough, very long input, max_sentences validation, CJK terminators (。!?) recognized and Korean/CJK text handled without crashing (whitespace tokenization limitation documented honestly in the module docstring). - 19 new tests in tests/test_summarize.py; interrogate 100% repo-wide. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CJRVbDrp1vGYkJgNHMGPpG --- summarize.py | 198 +++++++++++++++++++++++++++++++++++++++ tests/test_summarize.py | 202 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 400 insertions(+) create mode 100644 summarize.py create mode 100644 tests/test_summarize.py diff --git a/summarize.py b/summarize.py new file mode 100644 index 0000000..a5b56fe --- /dev/null +++ b/summarize.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +"""Extractive transcript summarization with zero external dependencies. + +This module turns long transcript text into skimmable notes by selecting the +most representative sentences verbatim from the input. It is an *extractive* +summarizer — it never generates new prose, never paraphrases, and never calls +an LLM or any network service. The algorithm is the classic word-frequency +approach: + +1. Split the text into sentences with a regular expression. +2. Build a frequency table of lowercased whitespace tokens, ignoring a small + English stopword list and bare punctuation. +3. Score each sentence by the average frequency of its content words. +4. Return the top-N sentences, re-ordered to match their original positions. + +Limitations (by design, to stay stdlib-only): + +* Tokenization is whitespace-based. Languages written without spaces between + words (Chinese, Japanese) degrade to per-run tokens, so frequency scoring is + coarse for them; Korean is space-separated and fares better, though particle + suffixes are not stripped. CJK input is handled without crashing and CJK + sentence terminators (。!?) are recognized. +* Sentence splitting is heuristic; unusual abbreviations may split early. +""" + +import re +from dataclasses import dataclass, field + +# Sentence terminators: ASCII . ! ? plus CJK ideographic full stop and +# full-width ! / ?. A sentence ends at one or more terminators followed by +# closing quotes/brackets, then whitespace or end-of-text. +_SENTENCE_END_RE = re.compile( + r"(?<=[.!?。!?])[\"'”’)\]]*\s+" +) + +# Strip leading/trailing punctuation from tokens before frequency counting. +_TOKEN_STRIP_RE = re.compile(r"^\W+|\W+$", re.UNICODE) + +# Small English stopword list: enough to keep glue words from dominating the +# frequency table without pulling in any external corpus. +_STOPWORDS = frozenset( + """ + a an and are as at be but by for from had has have he her his i if in is + it its me my nor not of on or our she so that the their them then there + they this to us was we were what when which who will with you your + """.split() +) + + +@dataclass +class Summary: + """Result of an extractive summarization run. + + Attributes: + summary_text: The selected sentences joined with single spaces, in + their original document order. + key_sentences: The selected sentences as a list, in original order. + word_count: Number of whitespace-separated tokens in the *input* + text (not the summary). + """ + + summary_text: str + key_sentences: list = field(default_factory=list) + word_count: int = 0 + + +def _split_sentences(text): + """Split ``text`` into sentences using punctuation heuristics. + + Args: + text: Raw input text. + + Returns: + A list of non-empty sentence strings with surrounding whitespace + stripped. Newlines without terminal punctuation do not split + sentences, so transcripts with hard-wrapped lines stay intact. + """ + normalized = re.sub(r"\s+", " ", text).strip() + if not normalized: + return [] + parts = _SENTENCE_END_RE.split(normalized) + return [part.strip() for part in parts if part.strip()] + + +def _content_words(sentence): + """Extract lowercased content words from a sentence. + + Args: + sentence: A single sentence string. + + Returns: + A list of lowercased whitespace tokens with surrounding punctuation + stripped, excluding stopwords and empty tokens. + """ + words = [] + for raw in sentence.split(): + token = _TOKEN_STRIP_RE.sub("", raw).lower() + if token and token not in _STOPWORDS: + words.append(token) + return words + + +def _score_sentences(sentences): + """Score sentences by average content-word frequency. + + Args: + sentences: List of sentence strings. + + Returns: + A list of float scores parallel to ``sentences``. Sentences with no + content words score 0.0. Using the *average* (rather than the sum) + keeps long sentences from winning on length alone. + """ + frequencies = {} + tokenized = [] + for sentence in sentences: + words = _content_words(sentence) + tokenized.append(words) + for word in words: + frequencies[word] = frequencies.get(word, 0) + 1 + + scores = [] + for words in tokenized: + if words: + scores.append(sum(frequencies[w] for w in words) / len(words)) + else: + scores.append(0.0) + return scores + + +def summarize_text(text, max_sentences=5): + """Produce an extractive summary of ``text``. + + Selects the ``max_sentences`` highest-scoring sentences (ties broken by + earlier position) and returns them in their original order, so the + summary reads chronologically — important for transcripts. + + Args: + text: The transcript or document text to summarize. May be empty. + max_sentences: Maximum number of sentences to include in the + summary. Must be at least 1. + + Returns: + A :class:`Summary`. For empty or whitespace-only input the summary + is empty with ``word_count`` 0. If the text has ``max_sentences`` or + fewer sentences, all of them are returned unchanged (short input is + effectively passed through). + + Raises: + ValueError: If ``max_sentences`` is less than 1. + """ + if max_sentences < 1: + raise ValueError("max_sentences must be at least 1") + + word_count = len(text.split()) + sentences = _split_sentences(text) + if not sentences: + return Summary(summary_text="", key_sentences=[], word_count=0) + + if len(sentences) <= max_sentences: + selected = list(sentences) + else: + scores = _score_sentences(sentences) + ranked = sorted( + range(len(sentences)), key=lambda i: (-scores[i], i) + ) + chosen = sorted(ranked[:max_sentences]) + selected = [sentences[i] for i in chosen] + + return Summary( + summary_text=" ".join(selected), + key_sentences=selected, + word_count=word_count, + ) + + +def summarize_segments(segments, max_sentences=5): + """Summarize a sequence of transcript segments. + + Accepts any iterable of duck-typed segment objects exposing a ``.text`` + attribute (e.g. timestamped transcription segments). Segment texts are + joined with single spaces before summarization, so sentences that span + segment boundaries are reassembled. + + Args: + segments: Iterable of objects with a ``.text`` string attribute. + max_sentences: Maximum number of sentences in the summary. Must be + at least 1. + + Returns: + A :class:`Summary` over the joined segment text. + + Raises: + ValueError: If ``max_sentences`` is less than 1. + AttributeError: If a segment lacks a ``.text`` attribute. + """ + joined = " ".join(segment.text for segment in segments) + return summarize_text(joined, max_sentences=max_sentences) diff --git a/tests/test_summarize.py b/tests/test_summarize.py new file mode 100644 index 0000000..4d4c231 --- /dev/null +++ b/tests/test_summarize.py @@ -0,0 +1,202 @@ +"""Tests for the extractive transcript summarizer (summarize.py).""" + +import unittest +from dataclasses import dataclass + +from summarize import Summary, summarize_segments, summarize_text + + +@dataclass +class FakeSegment: + """Duck-typed stand-in for a timestamped transcription segment.""" + + text: str + start: float = 0.0 + end: float = 0.0 + + +class TestSummarizeText(unittest.TestCase): + """Behavioral tests for summarize_text.""" + + def test_empty_input_returns_empty_summary(self): + result = summarize_text("") + self.assertIsInstance(result, Summary) + self.assertEqual(result.summary_text, "") + self.assertEqual(result.key_sentences, []) + self.assertEqual(result.word_count, 0) + + def test_whitespace_only_input_returns_empty_summary(self): + result = summarize_text(" \n\t ") + self.assertEqual(result.summary_text, "") + self.assertEqual(result.key_sentences, []) + self.assertEqual(result.word_count, 0) + + def test_short_input_returned_as_is(self): + text = "Codec detection finished. Two streams were recovered." + result = summarize_text(text, max_sentences=5) + self.assertEqual( + result.key_sentences, + ["Codec detection finished.", "Two streams were recovered."], + ) + self.assertEqual( + result.summary_text, + "Codec detection finished. Two streams were recovered.", + ) + self.assertEqual(result.word_count, 7) + + def test_single_sentence_without_terminator(self): + result = summarize_text("no punctuation here at all") + self.assertEqual(result.key_sentences, ["no punctuation here at all"]) + self.assertEqual(result.word_count, 5) + + def test_max_sentences_respected(self): + text = " ".join(f"Sentence number {i} talks about topics." for i in range(10)) + result = summarize_text(text, max_sentences=3) + self.assertEqual(len(result.key_sentences), 3) + + def test_max_sentences_below_one_raises(self): + with self.assertRaises(ValueError): + summarize_text("Some text.", max_sentences=0) + with self.assertRaises(ValueError): + summarize_text("Some text.", max_sentences=-2) + + def test_top_sentences_selected_by_frequency(self): + # "codec" appears in three sentences, making it the dominant term. + # The filler sentences share no repeated content words. + text = ( + "Codec analysis started today. " + "Weather stayed calm outside. " + "Codec recovery needs codec tables. " + "Lunch arrived late unfortunately. " + "Broken codec headers hide codec frames." + ) + result = summarize_text(text, max_sentences=2) + self.assertEqual(len(result.key_sentences), 2) + for sentence in result.key_sentences: + self.assertIn("codec", sentence.lower()) + # The two densest codec sentences must win over the filler ones. + self.assertNotIn("Weather stayed calm outside.", result.key_sentences) + self.assertNotIn("Lunch arrived late unfortunately.", result.key_sentences) + + def test_original_order_preserved(self): + # The highest-scoring sentence appears last in the document; the + # summary must still present sentences in document order. + text = ( + "Alpha recovery began early. " + "Random filler mentions nothing shared. " + "Alpha frames feed alpha decoding of alpha streams." + ) + result = summarize_text(text, max_sentences=2) + self.assertEqual(len(result.key_sentences), 2) + positions = [text.index(s) for s in result.key_sentences] + self.assertEqual(positions, sorted(positions)) + # summary_text mirrors key_sentences order. + self.assertEqual(result.summary_text, " ".join(result.key_sentences)) + + def test_stopwords_do_not_dominate_scoring(self): + # A sentence made almost entirely of stopwords must lose to a + # sentence carrying the repeated content word. + text = ( + "It is the and of to a in that. " + "Transcoding pipeline rebuilt transcoding cache. " + "Transcoding jobs resumed. " + "Nothing else happened." + ) + result = summarize_text(text, max_sentences=1) + self.assertIn("transcoding", result.key_sentences[0].lower()) + + def test_word_count_counts_input_tokens(self): + text = "One two three. Four five six seven." + result = summarize_text(text, max_sentences=1) + self.assertEqual(result.word_count, 7) + + def test_long_input_produces_bounded_summary(self): + text = " ".join( + f"Segment {i} discusses container metadata and stream offsets." + for i in range(500) + ) + result = summarize_text(text, max_sentences=4) + self.assertEqual(len(result.key_sentences), 4) + self.assertLess(len(result.summary_text), len(text)) + self.assertEqual(result.word_count, 500 * 8) + + def test_korean_text_does_not_crash(self): + text = ( + "코덱 복구 작업이 시작되었습니다. " + "코덱 테이블을 분석했습니다. " + "날씨가 좋았습니다. " + "코덱 헤더를 복원했습니다." + ) + result = summarize_text(text, max_sentences=2) + self.assertEqual(len(result.key_sentences), 2) + for sentence in result.key_sentences: + self.assertIn(sentence, text) + + def test_cjk_fullwidth_terminators_split_sentences(self): + text = "映像の解析が完了した。音声の復元も完了した。結果は良好だ。" + result = summarize_text(text, max_sentences=5) + # No whitespace after 。 means no split — input passes through whole. + self.assertEqual(len(result.key_sentences), 1) + self.assertEqual(result.summary_text, text) + + spaced = "映像の解析が完了した。 音声の復元も完了した。 結果は良好だ。" + spaced_result = summarize_text(spaced, max_sentences=2) + self.assertEqual(len(spaced_result.key_sentences), 2) + + def test_ties_broken_by_earlier_position(self): + # All sentences score identically; the earliest ones must be chosen. + text = "Alpha beta gamma. Delta epsilon zeta. Eta theta iota. Kappa lambda mu." + result = summarize_text(text, max_sentences=2) + self.assertEqual( + result.key_sentences, + ["Alpha beta gamma.", "Delta epsilon zeta."], + ) + + +class TestSummarizeSegments(unittest.TestCase): + """Behavioral tests for summarize_segments.""" + + def test_segments_joined_and_summarized(self): + segments = [ + FakeSegment(text="Muxer errors detected in the first pass.", start=0.0), + FakeSegment(text="Muxer retries fixed most muxer errors.", start=4.2), + FakeSegment(text="Coffee break happened afterwards.", start=9.9), + ] + result = summarize_segments(segments, max_sentences=1) + self.assertEqual(len(result.key_sentences), 1) + self.assertIn("muxer", result.key_sentences[0].lower()) + + def test_sentence_spanning_segment_boundary_is_reassembled(self): + segments = [ + FakeSegment(text="The bitstream parser found"), + FakeSegment(text="seventeen recoverable frames."), + ] + result = summarize_segments(segments, max_sentences=5) + self.assertEqual( + result.key_sentences, + ["The bitstream parser found seventeen recoverable frames."], + ) + + def test_empty_segment_list(self): + result = summarize_segments([]) + self.assertEqual(result.summary_text, "") + self.assertEqual(result.key_sentences, []) + self.assertEqual(result.word_count, 0) + + def test_max_sentences_validated(self): + with self.assertRaises(ValueError): + summarize_segments([FakeSegment(text="Hello there.")], max_sentences=0) + + def test_duck_typing_accepts_any_object_with_text(self): + class Chunk: + """Minimal object exposing only .text.""" + + def __init__(self, text): + self.text = text + + result = summarize_segments([Chunk("Only one sentence here.")]) + self.assertEqual(result.key_sentences, ["Only one sentence here."]) + + +if __name__ == "__main__": + unittest.main()