diff --git a/backend/src/alphapulse/calendar/jin10.py b/backend/src/alphapulse/calendar/jin10.py index ba2f253..5d95176 100644 --- a/backend/src/alphapulse/calendar/jin10.py +++ b/backend/src/alphapulse/calendar/jin10.py @@ -79,7 +79,8 @@ def normalize(body: dict, kind: str, day: date) -> list[Entry]: if identity in identities: raise ValueError("金十同一响应存在重复事件标识") identities.add(identity) - country = text(raw, "country", nullable=True) + country_value = raw.get("country") + country = None if isinstance(country_value, str) and not country_value.strip() else text(raw, "country", nullable=True) precision = "day" if kind == "holiday" else "minute" if kind == "holiday": if raw.get("date") != day.isoformat(): diff --git a/backend/src/alphapulse/calendar/umich.py b/backend/src/alphapulse/calendar/umich.py index 855aaa5..863f8ec 100644 --- a/backend/src/alphapulse/calendar/umich.py +++ b/backend/src/alphapulse/calendar/umich.py @@ -7,7 +7,7 @@ from urllib.parse import urljoin from parsel import Selector from pypdf import PdfReader -from pypdf.errors import PdfReadError +from pypdf.errors import LimitReachedError, PdfReadError from alphapulse.calendar.parsers import EASTERN, MONTHS, Scheduled, cells, compact, number @@ -16,6 +16,8 @@ def pdf_text(body: bytes, pages: int | None = None) -> str: try: reader = PdfReader(BytesIO(body)) text = "\n".join(page.extract_text() for page in reader.pages[:pages]) + except LimitReachedError as exc: + raise ValueError(f"官方 PDF 解析超过资源上限：{exc}") from exc except PdfReadError as exc: raise ValueError("官方 PDF 无法读取") from exc if not text.strip(): diff --git a/backend/src/alphapulse/calendar/worker.py b/backend/src/alphapulse/calendar/worker.py index a09457d..2150d34 100644 --- a/backend/src/alphapulse/calendar/worker.py +++ b/backend/src/alphapulse/calendar/worker.py @@ -359,7 +359,8 @@ class CalendarWorker: raise ValueError("EIA 公告统计周或时间与所选日程不符，未填入其他期次的数据") return response, raw_id if row.source == "dol": - if row.scheduled_at.astimezone(EASTERN).date() < datetime.now(EASTERN).date() and (not row.release_url or "/press/" not in urlparse(row.release_url).path): + # Prefer dated official PDFs even on release day; the rolling host may deny access. + if not row.release_url or "/press/" not in urlparse(row.release_url).path: try: response, raw_id = await self.archive_page(row) except ArchiveMissing: diff --git a/backend/src/alphapulse/core/us_news.py b/backend/src/alphapulse/core/us_news.py index 5258e49..bcac5d4 100644 --- a/backend/src/alphapulse/core/us_news.py +++ b/backend/src/alphapulse/core/us_news.py @@ -154,7 +154,9 @@ def article(source: str, html: str, url: str, *, release_date: str | None = None if not body: raise ValueError('官方正文结构未识别') for node in body.css('script, style, nav, header, footer, form'): - node.root.getparent().remove(node.root) + parent = node.root.getparent() + if parent is not None: + parent.remove(node.root) body = sorted(body, key=lambda node: len(node.xpath('string(.)').get()), reverse=True) content = body[0].get() text = ' '.join(body[0].xpath('.//text()').getall()).strip() diff --git a/backend/src/alphapulse/crawler/spiders/base.py b/backend/src/alphapulse/crawler/spiders/base.py index 1eff60f..0222a4f 100644 --- a/backend/src/alphapulse/crawler/spiders/base.py +++ b/backend/src/alphapulse/crawler/spiders/base.py @@ -247,6 +247,9 @@ class NewsSpider(RedisSpider): # Retain completed reads and explicit gaps, never failed lane positions. cursor = dict(source.history_cursor or {}) for key in ('transport', 'checked', 'document_retries', 'attachment_retries', 'bill_retries'): + # Treasury's HTML lanes must be reset again if the first JSON round fails. + if key == 'transport' and self.name == 'treasury': + continue if key in run.official_cursor: cursor[key] = run.official_cursor[key] source.history_cursor = cursor diff --git a/backend/src/alphapulse/crawler/spiders/us_official.py b/backend/src/alphapulse/crawler/spiders/us_official.py index fe36887..872fc0e 100644 --- a/backend/src/alphapulse/crawler/spiders/us_official.py +++ b/backend/src/alphapulse/crawler/spiders/us_official.py @@ -4,11 +4,13 @@ import json import re import time import uuid +from contextlib import nullcontext from datetime import UTC, datetime, timedelta from html import escape, unescape from email.utils import parsedate_to_datetime from lxml import etree +from pypdf import apply_configuration from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from scrapy import Request @@ -108,6 +110,11 @@ class USOfficialSpider(NewsSpider): def first_requests(self): self.run.completed_documents_safe = True + if self.name == 'treasury' and self.cursor.get('transport') != 'treasury-json-v1': + # HTML page numbers no longer select distinct pages. Retain completion and gap evidence. + self.cursor = {k: v for k, v in self.cursor.items() if k not in ('history', 'realtime')} + self.cursor['transport'] = 'treasury-json-v1' + self.run.official_cursor = self.cursor if self.name == 'doj' and self.cursor.get('transport') != 'doj-api-v1': # Retain retry evidence; HTML discovery cursors cannot use API pagination. self.cursor = {k: v for k, v in self.cursor.items() if k not in ('history', 'realtime', 'checked')} @@ -211,6 +218,8 @@ class USOfficialSpider(NewsSpider): self.cursor['realtime'] = {'entry': 0} return [] url = state.get('page') or spec.entries[state['entry']] + if self.name == 'treasury' and not state.get('page'): + url = 'https://home.treasury.gov/news-data/press-releases/manifest.json' if self.name == 'congress' and not state.get('page'): begin = self.run.history_target if lane == 'history' else self.run.cutoff url = configure_url(url, fromDateTime=begin.strftime('%Y-%m-%dT%H:%M:%SZ'), toDateTime=self.run.started_at.strftime('%Y-%m-%dT%H:%M:%SZ')) @@ -225,11 +234,62 @@ class USOfficialSpider(NewsSpider): state['page'] = url return [self.request_for(url, self.parse_listing, lane)] + def treasury_listing(self, response, lane): + state = self.cursor[lane] + data = response.json() + if not isinstance(data, dict) or data.get('category') != 'press-releases': + raise ValueError('Treasury JSON 索引结构未识别') + if urlsplit(response.url).path.endswith('/manifest.json'): + shards = data.get('shards') + if not isinstance(shards, list) or not shards or type(data.get('total')) is not int: + raise ValueError('Treasury JSON 分片清单不完整') + begin = self.run.history_target if lane == 'history' else self.run.cutoff + offset, selected, seen = 0, [], set() + for shard in shards: + if not isinstance(shard, dict) or any(type(shard.get(k)) is not int for k in ('count', 'offset', 'endOffset', 'startYear', 'endYear')): + raise ValueError('Treasury JSON 分片元数据无效') + if shard['count'] <= 0 or shard['offset'] != offset or shard['endOffset'] != offset + shard['count'] or shard['startYear'] > shard['endYear']: + raise ValueError('Treasury JSON 分片范围不连续') + url = official_url(self.name, shard['path'], response.url) + if not re.fullmatch(r'https://home\.treasury\.gov/news-data/press-releases/\d{4}-\d{4}\.json', url) or url in seen: + raise ValueError('Treasury JSON 分片地址无效或重复') + seen.add(url) + offset = shard['endOffset'] + if shard['endYear'] >= begin.year and shard['startYear'] <= self.run.started_at.year: + selected.append({'url': url, 'count': shard['count']}) + if offset != data['total'] or not selected: + raise ValueError('Treasury JSON 分片总数或目标年份不符') + state['treasury_shards'] = selected + return [], selected[0]['url'] + shards = state.get('treasury_shards', []) + items = data.get('items') + if not shards or shards[0]['url'] != response.url or not isinstance(items, list) or type(data.get('count')) is not int or len(items) != data['count'] or len(items) != shards[0]['count']: + raise ValueError('Treasury JSON 分片内容不完整') + docs = [] + for item in items: + url = official_url(self.name, item['url'], response.url) + path = urlsplit(url).path + # The official index also contains these category landing pages. + if path.rstrip('/') in {f'/news/press-releases/{name}' for name in ('readouts', 'statements-remarks', 'testimonies')}: + continue + if not re.fullmatch(r'/(?:news/)?press-releases/[a-z]+\d+/?', path): + raise ValueError('Treasury JSON 文章地址无效') + at, _ = published(item['datetime']) + if at.date().isoformat() != item['date']: + raise ValueError('Treasury JSON 文章日期不符') + docs.append({'url': url, 'kind': 'article', 'listing_date': at.isoformat()}) + if len({doc['url'] for doc in docs}) != len(docs): + raise ValueError('Treasury JSON 分片文章重复') + state['treasury_shards'] = shards[1:] + return docs, shards[1]['url'] if len(shards) > 1 else None + def parse_listing(self, response): lane = response.meta['lane'] state = self.cursor[lane] docs, next_url = [], None - if self.name == 'doj': + if self.name == 'treasury' and urlsplit(response.url).path.startswith('/news-data/'): + docs, next_url = self.treasury_listing(response, lane) + elif self.name == 'doj': data = response.json() resultset = data['metadata']['resultset'] if data['metadata']['responseInfo']['status'] != 200 or not isinstance(data['results'], list): @@ -583,7 +643,7 @@ class USOfficialSpider(NewsSpider): elif doc['kind'] == 'text' or doc.get('format') == 'pdf': at, precision = published(doc['listing_date'] if doc.get('format') == 'pdf' else doc['at']) if response.body.startswith(b'%PDF'): - content = '
' + escape(pdf_text(response.body)) + '
' + content = '
' + escape(self.pdf_text(response.body)) + '
' else: content = response.text validate_body(content) @@ -692,6 +752,12 @@ class USOfficialSpider(NewsSpider): self.run.stats['unclassified_attachment_gaps'] = self.run.stats.get('unclassified_attachment_gaps', 0) + 1 yield from self.complete(lane) + def pdf_text(self, body): + # Bound the known complex FinCEN report without changing other collectors' limits. + context = apply_configuration(stream_parsing_work_maximum_count=4_000_000) if self.name == 'fincen' else nullcontext() + with context: + return pdf_text(body) + 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') @@ -701,7 +767,7 @@ class USOfficialSpider(NewsSpider): 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) + text = self.pdf_text(response.body) attachment['status'] = 'complete' parsed['content_html'] += f"

{escape(attachment['title'])}

{escape(text)}
" except (ValueError, OSError) as exc: diff --git a/backend/tests/test_calendar_recovery.py b/backend/tests/test_calendar_recovery.py index 50ce1f8..e9f361c 100644 --- a/backend/tests/test_calendar_recovery.py +++ b/backend/tests/test_calendar_recovery.py @@ -268,3 +268,31 @@ def test_restart_honors_persisted_source_pause(trading, monkeypatch): await worker.http.aclose() await worker.bls.aclose() asyncio.run(run()) + +def test_dol_release_day_prefers_dated_archive_without_rolling_request(trading, monkeypatch): + from alphapulse.calendar import worker as module, dol + from test_dol import REPORT + class ReleaseDay(datetime): + @classmethod + def now(cls, tz=None): + return datetime(2026, 10, 29, 14, tzinfo=UTC).astimezone(tz) + monkeypatch.setattr(module, 'datetime', ReleaseDay) + monkeypatch.setattr(module, 'pdf_text', lambda _: REPORT) + async def run(): + worker = module.CalendarWorker() + row = SimpleNamespace(source='dol', family='initial_claims', scheduled_at=datetime(2026, 10, 29, 12, 30, tzinfo=UTC), + release_url=dol.LATEST, first_received_at=None) + page = httpx.Response(200, content=b'%PDF', request=httpx.Request('GET', 'https://oui.doleta.gov/press/2026/102926.pdf')) + worker.archive_page = AsyncMock(return_value=(page, 7)) + worker.fetch = AsyncMock(side_effect=AssertionError('rolling host must not be fetched')) + try: + assert await worker.result_page(row) == (page, 7) + worker.archive_page.assert_awaited_once_with(row) + worker.fetch.assert_not_awaited() + row.scheduled_at -= timedelta(days=7) + with pytest.raises(ValueError, match='所选日程不符'): + await worker.result_page(row) + finally: + await worker.http.aclose() + await worker.bls.aclose() + asyncio.run(run()) diff --git a/backend/tests/test_jin10_calendar.py b/backend/tests/test_jin10_calendar.py index ce5b8db..920283b 100644 --- a/backend/tests/test_jin10_calendar.py +++ b/backend/tests/test_jin10_calendar.py @@ -351,3 +351,22 @@ def test_cross_day_membership_does_not_withdraw_until_all_buckets_remove_it(trad await save_entry(session, original, 4, now) assert row.schedule_status == 'scheduled' asyncio.run(run()) + + +@pytest.mark.parametrize('country', [None, '', ' \t ']) +def test_optional_empty_country_is_unlabelled_without_inference(country): + body = deepcopy(FIXTURE['event:2025-01-08']) + body['list'] = body['list'][:1] + body['list'][0]['country'] = country + entry, = normalize(body, 'event', date(2025, 1, 8)) + assert entry.values['_jin10']['country_name'] == '未标注' + assert entry.values['_jin10']['country_code'] == 'UNKNOWN' + assert body['list'][0]['country'] == country + + +@pytest.mark.parametrize('country', [7, [], {}]) +def test_optional_country_still_rejects_non_string_values(country): + body = deepcopy(FIXTURE['event:2025-01-08']) + body['list'][0]['country'] = country + with pytest.raises(ValueError, match='country'): + normalize(body, 'event', date(2025, 1, 8)) diff --git a/backend/tests/test_us_official_news.py b/backend/tests/test_us_official_news.py index 21003d6..3ab2a53 100644 --- a/backend/tests/test_us_official_news.py +++ b/backend/tests/test_us_official_news.py @@ -1425,3 +1425,160 @@ def test_sec_historical_listing_stops_only_with_verified_dates(trading, dated, m else: assert any('/2024-199' in r.url for r in requests) assert obj.cursor['history']['next'].endswith('?page=1') + + +def test_fed_nested_body_selectors_remove_shared_noise_once(): + body = """

Federal Reserve announcement

+

Federal Reserve announcement

subscribe
+

The Federal Reserve announces a monetary policy decision affecting interest rates.

+ Official report
""" + parsed = article('fed', body, 'https://www.federalreserve.gov/newsevents/pressreleases/other20260513a.htm') + assert 'monetary policy decision' in parsed['content_html'] + assert 'subscribe' not in parsed['content_html'] and 'noise()' not in parsed['content_html'] + assert parsed['published_at'].date().isoformat() == '2026-05-13' + assert parsed['extra']['attachments'][0]['url'].endswith('/report.pdf') + + +def treasury_manifest(): + return {'category': 'press-releases', 'total': 3, 'shards': [ + {'path': '/news-data/press-releases/2026-2025.json', 'count': 2, 'offset': 0, 'endOffset': 2, 'startYear': 2025, 'endYear': 2026}, + {'path': '/news-data/press-releases/2024-2023.json', 'count': 1, 'offset': 2, 'endOffset': 3, 'startYear': 2023, 'endYear': 2024}]} + + +def test_treasury_json_replaces_only_discovery_cursors(trading): + obj = spider(trading, 'treasury') + obj.cursor.update(checked={'completed': 1}, document_retries={'gap': {'document': {}, 'attempts': 3}}, + attachment_retries={'attachment': {'attempts': 3}}, transport='old') + obj.cursor['history']['page'] = SOURCES['treasury'].entries[0] + '?page=2' + requests = obj.first_requests() + assert requests[0].url.endswith('/news-data/press-releases/manifest.json') + assert 'history' not in obj.cursor + assert obj.cursor['checked'] == {'completed': 1} + assert obj.cursor['document_retries']['gap']['attempts'] == 3 + assert obj.cursor['attachment_retries'] == {'attachment': {'attempts': 3}} + checkpoint = copy.deepcopy(obj.cursor) + obj.first_requests() + assert obj.cursor == checkpoint + + +def test_treasury_json_history_resumes_pending_full_articles(trading): + obj = spider(trading, 'treasury') + manifest = 'https://home.treasury.gov/news-data/press-releases/manifest.json' + request, = obj.parse_listing(response(manifest, json.dumps(treasury_manifest()), lane='history')) + assert request.url.endswith('/2026-2025.json') + # A restart must fetch the shard, not declare the empty manifest a finished history. + resumed = spider(trading, 'treasury') + resumed.cursor = copy.deepcopy(obj.cursor) + assert resumed.advance('history')[0].url == request.url + data = {'category': 'press-releases', 'count': 2, 'items': [ + {'url': '/news/press-releases/sb0002/', 'date': '2026-11-03', 'datetime': '2026-11-03T18:00:00Z'}, + {'url': '/news/press-releases/sb0001/', 'date': '2025-01-01', 'datetime': '2025-01-01T18:00:00Z'}]} + document, = resumed.parse_listing(response(request.url, json.dumps(data), lane='history')) + assert document.meta['document']['kind'] == 'article' + assert len(resumed.cursor['history']['pending']) == 2 + assert not resumed.run.history_done + assert resumed.complete('history')[0].url.endswith('/sb0001/') + assert list(resumed.complete('history')) == [] + assert resumed.run.history_done + + +@pytest.mark.parametrize('failure', ['total', 'offset', 'host', 'empty']) +def test_treasury_json_incomplete_manifest_cannot_complete_history(trading, failure): + data = treasury_manifest() + if failure == 'total': data['total'] = 4 + if failure == 'offset': data['shards'][1]['offset'] = 0 + if failure == 'host': data['shards'][0]['path'] = 'https://evil.example/shard.json' + if failure == 'empty': data['shards'] = [] + obj = spider(trading, 'treasury') + with pytest.raises(ValueError): + list(obj.parse_listing(response('https://home.treasury.gov/news-data/press-releases/manifest.json', json.dumps(data), lane='history'))) + assert not obj.run.history_done + + +@pytest.mark.parametrize('source, expected', [('fincen', 4_000_000), ('sec', 2_000_000)]) +def test_pdf_budget_is_bounded_scoped_and_restored(trading, monkeypatch, source, expected): + from pypdf import get_configuration + from alphapulse.crawler.spiders import us_official + def extract(body): + assert get_configuration().stream_parsing_work_maximum_count == expected + raise ValueError('known PDF error') + monkeypatch.setattr(us_official, 'pdf_text', extract) + with pytest.raises(ValueError, match='known PDF error'): + spider(trading, source).pdf_text(b'%PDF-test') + assert get_configuration().stream_parsing_work_maximum_count == 2_000_000 + + +def test_pdf_resource_failure_retains_reason_as_a_gap(trading, monkeypatch): + from pypdf.errors import LimitReachedError + from alphapulse.calendar import umich + def reader(_): + raise LimitReachedError('stream work budget exhausted') + monkeypatch.setattr(umich, 'PdfReader', reader) + obj = spider(trading, 'fincen') + doc = {'url': 'https://www.fincen.gov/system/files/report.pdf', 'kind': 'article', 'format': 'pdf', + 'listing_date': '2026-11-03', 'listing_url': SOURCES['fincen'].entries[1], 'title': 'Official report'} + obj.cursor['realtime'].update(pending=[doc], entry=len(SOURCES['fincen'].entries)) + async def collect(): + return [item async for item in obj.parse_document(Response(doc['url'], body=b'%PDF-test', + request=Request(doc['url'], meta={'lane': 'realtime', 'document': doc, 'raw_response_id': 12})))] + assert asyncio.run(collect()) == [] + assert 'stream work budget exhausted' in obj.cursor['document_retries'][doc['url']]['error'] + assert doc['url'] not in obj.cursor.get('checked', {}) + + +@pytest.mark.parametrize('failure', ['count', 'date', 'host', 'path']) +def test_treasury_invalid_shard_keeps_history_unfinished(trading, failure): + obj = spider(trading, 'treasury') + url = 'https://home.treasury.gov/news-data/press-releases/2026-2025.json' + obj.cursor['history']['treasury_shards'] = [{'url': url, 'count': 1}] + item = {'url': '/news/press-releases/sb0001/', 'date': '2025-01-01', 'datetime': '2025-01-01T18:00:00Z'} + data = {'category': 'press-releases', 'count': 1, 'items': [item]} + if failure == 'count': data['count'] = 2 + if failure == 'date': item['date'] = '2025-01-02' + if failure == 'host': item['url'] = 'https://evil.example/news/press-releases/sb0001/' + if failure == 'path': item['url'] = '/news/unknown/' + with pytest.raises(ValueError): + list(obj.parse_listing(response(url, json.dumps(data), lane='history'))) + assert not obj.run.history_done + assert obj.cursor['history']['treasury_shards'] == [{'url': url, 'count': 1}] + + +def test_treasury_shard_excludes_known_categories_and_retains_legacy_articles(trading): + obj = spider(trading, 'treasury') + url = 'https://home.treasury.gov/news-data/press-releases/2026-2023.json' + obj.cursor['history']['treasury_shards'] = [{'url': url, 'count': 4}] + items = [{'url': f'/news/press-releases/{name}/'} for name in ('readouts', 'statements-remarks', 'testimonies')] + items.append({'url': '/press-releases/jy1839/', 'date': '2023-10-24', 'datetime': '2023-10-24T18:00:00Z'}) + docs, next_url = obj.treasury_listing(response(url, json.dumps({'category': 'press-releases', 'count': 4, 'items': items}), lane='history'), 'history') + assert next_url is None and len(docs) == 1 + assert docs[0]['url'].endswith('/press-releases/jy1839/') + + +@pytest.mark.parametrize('failed', [False, True]) +def test_treasury_transport_migration_commits_only_with_json_lanes(trading, monkeypatch, failed): + from alphapulse.crawler.spiders import base + obj = spider(trading, 'treasury') + old = {'entry': 0, 'page': SOURCES['treasury'].entries[0] + '?page=2'} + source = SimpleNamespace(history_target=obj.run.history_target, history_cursor={'history': old, 'realtime': old}, 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.cursor = copy.deepcopy(source.history_cursor) + obj.first_requests() + obj.cursor['history'] = {'entry': 0, 'page': 'https://home.treasury.gov/news-data/press-releases/manifest.json'} + obj.run.history_position = obj.run.history_target + obj.run.errors = ['temporary failure'] if failed else [] + asyncio.run(obj._finish_run(obj.run)) + if failed: + assert 'transport' not in source.history_cursor + assert source.history_cursor['history'] == old + else: + assert source.history_cursor['transport'] == 'treasury-json-v1' + assert '/news-data/' in source.history_cursor['history']['page'] + resumed = spider(trading, 'treasury') + resumed.cursor = copy.deepcopy(source.history_cursor) + assert resumed.first_requests()[0].url.endswith('/news-data/press-releases/manifest.json')