diff --git a/backend/src/alphapulse/api/crawler.py b/backend/src/alphapulse/api/crawler.py index fd1f303..3881564 100644 --- a/backend/src/alphapulse/api/crawler.py +++ b/backend/src/alphapulse/api/crawler.py @@ -188,6 +188,15 @@ async def list_sources(session: Annotated[AsyncSession, Depends(get_session)]) - run_stats = {run.source_code: run.stats for run in latest_runs} statuses = await redis.hgetall(STATUS_KEY) db_size = await session.scalar(select(func.pg_database_size(func.current_database()))) + attachment_stats = {} + for source in sources: + retries = (source.history_cursor or {}).get('attachment_retries', {}) + pending = sum(value['attempts'] < 3 for value in retries.values()) + attachment_stats[source.code] = { + 'attachment_gaps': len(retries), + 'attachment_retry_pending': pending, + 'attachment_retry_exhausted': len(retries) - pending, + } return success(SourcesOut( manager_alive=bool(statuses), db_size_bytes=db_size, @@ -198,7 +207,7 @@ async def list_sources(session: Annotated[AsyncSession, Depends(get_session)]) - "region_code": REGIONS.get(s.code), "supports_history": s.code in HISTORY_SOURCES, "configuration_error": configuration_error(s.code), - "collection_stats": run_stats.get(s.code, {}), + "collection_stats": {**run_stats.get(s.code, {}), **attachment_stats[s.code]}, "process": json.loads(statuses[s.code]) if s.code in statuses else None, "today_new": today_counts.get(s.code, (0, 0))[0], "today_revised": today_counts.get(s.code, (0, 0))[1], diff --git a/backend/src/alphapulse/core/us_news.py b/backend/src/alphapulse/core/us_news.py index 51f2e0d..3be7383 100644 --- a/backend/src/alphapulse/core/us_news.py +++ b/backend/src/alphapulse/core/us_news.py @@ -23,7 +23,7 @@ SOURCES = { 'sec': OfficialSource('SEC·证券监管', ('https://www.sec.gov/newsroom/press-releases', 'https://www.sec.gov/newsroom/speeches-statements', 'https://www.sec.gov/rules-regulations/rulemaking-activity', 'https://www.sec.gov/rules-regulations/self-regulatory-organization-rulemaking'), r'/(?:newsroom/(?:press-releases|speeches-statements)/[^/]+|rules-regulations/.*\d.*)', '.node-details-layout__main-region__content .field--name-body', ('www.sec.gov',)), 'cftc': OfficialSource('CFTC·衍生品监管', ('https://www.cftc.gov/PressRoom/PressReleases', 'https://www.cftc.gov/PressRoom/SpeechesTestimony', 'https://www.cftc.gov/LawRegulation/CFTCStaffLetters/letters.htm'), r'/(?:PressRoom/(?:PressReleases/\d+-\d+|SpeechesTestimony/(?!index)[^/]+)|csl/)', 'article .field--name-body', ('www.cftc.gov',)), 'treasury': OfficialSource('美国财政部', ('https://home.treasury.gov/news/press-releases',), r'/news/press-releases/[a-z]+\d+/?$', '.field--name-field-news-body', ('home.treasury.gov', 'www.treasury.gov', 'ofac.treasury.gov')), - 'ofac': OfficialSource('OFAC·金融制裁', ('https://ofac.treasury.gov/recent-actions',), r'/recent-actions/\d{8}[^/]*$', '.field--name-body, .body-content', ('ofac.treasury.gov', 'home.treasury.gov')), + 'ofac': OfficialSource('OFAC·金融制裁', ('https://ofac.treasury.gov/recent-actions',), r'/recent-actions/\d{8}[^/]*$', 'article.node--type-ofac-recent-action .field--name-field-body', ('ofac.treasury.gov', 'home.treasury.gov')), 'fincen': OfficialSource('FinCEN·反洗钱', ('https://www.fincen.gov/news/press-releases', 'https://www.fincen.gov/news/enforcement-actions', 'https://www.fincen.gov/resources/advisoriesbulletinsfact-sheets'), r'/(?:news/news-releases/[^/]+|resources/advisories/[^/]+|news/enforcement-actions/[^/]+)', 'article .field--name-body', ('www.fincen.gov',)), 'whitehouse': OfficialSource('白宫·政策', ('https://www.whitehouse.gov/presidential-actions/', 'https://www.whitehouse.gov/fact-sheets/', 'https://www.whitehouse.gov/briefings-statements/', 'https://bidenwhitehouse.archives.gov/briefing-room/'), r'/(?:presidential-actions|fact-sheets|briefings-statements|briefing-room/[^/]+)/202[5-9]/\d{2}/\d{2}/|/(?:presidential-actions|fact-sheets|briefings-statements)/202[5-9]/\d{2}/[^/]+', '.wp-block-post-content, .body-content, .entry-content', ('www.whitehouse.gov', 'bidenwhitehouse.archives.gov')), 'congress': OfficialSource('美国国会·立法', ('https://api.congress.gov/v3/bill?format=json&limit=250', 'https://api.congress.gov/v3/committee-meeting?format=json&limit=250'), r'/bill/', '', ('api.congress.gov', 'www.congress.gov', 'www.govinfo.gov'), 900), @@ -32,11 +32,11 @@ SOURCES = { 'doj': OfficialSource('DOJ·加密机构执法', ('https://www.justice.gov/api/v1/press_releases.json?pagesize=50&sort=date&direction=DESC&fields=uuid,date',), r'/(?:opa|usao[^/]*|archives/opa)/(?:pr|speech)/[^/]+', '.field--name-body, .field--name-field-pr-body', ('www.justice.gov',)), 'fed': OfficialSource('美联储·政策与讲话', tuple(f'https://www.federalreserve.gov/newsevents/{kind}/{year}-{suffix}.htm' for year in range(datetime.now(UTC).year, 2024, -1) for kind, suffix in (('pressreleases', 'press'), ('speech', 'speeches'), ('testimony', 'testimony'))) + ('https://www.federalreserve.gov/monetarypolicy/fomccalendars.htm', 'https://www.federalreserve.gov/releases/h41/releaseDates.json'), r'/(?:newsevents/(?:pressreleases|speech|testimony)/[^/]+\.htm|monetarypolicy/(?:fomcminutes|fomcprojtabl)\d+\.htm|releases/h41/\d+/)', '#article, .col-xs-12.col-sm-8.col-md-8, #content', ('www.federalreserve.gov',), 60), } -RULE_VERSION = 'us-policy-v1' +RULE_VERSION = 'us-policy-v2' CRYPTO = re.compile(r'\b(?:crypto(?:currency|currencies|[ -]assets?)?|digital[ -]assets?|virtual[ -]currenc(?:y|ies)|stablecoins?|bitcoin|ethereum|blockchain|tokeniz(?:ed|ation))\b', re.I) POLICY = re.compile(r'\b(?:regulat\w*|rules?|guidance|legislation|acts?|sanctions?|licenses?|executive order|monetary policy|federal funds|interest rates?|balance sheet|liquidity|quantitative|refunding|borrowing|debt limit|tariffs?|fiscal|budget|appropriation|tax reform)\b', re.I) INSTITUTION = re.compile(r'\b(?:exchanges?|banks?|financial institutions?|payment processors?|custodians?|stablecoin issuers?|binance|coinbase|tether|circle|kraken|bybit|bitfinex)\b', re.I) -MACRO = re.compile(r'\b(?:monetary policy|federal funds|interest rates?|fomc|balance sheet|liquidity|quantitative|refunding|borrowing estimates|debt (?:limit|ceiling)|tariffs?|fiscal|budget|appropriation|tax reform|financial sanctions)\b', re.I) +MACRO = re.compile(r"\b(?:monetary policy|federal funds (?:rate|target)|interest rates?|discount rates?|(?:primary|secondary) credit rates?|reserve requirements?|interest on reserve balances|fomc|balance sheet|liquidity|quantitative (?:easing|tightening)|refunding|borrowing estimates|debt (?:limit|ceiling)|tariffs?|fiscal (?:policy|stimulus|outlook)|federal (?:budget|deficit)|president['’]s budget|budget of the united states|(?:continuing|consolidated|full[ -]year) appropriations?|government (?:shutdown|funding)|tax reform|financial sanctions)\b", re.I) def selection(source: str, title: str, text: str) -> list[str]: @@ -44,7 +44,8 @@ def selection(source: str, title: str, text: str) -> list[str]: crypto = bool(CRYPTO.search(content)) if crypto and (source not in ('doj', 'ofac', 'fincen') or POLICY.search(title) or INSTITUTION.search(content)): return ['crypto_policy' if POLICY.search(content) else 'crypto_institution'] - if source not in ('doj', 'cftc', 'sec') and MACRO.search(content): + macro_scope = title if source == 'federal_register' else content + if source not in ('doj', 'cftc', 'sec') and MACRO.search(macro_scope): return ['us_macro_policy'] if source == 'ofac' and INSTITUTION.search(content) and POLICY.search(content): return ['financial_sanctions'] diff --git a/backend/src/alphapulse/crawler/spiders/us_official.py b/backend/src/alphapulse/crawler/spiders/us_official.py index 9d30fa1..5224add 100644 --- a/backend/src/alphapulse/crawler/spiders/us_official.py +++ b/backend/src/alphapulse/crawler/spiders/us_official.py @@ -316,7 +316,7 @@ class USOfficialSpider(NewsSpider): for result in self.complete(lane): yield result return - parsed = {'title': data['title'], 'content_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'], + parsed = {'title': data['title'], 'content_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'], 'retry_document': doc, '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 @@ -514,7 +514,7 @@ class USOfficialSpider(NewsSpider): 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: + if reasons: 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) @@ -523,6 +523,8 @@ class USOfficialSpider(NewsSpider): yield item else: self.run.stats['excluded'] = self.run.stats.get('excluded', 0) + 1 + if incomplete: + self.run.stats['unclassified_attachment_gaps'] = self.run.stats.get('unclassified_attachment_gaps', 0) + 1 yield from self.complete(lane) def parse_attachment(self, response): @@ -532,6 +534,8 @@ class USOfficialSpider(NewsSpider): try: if response.status != 200: raise ValueError(f'HTTP {response.status}') + if response.headers.get('Content-Type', b'').lower().startswith(b'image/') and response.body.startswith((b'\x89PNG\r\n\x1a\n', b'\xff\xd8\xff', b'GIF87a', b'GIF89a')): + raise ValueError('官方图片附件尚未提取文字，请核对原文') text = pdf_text(response.body) attachment['status'] = 'complete' parsed['content_html'] += f"

{escape(attachment['title'])}

{escape(text)}
" diff --git a/backend/tests/fixtures/us_news/live_selection.json.gz b/backend/tests/fixtures/us_news/live_selection.json.gz new file mode 100644 index 0000000..cfae5e7 Binary files /dev/null and b/backend/tests/fixtures/us_news/live_selection.json.gz differ diff --git a/backend/tests/test_us_official_news.py b/backend/tests/test_us_official_news.py index d367e1b..f7e05af 100644 --- a/backend/tests/test_us_official_news.py +++ b/backend/tests/test_us_official_news.py @@ -8,7 +8,7 @@ from types import SimpleNamespace import pytest from scrapy import Request -from scrapy.http import HtmlResponse +from scrapy.http import HtmlResponse, Response from alphapulse.core.us_news import SOURCES, article, official_url, published, selection @@ -39,6 +39,14 @@ def response(url, body, **meta): ('treasury', 'Meeting with delegation', 'Greetings and cooperation', False), ('ofac', 'New sanctions', 'Sanctions against a financial institution', True), ('sec', 'Ethernet equipment', 'Method and hardware', False), + ('whitehouse', "The President's Budget for Fiscal Year 2027", 'The President submitted his budget request to Congress, proposing nationwide spending and revenue changes.', True), + ('whitehouse', 'The President’s Budget', 'Annual spending and revenue proposals.', True), + ('congress', 'Continuing Appropriations Act, 2027', 'This bill provides continuing appropriation for the operations of the United States.', True), + ('federal_register', 'Adjustments to the Reserve Bank Discount Rates', 'The Federal Reserve increases interest rates charged on loans to depository institutions.', True), + ('federal_register', 'Reserve Requirements of Depository Institutions', 'The Federal Reserve adjusts reserve requirements.', True), + ('federal_register', 'Interest on Reserve Balances', 'The Federal Reserve adjusts interest paid on reserve balances.', True), + ('federal_register', 'Hospital Budget Estimates for Fiscal Year 2027', 'The budget provides funding for hospital operations.', False), + ('whitehouse', 'Health research funding', 'The department budget supports quantitative research during the fiscal year.', False), ]) def test_selection_boundaries(source, title, text, expected): assert bool(selection(source, title, text)) is expected @@ -58,14 +66,102 @@ def test_dates_do_not_fabricate_publisher_time(): def test_ofac_original_body_and_attachment_boundaries(): - data = article('ofac', '''

Crypto sanctions

New sanctions against a virtual currency exchange institution.

- Licenseexternal
''', 'https://ofac.treasury.gov/recent-actions/20250101') + data = article('ofac', '''

Crypto sanctions

New sanctions against a virtual currency exchange institution.

+ Licenseexternal
''', 'https://ofac.treasury.gov/recent-actions/20250101') assert data['published_at'] == datetime(2025, 1, 1, tzinfo=UTC) assert data['extra']['published_at_precision'] == 'date' assert len(data['extra']['attachments']) == 2 assert data['extra']['attachments'][1]['status'] == 'unsupported_host' +def test_live_ofac_body_and_policy_selection(): + import gzip + from pathlib import Path + sample = json.loads(gzip.decompress((Path(__file__).parent / 'fixtures/us_news/live_selection.json.gz').read_bytes())) + parsed = article('ofac', sample['ofac']['body'], sample['ofac']['url']) + assert 'Iran General License FF-1' in parsed['content_html'] + assert "A Part of Treasury's Office" not in parsed['content_html'] + for case in sample['cases']: + assert bool(selection(case['source_code'], case['title'], case['content_text'])) is case['expected'], case['title'] + + +def test_unrelated_failed_attachment_is_not_news(trading): + obj = spider(trading, 'doj') + obj.run.history_deadline = time.monotonic() - 1 + doc = {'url': 'https://www.justice.gov/api/v1/press_releases/x.json', 'kind': 'doj'} + obj.cursor['history']['pending'] = [doc] + parsed = {'url': doc['url'], 'title': 'Federal prosecutor receives national award', 'external_id': 'award', 'content_html': '

The department recognized superior performance in Indian Country.

', 'published_at': datetime(2026, 11, 4, tzinfo=UTC), 'raw_response_id': 12, 'retry_document': doc, + 'extra': {'attachments': [{'url': 'https://www.justice.gov/media/1/dl', 'status': 'unavailable'}]}} + assert list(obj.attach_or_finish(parsed, 'history')) == [] + assert obj.run.stats['unclassified_attachment_gaps'] == 1 + assert obj.cursor['attachment_retries'][doc['url']]['attempts'] == 1 + + +@pytest.mark.parametrize('content_type,body', [('image/png', b'\x89PNG\r\n\x1a\n'), ('image/jpeg', b'\xff\xd8\xff'), ('image/gif', b'GIF89a')]) +@pytest.mark.parametrize('relevant', [True, False]) +def test_image_attachment_preserves_unknown_text_gap(trading, monkeypatch, content_type, body, relevant): + from alphapulse.crawler.spiders import us_official + obj = spider(trading, 'doj') + obj.run.history_deadline = time.monotonic() - 1 + url = 'https://www.justice.gov/media/1/dl' + doc = {'url': 'https://www.justice.gov/api/v1/press_releases/x.json', 'kind': 'doj'} + obj.cursor['history']['pending'] = [doc] + parsed = {'url': doc['url'], 'title': 'Cryptocurrency exchange charged' if relevant else 'Department announces an enforcement action', 'external_id': 'case', 'content_html': '

The cryptocurrency exchange was charged with money laundering.

' if relevant else '

See the attached consent order for the findings and penalty.

', 'published_at': datetime(2026, 11, 4, tzinfo=UTC), 'raw_response_id': 12, 'retry_document': doc, + 'extra': {'attachments': [{'url': url, 'title': 'Consent Order', 'status': 'pending'}]}} + monkeypatch.setattr(us_official, 'pdf_text', lambda body: pytest.fail('Images must not reach the PDF parser')) + res = Response(url, body=body, headers={'Content-Type': content_type}, request=Request(url, meta={'lane': 'history', 'parsed': parsed, 'raw_response_id': 13})) + items = list(obj.parse_attachment(res)) + if relevant: + item, = items + assert item['extra']['content_scope'] == 'partial' + else: + assert items == [] + assert obj.run.stats['unclassified_attachment_gaps'] == 1 + attachment = parsed['extra']['attachments'][0] + assert attachment['status'] == 'unavailable' + assert '图片' in attachment['error'] + assert attachment['raw_response_id'] == 13 + assert obj.cursor['attachment_retries'][doc['url']]['attempts'] == 1 + + +def test_source_api_keeps_attachment_gaps_visible_across_rounds(trading, monkeypatch): + from unittest.mock import AsyncMock + from alphapulse.api import crawler + from alphapulse.models import Source + + source = Source(code='doj', name='DOJ', list_url=SOURCES['doj'].entries[0], enabled=False, interval_seconds=300, backfill_hours=72, + consecutive_failures=0, last_success_at=None, last_error=None, paused_until=None, history_cursor={'attachment_retries': {'a': {'attempts': 1}, 'b': {'attempts': 2}, 'c': {'attempts': 3}}}, + history_target=None, history_position=None, history_done=False) + latest = SimpleNamespace(source_code='doj', stats={'discovered': 2, 'attachment_gaps': 999}) + + class Session: + async def scalars(self, statement): + rows = [source] if statement.column_descriptions[0]['entity'] is Source else [latest] + return SimpleNamespace(all=lambda: rows) + async def execute(self, statement): + return SimpleNamespace(all=lambda: []) if 'news_items' in str(statement) else [] + async def scalar(self, statement): + return 1234 + + monkeypatch.setattr(crawler.redis, 'hgetall', AsyncMock(return_value={})) + monkeypatch.setattr(crawler, 'calendar_sources', AsyncMock(return_value=[])) + for stats in ({'discovered': 2, 'attachment_gaps': 999}, {}): + latest.stats = stats + result = asyncio.run(crawler.list_sources(Session())).data.sources[0] + assert result.collection_stats['attachment_gaps'] == 3 + assert result.collection_stats['attachment_retry_pending'] == 2 + assert result.collection_stats['attachment_retry_exhausted'] == 1 + source.history_cursor['attachment_retries'].pop('a') + result = asyncio.run(crawler.list_sources(Session())).data.sources[0] + assert result.collection_stats['attachment_gaps'] == 2 + assert result.collection_stats['attachment_retry_pending'] == 1 + source.history_cursor = {} + latest.stats = {'attachment_gaps': 999} + result = asyncio.run(crawler.list_sources(Session())).data.sources[0] + assert result.collection_stats['attachment_gaps'] == 0 + assert result.collection_stats['attachment_retry_exhausted'] == 0 + + def test_challenge_is_failure_not_empty_success(trading): obj = spider(trading) with pytest.raises(ValueError, match='未识别'): @@ -730,6 +826,11 @@ def test_congress_meeting_includes_deduplicated_witness_pdfs(trading, monkeypatc assert req.url == pdf['url'] assert len(req.meta['parsed']['extra']['attachments']) == 1 assert req.meta['parsed']['extra']['attachments'][0]['title'] == 'Witness Statement' + obj.run.history_deadline = time.monotonic() - 1 + req.meta['raw_response_id'] = 13 + item, = obj.parse_attachment(Response(req.url, status=404, request=req)) + assert item['extra']['content_scope'] == 'partial' + assert obj.cursor['attachment_retries'][doc['url']]['document'] == doc def test_dated_but_unavailable_bill_text_remains_a_gap(trading): diff --git a/deploy/README.md b/deploy/README.md index 948bfaa..3d2416d 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -276,6 +276,22 @@ attachment text are searchable and versioned. Date-only releases are displayed a dates, not precise release instants. Backfilled documents describe the version available at capture time; they are not historical point-in-time evidence. +Selection rule `us-policy-v2` uses explicit national budget, continuing/consolidated +appropriations, discount-rate and reserve-policy phrases instead of standalone +`budget`, `fiscal` or `quantitative`. Federal Register macro screening uses the title +to avoid incidental references in unrelated regulations; crypto screening still +reads the complete body and extracted attachments. + +Image attachments remain an explicit text-extraction gap, including when their +format is PNG, JPEG or GIF. They do not establish completeness or irrelevance. +Relevant documents remain `partial`; documents whose relevance is still unknown +are kept in bounded attachment retries rather than inserted as unrelated news. +Congress meeting attachment failures retain the same retry evidence as articles. +The source API derives `attachment_gaps`, `attachment_retry_pending` and +`attachment_retry_exhausted` from the persisted cursor, independently of the latest +round's statistics. The console keeps these document counts visible across rounds; +success clears the gap, while three failed attempts remain visible for investigation. + Validation without production credentials: - `cd backend && uv run pytest && uvx ruff check --select F,E9,B src tests` diff --git a/web/src/pages/crawler/components/SourcePanel.vue b/web/src/pages/crawler/components/SourcePanel.vue index d278638..768b859 100644 --- a/web/src/pages/crawler/components/SourcePanel.vue +++ b/web/src/pages/crawler/components/SourcePanel.vue @@ -160,8 +160,11 @@ onMounted(loadLatest)
待核验法案 {{ source.collection_stats.unresolved_bills }}；官方资料尚不充分
+
+ 附件待核验文档 {{ source.collection_stats.attachment_gaps }}；待重试 {{ source.collection_stats.attachment_retry_pending }}；重试已耗尽 {{ source.collection_stats.attachment_retry_exhausted }}，请核对原文 +
- 本轮发现 {{ source.collection_stats.discovered }}；筛除 {{ source.collection_stats.excluded ?? 0 }}；附件缺口 {{ source.collection_stats.partial ?? 0 }} + 本轮发现 {{ source.collection_stats.discovered }}；筛除 {{ source.collection_stats.excluded ?? 0 }}；不完整包 {{ source.collection_stats.partial ?? 0 }}
今日新增 +{{ source.today_new }}