Add resources/excluded_phrases.txt (skip links, privacy/consent, legal boilerplate) and match it case- and punctuation-insensitively as a contiguous run of words. Phrases live in a data file so they can be extended without touching code.master
| @@ -41,6 +41,7 @@ Run from the project root — the resource paths are relative (`./resources/...` | |||
| - `resources/sources.txt` — one news source URL per line. | |||
| - `resources/stopwords.txt` — one stopword per line. | |||
| - `resources/excluded_phrases.txt` — boilerplate link text to ignore (see below). | |||
| - Defaults live at the top of `main.py` (`SIMILARITY_THRESHOLD = 0.75`, | |||
| `MIN_SOURCES = 2`) and can be overridden on the command line. | |||
| @@ -97,6 +98,11 @@ News stories covered by at least 2 distinct sources (published on or after 2026- | |||
| attributes or inner text) and associated with headlines in document order; the | |||
| association resets at each `<article>`/`<li>` boundary so undated headlines don't | |||
| inherit a neighboring story's date. | |||
| - **Boilerplate link text is filtered.** Navigation/footer/legal links like | |||
| "skip to content" or "your privacy choices" are dropped by matching against | |||
| `excluded_phrases.txt` (case- and punctuation-insensitive; an entry matches when | |||
| it appears anywhere in the link text as a run of words, so `privacy choices` | |||
| also catches "Your Privacy Choices"). Extend the file rather than editing code. | |||
| ## Tests | |||
| @@ -120,7 +126,8 @@ anya/ | |||
| │ ├── headline.py # Headline model (text, domain, published_at) | |||
| │ └── story.py # Story model (sources, representative, latest_date) | |||
| ├── resources/ | |||
| │ ├── sources.txt # one source URL per line | |||
| │ └── stopwords.txt # one stopword per line | |||
| │ ├── sources.txt # one source URL per line | |||
| │ ├── stopwords.txt # one stopword per line | |||
| │ └── excluded_phrases.txt # boilerplate link text to ignore | |||
| └── tests/ # unit tests | |||
| ``` | |||
| @@ -4,7 +4,7 @@ from datetime import date | |||
| from services.dates import filter_by_date_window, parse_date_arg | |||
| from services.headlines import prepare_headlines | |||
| from services.normalization import get_stopwords | |||
| from services.normalization import get_excluded_phrases, get_stopwords | |||
| from services.sources import get_sources | |||
| from services.stories import cluster_stories | |||
| @@ -16,6 +16,7 @@ logger = logging.getLogger(__name__) | |||
| SOURCE_FILE = './resources/sources.txt' | |||
| STOPWORDS_FILE = './resources/stopwords.txt' | |||
| EXCLUDED_PHRASES_FILE = './resources/excluded_phrases.txt' | |||
| SIMILARITY_THRESHOLD = 0.75 | |||
| MIN_SOURCES = 2 | |||
| @@ -86,9 +87,17 @@ def main(): | |||
| logger.critical("Failed to load stopwords from '%s': %s", STOPWORDS_FILE, e, exc_info=True) | |||
| return | |||
| try: | |||
| logger.info("Loading excluded phrases from '%s'", EXCLUDED_PHRASES_FILE) | |||
| excluded_phrases = get_excluded_phrases(EXCLUDED_PHRASES_FILE) | |||
| logger.info("Loaded %d excluded phrases successfully", len(excluded_phrases)) | |||
| except Exception as e: | |||
| logger.critical("Failed to load excluded phrases from '%s': %s", EXCLUDED_PHRASES_FILE, e, exc_info=True) | |||
| return | |||
| try: | |||
| logger.info("Fetching and preparing headlines from %d sources", len(sources)) | |||
| headlines = prepare_headlines(sources, stopwords) | |||
| headlines = prepare_headlines(sources, stopwords, excluded_phrases=excluded_phrases) | |||
| logger.info("Total prepared headlines available for comparison: %d", len(headlines)) | |||
| except Exception as e: | |||
| logger.critical("Failed during headline preparation: %s", e, exc_info=True) | |||
| @@ -0,0 +1,35 @@ | |||
| # Boilerplate link text that should never be treated as a headline. | |||
| # One phrase per line. Matching is case- and punctuation-insensitive, and a | |||
| # phrase matches whenever it appears as a contiguous run of words in a link's | |||
| # text (so "privacy choices" also matches "Your Privacy Choices"). | |||
| # Lines starting with '#' are ignored. | |||
| # Skip / navigation | |||
| skip to content | |||
| skip to main content | |||
| jump to content | |||
| jump to main content | |||
| main navigation | |||
| open menu | |||
| close menu | |||
| main menu | |||
| # Privacy / consent | |||
| your privacy choices | |||
| privacy choices | |||
| privacy policy | |||
| cookie settings | |||
| cookie preferences | |||
| manage cookies | |||
| manage consent | |||
| do not sell my personal information | |||
| do not sell or share my personal information | |||
| do not sell or share my personal data | |||
| # Legal | |||
| terms of service | |||
| terms and conditions | |||
| terms of use | |||
| conditions of use | |||
| accessibility statement | |||
| copyright notice | |||
| @@ -4,7 +4,7 @@ import re | |||
| import requests | |||
| from services.dates import parse_datetime | |||
| from services.normalization import normalize_headline | |||
| from services.normalization import is_excluded, normalize_headline | |||
| from structs.headline import Headline | |||
| logger = logging.getLogger(__name__) | |||
| @@ -86,7 +86,7 @@ def _extract_candidates(source_content): | |||
| return candidates | |||
| def is_headline(text, stopwords=None): | |||
| def is_headline(text, stopwords=None, excluded_phrases=None): | |||
| if not text or not isinstance(text, str): | |||
| return False | |||
| cleaned_text = text.strip() | |||
| @@ -99,6 +99,9 @@ def is_headline(text, stopwords=None): | |||
| if not any(c.isalnum() for c in cleaned_text): | |||
| logger.debug("Text rejected as headline (no alphanumeric characters): %r", cleaned_text) | |||
| return False | |||
| if excluded_phrases and is_excluded(cleaned_text, excluded_phrases): | |||
| logger.debug("Text rejected as non-headline (matches excluded phrase): %r", cleaned_text) | |||
| return False | |||
| if stopwords is not None: | |||
| normalized = normalize_headline(cleaned_text, stopwords) | |||
| if not normalized: | |||
| @@ -107,7 +110,7 @@ def is_headline(text, stopwords=None): | |||
| return True | |||
| def prepare_headlines(sources, stopwords, timeout=DEFAULT_TIMEOUT, headers=None): | |||
| def prepare_headlines(sources, stopwords, timeout=DEFAULT_TIMEOUT, headers=None, excluded_phrases=None): | |||
| request_headers = headers if headers is not None else DEFAULT_HEADERS | |||
| logger.info("Starting preparation of headlines for %d sources", len(sources) if sources else 0) | |||
| headlines = [] | |||
| @@ -153,7 +156,7 @@ def prepare_headlines(sources, stopwords, timeout=DEFAULT_TIMEOUT, headers=None) | |||
| if not cleaned_text: | |||
| logger.debug("Skipping empty tag text at position %d from '%s'", tag_idx, source_url) | |||
| continue | |||
| if not is_headline(cleaned_text, stopwords): | |||
| if not is_headline(cleaned_text, stopwords, excluded_phrases): | |||
| logger.debug("Skipping non-headline tag text at position %d from '%s': %r", tag_idx, source_url, cleaned_text) | |||
| continue | |||
| logger.debug("Processing tag [%d/%d] from '%s': %r", tag_idx, len(candidates), source_url, cleaned_text) | |||
| @@ -24,6 +24,72 @@ def get_stopwords(path): | |||
| raise | |||
| def get_excluded_phrases(path): | |||
| """Load boilerplate link phrases from a file (one per line, '#' comments). | |||
| These are strings that appear in navigation/footer links ("skip to content", | |||
| "your privacy choices", "terms of service") and should never be treated as | |||
| headlines. Returns a list of stripped, non-empty phrases in file order. | |||
| """ | |||
| logger.info("Attempting to load excluded phrases from file: '%s'", path) | |||
| try: | |||
| with open(path, 'r', encoding='utf-8-sig') as phrases_file: | |||
| lines = phrases_file.read().splitlines() | |||
| except FileNotFoundError: | |||
| logger.error("Excluded phrases file not found at path: '%s'", path, exc_info=True) | |||
| raise | |||
| except PermissionError: | |||
| logger.error("Permission denied accessing excluded phrases file: '%s'", path, exc_info=True) | |||
| raise | |||
| except Exception as e: | |||
| logger.error("Failed to load excluded phrases from '%s': %s", path, e, exc_info=True) | |||
| raise | |||
| phrases = [] | |||
| for line in lines: | |||
| stripped = line.strip() | |||
| if not stripped or stripped.startswith('#'): | |||
| continue | |||
| phrases.append(stripped) | |||
| logger.info("Successfully loaded %d excluded phrases from '%s'", len(phrases), path) | |||
| return phrases | |||
| def canonicalize(text): | |||
| """Lowercase, turn punctuation into spaces, and collapse whitespace. | |||
| Produces a comparable form of free text for phrase matching — e.g. | |||
| ``"Terms-of-Service."`` becomes ``"terms of service"``. | |||
| """ | |||
| text = (text or '').lower() | |||
| for char in string.punctuation: | |||
| text = text.replace(char, ' ') | |||
| return ' '.join(text.split()) | |||
| def is_excluded(text, excluded_phrases): | |||
| """Return True if ``text`` contains an excluded phrase as a contiguous | |||
| token subsequence (case-insensitive, punctuation-insensitive). | |||
| Matching is sub-phrase aware, so the entry ``privacy choices`` also matches | |||
| the link text "Your Privacy Choices" without needing a separate exact row. | |||
| """ | |||
| if not excluded_phrases: | |||
| return False | |||
| tokens = canonicalize(text).split() | |||
| if not tokens: | |||
| return False | |||
| for phrase in excluded_phrases: | |||
| phrase_tokens = canonicalize(phrase).split() | |||
| if not phrase_tokens: | |||
| continue | |||
| for start in range(len(tokens) - len(phrase_tokens) + 1): | |||
| if tokens[start:start + len(phrase_tokens)] == phrase_tokens: | |||
| logger.debug("Text excluded by phrase %r: %r", phrase, text) | |||
| return True | |||
| return False | |||
| def remove_stopwords(text, stopwords): | |||
| logger.debug("Removing stopwords from text (%d chars): %r (available stopwords: %d)", len(text), text, len(stopwords)) | |||
| # split() without arguments handles all whitespace (spaces, tabs, newlines) | |||
| @@ -0,0 +1,99 @@ | |||
| import os | |||
| import tempfile | |||
| import unittest | |||
| from unittest.mock import patch, MagicMock | |||
| from services.headlines import is_headline, prepare_headlines | |||
| from services.normalization import canonicalize, get_excluded_phrases, is_excluded | |||
| PHRASES = [ | |||
| "skip to content", | |||
| "privacy choices", | |||
| "terms of service", | |||
| "privacy policy", | |||
| ] | |||
| class TestCanonicalize(unittest.TestCase): | |||
| def test_lowercase_and_collapse(self): | |||
| self.assertEqual(canonicalize(" Terms-of-Service. "), "terms of service") | |||
| def test_empty(self): | |||
| self.assertEqual(canonicalize(""), "") | |||
| self.assertEqual(canonicalize(None), "") | |||
| class TestIsExcluded(unittest.TestCase): | |||
| def test_exact_match(self): | |||
| self.assertTrue(is_excluded("Terms of Service", PHRASES)) | |||
| def test_sub_phrase_match(self): | |||
| # "privacy choices" matches the longer "Your Privacy Choices". | |||
| self.assertTrue(is_excluded("Your Privacy Choices", PHRASES)) | |||
| def test_case_and_punctuation_insensitive(self): | |||
| self.assertTrue(is_excluded("SKIP-TO-CONTENT", PHRASES)) | |||
| def test_no_match(self): | |||
| self.assertFalse(is_excluded("Federal Reserve holds rates steady", PHRASES)) | |||
| def test_no_phrases(self): | |||
| self.assertFalse(is_excluded("anything here now", [])) | |||
| self.assertFalse(is_excluded("anything here now", None)) | |||
| class TestGetExcludedPhrases(unittest.TestCase): | |||
| def test_loads_skipping_comments_and_blanks(self): | |||
| path = None | |||
| try: | |||
| with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as f: | |||
| f.write("# a comment\n\nskip to content\n\nprivacy choices\n# another\n") | |||
| path = f.name | |||
| self.assertEqual(get_excluded_phrases(path), ["skip to content", "privacy choices"]) | |||
| finally: | |||
| if path: | |||
| os.unlink(path) | |||
| class TestIsHeadlineExcluded(unittest.TestCase): | |||
| def setUp(self): | |||
| self.stopwords = {"the", "a", "an", "in", "on", "and", "of", "to"} | |||
| def test_boilerplate_rejected(self): | |||
| for text in ["Your Privacy Choices", "Skip to Content", "Terms of Service"]: | |||
| with self.subTest(text=text): | |||
| self.assertFalse(is_headline(text, self.stopwords, PHRASES)) | |||
| def test_real_headline_kept(self): | |||
| self.assertTrue(is_headline("Senate passes major infrastructure bill", self.stopwords, PHRASES)) | |||
| class TestPrepareHeadlinesExcluded(unittest.TestCase): | |||
| def setUp(self): | |||
| self.stopwords = {"the", "a", "an", "in", "on", "and", "of", "to"} | |||
| @patch("services.headlines.requests.get") | |||
| def test_boilerplate_links_are_filtered_out(self, mock_get): | |||
| html = ( | |||
| "<html><body>" | |||
| "<a href='/skip'>Skip to Content</a>" | |||
| "<a href='/terms'>Terms of Service</a>" | |||
| "<a href='/privacy'>Your Privacy Choices</a>" | |||
| "<a href='/article'>Senate passes major infrastructure bill</a>" | |||
| "</body></html>" | |||
| ) | |||
| mock_response = MagicMock() | |||
| mock_response.status_code = 200 | |||
| mock_response.content = html.encode("utf-8") | |||
| mock_response.text = html | |||
| mock_get.return_value = mock_response | |||
| headlines = prepare_headlines(["https://example.com/news"], self.stopwords, excluded_phrases=PHRASES) | |||
| texts = [h.display_text for h in headlines] | |||
| self.assertEqual(texts, ["Senate passes major infrastructure bill"]) | |||
| if __name__ == "__main__": | |||
| unittest.main() | |||