Repair official collection and retire exhausted material tasks

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

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,

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]

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
 
 

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,

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

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)

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)

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)

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

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='<div id="report_heading"><span class="title">2026 Reports</span></div><a href="fetchdoc.php?docid=1">August Final Results</a>', 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()

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', '<a href="/press/2025/010225.pdf">old</a>'), 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())

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

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('<a href="csv/table1.csv">CSV</a>', url) == url.rsplit('/', 1)[0] + '/csv/table1.csv'
+    with pytest.raises(ValueError):
+        eia.csv_link('<a href="csv/table1.csv">A</a><a href="download/table1.csv">B</a>', url)

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': '<p>The cryptocurrency exchange pleaded guilty to violations of financial regulations.</p>',
+              'image': '<a href="/usao-sd/media/1470086">award.png</a>',
+              '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')

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.

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

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

web/src/pages/calendar/index.vue

@@ -26,7 +26,7 @@ const detail = ref<CalendarItem | null>(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<string, CalendarItem[]>()

web/src/pages/crawler/components/CalendarSourcePanel.vue

@@ -16,12 +16,13 @@ const date = (at: string | null) => at ? formatTime(at) : "尚无记录"
           日历
         </el-tag>
         <div class="flex-1" />
-        <el-tag :type="!source.worker_alive ? 'danger' : source.errors.length ? 'warning' : source.schedule_success_at ? 'success' : 'info'" size="small">
+        <el-tag :type="!source.enabled ? 'info' : !source.worker_alive ? 'danger' : source.errors.length ? 'warning' : source.schedule_success_at ? 'success' : 'info'" size="small">
-          {{ !source.worker_alive ? "进程离线" : source.errors.length ? "采集异常" : source.schedule_success_at ? "同步正常" : "待同步" }}
+          {{ !source.enabled ? "已停用" : !source.worker_alive ? "进程离线" : source.errors.length ? "采集异常" : source.schedule_success_at ? "同步正常" : "待同步" }}
         </el-tag>
       </div>
     </template>
     <dl class="grid grid-cols-[auto_1fr] gap-x-3 gap-y-2 text-xs text-slate-500">
+      <dt>已停止采集</dt><dd>{{ source.retired_count }} 场（历史资料保留）</dd>
       <dt>进程心跳</dt><dd>{{ date(source.heartbeat_at) }}</dd>
       <dt>日程最近检查</dt><dd>{{ date(source.schedule_checked_at) }}</dd>
       <dt>日程最近成功</dt><dd>{{ date(source.schedule_success_at) }}</dd>

web/src/pages/crawler/components/SourcePanel.vue

@@ -144,21 +144,24 @@ onMounted(loadLatest)
         </div>
       </div>
     </div>
-    <div v-if="source.collection_stats?.unresolved_bills" class="text-xs text-amber-600">
+    <div v-if="source.collection_stats?.active_gaps !== 0 && source.collection_stats?.unresolved_bills" class="text-xs text-amber-600">
       待核验法案 {{ source.collection_stats.unresolved_bills }}；待重试 {{ source.collection_stats.bill_retry_pending }}；重试已耗尽 {{ source.collection_stats.bill_retry_exhausted }}
     </div>
-    <div v-if="source.collection_stats?.attachment_gaps" class="text-xs text-amber-600">
+    <div v-if="source.collection_stats?.active_gaps !== 0 && source.collection_stats?.attachment_gaps" class="text-xs text-amber-600">
       附件待核验文档 {{ source.collection_stats.attachment_gaps }}；待重试 {{ source.collection_stats.attachment_retry_pending }}；重试已耗尽 {{ source.collection_stats.attachment_retry_exhausted }}，请核对原文
     </div>
-    <div v-if="source.collection_stats?.document_gaps" class="text-xs text-amber-600">
+    <div v-if="source.collection_stats?.active_gaps !== 0 && source.collection_stats?.document_gaps" class="text-xs text-amber-600">
       正文待核验 {{ source.collection_stats.document_gaps }}；待重试 {{ source.collection_stats.document_retry_pending }}；重试已耗尽 {{ source.collection_stats.document_retry_exhausted }}，请核对来源日志
     </div>
+    <p v-if="source.collection_stats?.retired_count" class="text-xs text-slate-500">
+      已停止采集 {{ source.collection_stats.retired_count }} 条，原因及原文保留在详情中。
+    </p>
     <details v-if="gaps.length" class="text-xs text-amber-700">
       <summary class="cursor-pointer">
         缺口详情（{{ gaps.length }}）
       </summary>
       <div v-for="gap in gaps" :key="gap.key" class="mt-2 border border-amber-200 rounded border-solid p-2">
-        <div>{{ gap.label }}；已尝试 {{ gap.attempts }} 次；{{ gap.exhausted ? "重试已耗尽" : `下次重试不早于 ${formatTime(gap.nextAt)}` }}</div>
+        <div>{{ gap.label }}；已尝试 {{ gap.attempts }} 次；{{ gap.retired ? "已停止采集" : gap.exhausted ? "重试已耗尽" : `下次重试不早于 ${formatTime(gap.nextAt)}` }}</div>
         <div class="my-1 whitespace-pre-wrap break-words">
           {{ gap.error }}
         </div>

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<SourceInfo, "history_cursor">) {
     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()
   })))

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")
+})