"""Offline feedparser evaluation plus an explicitly separate editorial example. No network requests, model calls, publishing, scheduler, database or credentials. Requires the two exact packages in the accompanying requirements file. """ import argparse from collections import defaultdict from datetime import datetime, timezone import hashlib import importlib.metadata import json from pathlib import Path import platform import statistics import time from urllib.parse import urlsplit import xml.etree.ElementTree as ET import feedparser EXPECTED_VERSION = '6.0.14' def digest(value): if isinstance(value, str): value = value.encode('utf-8') return hashlib.sha256(value).hexdigest() def parse_bytes(raw): parsed = feedparser.parse(raw) entries = [] for index, entry in enumerate(parsed.entries): fields = {name: entry.get(name) for name in ('id', 'link', 'title', 'summary', 'published', 'updated')} entries.append({**fields, 'index': index, 'guidislink': entry.get('guidislink'), 'published_parsed': list(entry['published_parsed']) if entry.get('published_parsed') else None, 'parsed_fields_sha256': digest(json.dumps(fields, sort_keys=True, ensure_ascii=False))}) exception = parsed.get('bozo_exception') # This is our example review policy, not a feedparser classification feature. hold = bool(parsed.get('bozo')) or not parsed.get('version') or not entries return {'feed_version': parsed.get('version'), 'bozo': bool(parsed.get('bozo')), 'bozo_exception_type': type(exception).__name__ if exception else None, 'entry_count': len(entries), 'entries': entries, 'example_batch_decision': 'hold-for-review' if hold else 'parsed-for-ledger-example'} def editorial_ledger(batches): """Illustrative identity/revision grouping; deliberately not a product feature. Retain every observation. Use feed+ID, then exact link, then a unique observation key. Do not normalize query strings, follow URLs or decide factual truth. """ groups, same_link = defaultdict(list), defaultdict(list) for batch_name, feed_url, parsed in batches: if parsed['example_batch_decision'] == 'hold-for-review': continue for entry in parsed['entries']: if entry['id']: identity = [feed_url, 'feed-id', entry['id']] elif entry['link']: identity = [feed_url, 'exact-link', entry['link']] else: identity = [feed_url, 'unidentified-observation', batch_name, entry['index']] observation = {'batch': batch_name, 'feed_url': feed_url, **entry, 'linked_document_final_url': None, 'linked_document_sha256': None, 'factual_verification': 'NOT_RUN'} groups[json.dumps(identity, ensure_ascii=False)].append(observation) if entry['link']: same_link[entry['link']].append(identity) records = [{'identity': json.loads(key), 'observation_count': len(observations), 'parsed_revision_count': len({e['parsed_fields_sha256'] for e in observations}), 'observations': observations} for key, observations in groups.items()] return {'observation_count': sum(r['observation_count'] for r in records), 'identity_count': len(records), 'records': records, 'same_link_different_identities': {link: len({json.dumps(i) for i in identities}) for link, identities in same_link.items() if len({json.dumps(i) for i in identities}) > 1}} def evaluate(public_raw, fixtures): public_url = 'https://signal.urdream.top/rss.xml' public = parse_bytes(public_raw) cases = {case['name']: parse_bytes(case['xml'].encode('utf-8')) for case in fixtures['cases']} urls = {case['name']: case['feed_url'] for case in fixtures['cases']} duplicate = editorial_ledger([('input-1', public_url, public), ('input-2-same-saved-bytes', public_url, parse_bytes(public_raw))]) revised = editorial_ledger([(name, urls[name], cases[name]) for name in ('revision-before', 'revision-after')]) shared = editorial_ledger([(name, urls[name], cases[name]) for name in ('shared-link-a', 'shared-link-b')]) query = editorial_ledger([('query-distinct', urls['query-distinct'], cases['query-distinct'])]) missing = editorial_ledger([('missing-identity', urls['missing-identity'], cases['missing-identity'])]) checks = [] def check(name, condition): checks.append({'name': name, 'passed': bool(condition)}) check('installed-version-pinned', feedparser.__version__ == EXPECTED_VERSION) check('public-rss20', public['feed_version'] == 'rss20') check('public-no-bozo', not public['bozo']) check('public-count-matches-raw-xml', public['entry_count'] == len(ET.fromstring(public_raw).findall('channel/item'))) check('public-snapshot-has-18-items', public['entry_count'] == 18) check('public-identifiers-all-unique', len({e['id'] for e in public['entries']}) == 18) check('public-links-correct-host', all(urlsplit(e['link']).netloc == 'signal.urdream.top' for e in public['entries'])) check('public-id-and-link-equal-in-this-snapshot', all(e['id'] == e['link'] for e in public['entries'])) check('public-guid-flags-not-permalinks', all(e['guidislink'] is False for e in public['entries'])) check('second-parse-does-not-delete-observations', duplicate['observation_count'] == 36) check('example-ledger-groups-18-identities', duplicate['identity_count'] == 18) check('example-ledger-keeps-two-origins-per-identity', all(r['observation_count'] == 2 for r in duplicate['records'])) check('example-ledger-sees-no-change-on-same-bytes', all(r['parsed_revision_count'] == 1 for r in duplicate['records'])) opaque = cases['opaque-guid']['entries'][0] check('opaque-guid-remains-separate-from-link', opaque['id'] == 'urn:example:record:42' and opaque['link'] == 'https://publisher.example/report') check('same-id-changed-summary-is-one-identity-two-revisions', revised['identity_count'] == 1 and revised['records'][0]['parsed_revision_count'] == 2) check('two-publishers-not-collapsed-by-common-link', shared['identity_count'] == 2 and len(shared['same_link_different_identities']) == 1) check('query-strings-preserved-as-distinct-links', query['identity_count'] == 2 and len({e['link'] for e in cases['query-distinct']['entries']}) == 2) check('unparsed-date-not-silently-assigned-a-time', cases['unparsed-date']['entries'][0]['published'] == 'not a date' and cases['unparsed-date']['entries'][0]['published_parsed'] is None) check('unknown-identities-not-merged-by-title', missing['identity_count'] == 2) check('malformed-input-sets-bozo', cases['truncated-xml']['bozo']) check('malformed-input-held-even-with-recovered-items', cases['truncated-xml']['entry_count'] > 0 and cases['truncated-xml']['example_batch_decision'] == 'hold-for-review') check('ordinary-html-not-mistaken-for-a-feed', cases['not-a-feed']['feed_version'] == '' and cases['not-a-feed']['example_batch_decision'] == 'hold-for-review') return {'public_input': public, 'synthetic_inputs': cases, 'editorial_example': {'duplicate_public_input': duplicate, 'same_id_revision': revised, 'common_link': shared, 'distinct_query': query, 'missing_identity': missing}, 'checks': checks, 'passed': all(c['passed'] for c in checks)} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--public-feed', required=True) parser.add_argument('--fixtures', required=True) parser.add_argument('--output', required=True) parser.add_argument('--expected-results', help='Optional offline comparison; ignores new timing/run metadata') args = parser.parse_args() started = time.perf_counter() raw = Path(args.public_feed).read_bytes() fixture_bytes = Path(args.fixtures).read_bytes() outcomes = evaluate(raw, json.loads(fixture_bytes)) samples = [] for _ in range(25): before = time.perf_counter_ns() feedparser.parse(raw) samples.append((time.perf_counter_ns() - before) / 1000000) expected_match = None if args.expected_results: expected_match = json.loads(Path(args.expected_results).read_text())['outcomes'] == outcomes report = {'schema': 'urdream-rss-ledger-execution-v1', 'status': 'PASS' if outcomes['passed'] and expected_match is not False else 'FAIL', 'run': {'executed_at': datetime.now(timezone.utc).isoformat(), 'platform': {'os': platform.system(), 'os_release': platform.release(), 'architecture': platform.machine(), 'python': platform.python_version()}, 'packages': {name: importlib.metadata.version(name) for name in ('feedparser', 'feedparser-sgmllib')}, 'public_feed_file': Path(args.public_feed).name, 'public_feed_sha256': digest(raw), 'fixture_file': Path(args.fixtures).name, 'fixture_sha256': digest(fixture_bytes), 'replay_core_matches_expected': expected_match, 'cached_parse_timing': {'iterations': 25, 'median_ms': round(statistics.median(samples), 4), 'min_ms': round(min(samples), 4), 'max_ms': round(max(samples), 4)}, 'total_elapsed_ms': round((time.perf_counter() - started) * 1000, 3)}, 'outcomes': outcomes, 'not_run': ['Linked article retrieval or redirects', 'Fact verification or editorial acceptance', 'Live conditional GET/304 behavior', 'Persistent crawler or scheduled ingestion', 'AI/model generation or evaluation', 'Production deployment, VPS load or network-proxy impact'], 'boundary': 'Real saved public RSS input plus explicitly synthetic fixtures. Ledger grouping and hold decisions are our illustrative Python layer, not feedparser product features. Timing is one local cached-byte experiment.'} Path(args.output).write_text(json.dumps(report, indent=2, ensure_ascii=False) + '\n') print(json.dumps({'status': report['status'], 'checks': len(outcomes['checks']), 'passed': sum(c['passed'] for c in outcomes['checks']), 'public_entries': outcomes['public_input']['entry_count'], 'replay_core_matches_expected': expected_match, 'cached_parse_timing': report['run']['cached_parse_timing']})) return 0 if report['status'] == 'PASS' else 1 if __name__ == '__main__': raise SystemExit(main())