Add resumable official news and calendar backfill to January 2025

backend/src/alphapulse/calendar/backfill.py

@@ -129,15 +129,17 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
 
     def report_error(source: str, url: str, exc: Exception) -> None:
         failure = {"source": source, "url": url, "error": str(exc)}
-        if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code in (429, 503):
+        if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code in (403, 429, 503):
             stopped_sources.add(source)
             failure.update(retry_after=exc.response.headers.get("Retry-After"), stop_reason="已停止该来源，历史覆盖不完整。")
         errors.append(failure)
 
     ctx = ssl.create_default_context()
     ctx.set_ciphers("DEFAULT")
-    async with httpx.AsyncClient(timeout=15, follow_redirects=False) as client, httpx.AsyncClient(timeout=15, verify=ctx) as bls:
+    async with httpx.AsyncClient(timeout=20, follow_redirects=False) as client, httpx.AsyncClient(timeout=15, verify=ctx) as bls:
         async def fetch(source: str, url: str, *, query: dict | None = None) -> Document:
+            if source in stopped_sources:
+                raise ValueError("该来源本轮已停止，保留冷却等待下次尝试")
             parsed = urlparse(url)
             if parsed.scheme != "https" or parsed.hostname not in HOSTS[source]:
                 raise ValueError("归档链接不是对应机构官方 HTTPS 地址")
@@ -156,6 +158,8 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
                     query = None
             else:
                 raise ValueError("官方归档重定向超过 5 次")
+            if response.status_code in (403, 429, 503):
+                stopped_sources.add(source)
             response.raise_for_status()
             return Document(source, str(response.url), response.content, datetime.now(UTC), headers=dict(response.headers))
 
@@ -191,7 +195,7 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
                             documents.append(await fetch(source, link))
                             known.add(link)
                         except (httpx.HTTPError, ValueError) as exc:
-                            if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code in (429, 503):
+                            if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code in (403, 429, 503):
                                 raise
                             report_error(source, link, exc)
                     url = following
@@ -203,7 +207,7 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
                 documents.append(await fetch("dol", dol.LATEST))
             except (httpx.HTTPError, ValueError) as exc:
                 report_error("dol", dol.LATEST, exc)
-        if "adp" in selected:
+        if "adp" in selected and "adp" not in stopped_sources:
             url = adp.INDEX
             visited = set()
             while url:
@@ -221,7 +225,7 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
                 except (httpx.HTTPError, ValueError) as exc:
                     report_error("adp", str(exc.request.url) if isinstance(exc, httpx.HTTPError) else url, exc)
                     break
-        if "eia" in selected:
+        if "eia" in selected and "eia" not in stopped_sources:
             try:
                 documents.append(await fetch("eia", eia.SCHEDULE))
                 index = await fetch("eia", eia.ARCHIVE)
@@ -271,7 +275,7 @@ async def fetch_documents(start: date, end: date, known: set[str], errors: list[
                             documents.append(await fetch("umich", link))
                             known.add(link)
                         except (httpx.HTTPError, ValueError) as exc:
-                            if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code in (429, 503):
+                            if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code in (403, 429, 503):
                                 raise
                             report_error("umich", link, exc)
             except (httpx.HTTPError, ValueError) as exc:
@@ -290,6 +294,8 @@ async def save_announcement(session, document: Document, family: str, announceme
         raise ValueError("同日同系列匹配到多个日历事件，未写入")
     raw_id = document.raw_id
     row = candidates[0] if candidates else None
+    if row is not None and row.collection_retired_at is not None:
+        return False
     # 回补不回滚已采集的结果；只为已知事件补充缺失的统计期和发布阶段。
     if row is not None and row.first_received_at is not None:
         if (row.period is not None or announcement.period is None) and (row.stage is not None or announcement.stage is None):
@@ -318,9 +324,13 @@ async def save_announcement(session, document: Document, family: str, announceme
     return True
 
 
-async def run(start: date, end: date, *, commit: bool = False, stored_only: bool = False, sources: set[str] | None = None) -> dict:
+async def run(start: date, end: date, *, commit: bool = False, stored_only: bool = False, sources: set[str] | None = None, until: datetime | None = None) -> dict:
     begin = datetime.combine(start, time.min, LOCAL).astimezone(UTC)
     finish = datetime.combine(end + timedelta(days=1), time.min, LOCAL).astimezone(UTC)
+    if until is not None:
+        if until.tzinfo is None or until <= begin:
+            raise ValueError("回补终点须带时区且晚于起点")
+        finish = min(finish, until)
     errors = []
     selected = sources if sources is not None else US_SOURCES
     async with SessionLocal() as session:

backend/src/alphapulse/crawler/history_capture.py

@@ -0,0 +1,200 @@
+"""可恢复历史摘要导出；输出真实响应供既有 import_history 导入，不写业务表。"""
+import argparse
+import asyncio
+import base64
+import gzip
+import json
+import re
+from datetime import UTC, datetime
+from pathlib import Path
+from urllib.parse import urljoin, urlparse
+
+import httpx
+from parsel import Selector
+from sqlalchemy import select
+
+from alphapulse.calendar.backfill import fetch_documents, family_for
+from alphapulse.calendar.archives import parse_announcements, publication_time
+from alphapulse.crawler.europe_history import capture as european_capture
+from alphapulse.crawler.import_history import ArchivedArticle, Capture
+from alphapulse.core.db import SessionLocal, engine
+from alphapulse.core.official_sources import FEEDS, relevant_news
+from alphapulse.models import RawBody, RawResponse
+
+SOURCES = ('coindesk', 'bls', 'bea', *FEEDS)
+
+
+def timeline_articles(data):
+    if not isinstance(data.get('articles'), list) or not isinstance(data.get('hasMore'), bool):
+        raise ValueError('CoinDesk 时间线结构未识别')
+    return [ArchivedArticle(guid=a['_id'], url=urljoin('https://www.coindesk.com', a['pathname']),
+                            title=a['title'], summary=a.get('description') or '',
+                            published_at=a['articleDates']['displayDate']) for a in data['articles']]
+
+
+def ecb_articles(data):
+    if not isinstance(data, list) or len(data) % 13:
+        raise ValueError('ECB 官方索引记录宽度已变化')
+    articles = []
+    for i in range(0, len(data), 13):
+        row = data[i:i + 13]
+        at = datetime.fromtimestamp(row[1], UTC)
+        for path in row[9]:
+            url = urljoin('https://www.ecb.europa.eu', path)
+            title = row[10]['Title']
+            if relevant_news('ecb', title, url) and not url.endswith('.pdf'):
+                articles.append(ArchivedArticle(guid=url, url=url, title=title, summary='', published_at=at))
+    return articles
+
+
+def package(source, url, body, at, headers, articles):
+    return Capture(source=source, url=url, status=200, headers=headers,
+                   body=base64.b64encode(body).decode(), body_encoding='base64',
+                   fetched_at=at, articles=articles)
+
+
+async def export(source, begin, end, output):
+    errors, captures = [], []
+    existing = []
+    async with SessionLocal() as session:
+        rows = (await session.execute(select(RawResponse, RawBody.body).join(
+            RawBody, RawBody.sha256 == RawResponse.body_sha256).where(
+                RawResponse.source_code == source, RawResponse.status == 200,
+                RawResponse.url.contains('/chunk_') if source == 'ecb' else
+                RawResponse.url.contains('/api/v1/articles/timeline'))
+            .distinct(RawResponse.url, RawResponse.body_sha256)
+            .order_by(RawResponse.url, RawResponse.body_sha256, RawResponse.id))).all() if source in ('ecb', 'coindesk') else []
+        for raw, body in rows:
+            body = gzip.decompress(body)
+            try:
+                articles = ecb_articles(json.loads(body)) if source == 'ecb' else timeline_articles(json.loads(body))
+                captures.append(package(source, raw.url, body, raw.fetched_at, raw.headers, articles))
+                existing.append(raw.url)
+            except (ValueError, KeyError, TypeError, IndexError) as exc:
+                errors.append({'url': raw.url, 'error': str(exc)})
+    async with httpx.AsyncClient(timeout=20, follow_redirects=False) as client:
+        async def get(url, host):
+            await asyncio.sleep(1)
+            for _ in range(6):
+                parsed = urlparse(url)
+                if parsed.scheme != 'https' or parsed.hostname != host:
+                    raise ValueError('归档链接或重定向不属于预期官方域名')
+                r = await client.get(url, headers={'User-Agent': 'AlphaPulse historical news collector'})
+                if not r.is_redirect:
+                    break
+                url = urljoin(url, r.headers['location'])
+            else:
+                raise ValueError('归档重定向超过五次')
+            r.raise_for_status()
+            return r
+        try:
+            if source == 'coindesk':
+                url, visited = 'https://www.coindesk.com/api/v1/articles/timeline?size=16&lang=en', set()
+                while url:
+                    if url in visited:
+                        raise ValueError('CoinDesk 历史分页形成循环')
+                    visited.add(url)
+                    r = await get(url, 'www.coindesk.com'); data = r.json()
+                    articles = timeline_articles(data)
+                    captures.append(package(source, str(r.url), r.content, datetime.now(UTC), dict(r.headers), articles))
+                    if not articles:
+                        if data['hasMore']:
+                            raise ValueError('CoinDesk 空页仍声明有后续数据')
+                        break
+                    dates = [a.published_at for a in articles]
+                    if dates != sorted(dates, reverse=True):
+                        raise ValueError('CoinDesk 时间线日期非倒序')
+                    if min(dates) < begin or not data['hasMore']:
+                        break
+                    a = data['articles'][-1]
+                    url = str(httpx.URL('https://www.coindesk.com/api/v1/articles/timeline', params={
+                        'size': 16, 'lang': 'en', 'lastDisplayDate': a['articleDates']['displayDate'], 'lastId': a['_id']}))
+            elif source == 'bea':
+                url, visited = 'https://www.bea.gov/news/archive?field_related_product_target_id=All&created_1=All&title=', set()
+                while url:
+                    if url in visited:
+                        raise ValueError('BEA 归档分页形成循环')
+                    visited.add(url)
+                    r = await get(url, 'www.bea.gov'); sel = Selector(text=r.text)
+                    links = list(dict.fromkeys(urljoin(str(r.url), u) for u in sel.css('a[href]::attr(href)').getall()
+                        if any('/news/' + str(y) + '/' in u for y in range(begin.year, end.year + 1))))
+                    for link in links:
+                        page = await get(link, 'www.bea.gov'); article = Selector(text=page.text)
+                        stamp = article.css('.field--name-field-release-date').xpath('string(.)').get()
+                        if not stamp:
+                            raise ValueError('BEA 公告缺少正式发布日期')
+                        a = ArchivedArticle(guid=link, url=link, title=article.css('h1').xpath('string(.)').get(),
+                            summary=article.css('meta[name="description"]::attr(content)').get() or '', published_at=publication_time(stamp))
+                        if begin <= a.published_at < end:
+                            captures.append(package(source, link, page.content, datetime.now(UTC), dict(page.headers), [a]))
+                    following = sel.css('.pager__item--next a::attr(href)').get()
+                    years = re.findall(r'/news/(20\d\d)/', r.text)
+                    url = urljoin(str(r.url), following) if following and (not years or max(map(int, years)) >= begin.year) else None
+            elif source == 'ecb':
+                if not existing:
+                    raise ValueError('ECB 缺少已验证的官方索引入口')
+                for url in dict.fromkeys(existing):
+                    r = await get(url, 'www.ecb.europa.eu')
+                    captures.append(package(source, str(r.url), r.content, datetime.now(UTC), dict(r.headers), ecb_articles(r.json())))
+                errors.append({'error': 'ECB 已存索引已重验；动态索引版本至任务终点的完整性仍未证实'})
+            elif source == 'boe':
+                r = await get('https://www.bankofengland.co.uk/sitemap/speeches', 'www.bankofengland.co.uk')
+                links = list(dict.fromkeys(urljoin(str(r.url), p) for p in Selector(text=r.text).css('a[href]::attr(href)').getall() if '/speech/' in p))
+                if not links:
+                    raise ValueError('BoE 讲话索引没有可识别的链接')
+                for url in links:
+                    page = await get(url, 'www.bankofengland.co.uk');sel = Selector(text=page.text)
+                    dates = sel.css('time::attr(datetime), meta[property="article:published_time"]::attr(content)').getall()
+                    if not dates:
+                        raise ValueError('BoE 讲话缺少可识别的正式日期')
+                    a = ArchivedArticle(guid=url, url=url, title=sel.css('h1').xpath('string(.)').get(),
+                                        summary=sel.css('.page-content p').xpath('string(.)').get() or '', published_at=dates[0])
+                    if begin <= a.published_at < end:
+                        captures.append(package(source, url, page.content, datetime.now(UTC), dict(page.headers), [a]))
+        except (httpx.HTTPError, ValueError, KeyError, TypeError, IndexError) as exc:
+            item = {'error': f'{type(exc).__name__}: {exc}'}
+            if isinstance(exc, httpx.HTTPStatusError):
+                item['retry_after'] = exc.response.headers.get('Retry-After')
+            errors.append(item)
+    if source != 'coindesk' and not (source in ('boe', 'ecb') and any('HTTPStatusError' in e.get('error', '') for e in errors)):
+        documents = await fetch_documents(begin.date(), end.date(), set(), errors, sources={source})
+        for doc in documents:
+            try:
+                if source in FEEDS:
+                    captures.append(european_capture(doc))
+                else:
+                    family = family_for(doc)
+                    if family:
+                        announcements = parse_announcements(family, doc.body, pdf=doc.body.startswith(b'%PDF'))
+                        title = Selector(text=doc.body.decode(errors='replace')).css('h1').xpath('string(.)').get() or family
+                        articles = [ArchivedArticle(guid=doc.url, url=doc.url, title=title, summary='', published_at=a.at) for a in announcements]
+                        captures.append(package(source, doc.url, doc.body, doc.captured_at, doc.headers, articles))
+            except (ValueError, KeyError, TypeError, IndexError) as exc:
+                errors.append({'url': doc.url, 'error': str(exc)})
+    with gzip.open(output, 'xt', encoding='utf-8') as stream:
+        for capture in captures:
+            capture.articles = [a for a in capture.articles if begin <= a.published_at < end]
+            stream.write(capture.model_dump_json() + '\n')
+    return {'source': source, 'responses': len(captures), 'articles': sum(len(c.articles) for c in captures),
+            'errors': errors, 'output': str(output)}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('source', choices=SOURCES)
+    parser.add_argument('--start', required=True, type=datetime.fromisoformat)
+    parser.add_argument('--end', required=True, type=datetime.fromisoformat)
+    parser.add_argument('--output', required=True, type=Path)
+    args = parser.parse_args()
+    if args.start.tzinfo is None or args.end.tzinfo is None or args.start >= args.end:
+        parser.error('范围须带时区且开始早于结束')
+    async def run():
+        try:
+            return await export(args.source, args.start, args.end, args.output)
+        finally:
+            await engine.dispose()
+    print(json.dumps(asyncio.run(run()), ensure_ascii=False))
+
+
+if __name__ == '__main__':
+    main()

backend/src/alphapulse/crawler/spiders/us_official.py

@@ -298,8 +298,9 @@ class USOfficialSpider(NewsSpider):
                             if not title:
                                 raise ValueError('FinCEN PDF 列表缺少标题')
                             doc.update(format='pdf', title=title, listing_url=response.url)
-                    elif self.name in ('treasury', 'cftc', 'fincen', 'whitehouse'):
+                    elif self.name in ('sec', 'treasury', 'cftc', 'fincen', 'whitehouse'):
                         container = {
+                            'sec': 'ancestor::tr[1]',
                             'treasury': 'ancestor::div[h3][1]',
                             'cftc': 'ancestor::tr[1]',
                             'fincen': 'ancestor::div[contains(@class,"fincen-news-article__content")][1]',
@@ -332,7 +333,7 @@ class USOfficialSpider(NewsSpider):
                     return True
                 return day >= begin.date()
             docs = [doc for doc in docs if in_range(doc)]
-        if self.name in ('treasury', 'cftc', 'fincen', 'whitehouse') and docs:
+        if self.name in ('sec', 'treasury', 'cftc', 'fincen', 'whitehouse') and docs:
             begin = self.run.history_target if lane == 'history' else self.run.cutoff
             dates = [published(d['listing_date'])[0] for d in docs if d.get('listing_date')]
             if len(dates) == len(docs) and dates == sorted(dates, reverse=True) and dates[-1].date() < begin.date():

backend/tests/test_history_capture.py

@@ -0,0 +1,31 @@
+from datetime import UTC, datetime
+import pytest
+
+
+
+def test_timeline_preserves_actual_publication_and_provider_identity(trading):
+    from alphapulse.crawler.history_capture import timeline_articles
+    result = timeline_articles({'hasMore': False, 'articles': [{'_id': 'original', 'pathname': '/markets/story',
+        'title': 'Official title', 'description': 'Public summary',
+        'articleDates': {'displayDate': '2024-12-31T17:00:00Z'}}]})
+    assert result[0].guid == 'original'
+    assert result[0].published_at == datetime(2024, 12, 31, 17, tzinfo=UTC)
+    assert result[0].summary == 'Public summary'
+
+
+def test_unknown_timeline_shape_is_not_empty_success(trading):
+    from alphapulse.crawler.history_capture import timeline_articles
+    with pytest.raises(ValueError):
+        timeline_articles({'articles': []})
+
+
+def test_ecb_index_keeps_existing_news_scope(trading):
+    from alphapulse.crawler.history_capture import ecb_articles
+    row = [1, 1735732800, 2025, 0, 1, None, None, None, None,
+           ['/press/key/date/2025/html/example.en.html', '/stats/example.html'], {'Title': 'Speech'}, [], []]
+    result = ecb_articles(row)
+    assert len(result) == 1
+    assert '/press/key/' in str(result[0].url)
+    assert result[0].published_at == datetime.fromtimestamp(1735732800, UTC)
+    with pytest.raises(ValueError):
+        ecb_articles(row[:-1])

backend/tests/test_us_official_news.py

@@ -1411,3 +1411,17 @@ def test_attachment_http_failure_spends_durable_retry_budget(trading):
     assert item['extra']['content_scope'] == 'partial'
     assert obj.cursor['attachment_retries'][url]['attempts'] == 1
     assert '503' in obj.cursor['attachment_retries'][url]['error']
+
+
+@pytest.mark.parametrize('dated', [True, False])
+def test_sec_historical_listing_stops_only_with_verified_dates(trading, dated, monkeypatch):
+    monkeypatch.setattr('alphapulse.crawler.spiders.us_official.configuration_error', lambda code: None)
+    obj = spider(trading, 'sec')
+    stamp = '<time datetime="2024-12-30T12:00:00Z">Dec 30</time>' if dated else ''
+    body = '<h1>Press releases</h1><table><tr><td>' + stamp + '</td><td><a href="/newsroom/press-releases/2024-199">Release</a></td></tr></table><a rel="next" href="?page=1">Next</a>'
+    requests = list(obj.parse_listing(response(SOURCES['sec'].entries[0], body, lane='history')))
+    if dated:
+        assert all('/2024-199' not in r.url and '?page=1' not in r.url for r in requests)
+    else:
+        assert any('/2024-199' in r.url for r in requests)
+        assert obj.cursor['history']['next'].endswith('?page=1')

deploy/README.md

@@ -431,3 +431,23 @@ repair budget for existing news gaps. It preserves the previous retry evidence
 and records a durable initialization marker. Repeating it cannot renew the budget.
 Set the authorized SEC contact in the private instance environment before enabling
 SEC; do not put the address or credentials in tracked configuration.
+
+### Authorized 2025 history backfill
+
+`scripts/backfill_sources.py` previews the current baseline without writing data.
+`--execute` creates or resumes `/docker/gitea/dev/AlphaPulse/backfill-20250101/state.json`;
+`--once` runs one eligible pass. The frozen start is 2025-01-01 Asia/Shanghai.
+Both WallstreetCN sources are excluded. Existing US official history lanes are
+activated without resetting document retry budgets. Nine other news sources use
+real archive captures and the existing idempotent importer; all thirteen official
+calendar sources use the archive backfill. Jin10 retains its existing daily gap
+scheduler and validated checkpoints. News AI must remain paused.
+
+Each external archive phase retains its reports and capture files. Actual failed
+phases wait at least one hour (or a longer Retry-After) and stop after three
+attempts; rerunning the driver preserves this budget. Already collected full news
+and retired calendar results are retained. A source can remain limited even when
+some documents were imported: ECB dynamic index version coverage, inaccessible
+BoE speeches and CoinDesk access blocks are reported rather than labeled complete.
+The deployment lock serializes writes; crawler/calendar processes are restored in
+finally blocks. Do not start a second driver or remove its state to renew quotas.

scripts/backfill_sources.py

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+"""Resume the authorized 2025 news/calendar backfill. Default: read-only preview."""
+import argparse
+from datetime import UTC, datetime, timedelta
+import json
+from pathlib import Path
+import time
+from zoneinfo import ZoneInfo
+
+from ci import locked, require, run, save
+import deploy
+
+NEWS = ('coindesk', 'bls', 'bea', 'ecb', 'eurostat', 'boe', 'ons', 'destatis', 'insee')
+CALENDAR = ('bls', 'bea', 'fed', 'umich', 'dol', 'eia', 'adp', 'ecb', 'eurostat', 'boe', 'ons', 'destatis', 'insee')
+BEGIN = '2025-01-01T00:00:00+08:00'
+
+
+def python(code):
+    return run('docker', 'exec', '-i', 'AlphaPulse-dev-api', 'python', '-', input=code)
+
+
+def snapshot():
+    return json.loads(python('''
+import asyncio,json
+from sqlalchemy import select,func
+from alphapulse.core.db import SessionLocal,engine
+from alphapulse.models import Source,NewsItem
+from alphapulse.models.risk import RiskSettings
+async def main():
+ async with SessionLocal() as s:
+  setting=await s.get(RiskSettings,1)
+  rows=(await s.execute(select(NewsItem.source_code,func.count()).group_by(NewsItem.source_code))).all()
+  print(json.dumps({'news_ai_enabled':bool(setting and setting.enabled),'news_counts':dict(rows)}))
+ await engine.dispose()
+asyncio.run(main())
+'''))
+
+
+def initialize():
+    python('''
+import asyncio,copy
+from datetime import datetime
+from sqlalchemy import select
+from alphapulse.core.db import SessionLocal,engine
+from alphapulse.core.us_news import SOURCES
+async def main():
+ async with SessionLocal() as s:
+  from alphapulse.models import Source
+  for source in await s.scalars(select(Source).where(Source.code.in_(SOURCES)).with_for_update()):
+   target=datetime.fromisoformat('2025-01-01T00:00:00+08:00')
+   if source.history_target==target: continue
+   source.history_target=target;source.history_done=False
+   cursor=copy.deepcopy(source.history_cursor or {})
+   cursor.pop('history',None)
+   if source.code=='sec': cursor['realtime']={'entry':0}
+   source.history_cursor=cursor
+  await s.commit()
+ await engine.dispose()
+asyncio.run(main())
+''')
+
+
+def attempt(task, state, directory, command):
+    kind, source = task.split(':')
+    if kind == 'news':
+        name = f'{source}-{state["attempts"]}.jsonl.gz'
+        target = '/tmp/alphapulse-' + name
+        result = json.loads(run('docker', 'exec', 'AlphaPulse-dev-api', 'python', '-m',
+            'alphapulse.crawler.history_capture', source, '--start', BEGIN, '--end', state['end'], '--output', target))
+        run('docker', 'cp', 'AlphaPulse-dev-api:' + target, directory / name)
+        run(*command, 'stop', 'crawler')
+        try:
+            run('docker', 'exec', 'AlphaPulse-dev-api', 'python', '-m', 'alphapulse.crawler.import_history',
+                source, target, '--start', BEGIN)
+        finally:
+            run(*command, 'up', '-d', 'crawler')
+        run('docker', 'exec', 'AlphaPulse-dev-api', 'rm', target)
+        return result
+    # Existing calendar backfill requires sole ownership while matching/creating releases.
+    run(*command, 'stop', 'calendar')
+    try:
+        value = python('''
+import asyncio,json
+from datetime import date,datetime
+from alphapulse.calendar.backfill import run
+from alphapulse.core.db import engine
+async def main():
+ try:
+  result=await run(date(2025,1,1),date.fromisoformat(END),commit=True,sources={SOURCE},until=datetime.fromisoformat(UNTIL))
+  print(json.dumps(result,ensure_ascii=False))
+ finally: await engine.dispose()
+asyncio.run(main())
+'''.replace('END', repr(datetime.fromisoformat(state['end']).astimezone(ZoneInfo('Asia/Shanghai')).date().isoformat())).replace('SOURCE', repr(source)).replace('UNTIL', repr(state['end'])))
+        return json.loads(value)
+    finally:
+        run(*command, 'up', '-d', 'calendar')
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--execute', action='store_true')
+    parser.add_argument('--once', action='store_true', help='One pass; leave cooldown tasks for a later resume')
+    args = parser.parse_args()
+    directory = deploy.INSTANCE / 'backfill-20250101'
+    if not args.execute:
+        print(json.dumps({'start': BEGIN, 'exclude': ['wscn_live', 'wscn_news'], 'baseline': snapshot(),
+                          'state': str(directory / 'state.json')}, ensure_ascii=False))
+        return
+    directory.mkdir(exist_ok=True)
+    with locked(directory / 'task.lock'):
+        path = directory / 'state.json'
+        state = json.loads(path.read_text()) if path.exists() else {'start': BEGIN,
+            'end': datetime.now(UTC).isoformat(), 'baseline': snapshot(), 'initialized': False,
+            'tasks': {f'{kind}:{source}': {'attempts': 0, 'status': 'pending', 'next_at': 0}
+                      for kind, codes in [('news', NEWS), ('calendar', CALENDAR)] for source in codes}}
+        require(not snapshot()['news_ai_enabled'], 'News AI must remain paused for collection backfill')
+        save(path, state)
+        while True:
+            with locked(deploy.INSTANCE / '.deploy.lock'):
+                require(not (deploy.INSTANCE / 'deployment.pending.json').exists(), 'Deployment in progress; resume later')
+                release = json.loads((deploy.INSTANCE / 'deployment.json').read_text())
+                command = deploy.compose(deploy.INSTANCE, release)
+                if not state['initialized']:
+                    run(*command, 'stop', 'crawler')
+                    try:
+                        initialize()
+                        state['initialized'] = True
+                        save(path, state)
+                    finally:
+                        run(*command, 'up', '-d', 'crawler')
+                for name, task in state['tasks'].items():
+                    if task['status'] in ('done', 'retired') or task['next_at'] > time.time():
+                        continue
+                    task['attempts'] += 1
+                    task['status'] = 'running'
+                    save(path, state)
+                    try:
+                        result = attempt(name, {**task, 'end': state['end']}, directory, command)
+                        save(directory / (name.replace(':', '-') + f'-{task["attempts"]}.json'), result)
+                        errors = result.get('errors', []) + result.get('missing_announcements', [])
+                        task['status'] = 'waiting' if errors else 'done'
+                        task['errors'] = errors
+                    except (RuntimeError, ValueError, OSError) as exc:
+                        task.update(status='waiting', errors=[{'error': str(exc)[-4000:]}])
+                    if task['status'] == 'waiting':
+                        if task['attempts'] >= 3:
+                            task['status'] = 'retired'
+                        deadline = datetime.now(UTC) + timedelta(hours=1)
+                        from email.utils import parsedate_to_datetime
+                        for error in task['errors']:
+                            value = error.get('retry_after')
+                            if value:
+                                try:
+                                    at = datetime.now(UTC) + timedelta(seconds=int(value)) if value.isdigit() else parsedate_to_datetime(value)
+                                    deadline = max(deadline, at)
+                                except (ValueError, TypeError, OverflowError):
+                                    pass
+                        task['next_at'] = deadline.timestamp()
+                    save(path, state)
+                    print(name, task['status'], 'attempt', task['attempts'], flush=True)
+            pending = [v for v in state['tasks'].values() if v['status'] not in ('done', 'retired')]
+            if args.once or not pending:
+                print(json.dumps({'tasks': {k: v['status'] for k, v in state['tasks'].items()}, 'snapshot': snapshot()}))
+                return
+            time.sleep(min(30, max(1, min(v['next_at'] for v in pending) - time.time())))
+
+
+if __name__ == '__main__':
+    main()