diff --git a/backend/src/alphapulse/calendar/backfill.py b/backend/src/alphapulse/calendar/backfill.py index 4586400..8266152 100644 --- a/backend/src/alphapulse/calendar/backfill.py +++ b/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: diff --git a/backend/src/alphapulse/crawler/history_capture.py b/backend/src/alphapulse/crawler/history_capture.py new file mode 100644 index 0000000..c2f922f --- /dev/null +++ b/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() diff --git a/backend/src/alphapulse/crawler/spiders/us_official.py b/backend/src/alphapulse/crawler/spiders/us_official.py index 7ebb794..fe36887 100644 --- a/backend/src/alphapulse/crawler/spiders/us_official.py +++ b/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(): diff --git a/backend/tests/test_history_capture.py b/backend/tests/test_history_capture.py new file mode 100644 index 0000000..2cd62e3 --- /dev/null +++ b/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]) diff --git a/backend/tests/test_us_official_news.py b/backend/tests/test_us_official_news.py index 736c098..21003d6 100644 --- a/backend/tests/test_us_official_news.py +++ b/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 = '' if dated else '' + body = '

Press releases

' + stamp + 'Release
' + 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') diff --git a/deploy/README.md b/deploy/README.md index 7a081ea..fb7ad86 100644 --- a/deploy/README.md +++ b/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. diff --git a/scripts/backfill_sources.py b/scripts/backfill_sources.py new file mode 100644 index 0000000..590dd91 --- /dev/null +++ b/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()