{record["body"]}{attachment_html}
' + escape(json.dumps(data, ensure_ascii=False, indent=2)) + '', 'published_at': at, 'url': response.url, 'external_id': identity(response.url), 'raw_response_id': response.meta['raw_response_id'], + 'extra': {'content_scope': 'article', 'published_at_precision': precision, 'published_at_basis': 'official_update', 'published_date': at.date().isoformat(), 'official_meeting': data, 'rule_version': RULE_VERSION, 'attachments': [{'url': official_url(self.name, d['url']), 'title': d.get('name') or d.get('documentType', ''), 'status': 'pending'} for d in {d['url']: d for d in data.get('meetingDocuments', []) + data.get('witnessDocuments', []) if d.get('format') == 'PDF'}.values()]}} + for result in self.attach_or_finish(parsed, lane): + yield result + return + if doc['kind'] == 'bill': + data = response.json()['bill'] + base = response.url.split('?')[0] + state = self.cursor[lane] + previous = self.cursor.setdefault('bill_retries', {}).get(base) + version = data.get('updateDateIncludingText') or data.get('updateDate') + if previous and previous.get('version') != version: + self.cursor['bill_retries'].pop(base) + if previous and previous.get('version') == version and (previous['attempts'] >= 3 or previous['next_at'] > self.run.started_at.timestamp()): + for result in self.complete(lane): + yield result + return + reasons = selection(self.name, data['title'], json.dumps(data.get('policyArea', {}))) + state['bill'] = {'url': base, 'document_url': doc['url'], 'title': data['title'], 'reasons': reasons, 'bodies': 0, 'version': version, 'retry': bool(doc.get('bill_retry') or doc.get('attachment_retry'))} + kinds = ([] if reasons else ['subjects', 'summaries']) + ['texts', 'actions', 'bill_finish'] + state['pending'][0:1] = [{'url': base + '/' + ('text' if kind == 'texts' else kind) + '?format=json&limit=250', 'kind': kind, 'title': data['title'], 'bill': base} for kind in kinds] + for result in self.advance(lane): + yield result + return + if doc['kind'] in ('subjects', 'summaries'): + data = response.json() + key = 'subjects' if doc['kind'] == 'subjects' else 'summaries' + if key not in data: + raise ValueError('官方法案补充资料结构未识别') + state = self.cursor[lane] + state['bill']['reasons'] = state['bill']['reasons'] or selection(self.name, doc['title'], json.dumps(data[key])) + children = [] + if next_url := data.get('pagination', {}).get('next'): + children.append({**doc, 'url': official_url(self.name, next_url)}) + state['pending'][0:1] = children + for result in self.advance(lane): + yield result + return + if doc['kind'] in ('actions', 'texts'): + data = response.json() + children = [] + if doc['kind'] == 'actions': + for action in data['actions']: + key = identity(doc['bill'] + json.dumps(action, sort_keys=True)) + at, precision = published(action['actionDate']) + begin = datetime(2025, 1, 1, tzinfo=UTC) if self.cursor[lane]['bill'].get('retry') else (self.run.history_target if lane == 'history' else self.run.cutoff) + if at < begin: + continue + item = self.item(response, external_id=key, url=congress_bill_url(doc['bill']), title=doc['title'], summary='', content_html=f"
{escape(action['text'])}
", published_at=at, + extra={'content_scope': 'official_action', 'published_at_precision': precision, 'published_date': at.date().isoformat(), 'bill': doc['bill'], 'official_action': action, 'rule_version': RULE_VERSION}) + yield item + else: + for version in data['textVersions']: + if not version.get('date'): + self.cursor[lane]['bill']['missing_material'] = True + continue + formats = version.get('formats', []) + chosen = next((f for f in formats if f['type'] == 'Formatted Text'), None) or next((f for f in formats if f['type'] == 'Formatted XML'), None) or next((f for f in formats if f['type'] == 'PDF'), None) + if not chosen: + self.cursor[lane]['bill']['missing_material'] = True + if chosen: + children.append({'url': chosen['url'], 'kind': 'text', 'title': doc['title'] + ' — ' + version['type'], 'at': version['date'], 'bill': doc['bill']}) + next_url = data.get('pagination', {}).get('next') + if next_url: + children.append({**doc, 'url': official_url(self.name, next_url)}) + self.cursor[lane]['pending'][0:1] = children + for result in self.advance(lane): + yield result + return + if doc['kind'] == 'register': + data = response.json() + url = data.get('body_html_url') or data.get('raw_text_url') + if not url: + raise ValueError('规则文档缺少正文地址') + self.cursor[lane]['pending'][0] = {'url': official_url(self.name, url), 'kind': 'text', 'title': data['title'], 'at': data['publication_date'], 'canonical': data['html_url'], 'document_number': data['document_number'], 'effective_on': data.get('effective_on'), 'pdf': data.get('pdf_url')} + for result in self.advance(lane): + yield result + return + if doc['kind'] == 'doj': + data = response.json() + records = data['results'] + if data['metadata']['responseInfo']['status'] != 200 or len(records) != 1: + raise ValueError('DOJ API 正文结构未识别') + record = records[0] + if str(uuid.UUID(record['uuid'])) != doc['url'].rsplit('/', 1)[1].removesuffix('.json'): + raise ValueError('DOJ API 正文 UUID 不符') + canonical = official_url(self.name, record['url'].replace('https://justice.gov/', 'https://www.justice.gov/', 1)) + title = unescape(record['title']).strip() + if not title: + raise ValueError('DOJ 正文缺少标题') + # The API release date is a calendar date, not a release instant. + date = datetime.fromtimestamp(int(record['date']), UTC).date().isoformat() + attachment_html = record.get('attachment') or '' + api_attachments = [] + if isinstance(attachment_html, list): + for attachment in attachment_html: + file_id = attachment['file']['id'] + if not str(file_id).isdigit(): + raise ValueError('DOJ API 附件编号无效') + api_attachments.append({'url': f'https://www.justice.gov/media/{file_id}/dl', 'title': attachment.get('description') or 'Official attachment', 'status': 'pending'}) + attachment_html = '' + elif not isinstance(attachment_html, str): + raise ValueError('DOJ API 附件结构未识别') + html = f'' + escape(pdf_text(response.body)) + '' + else: + content = response.text + validate_body(content) + if '<' not in content: + content = '
' + escape(content) + '' + parsed = {'title': doc['title'], 'content_html': content, 'published_at': at, 'extra': {'content_scope': 'article', 'language': 'en', 'published_at_precision': precision, 'published_date': at.date().isoformat(), 'rule_version': RULE_VERSION, 'official_document': doc, 'attachments': []}} + if doc.get('pdf'): + parsed['extra']['attachments'].append({'url': official_url(self.name, doc['pdf']), 'title': 'Official PDF', 'status': 'pending'}) + else: + parsed = article(self.name, response.text, response.url) + if doc.get('bill'): + context = self.cursor[lane]['bill'] + context['bodies'] += 1 + if not context['reasons']: + context['reasons'] = selection(self.name, context['title'], html_to_text(parsed['content_html'])) + if context['reasons']: + # A later version can establish relevance for earlier versions. + self.cursor[lane]['pending'][1:1] = context.pop('unmatched_texts', []) + else: + context.setdefault('unmatched_texts', []).append(copy.deepcopy(doc)) + parsed['extra']['selection_reasons'] = context['reasons'] + begin = self.run.history_target if lane == 'history' else self.run.cutoff + if doc.get('attachment_retry') or (doc.get('bill') and self.cursor[lane]['bill'].get('retry')): + begin = datetime(2025, 1, 1, tzinfo=UTC) + if parsed['published_at'] < begin: + self.run.stats['outside_range'] = self.run.stats.get('outside_range', 0) + 1 + for result in self.complete(lane): + yield result + return + if lane == 'history': + self.run.history_position = min(self.run.history_position or self.run.started_at, parsed['published_at']) + parsed['url'] = doc.get('canonical') or response.url + parsed['external_id'] = identity(parsed['url']) + if self.name == 'fed': + # Preserve pre-existing RSS GUID identities during full-text enrichment. + async with SessionLocal() as session: + existing = await session.scalar(select(NewsItem.external_id).where(NewsItem.source_code == 'fed', NewsItem.url == parsed['url']).limit(1)) + if existing: + parsed['external_id'] = existing + parsed['raw_response_id'] = response.meta['raw_response_id'] + parsed['retry_document'] = {'url': doc['bill'], 'kind': 'bill'} if doc.get('bill') else doc + for result in self.attach_or_finish(parsed, lane): + yield result + + def finish_bill(self, lane): + context = self.cursor[lane].pop('bill') + retries = self.cursor.setdefault('bill_retries', {}) + if not context['bodies'] or context.get('missing_material'): + previous = retries.get(context['url'], {}) + retries[context['url']] = {'document': {'url': context['url'], 'kind': 'bill'}, 'version': context['version'], 'attempts': previous.get('attempts', 0) + 1, 'next_at': (self.run.started_at + timedelta(hours=1)).timestamp()} + else: + retries.pop(context['url'], None) + if not context['reasons']: + self.run.stats['excluded'] = self.run.stats.get('excluded', 0) + 1 + self.run.stats['unresolved_bills'] = len(retries) + self.cursor[lane]['pending'].pop(0) + if lane == 'realtime': + self.cursor.setdefault('checked', {})[context['document_url']] = self.run.started_at.timestamp() + + def attach_or_finish(self, parsed, lane): + pending = next((a for a in parsed['extra']['attachments'] if a['status'] == 'pending'), None) + if pending: + request = self.request_for(pending['url'], self.parse_attachment, lane, parsed=parsed) + request.meta['handle_httpstatus_list'] = [404, 410] + yield request + return + text = html_to_text(parsed['content_html']) + reasons = selection(self.name, parsed['title'], text) or parsed['extra'].get('selection_reasons', []) + incomplete = any(a['status'] != 'complete' for a in parsed['extra']['attachments']) + retry_doc = parsed.pop('retry_document', None) + retries = self.cursor.setdefault('attachment_retries', {}) + if incomplete and retry_doc: + previous = retries.get(parsed['url'], {}) + retries[parsed['url']] = {'document': retry_doc, 'attempts': previous.get('attempts', 0) + 1, 'next_at': (self.run.started_at + timedelta(hours=1)).timestamp()} + elif not incomplete: + retries.pop(parsed['url'], None) + if reasons or incomplete: + parsed['extra']['selection_reasons'] = reasons + parsed['extra']['content_scope'] = 'partial' if incomplete else 'article' + self.run.stats['partial'] = self.run.stats.get('partial', 0) + int(incomplete) + raw_id = parsed.pop('raw_response_id') + item = self.item(Response(parsed['url'], request=Request(parsed['url'], meta={'raw_response_id': raw_id})), summary='', **parsed) + yield item + else: + self.run.stats['excluded'] = self.run.stats.get('excluded', 0) + 1 + yield from self.complete(lane) + + def parse_attachment(self, response): + parsed = response.meta['parsed'] + attachment = next(a for a in parsed['extra']['attachments'] if a['url'] == response.url or a['status'] == 'pending') + attachment['raw_response_id'] = response.meta['raw_response_id'] + try: + if response.status != 200: + raise ValueError(f'HTTP {response.status}') + text = pdf_text(response.body) + attachment['status'] = 'complete' + parsed['content_html'] += f"
{escape(text)}The law establishes regulatory requirements for stablecoins and digital assets.
') + if path == '/v3/committee-meeting/119': + data = {'committeeMeetings': [{'url': 'https://api.congress.gov/v3/committee-meeting/119/house/1'}]} + elif '/committee-meeting/' in path: + data = {'committeeMeeting': {'title': 'Digital asset regulation', 'updateDate': '2025-01-01T12:00:00Z'}} + elif path.endswith('/text'): + data = {'textVersions': [{'date': '2025-01-01', 'type': 'Introduced', 'formats': [{'type': 'Formatted Text', 'url': 'https://www.congress.gov' + path + '.html'}]}]} + elif path.endswith('/actions'): + data = {'actions': [{'actionDate': '2025-01-01', 'text': 'Passed'}]} + else: + data = {'bill': {'title': 'Digital asset regulation', 'updateDate': '2025-01-01'}} + return httpx.Response(200, json=data) + with httpx.Client(transport=httpx.MockTransport(handle)) as client: + capture = module['Capture'](tmp_path / 'cache.gz', 'test', client=client) + report = asyncio.run(module['run'](capture)) + assert report['accepted'] is True + assert len(report['cases']) == 4 + assert sum(case['items'] for case in report['cases']) == 7 + resumed = module['Capture'](capture.path, 'test', client=client) + assert asyncio.run(module['run'](resumed))['accepted'] is True + assert resumed.requests == 0 diff --git a/backend/tests/test_news_sources.py b/backend/tests/test_news_sources.py index 4d84cff..1395ce1 100644 --- a/backend/tests/test_news_sources.py +++ b/backend/tests/test_news_sources.py @@ -33,7 +33,7 @@ ITEM = """Virtual currency exchange sanctions.
', 'published_at': datetime(2025, 1, 1, tzinfo=UTC), 'raw_response_id': 12, 'retry_document': doc, + 'extra': {'attachments': [{'url': 'https://ofac.treasury.gov/a.pdf', 'title': 'License', 'status': 'unavailable'}]}} + item, = obj.attach_or_finish(parsed, 'history') + assert item['extra']['content_scope'] == 'partial' + assert obj.cursor['attachment_retries'][url]['attempts'] == 1 + assert obj.cursor['attachment_retries'][url]['document'] == doc + + +@pytest.mark.parametrize('canceled,failed', [(False, False), (True, False), (False, True)]) +def test_checkpoint_does_not_advance_failed_or_canceled_history(trading, monkeypatch, canceled, failed): + from alphapulse.crawler.spiders import base + now = datetime.now(UTC) + target = now - timedelta(days=300) + source = SimpleNamespace(history_target=None if canceled else target, history_cursor={'history': {'entry': 0}}, consecutive_failures=0) + class Session: + async def __aenter__(self): + return self + async def __aexit__(self, *args): + pass + async def execute(self, *args): + pass + async def get(self, *args, **kwargs): + return source + async def commit(self): + pass + monkeypatch.setattr(base, 'SessionLocal', Session) + obj = spider(trading) + run = base.RunState(uuid.uuid4(), 'schedule', now, target, history_target=target, history_position=now, official_cursor={'history': {'entry': 5}, 'realtime': {'entry': 2}}) + if failed: + run.errors.append('write failed') + asyncio.run(obj._finish_run(run)) + assert source.history_cursor['history']['entry'] == (0 if canceled or failed else 5) + + +def test_congress_action_links_use_public_bill_type(trading): + from alphapulse.crawler.spiders.us_official import congress_bill_url + assert congress_bill_url('https://api.congress.gov/v3/bill/119/hr/1') == 'https://www.congress.gov/bill/119th-congress/house-bill/1' + assert congress_bill_url('https://api.congress.gov/v3/bill/119/s/1582') == 'https://www.congress.gov/bill/119th-congress/senate-bill/1582' + + +def test_tick_during_active_round_does_not_replace_its_cursors(trading): + obj = spider(trading) + obj.crawler = SimpleNamespace(engine=SimpleNamespace(spider_is_idle=lambda: False)) + before = copy.deepcopy(obj.cursor) + assert obj.make_request_from_data(json.dumps({'trigger': 'schedule', 'official_cursor': {'history': {'entry': 99}}}).encode()) == [] + assert obj.cursor == before + + +def test_realtime_keeps_next_page(trading): + obj = spider(trading) + url = SOURCES['ofac'].entries[0] + list(obj.parse_listing(response(url, 'Enable JavaScript and cookies to continue.
']) +def test_text_challenge_never_emits_or_advances(trading, body): + obj = spider(trading, 'federal_register') + doc = {'url': 'https://www.federalregister.gov/body', 'kind': 'text', 'title': 'Digital asset regulation', 'at': '2026-11-04'} + obj.cursor['history']['pending'] = [doc] + async def collect(): + return [v async for v in obj.parse_document(response(doc['url'], body, lane='history', document=doc))] + with pytest.raises(ValueError): + asyncio.run(collect()) + assert obj.cursor['history']['pending'] == [doc] + + +def test_genius_bill_requires_evidence_before_exclusion(trading, monkeypatch): + from alphapulse.crawler.spiders import us_official + monkeypatch.setattr(us_official, 'get_settings', lambda: SimpleNamespace(congress_api_key='test', crawler_contact_email='test@example.invalid')) + obj = spider(trading, 'congress') + doc = {'url': 'https://api.congress.gov/v3/bill/119/s/1582', 'kind': 'bill'} + obj.cursor['history']['pending'] = [doc] + async def collect(): + return [v async for v in obj.parse_document(response(doc['url'], json.dumps({'bill': {'title': 'GENIUS Act', 'policyArea': {'name': 'Finance and Financial Sector'}}}), lane='history', document=doc))] + req, = asyncio.run(collect()) + assert '/subjects?' in req.url + assert not obj.run.stats.get('excluded') + + +def test_real_titles_are_article_headlines(): + import gzip + from pathlib import Path + samples = json.load(gzip.open(Path(__file__).parent / 'fixtures/us_news/public_samples.json.gz', 'rt')) + for s in samples: + if s['source'] == 'fed' and '2026-press.htm' in s['url']: + continue + title = article(s['source'], s['body'], s['url'])['title'] + assert title not in ('Board of Governors of the Federal Reserve System', 'U.S. Department of the Treasury') + assert not title.startswith('Release Number ') + + +@pytest.mark.parametrize('enabled,history_enabled,save_failed', [(True, True, False), (False, True, False), (True, False, False), (True, True, True)]) +def test_round_waits_for_commit_and_reads_fresh_state(trading, monkeypatch, enabled, history_enabled, save_failed): + from alphapulse.crawler.spiders import us_official + fresh = {'history': {'entry': 0, 'target': '2025-01-01T00:00:00+00:00', 'pending': [{'url': 'https://ofac.treasury.gov/recent-actions/20250101', 'kind': 'article'}]}} + source = SimpleNamespace(enabled=enabled, paused_until=None, history_cursor=fresh, history_target=datetime(2025, 1, 1, tzinfo=UTC) if history_enabled else None, history_done=False, history_position=None, backfill_hours=72) + reads = [] + class Session: + async def __aenter__(self): return self + async def __aexit__(self, *args): pass + async def get(self, *args): + reads.append(True) + return source + monkeypatch.setattr(us_official, 'SessionLocal', Session) + async def run(): + obj = spider(trading) + obj.run = None + obj.crawler = SimpleNamespace(engine=SimpleNamespace(crawl=reads.append)) + obj._insert_run = lambda run: asyncio.sleep(0) + saved = asyncio.Event() + async def finish(): + await saved.wait() + if save_failed: raise RuntimeError('test write failure') + obj.finish_task = asyncio.create_task(finish()) + boot = asyncio.create_task(obj.start_round('schedule')) + await asyncio.sleep(0) + assert not reads + saved.set() + await boot + requests = [r for r in reads if isinstance(r, Request)] + if enabled and not save_failed: + assert bool([r for r in requests if r.meta['lane'] == 'history']) is history_enabled + if history_enabled: + assert requests[-1].url.endswith('20250101') + else: + assert not requests and obj.run is None + await asyncio.gather(*obj._tasks) + asyncio.run(run()) + + +def test_register_realtime_finishes_all_pages_before_changing_day(trading): + obj = spider(trading, 'federal_register') + obj.cursor['realtime']['day'] = '2026-11-04' + url = SOURCES[obj.name].entries[0] + list(obj.parse_listing(response(url, json.dumps({'count': 1001, 'results': [{'document_number': '2026-1234'}], 'next_page_url': url + '?page=2'}), lane='realtime'))) + assert obj.cursor['realtime']['day'] == '2026-11-04' + obj.cursor['realtime']['pending'] = [] + req, = obj.advance('realtime') + assert req.url.endswith('?page=2') + list(obj.parse_listing(response(req.url, json.dumps({'count': 1001, 'results': [{'document_number': '2026-1235'}]}), lane='realtime'))) + assert obj.cursor['realtime']['day'] == '2026-11-04' + obj.cursor['realtime']['pending'] = [] + obj.advance('realtime') + assert obj.cursor['realtime']['day'] == '2026-11-03' + + +def test_pagination_cycle_does_not_claim_success(trading): + obj = spider(trading) + url = SOURCES[obj.name].entries[0] + obj.cursor['history']['visited_pages'] = [url] + with pytest.raises(ValueError, match='循环'): + list(obj.parse_listing(response(url + '?page=2', f'This digital asset regulation establishes requirements for stablecoin issuers.
', lane='history', document=doc))] + item, = asyncio.run(collect()) + assert item['extra']['content_scope'] == 'article' + + +def test_bill_full_text_can_supply_relevance_and_resume(trading, monkeypatch): + from alphapulse.crawler.spiders import us_official + monkeypatch.setattr(us_official, 'get_settings', lambda: SimpleNamespace(congress_api_key='test', crawler_contact_email='test@example.invalid')) + obj = spider(trading, 'congress') + base = 'https://api.congress.gov/v3/bill/119/s/1582' + obj.cursor['history']['pending'] = [{'url': base, 'kind': 'bill'}] + async def parse(body): + doc = obj.cursor['history']['pending'][0] + return [v async for v in obj.parse_document(response(doc['url'], body, lane='history', document=doc))] + asyncio.run(parse(json.dumps({'bill': {'title': 'GENIUS Act', 'updateDate': '2026-11-04'}}))) + asyncio.run(parse(json.dumps({'subjects': {'legislativeSubjects': []}}))) + asyncio.run(parse(json.dumps({'summaries': []}))) + asyncio.run(parse(json.dumps({'textVersions': [{'type': 'Enrolled', 'date': '2026-11-04', 'formats': [{'type': 'Formatted Text', 'url': 'https://www.congress.gov/text'}]}]}))) + obj.cursor = json.loads(json.dumps(obj.cursor)) + output = asyncio.run(parse('The bill establishes the regulatory framework for payment stablecoins and digital assets.
')) + assert output[0]['title'] == 'GENIUS Act — Enrolled' + assert '/actions?' in output[1].url + output = asyncio.run(parse(json.dumps({'actions': [{'actionDate': '2026-11-04', 'text': 'Became law'}]}))) + assert output[0]['extra']['content_scope'] == 'official_action' + assert not obj.cursor.get('bill_retries') + + +def test_missing_bill_material_is_a_retryable_gap(trading, monkeypatch): + from alphapulse.crawler.spiders import us_official + monkeypatch.setattr(us_official, 'get_settings', lambda: SimpleNamespace(congress_api_key='test', crawler_contact_email='test@example.invalid')) + obj = spider(trading, 'congress') + base = 'https://api.congress.gov/v3/bill/119/s/1582' + obj.cursor['history']['pending'] = [{'url': base, 'kind': 'bill'}] + async def parse(data): + doc = obj.cursor['history']['pending'][0] + return [v async for v in obj.parse_document(response(doc['url'], json.dumps(data), lane='history', document=doc))] + asyncio.run(parse({'bill': {'title': 'GENIUS Act'}})) + asyncio.run(parse({'subjects': {}})) + asyncio.run(parse({'summaries': []})) + asyncio.run(parse({'textVersions': []})) + assert obj.cursor['bill_retries'][base]['attempts'] == 1 + assert obj.run.stats['unresolved_bills'] == 1 + assert not obj.run.stats.get('excluded') + + +@pytest.mark.parametrize('attempts,version,expected_kind', [(3, 'v1', 'listing'), (1, 'v1', 'listing'), (3, 'v2', 'subjects')]) +def test_bill_retries_are_bounded_until_official_update(trading, monkeypatch, attempts, version, expected_kind): + from alphapulse.crawler.spiders import us_official + monkeypatch.setattr(us_official, 'get_settings', lambda: SimpleNamespace(congress_api_key='test', crawler_contact_email='test@example.invalid')) + obj = spider(trading, 'congress') + url = 'https://api.congress.gov/v3/bill/119/s/1582' + doc = {'url': url, 'kind': 'bill'} + obj.cursor['history']['pending'] = [doc] + obj.cursor['bill_retries'] = {url: {'version': 'v1', 'attempts': attempts, 'next_at': (obj.run.started_at + timedelta(minutes=30)).timestamp()}} + async def collect(): + return [v async for v in obj.parse_document(response(url, json.dumps({'bill': {'title': 'GENIUS Act', 'updateDate': version}}), lane='history', document=doc))] + req, = asyncio.run(collect()) + assert ('/subjects?' in req.url) is (expected_kind == 'subjects') + + +def test_fed_archive_is_not_discovered_as_article(trading): + obj = spider(trading, 'fed') + url = SOURCES['fed'].entries[0] + req, = obj.parse_listing(response(url, 'The requested official document could not be located.
', 'This legislation establishes new administrative requirements for federal agencies.
')) + assert len(output) == 1 and isinstance(output[0], Request) + output = asyncio.run(parse(second, 'The legislation creates a regulatory framework for stablecoins and digital assets.
')) + assert output[0]['extra']['selection_reasons'] + assert output[1].url == first['url'] + obj.run.history_deadline = time.monotonic() - 1 + output = asyncio.run(parse(first, 'This legislation establishes new administrative requirements for federal agencies.
')) + assert output[0]['title'] == 'GENIUS Act' + + +def test_bill_retry_keeps_historical_text_outside_realtime_window(trading, monkeypatch): + from alphapulse.crawler.spiders import us_official + monkeypatch.setattr(us_official, 'get_settings', lambda: SimpleNamespace(congress_api_key='test', crawler_contact_email='test@example.invalid')) + obj = spider(trading, 'congress') + base = 'https://api.congress.gov/v3/bill/119/s/1582' + doc = {'url': 'https://www.congress.gov/old', 'kind': 'text', 'title': 'GENIUS Act', 'at': '2025-01-01', 'bill': base} + obj.cursor['realtime'].update(bill={'url': base, 'title': 'GENIUS Act', 'reasons': ['crypto_policy'], 'bodies': 0, 'retry': True}, pending=[doc]) + async def collect(): + return [v async for v in obj.parse_document(response(doc['url'], 'The law establishes requirements for stablecoin issuers and digital assets.
', lane='realtime', document=doc))] + assert asyncio.run(collect())[0]['published_at'].year == 2025 + + +def test_failed_checkpoint_save_recovers_on_later_trigger(trading, monkeypatch): + from alphapulse.crawler.spiders import us_official + source = SimpleNamespace(enabled=True, paused_until=None, history_cursor={}, history_target=None, history_done=False, backfill_hours=72) + class Session: + async def __aenter__(self): return self + async def __aexit__(self, *args): pass + async def get(self, *args): return source + monkeypatch.setattr(us_official, 'SessionLocal', Session) + async def check(): + obj = spider(trading) + requests, attempts = [], [] + obj.crawler = SimpleNamespace(engine=SimpleNamespace(crawl=requests.append)) + obj._insert_run = lambda run: asyncio.sleep(0) + original = obj.run + async def finish(run): + attempts.append(run) + if len(attempts) == 1: + raise RuntimeError('temporary checkpoint failure') + obj._finish_run = finish + obj._close_run() + await asyncio.gather(obj.finish_task, return_exceptions=True) + await obj.start_round('schedule') + assert attempts == [original, original] + assert requests and obj.run is not None + await asyncio.gather(*obj._tasks) + asyncio.run(check()) + + +def test_queued_tick_obeys_newly_saved_backoff(trading, monkeypatch): + from alphapulse.crawler.spiders import us_official + source = SimpleNamespace(enabled=True, paused_until=datetime.now(UTC) + timedelta(minutes=5), history_cursor={}, history_target=None, history_done=False, backfill_hours=72) + class Session: + async def __aenter__(self): return self + async def __aexit__(self, *args): pass + async def get(self, *args): return source + monkeypatch.setattr(us_official, 'SessionLocal', Session) + async def check(): + obj = spider(trading) + obj.run = None + requests = [] + obj.crawler = SimpleNamespace(engine=SimpleNamespace(crawl=requests.append)) + obj._insert_run = lambda run: asyncio.sleep(0) + await obj.start_round('schedule') + assert not requests and obj.run is None + await asyncio.gather(*obj._tasks) + asyncio.run(check()) + + +def test_checkpoint_retry_budget_preserves_original_round(trading, monkeypatch): + from alphapulse.crawler.spiders import us_official + monkeypatch.setattr(us_official, 'SessionLocal', lambda: pytest.fail('Uncommitted checkpoint must block source reads')) + async def check(): + obj = spider(trading) + original, attempts = obj.run, [] + async def fail(run): + attempts.append(run) + raise RuntimeError('persistent checkpoint failure') + obj._finish_run = fail + obj._close_run() + await asyncio.gather(obj.finish_task, return_exceptions=True) + for _ in range(5): + await obj.start_round('schedule') + assert attempts == [original] * 3 + assert obj.run is None and not obj._starting + asyncio.run(check()) + + +@pytest.mark.parametrize('failure', ['', 'count', 'date', 'malformed']) +def test_register_xml_validates_and_resumes(trading, failure): + obj = spider(trading, 'federal_register') + doc = {'url': 'https://www.govinfo.gov/issue.xml', 'kind': 'register_issue', 'day': '2025-01-01', 'count': 2001} + obj.cursor['history'].update(day=doc['day'], pending=[doc]) + xml = 'The cryptocurrency exchange pleaded guilty to violations of financial regulations.
' + record = {'uuid': identifier, 'url': 'https://justice.gov/opa/pr/test', 'date': '1735732800', 'title': 'Exchange & settlement', 'body': body, 'changed': ''} + async def collect(): + return [v async for v in obj.parse_document(response(doc['url'], json.dumps({'metadata': {'responseInfo': {'status': 200}}, 'results': [record]}), lane='history', document=doc))] + if bad_body: + with pytest.raises(ValueError): asyncio.run(collect()) + assert obj.cursor['history']['pending'] == [doc] + else: + item, = asyncio.run(collect()) + assert item['url'] == 'https://www.justice.gov/opa/pr/test' + assert item['title'] == 'Exchange & settlement' + assert item['extra']['published_at_precision'] == 'date' + assert item['source_updated_at'] == datetime(2025,1,2,12,tzinfo=UTC) + + +def test_doj_old_discovery_cursor_is_rebuilt(trading): + obj = spider(trading, 'doj') + obj.cursor.update(checked={'old': 1}, history={'entry': 2}, attachment_retries={}) + obj.first_requests() + assert obj.cursor['transport'] == 'doj-api-v1' + assert 'history' not in obj.cursor and obj.cursor['realtime']['entry'] == 0 + assert obj.run.history_target == datetime(2025,1,1,tzinfo=UTC) + + +def test_bill_undated_text_is_visible_retry_gap(trading): + obj = spider(trading, 'congress') + obj.run.history_deadline = time.monotonic() - 1 + base = 'https://api.congress.gov/v3/bill/119/s/1582' + doc = {'url': base + '/text', 'kind': 'texts', 'title': 'GENIUS Act', 'bill': base} + finish = {'url': base, 'kind': 'bill_finish'} + obj.cursor['history'].update(bill={'url': base, 'document_url': base, 'title': 'GENIUS Act', 'version': 'v1', 'reasons': ['crypto_policy'], 'bodies': 1}, pending=[doc, finish]) + async def collect(): + return [v async for v in obj.parse_document(response(doc['url'], json.dumps({'textVersions': [{'date': None, 'formats': []}]}), lane='history', document=doc))] + asyncio.run(collect()) + obj.finish_bill('history') + assert obj.cursor['bill_retries'][base]['attempts'] == 1 + assert obj.run.stats['unresolved_bills'] == 1 + + +def test_public_api_and_pdf_fixtures_replay_without_io(trading): + import runpy + from pathlib import Path + script = runpy.run_path(str(Path(__file__).resolve().parents[2] / 'scripts/smoke_us_news.py')) + items = asyncio.run(script['public_api_items']()) + assert len(items) == 3 + assert all(item['extra']['content_scope'] == 'article' for item in items) + + + +def test_congress_official_xml_crosscheck(trading): + import runpy + from pathlib import Path + script = runpy.run_path(str(Path(__file__).resolve().parents[2] / 'scripts/smoke_us_news.py')) + assert len(asyncio.run(script['congress_reference_items']())) == 3 + + +def test_congress_meeting_includes_deduplicated_witness_pdfs(trading, monkeypatch): + from alphapulse.crawler.spiders import us_official + monkeypatch.setattr(us_official, 'get_settings', lambda: SimpleNamespace(congress_api_key='test', crawler_contact_email='test@example.invalid')) + obj = spider(trading, 'congress') + doc = {'url': 'https://api.congress.gov/v3/committee-meeting/119/house/1', 'kind': 'meeting'} + obj.cursor['history']['pending'] = [doc] + pdf = {'url': 'https://www.congress.gov/witness.pdf', 'format': 'PDF', 'documentType': 'Witness Statement'} + data = {'committeeMeeting': {'title': 'Digital asset regulation', 'updateDate': '2025-01-01T12:00:00Z', 'meetingDocuments': [pdf], 'witnessDocuments': [pdf]}} + async def collect(): + return [v async for v in obj.parse_document(response(doc['url'], json.dumps(data), document=doc, lane='history'))] + req, = asyncio.run(collect()) + assert req.url == pdf['url'] + assert len(req.meta['parsed']['extra']['attachments']) == 1 + assert req.meta['parsed']['extra']['attachments'][0]['title'] == 'Witness Statement' + + +def test_dated_but_unavailable_bill_text_remains_a_gap(trading): + obj = spider(trading, 'congress') + obj.run.history_deadline = time.monotonic() - 1 + base = 'https://api.congress.gov/v3/bill/119/s/1582' + doc = {'url': base + '/text', 'kind': 'texts', 'title': 'GENIUS Act', 'bill': base} + obj.cursor['history'].update(bill={'url': base, 'document_url': base, 'title': 'GENIUS Act', 'version': 'v1', 'reasons': ['crypto_policy'], 'bodies': 1}, pending=[doc, {'url': base, 'kind': 'bill_finish'}]) + async def collect(): + return [v async for v in obj.parse_document(response(doc['url'], json.dumps({'textVersions': [{'date': '2025-01-01', 'formats': []}]}), lane='history', document=doc))] + asyncio.run(collect()) + obj.finish_bill('history') + assert obj.run.stats['unresolved_bills'] == 1 + + + +def test_real_congress_api_fixture_replays_with_visible_source_gap(trading, tmp_path): + import runpy + import shutil + from pathlib import Path + scripts = tmp_path / 'scripts' + scripts.mkdir() + original = Path(__file__).resolve().parents[2] / 'scripts' + for name in ('smoke_us_news.py', 'capture_congress.py'): + shutil.copyfile(original / name, scripts / name) + namespace = runpy.run_path(str(scripts / 'smoke_us_news.py')) + items = asyncio.run(namespace['congress_api_items']()) + assert len(items) == 100 + meetings = [item for item in items if item['extra'].get('official_meeting')] + assert len(meetings) == 1 + assert meetings[0]['extra']['content_scope'] == 'partial' + attachments = meetings[0]['extra']['attachments'] + assert len(attachments) == 15 + assert len({a['url'] for a in attachments}) == 15 + assert sum(a['status'] == 'complete' for a in attachments) == 14 + assert sum(a['status'] == 'unavailable' for a in attachments) == 1 + assert any(item['extra']['content_scope'] == 'official_action' for item in items) diff --git a/deploy/.env.example b/deploy/.env.example index d540e7d..5bd2004 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -10,3 +10,7 @@ ADMIN_PASSWORD_HASH='Enable JavaScript and cookies to continue.
', encoding='utf-8', request=request) + try: + async for item in obj.parse_document(invalid): + await pipeline.process_item(item) + except ValueError: + pass + else: + raise AssertionError('Challenge must fail before the pipeline') + async with SessionLocal() as session: + news = await session.scalar(select(NewsItem).where(NewsItem.external_id == 'legacy-rss-guid')) + assert (news.content_hash, news.revision_no) == baseline + assert await session.scalar(select(func.count()).select_from(NewsRevision)) == 51 + # A failed checkpoint commit must recover before a new round can read it. + async with SessionLocal() as session: + source = await session.get(Source, 'ofac') + source.enabled = True + await session.commit() + obj = USOfficialSpider(name='ofac') + obj.run = RunState(uuid.uuid4(), 'schedule', now, now-timedelta(hours=72), official_cursor={'checkpoint_probe': 'committed'}) + obj.crawler = SimpleNamespace(engine=SimpleNamespace(crawl=lambda request: None)) + finish = obj._finish_run + attempts = [] + async def transient_failure(run): + attempts.append(run.id) + if len(attempts) == 1: + raise RuntimeError('isolated checkpoint failure') + await finish(run) + obj._finish_run = transient_failure + original_id = obj.run.id + obj._close_run() + await asyncio.gather(obj.finish_task, return_exceptions=True) + await obj.start_round('schedule') + await asyncio.gather(*obj._tasks) + assert attempts == [original_id, original_id] and obj.run is not None + assert obj.cursor['checkpoint_probe'] == 'committed' + async with SessionLocal() as session: + saved = await session.get(CrawlRun, original_id) + assert saved.status == 'success' + assert (await session.get(Source, 'ofac')).history_cursor['checkpoint_probe'] == 'committed' + # API/PDF packages use the real parser and database revision pipeline too. + api_items = await public_api_items() + await congress_reference_items() + await congress_api_items() + with patch.object(pipelines, 'redis', SimpleNamespace(publish=AsyncMock())): + for item in api_items: + await pipeline.process_item(dict(item)) + async with SessionLocal() as session: + baseline_count = await session.scalar(select(func.count()).select_from(NewsRevision)) + for item in api_items: + await pipeline.process_item(dict(item)) + async with SessionLocal() as session: + assert await session.scalar(select(func.count()).select_from(NewsItem)) == 50 + len(api_items) + assert await session.scalar(select(func.count()).select_from(NewsRevision)) == baseline_count + revised = copy.deepcopy(api_items[0]) + revised['title'] += ' Updated' + with patch.object(pipelines, 'redis', SimpleNamespace(publish=AsyncMock())): + await pipeline.process_item(dict(revised)) + async with SessionLocal() as session: + assert await session.scalar(select(func.count()).select_from(NewsRevision)) == baseline_count + 1 + await engine.dispose() + print('US source migration, HTML, API/PDF, GovInfo and Congress packages, API/PDF replay and revision, negative screening, daily XML, RSS identity, challenge rejection and checkpoint recovery passed') + + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/web/src/common/apis/crawler/type.ts b/web/src/common/apis/crawler/type.ts index 61dd9bb..96e4181 100644 --- a/web/src/common/apis/crawler/type.ts +++ b/web/src/common/apis/crawler/type.ts @@ -17,6 +17,9 @@ export interface SourceInfo { last_success_at: string | null last_error: string | null paused_until: string | null + collection_stats?: Record