diff --git a/backend/alembic/versions/20261105_db2345678901_collection_retirement.py b/backend/alembic/versions/20261105_db2345678901_collection_retirement.py new file mode 100644 index 0000000..28f3bf9 --- /dev/null +++ b/backend/alembic/versions/20261105_db2345678901_collection_retirement.py @@ -0,0 +1,18 @@ +"""Persist bounded calendar collection failures.""" +from alembic import op +import sqlalchemy as sa + +revision = "db2345678901" +down_revision = "da1234567890" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("economic_releases", sa.Column("collection_failures", sa.Integer(), nullable=False, server_default="0")) + op.add_column("economic_releases", sa.Column("collection_retired_at", sa.DateTime(timezone=True), nullable=True)) + + +def downgrade(): + op.drop_column("economic_releases", "collection_retired_at") + op.drop_column("economic_releases", "collection_failures") diff --git a/backend/src/alphapulse/api/crawler.py b/backend/src/alphapulse/api/crawler.py index 90eeb73..b87abab 100644 --- a/backend/src/alphapulse/api/crawler.py +++ b/backend/src/alphapulse/api/crawler.py @@ -65,6 +65,8 @@ class CalendarSourceOut(BaseModel): code: str name: str region_code: str + enabled: bool + retired_count: int worker_alive: bool heartbeat_at: datetime | None schedule_checked_at: datetime | None @@ -124,13 +126,14 @@ async def calendar_sources(session: AsyncSession) -> list[CalendarSourceOut]: now = datetime.now(UTC) raw = await redis.get(f"{KEY_PREFIX}calendar:status") heartbeat = json.loads(raw) if raw else None - pending = EconomicRelease.first_received_at.is_(None) & (EconomicRelease.schedule_status == "scheduled") + pending = EconomicRelease.first_received_at.is_(None) & (EconomicRelease.schedule_status == "scheduled") & EconomicRelease.collection_retired_at.is_(None) published = EconomicRelease.first_received_at.is_not(None) rows = (await session.execute(select( EconomicRelease.source, func.max(EconomicRelease.schedule_checked_at).label("schedule_success_at"), func.max(EconomicRelease.first_received_at).label("latest_result_at"), func.min(EconomicRelease.scheduled_at).filter(pending & (EconomicRelease.scheduled_at > now)).label("next_event_at"), + func.count().filter(EconomicRelease.collection_retired_at.is_not(None)).label("retired_count"), func.count().filter(pending & EconomicRelease.scheduled_at.between(now - timedelta(days=45), now)).label("overdue_count"), func.min(EconomicRelease.scheduled_at).filter(published).label("history_start"), func.max(EconomicRelease.scheduled_at).filter(published).label("history_end"), @@ -138,7 +141,7 @@ async def calendar_sources(session: AsyncSession) -> list[CalendarSourceOut]: stats = {row["source"]: dict(row) for row in rows} result_errors = {} for source, error in await session.execute(select(EconomicRelease.source, EconomicRelease.error).where( - EconomicRelease.error.is_not(None), EconomicRelease.schedule_status == "scheduled", + EconomicRelease.error.is_not(None), EconomicRelease.collection_retired_at.is_(None), EconomicRelease.schedule_status == "scheduled", EconomicRelease.scheduled_at >= now - timedelta(days=45), ).order_by(EconomicRelease.last_checked_at.desc())): result_errors.setdefault(source, error) @@ -158,9 +161,12 @@ async def calendar_sources(session: AsyncSession) -> list[CalendarSourceOut]: errors = [s["error"] for s in checks if s["error"]] if source.code in result_errors: errors.append(result_errors[source.code]) + if not source.enabled and source.last_error: + errors.append(source.last_error) data = stats.get(source.code, {}) output.append(CalendarSourceOut( code=source.code, name=source.name, region_code=REGIONS[source.code], worker_alive=source_alive, + enabled=source.enabled, retired_count=data.get("retired_count", 0), heartbeat_at=current_heartbeat["at"] if current_heartbeat else None, schedule_checked_at=max((s["checked_at"] for s in checks), default=None), **{key: data.get(key) for key in ("schedule_success_at", "latest_result_at", "next_event_at", "history_start", "history_end")}, @@ -197,6 +203,8 @@ async def list_sources(session: Annotated[AsyncSession, Depends(get_session)]) - bills = (source.history_cursor or {}).get('bill_retries', {}) bill_pending = sum(value['attempts'] < 3 for value in bills.values()) attachment_stats[source.code] = { + 'retired_count': sum(bool(v.get('retired_at')) for group in (retries, documents, bills) for v in group.values()), + 'active_gaps': sum(not v.get('retired_at') for group in (retries, documents, bills) for v in group.values()), 'attachment_gaps': len(retries), 'attachment_retry_pending': pending, 'attachment_retry_exhausted': len(retries) - pending, diff --git a/backend/src/alphapulse/calendar/eia.py b/backend/src/alphapulse/calendar/eia.py index 5b71f42..10fa55e 100644 --- a/backend/src/alphapulse/calendar/eia.py +++ b/backend/src/alphapulse/calendar/eia.py @@ -77,7 +77,7 @@ def archive_links(body: str, start: date, end: date) -> list[str]: def csv_link(body: str, base: str) -> str: links = [urljoin(base, link) for link in Selector(text=body).css("a[href]::attr(href)").getall() - if link == "download/table1.csv"] + if link in ("download/table1.csv", "csv/table1.csv")] if len(set(links)) != 1: raise ValueError("EIA 归档未唯一列示 Table 1 CSV") return links[0] diff --git a/backend/src/alphapulse/calendar/service.py b/backend/src/alphapulse/calendar/service.py index 3af6852..83034fa 100644 --- a/backend/src/alphapulse/calendar/service.py +++ b/backend/src/alphapulse/calendar/service.py @@ -109,6 +109,8 @@ def save_result(session: AsyncSession, row: EconomicRelease, values: dict, url: if ((period is not None and row.period is not None and period != row.period) or (stage is not None and row.stage is not None and stage != row.stage)): raise ValueError("公告统计期或发布阶段与日历不符") + row.collection_failures = 0 + row.collection_retired_at = None if row.values != values or row.release_url != url or (period is not None and row.period != period) or (stage is not None and row.stage != stage): row.values = values row.release_url = url @@ -177,6 +179,8 @@ def release_rows(row: EconomicRelease, now: datetime) -> list[dict]: state = "withdrawn" elif row.first_received_at is not None: state = "released" + elif getattr(row, "collection_retired_at", None) is not None: + state = "retired" elif now - row.scheduled_at > timedelta(days=45): state = "historical" elif row.error: @@ -214,7 +218,7 @@ def release_rows(row: EconomicRelease, now: datetime) -> list[dict]: "impact_basis": "forecast" if forecast is not None else "previous" if previous is not None else None, "impact": direction, "explanation": explanation, "release_url": row.release_url, "schedule_url": row.schedule_url, "first_received_at": row.first_received_at, - "last_checked_at": row.last_checked_at, "error": row.error, "revision_no": row.revision_no}) + "last_checked_at": row.last_checked_at, "collection_retired_at": getattr(row, "collection_retired_at", None), "error": row.error, "revision_no": row.revision_no}) return result diff --git a/backend/src/alphapulse/calendar/worker.py b/backend/src/alphapulse/calendar/worker.py index 9d96b18..ab8d780 100644 --- a/backend/src/alphapulse/calendar/worker.py +++ b/backend/src/alphapulse/calendar/worker.py @@ -11,20 +11,21 @@ import httpx from parsel import Selector from sqlalchemy import select -from alphapulse.calendar.catalog import SCHEDULE_URLS, UMICH_CSV, UMICH_HISTORY, UMICH_HOME, UMICH_INFO +from alphapulse.calendar.catalog import SCHEDULE_URLS, UMICH_CSV, UMICH_HISTORY, UMICH_HOME, UMICH_INFO, UMICH_REPORTS from alphapulse.calendar import adp, dol, eia, europe -from alphapulse.calendar.archives import ArchiveMissing +from alphapulse.calendar.archives import ArchiveMissing, parse_umich_report from alphapulse.calendar.parsers import EASTERN, MONTHS, bea_release_link, parse_fed_calendar, parse_ics, parse_values, release_date, bls_periods, release_metadata from alphapulse.calendar.service import apply_bls_periods, record_revision, next_poll, save_raw, save_result, save_schedule, save_series -from alphapulse.calendar.umich import current_result, document_links, monthly_series, parse_schedule, pdf_text +from alphapulse.calendar.umich import current_result, document_links, monthly_series, parse_schedule, pdf_text, report_links from alphapulse.core.db import SessionLocal from alphapulse.core.official_sources import FEEDS, HOSTS, feed_entries from alphapulse.core.redis import KEY_PREFIX, redis -from alphapulse.models import EconomicRelease +from alphapulse.models import EconomicRelease, Source logger = logging.getLogger("alphapulse.calendar") STATUS_KEY = f"{KEY_PREFIX}calendar:status" HISTORY_DAYS = 45 +BOUNDED_SOURCES = frozenset({"dol", "eia", "umich", "adp"}) def retry_after_at(value: str | None, now: datetime) -> datetime | None: @@ -95,12 +96,13 @@ class CalendarWorker: self.index_cache[source, url] = (datetime.now(UTC) + timedelta(hours=1), page) return page - async def adp_links(self, start, end): + async def adp_links(self, start, end, first_page=None): url, visited = adp.INDEX, set() for _ in range(8): if url in visited: raise ValueError('ADP 官方分页形成循环') visited.add(url) - response, _ = await self.index('adp', url) + response, _ = first_page if first_page is not None else await self.index('adp', url) + first_page = None links, following = adp.release_links(response.text, url, start, end) if links or not following: return links url = following @@ -112,7 +114,13 @@ class CalendarWorker: module = dol if source == 'dol' else eia index_url = dol.SCHEDULE if source == 'dol' else eia.ARCHIVE response, _ = await self.index(source, index_url) - links = module.archive_links(response.text, day, day) + try: + links = module.archive_links(response.text, day, day) + except ArchiveMissing: + if source != 'dol': + raise + # The index can lag behind published PDFs; validate the PDF date below. + links = [f'https://oui.doleta.gov/press/{day:%Y}/{day:%m%d%y}.pdf'] if len(links) != 1: raise ValueError(f'{source} 当期官方归档不唯一') url = links[0] if source == 'eia': @@ -140,7 +148,7 @@ class CalendarWorker: if delay > 0: await asyncio.sleep(delay) async with self.fetch_slots: client = self.bls if source == "bls" else self.http - response = await client.post(url, json=query) if query is not None else await client.get(url) + response = await client.post(url, json=query) if query is not None else await client.get(url, **({"timeout": httpx.Timeout(20, connect=8)} if source == "adp" else {})) now = datetime.now(UTC) self.host_ready[host] = now + timedelta(seconds=1) if response.status_code in (403, 429, 503): @@ -169,6 +177,10 @@ class CalendarWorker: urls += [("dol", dol.SCHEDULE)] urls += [("eia", eia.SCHEDULE)] urls += [("adp", adp.INDEX)] + async with SessionLocal() as session: + disabled = set(await session.scalars(select(Source.code).where(Source.code.in_(BOUNDED_SOURCES), Source.enabled.is_(False)))) + urls = [(s, u) for s, u in urls if s not in disabled] + checked = {} # 启动时明确各来源尚未就绪；月份滚动后移除不再检查的旧月状态。 self.sources = {url: self.sources.get(url, {"source": source, "checked_at": now.isoformat(), "error": "尚未同步"}) for source, url in urls} @@ -184,7 +196,7 @@ class CalendarWorker: response, raw_id = await self.fetch(source, fetch_url, query=europe.INSEE_QUERY if source == "insee" else None) response.raise_for_status() if source == "adp": - links = await self.adp_links(datetime.min.date(), now.astimezone(EASTERN).date()) + links = await self.adp_links(datetime.min.date(), now.astimezone(EASTERN).date(), (response, raw_id)) if not links: raise ValueError("ADP 官方新闻索引缺少近期月报") report, evidence = await self.fetch(source, links[0]) @@ -223,8 +235,11 @@ class CalendarWorker: async with SessionLocal() as session: await save_schedule(session, source, url, events, raw_id, datetime.now(UTC)) await session.commit() + checked.setdefault(source, []).append(None) self.sources[url] = {"source": source, "checked_at": datetime.now(UTC).isoformat(), "error": None} except (httpx.HTTPError, ValueError, KeyError, TypeError, IndexError, StopIteration) as exc: + if not isinstance(exc, SourceCooling): + checked.setdefault(source, []).append(str(exc)) state = {"source": source, "checked_at": datetime.now(UTC).isoformat(), "error": str(exc)} retry_at = getattr(exc, 'retry_at', None) if isinstance(exc, httpx.HTTPStatusError): @@ -237,6 +252,24 @@ class CalendarWorker: self.sources[url] = state logger.exception("日程采集失败：%s", url) + async with SessionLocal() as session: + for code, outcomes in checked.items(): + if code not in BOUNDED_SOURCES: + continue + source = await session.get(Source, code, with_for_update=True) + if source is None: + continue + if None in outcomes: + source.consecutive_failures = 0 + source.last_success_at = datetime.now(UTC) + source.last_error = None + else: + source.consecutive_failures += 1 + source.last_error = outcomes[0] + if source.consecutive_failures >= 3: + source.enabled = False + await session.commit() + async def umich_schedule(self, info: str, now: datetime) -> None: history, _ = await self.fetch("umich", UMICH_HISTORY) history.raise_for_status() @@ -351,7 +384,17 @@ class CalendarWorker: return None if row.first_received_at is not None: return None # 最新页已换期；保留已取得的当期值，不误报历史缺口。 - raise ValueError("密歇根当前页已晚于所选历史期次，未取得当期公告") + index, _ = await self.index('umich', f'{UMICH_REPORTS}?year={row.period[:4]}') + url = report_links(index.text, int(row.period[:4])).get((row.period, row.stage)) + if url is None: + raise ArchiveMissing("密歇根官方年度归档尚未列出所选期次") + response, raw_id = await self.fetch('umich', url) + response.raise_for_status() + announcement = parse_umich_report(await asyncio.to_thread(pdf_text, response.content)) + if (announcement.at, announcement.period, announcement.stage) != (row.scheduled_at, row.period, row.stage): + raise ValueError("密歇根归档日期、月份或阶段与所选期次不符") + response.extensions['calendar_announcement'] = announcement + return response, raw_id if row.source == "bls": code = "empsit" if row.family == "jobs" else row.family # 本机 BLS archive 路径返回 403，最新公告页可访问；下面必须核对正式发布日期。 @@ -439,6 +482,8 @@ class CalendarWorker: async def poll_one(self, release_id: int) -> None: async with SessionLocal() as session: row = await session.scalar(select(EconomicRelease).where(EconomicRelease.id == release_id).with_for_update()) + if row is None or row.collection_retired_at is not None: + return now = datetime.now(UTC) retry_at = None try: @@ -458,18 +503,29 @@ class CalendarWorker: announcement = europe.parse_announcement(row.family, response.text) period, stage, values = announcement.period, announcement.stage, announcement.values elif row.family == "umich": - period, stage, values = current_result(response.text) + if announcement := response.extensions.get('calendar_announcement'): + period, stage, values = announcement.period, announcement.stage, announcement.values + else: + period, stage, values = current_result(response.text) else: values = parse_values(row.family, response.text) text = " ".join(Selector(text=response.text).css("h1,#bodytext").xpath("string(.)").getall()) period, stage = release_metadata(row.family, text) save_result(session, row, values, response.extensions.get("calendar_url", str(response.url)), raw_id, datetime.now(UTC), period=period, stage=stage) + if page is not None: + row.collection_failures = 0 row.error = None except (httpx.HTTPError, ValueError, KeyError, TypeError, IndexError) as exc: row.error = f"{type(exc).__name__}: {exc}" - retry_at = getattr(exc, 'retry_at', None) + if row.source in BOUNDED_SOURCES and not isinstance(exc, SourceCooling): + row.collection_failures += 1 + if row.collection_failures >= 3: + row.collection_retired_at = datetime.now(UTC) + if row.source in BOUNDED_SOURCES: + retry_at = datetime.now(UTC) + timedelta(hours=1) + retry_at = max(filter(None, (retry_at, getattr(exc, 'retry_at', None))), default=None) if isinstance(exc, httpx.HTTPStatusError): - retry_at = self.cooldowns.get(exc.request.url.host, retry_at) + retry_at = max(filter(None, (retry_at, self.cooldowns.get(exc.request.url.host))), default=None) if retry_at: row.error += f"；下次尝试 {retry_at.isoformat()}" if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code in (429, 503): retry_at = max(retry_at or datetime.now(UTC), retry_after_at(exc.response.headers.get("Retry-After"), datetime.now(UTC)) or datetime.now(UTC)) @@ -483,7 +539,8 @@ class CalendarWorker: async def results(self) -> None: now = datetime.now(UTC) async with SessionLocal() as session: - ids = list(await session.scalars(select(EconomicRelease.id).where( + ids = list(await session.scalars(select(EconomicRelease.id).join(Source, Source.code == EconomicRelease.source).where( + Source.enabled.is_(True), EconomicRelease.collection_retired_at.is_(None), EconomicRelease.source != "jin10", EconomicRelease.schedule_status == "scheduled", EconomicRelease.scheduled_at <= now, EconomicRelease.scheduled_at >= now - timedelta(days=HISTORY_DAYS), EconomicRelease.next_check_at <= now, diff --git a/backend/src/alphapulse/crawler/repair_collection.py b/backend/src/alphapulse/crawler/repair_collection.py new file mode 100644 index 0000000..b7455a9 --- /dev/null +++ b/backend/src/alphapulse/crawler/repair_collection.py @@ -0,0 +1,51 @@ +"""本次采集修复初始化：默认只读，--execute 幂等建立一次新预算。运行前停止 crawler。""" +import argparse +import asyncio +import copy +import json +from datetime import UTC, datetime + +from sqlalchemy import select, text + +from alphapulse.core.db import SessionLocal, engine +from alphapulse.models import Source + +REPAIR = 'collection-repair-20261105-v1' +CODES = ('sec', 'doj', 'congress') + + +async def run(execute=False): + output = [] + async with SessionLocal() as session: + if not execute: + await session.execute(text('SET TRANSACTION READ ONLY')) + query = select(Source).where(Source.code.in_(CODES)).order_by(Source.code) + if execute: + query = query.with_for_update() + for source in await session.scalars(query): + cursor = copy.deepcopy(source.history_cursor or {}) + if cursor.get('collection_repair') == REPAIR: + output.append({'source': source.code, 'already_initialized': True}) + continue + affected = [] + for key in ('document_retries', 'attachment_retries', 'bill_retries'): + for url, entry in cursor.get(key, {}).items(): + affected.append({'kind': key, 'url': url, 'previous_attempts': entry['attempts']}) + entry['repair_previous'] = {k: v for k, v in entry.items() if k != 'document'} + entry['attempts'] = 0 + entry['next_at'] = datetime.now(UTC).timestamp() + entry.pop('retired_at', None) + cursor['collection_repair'] = REPAIR + if execute: + source.history_cursor = cursor + output.append({'source': source.code, 'items': affected}) + if execute: + await session.commit() + await engine.dispose() + return {'mode': 'execute' if execute else 'dry-run', 'sources': output} + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--execute', action='store_true') + print(json.dumps(asyncio.run(run(parser.parse_args().execute)), ensure_ascii=False)) diff --git a/backend/src/alphapulse/crawler/spiders/base.py b/backend/src/alphapulse/crawler/spiders/base.py index cba2150..1eff60f 100644 --- a/backend/src/alphapulse/crawler/spiders/base.py +++ b/backend/src/alphapulse/crawler/spiders/base.py @@ -253,16 +253,22 @@ class NewsSpider(RedisSpider): if history_saved: source.history_position = run.history_position source.history_done = run.history_done - if failed: + bounded = self.name in ('sec', 'doj', 'congress') + source_failed = failed and (not bounded or not run.stats.get('source_readable')) + if source_failed: source.consecutive_failures += 1 delay = min(BACKOFF_BASE_SECONDS * 2 ** (source.consecutive_failures - 1), BACKOFF_MAX_SECONDS) if run.cooldown_until: delay = max(delay, int((run.cooldown_until - now).total_seconds()) + 1) + if bounded: + delay = max(delay, 3600) + if source.consecutive_failures >= 3: + source.enabled = False source.paused_until = now + timedelta(seconds=delay) source.last_error = run.errors[0][:1000] else: source.consecutive_failures = 0 - source.paused_until = None + source.paused_until = run.cooldown_until if bounded else None source.last_error = None source.last_success_at = now await session.commit() @@ -270,6 +276,7 @@ class NewsSpider(RedisSpider): elapsed = (now - run.started_at).total_seconds() summary = f"请求 {run.requests}，条目 {run.items_seen}，新增 {run.items_new}，修订 {run.items_revised}，用时 {elapsed:.1f}s" if failed: + delay = max(0, int((source.paused_until - now).total_seconds())) if source.paused_until else 0 self.logger.error("本轮失败：%s；错误 %d 个，暂停 %ds", summary, len(run.errors), delay) else: self.logger.info("本轮完成：%s", summary) diff --git a/backend/src/alphapulse/crawler/spiders/us_official.py b/backend/src/alphapulse/crawler/spiders/us_official.py index 064fffd..02cf08d 100644 --- a/backend/src/alphapulse/crawler/spiders/us_official.py +++ b/backend/src/alphapulse/crawler/spiders/us_official.py @@ -6,6 +6,7 @@ import time import uuid from datetime import UTC, datetime, timedelta from html import escape, unescape +from email.utils import parsedate_to_datetime from lxml import etree from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit @@ -155,6 +156,8 @@ class USOfficialSpider(NewsSpider): def request_for(self, url, callback, lane, **meta): url = official_url(self.name, url) request = self.request(url, callback, lane=lane, official_source=self.name, **meta) + if self.name in ('sec', 'doj', 'congress'): + request.meta['dont_retry'] = True request.priority = 10 if lane == 'realtime' else -10 # Credentials never travel to document hosts or into persisted URLs. if urlsplit(url).hostname == 'api.congress.gov': @@ -317,6 +320,7 @@ class USOfficialSpider(NewsSpider): docs = list({d['url']: d for d in docs}.values()) if not docs: raise ValueError('官方列表未发现文档，不能认定历史遍历完成') + self.run.stats['source_readable'] = 1 if self.name == 'fed': begin = self.run.history_target if lane == 'history' else self.run.cutoff def in_range(doc): @@ -380,11 +384,23 @@ class USOfficialSpider(NewsSpider): self.run.errors.append(error) self.logger.error('文档缺口：%s；%s', doc['url'], error) state['pending'].pop(0) + self.retire_exhausted() return self.advance(request.meta['lane']) def on_request_error(self, failure): status = failure.value.response.status if failure.check(HttpError) else None - if status not in (403, 429): + if self.name in ('sec', 'doj', 'congress') and status in (403, 429, 503): + if status in (403, 429): + self.block_until(datetime.now(UTC) + timedelta(hours=1)) + value = failure.value.response.headers.get('Retry-After', b'').decode('ascii', errors='ignore') + if value: + try: + until = datetime.now(UTC) + timedelta(seconds=int(value)) if value.isdigit() else parsedate_to_datetime(value) + if until.tzinfo is not None: + self.block_until(until) + except (ValueError, TypeError, OverflowError): + self.logger.warning('忽略无效 Retry-After 响应头') + if status not in (403, 429) or self.name in ('sec', 'doj', 'congress'): requests = self.document_failed(failure.request, f'请求失败 {failure.request.url}：{status or repr(failure.value)}') if requests is not None: return requests @@ -539,6 +555,10 @@ class USOfficialSpider(NewsSpider): file_id = attachment['file']['id'] if not str(file_id).isdigit(): raise ValueError('DOJ API 附件编号无效') + if (f'/media/{file_id}' in (record.get('image') or '') and + re.match(r'Photo[-\s]', attachment.get('alt', ''), re.I) and + 'left to right' in attachment.get('alt', '').lower()): + continue # Explicitly captioned illustration, not a text document. api_attachments.append({'url': f'https://www.justice.gov/media/{file_id}/dl', 'title': attachment.get('description') or 'Official attachment', 'status': 'pending'}) attachment_html = '' elif not isinstance(attachment_html, str): @@ -623,18 +643,23 @@ class USOfficialSpider(NewsSpider): retries.pop(context['url'], None) if not context['reasons']: self.run.stats['excluded'] = self.run.stats.get('excluded', 0) + 1 + self.retire_exhausted() self.run.stats['unresolved_bills'] = len(retries) self.cursor[lane]['pending'].pop(0) if lane == 'realtime': self.cursor.setdefault('checked', {})[context['document_url']] = self.run.started_at.timestamp() def attach_or_finish(self, parsed, lane): + if self.cursor.get('attachment_retries', {}).get(parsed['url'], {}).get('retired_at'): + yield from self.complete(lane) + return pending = next((a for a in parsed['extra']['attachments'] if a['status'] == 'pending'), None) if pending: request = self.request_for(pending['url'], self.parse_attachment, lane, parsed=parsed) request.meta['handle_httpstatus_list'] = [404, 410] yield request return + self.retire_exhausted() text = html_to_text(parsed['content_html']) reasons = selection(self.name, parsed['title'], text) or parsed['extra'].get('selection_reasons', []) incomplete = any(a['status'] != 'complete' for a in parsed['extra']['attachments']) @@ -646,6 +671,7 @@ class USOfficialSpider(NewsSpider): retries[parsed['url']] = {'document': retry_doc, 'attempts': previous.get('attempts', 0) + 1, 'next_at': (datetime.now(UTC) + timedelta(hours=1)).timestamp(), 'error': error} elif not incomplete: retries.pop(parsed['url'], None) + self.retire_exhausted() if reasons: parsed['extra']['selection_reasons'] = reasons parsed['extra']['content_scope'] = 'partial' if incomplete else 'article' @@ -676,6 +702,14 @@ class USOfficialSpider(NewsSpider): self.logger.warning('附件缺口：%s', response.url) yield from self.attach_or_finish(parsed, response.meta['lane']) + def retire_exhausted(self): + if self.name not in ('sec', 'doj', 'congress'): + return + for key in ('document_retries', 'attachment_retries', 'bill_retries'): + for entry in self.cursor.get(key, {}).values(): + if entry['attempts'] >= 3: + entry.setdefault('retired_at', datetime.now(UTC).isoformat()) + def complete(self, lane): doc = self.cursor[lane]['pending'].pop(0) self.cursor.get('document_retries', {}).pop(doc['url'], None) diff --git a/backend/src/alphapulse/models/calendar.py b/backend/src/alphapulse/models/calendar.py index 1de9002..88912de 100644 --- a/backend/src/alphapulse/models/calendar.py +++ b/backend/src/alphapulse/models/calendar.py @@ -27,6 +27,8 @@ class EconomicRelease(Base): first_received_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_checked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) next_check_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) + collection_failures: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + collection_retired_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) error: Mapped[str | None] = mapped_column(Text) # 按指标代码保存实际值、官方报告内前值和引用；数字以十进制字符串存储。 values: Mapped[dict] = mapped_column(JSONB, default=dict) diff --git a/backend/tests/test_calendar.py b/backend/tests/test_calendar.py index f16bf7c..b8af48f 100644 --- a/backend/tests/test_calendar.py +++ b/backend/tests/test_calendar.py @@ -178,7 +178,7 @@ def test_rolling_page_rollover_preserves_received_result_without_source_error(tr async def check(): first_received = datetime(2026, 9, 24, 14, tzinfo=UTC) values = {"cpi_yoy" if source == "bls" else "umich_sentiment": {"actual": "2.5", "previous": "2.4"}} - row = SimpleNamespace(id=1, source=source, family="cpi" if source == "bls" else "umich", + row = SimpleNamespace(id=1, collection_failures=0, collection_retired_at=None, source=source, family="cpi" if source == "bls" else "umich", scheduled_at=first_received, first_received_at=first_received, period="2026-09", stage="final" if source == "umich" else None, values=values.copy(), revision_no=3, error="旧的采集错误") @@ -275,7 +275,7 @@ def test_calendar_api_requires_auth_validates_range_and_exposes_unavailable_work asyncio.run(check()) -def test_worker_initializes_all_sources_and_drops_expired_month_status(trading): +def test_worker_initializes_all_sources_and_drops_expired_month_status(trading, schedule_session): import httpx from alphapulse.calendar.worker import CalendarWorker async def check(): @@ -313,7 +313,7 @@ def test_limited_result_preserves_response_and_defers_single_poll(trading, monke async def check(): now = datetime.now(UTC) - row = SimpleNamespace(id=1, source="eurostat", family="ea_hicp", scheduled_at=now, + row = SimpleNamespace(id=1, collection_failures=0, collection_retired_at=None, source="eurostat", family="ea_hicp", scheduled_at=now, first_received_at=None, release_url="https://ec.europa.eu/eurostat/product?code=2-16102026-ap") session = SimpleNamespace(scalar=AsyncMock(return_value=row), commit=AsyncMock()) class Context: @@ -344,7 +344,7 @@ def test_limited_result_preserves_response_and_defers_single_poll(trading, monke asyncio.run(check()) -def test_schedule_honors_server_pause_without_blocking_other_sources(trading): +def test_schedule_honors_server_pause_without_blocking_other_sources(trading, schedule_session): import httpx from alphapulse.calendar.worker import CalendarWorker async def check(): @@ -367,3 +367,14 @@ def test_schedule_honors_server_pause_without_blocking_other_sources(trading): await worker.http.aclose() await worker.bls.aclose() asyncio.run(check()) + + +@pytest.fixture +def schedule_session(trading, monkeypatch): + from alphapulse.calendar import worker as module + session = AsyncMock() + session.scalars.return_value = [] + session.get.return_value = None + session.__aenter__.return_value = session + monkeypatch.setattr(module, 'SessionLocal', lambda: session) + return session diff --git a/backend/tests/test_calendar_history.py b/backend/tests/test_calendar_history.py index 279330b..c0efe9a 100644 --- a/backend/tests/test_calendar_history.py +++ b/backend/tests/test_calendar_history.py @@ -220,7 +220,8 @@ def test_umich_poll_waits_for_final_instead_of_reusing_preliminary(trading): try: assert await worker.result_page(row) is None row.period = "2026-09" - with pytest.raises(ValueError, match="历史期次"): + worker.index = AsyncMock(return_value=(httpx.Response(200, text='
2026 Reports
August Final Results', request=httpx.Request('GET', 'https://data.sca.isr.umich.edu/reports.php?year=2026')), 2)) + with pytest.raises(ValueError, match="年度归档"): await worker.result_page(row) finally: await worker.http.aclose() diff --git a/backend/tests/test_calendar_recovery.py b/backend/tests/test_calendar_recovery.py index 67250e5..c05d179 100644 --- a/backend/tests/test_calendar_recovery.py +++ b/backend/tests/test_calendar_recovery.py @@ -203,3 +203,49 @@ def test_missing_archive_checks_latest_without_accepting_wrong_week(trading, mon await worker.http.aclose() await worker.bls.aclose() asyncio.run(run()) + + +def test_result_retirement_is_persistent_and_cooling_does_not_spend_budget(trading, monkeypatch): + from alphapulse.calendar import worker as module + + async def run(): + row = SimpleNamespace(source='umich', collection_failures=0, collection_retired_at=None, + scheduled_at=datetime(2026, 10, 9, tzinfo=UTC), first_received_at=None, + family='umich', error=None) + session = AsyncMock() + session.scalar.return_value = row + session.__aenter__.return_value = session + monkeypatch.setattr(module, 'SessionLocal', lambda: session) + worker = module.CalendarWorker() + try: + worker.result_page = AsyncMock(side_effect=module.SourceCooling('example', datetime.now(UTC) + timedelta(hours=2))) + await worker.poll_one(1) + assert row.collection_failures == 0 + assert row.next_check_at > datetime.now(UTC) + timedelta(hours=1) + worker.result_page.side_effect = ValueError('official archive missing') + for _ in range(3): + await worker.poll_one(1) + assert row.collection_failures == 3 and row.collection_retired_at is not None + calls = worker.result_page.await_count + await worker.poll_one(1) + assert worker.result_page.await_count == calls + finally: + await worker.http.aclose() + await worker.bls.aclose() + asyncio.run(run()) + + +def test_dol_lagging_index_uses_only_expected_official_archive(trading): + from alphapulse.calendar.worker import CalendarWorker + async def run(): + worker = CalendarWorker() + response = lambda url, body: httpx.Response(200, text=body, request=httpx.Request('GET', url)) + worker.index = AsyncMock(return_value=(response('https://oui.doleta.gov/unemploy/archive.asp', 'old'), 1)) + worker.fetch = AsyncMock(return_value=(response('https://oui.doleta.gov/press/2026/102926.pdf', 'PDF'), 2)) + try: + await worker.archive_page(SimpleNamespace(source='dol', scheduled_at=datetime(2026, 10, 29, 12, 30, tzinfo=UTC))) + worker.fetch.assert_awaited_once_with('dol', 'https://oui.doleta.gov/press/2026/102926.pdf') + finally: + await worker.http.aclose() + await worker.bls.aclose() + asyncio.run(run()) diff --git a/backend/tests/test_collection_repair.py b/backend/tests/test_collection_repair.py new file mode 100644 index 0000000..079ffd0 --- /dev/null +++ b/backend/tests/test_collection_repair.py @@ -0,0 +1,30 @@ +"""Repair initialization never renews an already initialized retry budget.""" +import asyncio +import copy +from types import SimpleNamespace +from unittest.mock import AsyncMock + + +def test_preview_and_repeated_execution_preserve_retry_evidence(trading, monkeypatch): + from alphapulse.crawler import repair_collection as module + original = {'document_retries': {'https://example.invalid/doc': { + 'document': {'url': 'https://example.invalid/doc'}, 'attempts': 3, + 'next_at': 1, 'error': 'HTTP 503'}}} + source = SimpleNamespace(code='doj', history_cursor=copy.deepcopy(original)) + session = AsyncMock() + session.scalars.return_value = [source] + session.__aenter__.return_value = session + monkeypatch.setattr(module, 'SessionLocal', lambda: session) + monkeypatch.setattr(module, 'engine', SimpleNamespace(dispose=AsyncMock())) + async def check(): + await module.run() + assert source.history_cursor == original + session.commit.assert_not_awaited() + await module.run(True) + entry = source.history_cursor['document_retries']['https://example.invalid/doc'] + assert entry['attempts'] == 0 and entry['repair_previous']['attempts'] == 3 + entry['attempts'] = 2 + result = await module.run(True) + assert result['sources'][0]['already_initialized'] + assert entry['attempts'] == 2 + asyncio.run(check()) diff --git a/backend/tests/test_eia.py b/backend/tests/test_eia.py index 8ba34e4..c7a6641 100644 --- a/backend/tests/test_eia.py +++ b/backend/tests/test_eia.py @@ -123,3 +123,10 @@ def test_history_preview_requires_matching_archive_date_without_writes(trading, assert result["announcements"].get("eia_stocks", 0) == int(valid) assert bool(result["errors"]) != valid save.assert_not_awaited() + + +def test_current_official_archive_csv_directory(): + url = 'https://www.eia.gov/petroleum/supply/weekly/archive/2026/2026_10_15/wpsr_2026_10_15.php' + assert eia.csv_link('CSV', url) == url.rsplit('/', 1)[0] + '/csv/table1.csv' + with pytest.raises(ValueError): + eia.csv_link('AB', url) diff --git a/backend/tests/test_us_official_news.py b/backend/tests/test_us_official_news.py index 3418ffa..06b1af3 100644 --- a/backend/tests/test_us_official_news.py +++ b/backend/tests/test_us_official_news.py @@ -1343,3 +1343,36 @@ def test_real_congress_api_fixture_replays_with_visible_source_gap(trading, tmp_ assert sum(a['status'] == 'complete' for a in attachments) == 14 assert sum(a['status'] == 'unavailable' for a in attachments) == 1 assert any(item['extra']['content_scope'] == 'official_action' for item in items) + + +def test_terminal_news_gap_survives_rediscovery(trading): + obj = spider(trading, 'doj') + url = 'https://www.justice.gov/opa/pr/example' + obj.cursor['attachment_retries'] = {url: {'attempts': 3, 'error': 'unsupported image'}} + obj.retire_exhausted() + assert obj.cursor['attachment_retries'][url]['retired_at'] + obj.complete = lambda lane: [] + assert list(obj.attach_or_finish({'url': url}, 'realtime')) == [] + + +def test_selected_news_requests_have_one_retry_owner(trading): + obj = spider(trading, 'doj') + req = obj.request_for(SOURCES['doj'].entries[0], obj.parse_listing, 'realtime') + assert req.meta['dont_retry'] is True + + +def test_doj_captioned_photo_is_not_a_document_gap(trading): + obj = spider(trading, 'doj') + obj.run.history_deadline = time.monotonic() - 1 + identifier = str(uuid.uuid4()) + doc = {'url': f'https://www.justice.gov/api/v1/press_releases/{identifier}.json', 'kind': 'doj'} + obj.cursor['history']['pending'] = [doc] + record = {'uuid': identifier, 'url': 'https://www.justice.gov/opa/pr/test', 'date': '1735732800', + 'title': 'Cryptocurrency exchange settlement', 'body': '

The cryptocurrency exchange pleaded guilty to violations of financial regulations.

', + 'image': 'award.png', + 'attachment': [{'file': {'id': '1470086'}, 'alt': 'Photo- Left to right: award recipients.'}]} + async def collect(): + return [v async for v in obj.parse_document(response(doc['url'], json.dumps({'metadata': {'responseInfo': {'status': 200}}, 'results': [record]}), lane='history', document=doc))] + item, = asyncio.run(collect()) + assert item['extra']['attachments'] == [] + assert not obj.cursor.get('attachment_retries') diff --git a/deploy/README.md b/deploy/README.md index d81bf8c..7a081ea 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -413,3 +413,21 @@ callback chain finished, including explicit partial packages; it does not mean full source quality passed. Exhausted candidate/request budgets remain incomplete. Mocked tests and real captured responses are separate evidence; neither proves complete 2025 coverage or authorizes production activation. + +### Bounded collection failures (November 2026) + +SEC, DOJ and Congress use one durable retry layer (three failed attempts, at least +one hour apart). EIA, DOL, Michigan and ADP retain the same budget per calendar +release in `collection_failures` / `collection_retired_at`. Cooling does not spend +an attempt. Retirement retains evidence and values; new releases are independent. +A source is disabled only after three failed discovery/schedule checks, not for +one document gap. Explicit official photo captions can exclude illustrations from +text extraction. Missing official dates are never replaced by update timestamps. + +Migration `db2345678901` is additive and must precede activation. With the crawler +stopped under the deployment lock, preview `python -m +alphapulse.crawler.repair_collection`; append `--execute` once to establish the +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. diff --git a/web/src/common/apis/calendar/type.ts b/web/src/common/apis/calendar/type.ts index 3bda3fb..a360ea4 100644 --- a/web/src/common/apis/calendar/type.ts +++ b/web/src/common/apis/calendar/type.ts @@ -12,7 +12,7 @@ export interface CalendarItem { unit: string | null importance: number scheduled_at: string - status: "scheduled" | "waiting" | "delayed" | "released" | "withdrawn" | "error" | "historical" | "unavailable" | "recorded" + status: "retired" | "scheduled" | "waiting" | "delayed" | "released" | "withdrawn" | "error" | "historical" | "unavailable" | "recorded" actual: string | null previous: string | null forecast: string | null diff --git a/web/src/common/apis/crawler/type.ts b/web/src/common/apis/crawler/type.ts index ddf4218..fa14c6d 100644 --- a/web/src/common/apis/crawler/type.ts +++ b/web/src/common/apis/crawler/type.ts @@ -37,6 +37,7 @@ export interface SourceInfo { } export interface CollectionRetry { + retired_at?: string document: { url: string, canonical?: string } attempts: number next_at: number @@ -51,6 +52,8 @@ export interface SourcesInfo { } export interface CalendarSourceInfo { + enabled: boolean + retired_count: number code: string name: string region_code: string diff --git a/web/src/pages/calendar/index.vue b/web/src/pages/calendar/index.vue index 7405cd5..0dae498 100644 --- a/web/src/pages/calendar/index.vue +++ b/web/src/pages/calendar/index.vue @@ -26,7 +26,7 @@ const detail = ref(null) const country = ref("") const impactLabels = { pending: "待公布", bullish: "利多", bearish: "利空", neutral: "影响较小", uncertain: "待判断" } const countries = computed(() => data.value?.countries ?? []) -const labels = { scheduled: "待公布", waiting: "获取中", delayed: "取数延迟", released: "已公布", withdrawn: "已撤回", error: "采集异常", historical: "仅有日程", unavailable: "官方未公布", recorded: "已收录" } +const labels = { retired: "已停止采集", scheduled: "待公布", waiting: "获取中", delayed: "取数延迟", released: "已公布", withdrawn: "已撤回", error: "采集异常", historical: "仅有日程", unavailable: "官方未公布", recorded: "已收录" } const formatted = (time: string) => dayjs(time).tz(zone).format("MM-DD HH:mm:ss") const groups = computed(() => { const result = new Map() diff --git a/web/src/pages/crawler/components/CalendarSourcePanel.vue b/web/src/pages/crawler/components/CalendarSourcePanel.vue index 3767b35..ac8e40b 100644 --- a/web/src/pages/crawler/components/CalendarSourcePanel.vue +++ b/web/src/pages/crawler/components/CalendarSourcePanel.vue @@ -16,12 +16,13 @@ const date = (at: string | null) => at ? formatTime(at) : "尚无记录" 日历
- - {{ !source.worker_alive ? "进程离线" : source.errors.length ? "采集异常" : source.schedule_success_at ? "同步正常" : "待同步" }} + + {{ !source.enabled ? "已停用" : !source.worker_alive ? "进程离线" : source.errors.length ? "采集异常" : source.schedule_success_at ? "同步正常" : "待同步" }}
+
已停止采集
{{ source.retired_count }} 场（历史资料保留）
进程心跳
{{ date(source.heartbeat_at) }}
日程最近检查
{{ date(source.schedule_checked_at) }}
日程最近成功
{{ date(source.schedule_success_at) }}
diff --git a/web/src/pages/crawler/components/SourcePanel.vue b/web/src/pages/crawler/components/SourcePanel.vue index 1e14eff..58eb523 100644 --- a/web/src/pages/crawler/components/SourcePanel.vue +++ b/web/src/pages/crawler/components/SourcePanel.vue @@ -144,21 +144,24 @@ onMounted(loadLatest) -
+
待核验法案 {{ source.collection_stats.unresolved_bills }}；待重试 {{ source.collection_stats.bill_retry_pending }}；重试已耗尽 {{ source.collection_stats.bill_retry_exhausted }}
-
+
附件待核验文档 {{ source.collection_stats.attachment_gaps }}；待重试 {{ source.collection_stats.attachment_retry_pending }}；重试已耗尽 {{ source.collection_stats.attachment_retry_exhausted }}，请核对原文
-
+
正文待核验 {{ source.collection_stats.document_gaps }}；待重试 {{ source.collection_stats.document_retry_pending }}；重试已耗尽 {{ source.collection_stats.document_retry_exhausted }}，请核对来源日志
+

+ 已停止采集 {{ source.collection_stats.retired_count }} 条，原因及原文保留在详情中。 +

缺口详情（{{ gaps.length }}）
-
{{ gap.label }}；已尝试 {{ gap.attempts }} 次；{{ gap.exhausted ? "重试已耗尽" : `下次重试不早于 ${formatTime(gap.nextAt)}` }}
+
{{ gap.label }}；已尝试 {{ gap.attempts }} 次；{{ gap.retired ? "已停止采集" : gap.exhausted ? "重试已耗尽" : `下次重试不早于 ${formatTime(gap.nextAt)}` }}
{{ gap.error }}
diff --git a/web/src/pages/crawler/sourceStatus.ts b/web/src/pages/crawler/sourceStatus.ts index b7eda29..02db4a8 100644 --- a/web/src/pages/crawler/sourceStatus.ts +++ b/web/src/pages/crawler/sourceStatus.ts @@ -15,7 +15,7 @@ export function sourceStatus(source: SourceInfo, restarting = false, now = Date. return { type: "primary", text: "采集中" } as const if (source.consecutive_failures) return { type: "danger", text: "失败待重试" } as const - if (source.collection_stats?.document_gaps || source.collection_stats?.attachment_gaps || source.collection_stats?.unresolved_bills) + if ((source.collection_stats?.active_gaps ?? (source.collection_stats?.document_gaps || source.collection_stats?.attachment_gaps || source.collection_stats?.unresolved_bills))) return { type: "warning", text: "存在采集缺口" } as const return { type: "info", text: "等待调度" } as const } @@ -49,6 +49,7 @@ export function collectionGaps(source: Pick) { url: originalUrl(retry.document.canonical ?? retry.document.url), error: retry.error || "历史记录未保存具体原因，请核对原文", attempts: retry.attempts, + retired: !!retry.retired_at, exhausted: retry.attempts >= 3, nextAt: new Date(retry.next_at * 1000).toISOString() }))) diff --git a/web/tests/crawler.test.ts b/web/tests/crawler.test.ts index a91d87d..9b4c192 100644 --- a/web/tests/crawler.test.ts +++ b/web/tests/crawler.test.ts @@ -102,3 +102,32 @@ test("gap links preserve canonical pages and do not rewrite unrelated URLs", () assert.equal(collectionGaps(source)[0].url, url) } }) + +test("retired gaps retain evidence without an active warning", () => { + const source = { + enabled: true, + consecutive_failures: 0, + code: "doj", + name: "DOJ", + region_code: "US", + list_url: "https://www.justice.gov/", + interval_seconds: 60, + backfill_hours: 72, + last_success_at: null, + last_error: null, + paused_until: null, + supports_history: true, + history_target: null, + history_position: null, + history_done: false, + today_new: 0, + today_revised: 0, + total_items: 0, + process: { state: "running", pid: 1, started_at: null, next_run_at: "2026-11-05T00:00:00Z" }, + collection_stats: { active_gaps: 0, document_gaps: 1, retired_count: 1 }, + history_cursor: { document_retries: { old: { document: { url: "https://www.justice.gov/old" }, attempts: 3, next_at: 1, error: "HTTP 503", retired_at: "2026-11-05T00:00:00Z" } } } + } as SourceInfo + assert.equal(sourceStatus(source).text, "等待调度") + assert.equal(collectionGaps(source)[0].retired, true) + assert.equal(collectionGaps(source)[0].error, "HTTP 503") +})