Fix official news discovery and calendar recovery

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():

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():

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:

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()

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

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 = '<pre>' + escape(pdf_text(response.body)) + '</pre>'
+                content = '<pre>' + escape(self.pdf_text(response.body)) + '</pre>'
             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"<section><h2>{escape(attachment['title'])}</h2><pre>{escape(text)}</pre></section>"
         except (ValueError, OSError) as exc:

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())

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))

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 = """<h1>Federal Reserve announcement</h1><time datetime="2026-05-13"></time>
+    <div id="content"><div id="article"><h3>Federal Reserve announcement</h3><form>subscribe</form><script>noise()</script>
+    <p>The Federal Reserve announces a monetary policy decision affecting interest rates.</p>
+    <a href="/report.pdf">Official report</a></div></div>"""
+    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')