From d5d350b7ec0eb61ee2c2f460cce5ff9907cef49c Mon Sep 17 00:00:00 2001 From: eezd Date: Thu, 5 Nov 2026 12:44:42 +0800 Subject: [PATCH] Expose persistent collection gaps and public Congress links Retain bill and attachment failure reasons and derive retry counts from saved cursors. Show gap details and retry exhaustion in the collection console while preserving historical records and retry limits. Convert Congress API bill links to public pages and cover diagnostics, mixed dated versions, persistent counts and legacy records with regression tests. --- backend/src/alphapulse/api/crawler.py | 5 ++ .../alphapulse/crawler/spiders/us_official.py | 10 ++- backend/tests/test_us_official_news.py | 33 +++++++-- deploy/README.md | 7 ++ web/src/common/apis/crawler/type.ts | 15 +++- .../pages/crawler/components/SourcePanel.vue | 17 ++++- web/src/pages/crawler/sourceStatus.ts | 38 +++++++++- web/tests/crawler.test.ts | 71 ++++++++++++++++++- 8 files changed, 182 insertions(+), 14 deletions(-) diff --git a/backend/src/alphapulse/api/crawler.py b/backend/src/alphapulse/api/crawler.py index 7b319d2..90eeb73 100644 --- a/backend/src/alphapulse/api/crawler.py +++ b/backend/src/alphapulse/api/crawler.py @@ -194,6 +194,8 @@ async def list_sources(session: Annotated[AsyncSession, Depends(get_session)]) - pending = sum(value['attempts'] < 3 for value in retries.values()) documents = (source.history_cursor or {}).get('document_retries', {}) document_pending = sum(value['attempts'] < 3 for value in documents.values()) + bills = (source.history_cursor or {}).get('bill_retries', {}) + bill_pending = sum(value['attempts'] < 3 for value in bills.values()) attachment_stats[source.code] = { 'attachment_gaps': len(retries), 'attachment_retry_pending': pending, @@ -201,6 +203,9 @@ async def list_sources(session: Annotated[AsyncSession, Depends(get_session)]) - 'document_gaps': len(documents), 'document_retry_pending': document_pending, 'document_retry_exhausted': len(documents) - document_pending, + 'unresolved_bills': len(bills), + 'bill_retry_pending': bill_pending, + 'bill_retry_exhausted': len(bills) - bill_pending, } return success(SourcesOut( manager_alive=bool(statuses), diff --git a/backend/src/alphapulse/crawler/spiders/us_official.py b/backend/src/alphapulse/crawler/spiders/us_official.py index 5b79ad4..064fffd 100644 --- a/backend/src/alphapulse/crawler/spiders/us_official.py +++ b/backend/src/alphapulse/crawler/spiders/us_official.py @@ -493,11 +493,13 @@ class USOfficialSpider(NewsSpider): for version in data['textVersions']: if not version.get('date'): self.cursor[lane]['bill']['missing_material'] = True + self.cursor[lane]['bill'].setdefault('material_errors', []).append(f"{version.get('type', '正文版本')} 缺少官方日期") continue formats = version.get('formats', []) chosen = next((f for f in formats if f['type'] == 'Formatted Text'), None) or next((f for f in formats if f['type'] == 'Formatted XML'), None) or next((f for f in formats if f['type'] == 'PDF'), None) if not chosen: self.cursor[lane]['bill']['missing_material'] = True + self.cursor[lane]['bill'].setdefault('material_errors', []).append(f"{version.get('type', '正文版本')} 缺少支持的正文格式") if chosen: children.append({'url': chosen['url'], 'kind': 'text', 'title': doc['title'] + ' — ' + version['type'], 'at': version['date'], 'bill': doc['bill']}) next_url = data.get('pagination', {}).get('next') @@ -613,7 +615,10 @@ class USOfficialSpider(NewsSpider): retries = self.cursor.setdefault('bill_retries', {}) if not context['bodies'] or context.get('missing_material'): previous = retries.get(context['url'], {}) - retries[context['url']] = {'document': {'url': context['url'], 'kind': 'bill'}, 'version': context['version'], 'attempts': previous.get('attempts', 0) + 1, 'next_at': (datetime.now(UTC) + timedelta(hours=1)).timestamp()} + errors = list(context.get('material_errors', [])) + if not context['bodies']: + errors.append('没有可用的官方正文') + retries[context['url']] = {'document': {'url': context['url'], 'kind': 'bill'}, 'version': context['version'], 'attempts': previous.get('attempts', 0) + 1, 'next_at': (datetime.now(UTC) + timedelta(hours=1)).timestamp(), 'error': '；'.join(dict.fromkeys(errors)) or '官方正文材料不完整'} else: retries.pop(context['url'], None) if not context['reasons']: @@ -637,7 +642,8 @@ class USOfficialSpider(NewsSpider): retries = self.cursor.setdefault('attachment_retries', {}) if incomplete and retry_doc: previous = retries.get(parsed['url'], {}) - retries[parsed['url']] = {'document': retry_doc, 'attempts': previous.get('attempts', 0) + 1, 'next_at': (datetime.now(UTC) + timedelta(hours=1)).timestamp()} + error = '；'.join(f"{a['url']}：{a.get('error') or '附件尚未完成文字提取'}" for a in parsed['extra']['attachments'] if a['status'] != 'complete') + 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) if reasons: diff --git a/backend/tests/test_us_official_news.py b/backend/tests/test_us_official_news.py index e504ce1..3418ffa 100644 --- a/backend/tests/test_us_official_news.py +++ b/backend/tests/test_us_official_news.py @@ -556,6 +556,8 @@ def test_image_attachment_preserves_unknown_text_gap(trading, monkeypatch, conte attachment = parsed['extra']['attachments'][0] assert attachment['status'] == 'unavailable' assert '图片' in attachment['error'] + assert attachment['error'] in obj.cursor['attachment_retries'][doc['url']]['error'] + assert url in obj.cursor['attachment_retries'][doc['url']]['error'] assert attachment['raw_response_id'] == 13 assert obj.cursor['attachment_retries'][doc['url']]['attempts'] == 1 @@ -566,7 +568,7 @@ def test_source_api_keeps_attachment_gaps_visible_across_rounds(trading, monkeyp from alphapulse.models import Source source = Source(code='doj', name='DOJ', list_url=SOURCES['doj'].entries[0], enabled=False, interval_seconds=300, backfill_hours=72, - consecutive_failures=0, last_success_at=None, last_error=None, paused_until=None, history_cursor={'attachment_retries': {'a': {'attempts': 1}, 'b': {'attempts': 2}, 'c': {'attempts': 3}}, 'document_retries': {'d': {'attempts': 1}, 'e': {'attempts': 3}}}, + consecutive_failures=0, last_success_at=None, last_error=None, paused_until=None, history_cursor={'attachment_retries': {'a': {'attempts': 1}, 'b': {'attempts': 2}, 'c': {'attempts': 3}}, 'document_retries': {'d': {'attempts': 1}, 'e': {'attempts': 3}}, 'bill_retries': {'f': {'attempts': 1}, 'g': {'attempts': 3}}}, history_target=None, history_position=None, history_done=False) latest = SimpleNamespace(source_code='doj', status='success', stats={'discovered': 2, 'attachment_gaps': 999}) @@ -581,7 +583,7 @@ def test_source_api_keeps_attachment_gaps_visible_across_rounds(trading, monkeyp monkeypatch.setattr(crawler.redis, 'hgetall', AsyncMock(return_value={})) monkeypatch.setattr(crawler, 'calendar_sources', AsyncMock(return_value=[])) - for stats in ({'discovered': 2, 'attachment_gaps': 999}, {}): + for stats in ({'discovered': 2, 'attachment_gaps': 999, 'unresolved_bills': 999}, {}): latest.stats = stats result = asyncio.run(crawler.list_sources(Session())).data.sources[0] assert result.collection_stats['attachment_gaps'] == 3 @@ -590,6 +592,14 @@ def test_source_api_keeps_attachment_gaps_visible_across_rounds(trading, monkeyp assert result.collection_stats['document_gaps'] == 2 assert result.collection_stats['document_retry_pending'] == 1 assert result.collection_stats['document_retry_exhausted'] == 1 + assert result.collection_stats['unresolved_bills'] == 2 + assert result.collection_stats['bill_retry_pending'] == 1 + assert result.collection_stats['bill_retry_exhausted'] == 1 + latest.status = 'running' + running = asyncio.run(crawler.list_sources(Session())).data.sources[0] + assert running.collection_stats['unresolved_bills'] == 2 + assert running.collection_stats['round_running'] == 1 + latest.status = 'success' assert result.collection_stats['round_running'] == 0 source.history_cursor['attachment_retries'].pop('a') result = asyncio.run(crawler.list_sources(Session())).data.sources[0] @@ -601,6 +611,9 @@ def test_source_api_keeps_attachment_gaps_visible_across_rounds(trading, monkeyp assert result.collection_stats['attachment_gaps'] == 0 assert result.collection_stats['attachment_retry_exhausted'] == 0 assert result.collection_stats['document_gaps'] == 0 + assert result.collection_stats['unresolved_bills'] == 0 + assert result.collection_stats['bill_retry_pending'] == 0 + assert result.collection_stats['bill_retry_exhausted'] == 0 latest.status = 'running' result = asyncio.run(crawler.list_sources(Session())).data.sources[0] assert result.collection_stats['round_running'] == 1 @@ -979,7 +992,8 @@ def test_fragment_is_valid_without_page_heading(trading): assert item['extra']['content_scope'] == 'article' -def test_bill_full_text_can_supply_relevance_and_resume(trading, monkeypatch): +@pytest.mark.parametrize('undated', [False, True]) +def test_bill_full_text_can_supply_relevance_and_resume(trading, monkeypatch, undated): from alphapulse.crawler.spiders import us_official monkeypatch.setattr(us_official, 'get_settings', lambda: SimpleNamespace(congress_api_key='test', crawler_contact_email='test@example.invalid')) obj = spider(trading, 'congress') @@ -991,14 +1005,20 @@ def test_bill_full_text_can_supply_relevance_and_resume(trading, monkeypatch): asyncio.run(parse(json.dumps({'bill': {'title': 'GENIUS Act', 'updateDate': '2026-11-04'}}))) asyncio.run(parse(json.dumps({'subjects': {'legislativeSubjects': []}}))) asyncio.run(parse(json.dumps({'summaries': []}))) - asyncio.run(parse(json.dumps({'textVersions': [{'type': 'Enrolled', 'date': '2026-11-04', 'formats': [{'type': 'Formatted Text', 'url': 'https://www.congress.gov/text'}]}]}))) + versions = [{'type': 'Enrolled', 'date': '2026-11-04', 'formats': [{'type': 'Formatted Text', 'url': 'https://www.congress.gov/text'}]}] + if undated: + versions.insert(0, {'type': 'Enrolled Bill', 'date': None, 'formats': []}) + asyncio.run(parse(json.dumps({'textVersions': versions}))) obj.cursor = json.loads(json.dumps(obj.cursor)) output = asyncio.run(parse('

The bill establishes the regulatory framework for payment stablecoins and digital assets.

')) assert output[0]['title'] == 'GENIUS Act — Enrolled' assert '/actions?' in output[1].url output = asyncio.run(parse(json.dumps({'actions': [{'actionDate': '2026-11-04', 'text': 'Became law'}]}))) assert output[0]['extra']['content_scope'] == 'official_action' - assert not obj.cursor.get('bill_retries') + if undated: + assert obj.cursor['bill_retries'][base]['error'] == 'Enrolled Bill 缺少官方日期' + else: + assert not obj.cursor.get('bill_retries') def test_missing_bill_material_is_a_retryable_gap(trading, monkeypatch): @@ -1019,6 +1039,7 @@ def test_missing_bill_material_is_a_retryable_gap(trading, monkeypatch): assert obj.cursor['bill_retries'][base]['attempts'] == 1 assert obj.run.stats['unresolved_bills'] == 1 assert not obj.run.stats.get('excluded') + assert obj.cursor['bill_retries'][base]['error'] == '没有可用的官方正文' assert (before + timedelta(hours=1)).timestamp() <= obj.cursor['bill_retries'][base]['next_at'] <= (after + timedelta(hours=1)).timestamp() @@ -1245,6 +1266,7 @@ def test_bill_undated_text_is_visible_retry_gap(trading): obj.finish_bill('history') assert obj.cursor['bill_retries'][base]['attempts'] == 1 assert obj.run.stats['unresolved_bills'] == 1 + assert '缺少官方日期' in obj.cursor['bill_retries'][base]['error'] def test_public_api_and_pdf_fixtures_replay_without_io(trading): @@ -1296,6 +1318,7 @@ def test_dated_but_unavailable_bill_text_remains_a_gap(trading): asyncio.run(collect()) obj.finish_bill('history') assert obj.run.stats['unresolved_bills'] == 1 + assert '缺少支持的正文格式' in obj.cursor['bill_retries'][base]['error'] diff --git a/deploy/README.md b/deploy/README.md index e5c052a..d81bf8c 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -304,6 +304,13 @@ format is PNG, JPEG or GIF. They do not establish completeness or irrelevance. Relevant documents remain `partial`; documents whose relevance is still unknown are kept in bounded attachment retries rather than inserted as unrelated news. Congress meeting attachment failures retain the same retry evidence as articles. +Bill and attachment retry records retain an optional `error` explaining missing +dates, unavailable text formats or extraction failures. The console shows these +reasons, original links, attempt counts and pending/exhausted state even while a +new round runs. Historical records without a reason remain visible as such; this +change neither resets their retry quota nor rewrites historical records. +`unresolved_bills`, `bill_retry_pending` and `bill_retry_exhausted` are derived from +the persisted bill retry cursor, independently of the latest round's statistics. The source API derives `attachment_gaps`, `attachment_retry_pending` and `attachment_retry_exhausted` from the persisted cursor, independently of the latest round's statistics. The console keeps these document counts visible across rounds; diff --git a/web/src/common/apis/crawler/type.ts b/web/src/common/apis/crawler/type.ts index 96e4181..ddf4218 100644 --- a/web/src/common/apis/crawler/type.ts +++ b/web/src/common/apis/crawler/type.ts @@ -19,7 +19,13 @@ export interface SourceInfo { paused_until: string | null collection_stats?: Record configuration_error?: string | null - history_cursor?: { history?: { scanned?: boolean }, realtime?: { entry: number } } + history_cursor?: { + history?: { scanned?: boolean } + realtime?: { entry: number } + document_retries?: Record + attachment_retries?: Record + bill_retries?: Record + } supports_history: boolean history_target: string | null history_position: string | null @@ -30,6 +36,13 @@ export interface SourceInfo { total_items: number } +export interface CollectionRetry { + document: { url: string, canonical?: string } + attempts: number + next_at: number + error?: string +} + export interface SourcesInfo { manager_alive: boolean db_size_bytes: number diff --git a/web/src/pages/crawler/components/SourcePanel.vue b/web/src/pages/crawler/components/SourcePanel.vue index 73b9109..1e14eff 100644 --- a/web/src/pages/crawler/components/SourcePanel.vue +++ b/web/src/pages/crawler/components/SourcePanel.vue @@ -7,7 +7,7 @@ import { useSessionScope } from "@@/composables/useSessionScope" import { formatTime } from "@@/utils/text" import { CircleCheck, RefreshRight } from "@element-plus/icons-vue" import dayjs from "dayjs" -import { sourceStatus } from "../sourceStatus" +import { collectionGaps, sourceStatus } from "../sourceStatus" /** logs 只用来发现「本轮完成」以刷新最新入库；alert 为连续失败达到告警阈值 */ const props = defineProps<{ @@ -89,6 +89,7 @@ async function cancelHistory() { } const statusTag = computed(() => sourceStatus(props.source, props.restarting)) +const gaps = computed(() => collectionGaps(props.source)) onMounted(loadLatest) @@ -144,7 +145,7 @@ onMounted(loadLatest)
- 待核验法案 {{ source.collection_stats.unresolved_bills }}；官方资料尚不充分 + 待核验法案 {{ 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 }}，请核对原文 @@ -152,6 +153,18 @@ onMounted(loadLatest)
正文待核验 {{ source.collection_stats.document_gaps }}；待重试 {{ source.collection_stats.document_retry_pending }}；重试已耗尽 {{ source.collection_stats.document_retry_exhausted }}，请核对来源日志
+
+ + 缺口详情（{{ gaps.length }}） + +
+
{{ gap.label }}；已尝试 {{ gap.attempts }} 次；{{ gap.exhausted ? "重试已耗尽" : `下次重试不早于 ${formatTime(gap.nextAt)}` }}
+
+ {{ gap.error }} +
+ 查看原文：{{ gap.url }} +
+
本轮发现 {{ source.collection_stats.discovered }}；筛除 {{ source.collection_stats.excluded ?? 0 }}；不完整包 {{ source.collection_stats.partial ?? 0 }}
diff --git a/web/src/pages/crawler/sourceStatus.ts b/web/src/pages/crawler/sourceStatus.ts index f595fce..b7eda29 100644 --- a/web/src/pages/crawler/sourceStatus.ts +++ b/web/src/pages/crawler/sourceStatus.ts @@ -15,7 +15,41 @@ 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) - return { type: "warning", text: "存在文档缺口" } as const + if (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 } + +function originalUrl(url: string) { + const bill = /^https:\/\/api\.congress\.gov\/v3\/bill\/(\d+)\/([a-z]+)\/(\d+)\/?(?:\?[^#]*)?(?:#.*)?$/i.exec(url) + const billTypes: Record = { + hr: "house-bill", + s: "senate-bill", + hjres: "house-joint-resolution", + sjres: "senate-joint-resolution", + hconres: "house-concurrent-resolution", + sconres: "senate-concurrent-resolution", + hres: "house-resolution", + sres: "senate-resolution" + } + if (bill && billTypes[bill[2].toLowerCase()]) + return `https://www.congress.gov/bill/${bill[1]}th-congress/${billTypes[bill[2].toLowerCase()]}/${bill[3]}` + return url +} + +export function collectionGaps(source: Pick) { + const groups = [ + ["document_retries", "正文"], + ["attachment_retries", "附件"], + ["bill_retries", "法案"] + ] as const + return groups.flatMap(([key, label]) => Object.entries(source.history_cursor?.[key] ?? {}).map(([url, retry]) => ({ + key: `${key}:${url}`, + label, + url: originalUrl(retry.document.canonical ?? retry.document.url), + error: retry.error || "历史记录未保存具体原因，请核对原文", + attempts: retry.attempts, + 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 341cda9..a91d87d 100644 --- a/web/tests/crawler.test.ts +++ b/web/tests/crawler.test.ts @@ -1,7 +1,7 @@ import type { SourceInfo } from "../src/common/apis/crawler/type.ts" import assert from "node:assert/strict" import { test } from "node:test" -import { sourceStatus } from "../src/pages/crawler/sourceStatus.ts" +import { collectionGaps, sourceStatus } from "../src/pages/crawler/sourceStatus.ts" test("crawler status distinguishes process health from collection results", () => { const source: SourceInfo = { @@ -30,8 +30,75 @@ test("crawler status distinguishes process health from collection results", () = assert.equal(sourceStatus({ ...source, collection_stats: { round_running: 1 } }).text, "采集中") assert.equal(sourceStatus({ ...source, consecutive_failures: 2 }).text, "失败待重试") assert.equal(sourceStatus({ ...source, paused_until: "2026-11-06T00:00:00Z" }, false, Date.parse("2026-11-05")).text, "失败退避中") - assert.equal(sourceStatus({ ...source, collection_stats: { document_gaps: 2 } }).text, "存在文档缺口") + assert.equal(sourceStatus({ ...source, collection_stats: { document_gaps: 2 } }).text, "存在采集缺口") + const gapStats: Record[] = [{ attachment_gaps: 1 }, { unresolved_bills: 8 }] + for (const collection_stats of gapStats) { + assert.equal(sourceStatus({ ...source, collection_stats }).text, "存在采集缺口") + assert.equal(sourceStatus({ ...source, collection_stats: { ...collection_stats, round_running: 1 } }).text, "采集中") + } assert.equal(sourceStatus({ ...source, enabled: false, consecutive_failures: 3 }).text, "已停用") assert.equal(sourceStatus({ ...source, configuration_error: "待配置 CRAWLER_CONTACT_EMAIL" }).type, "warning") assert.equal(sourceStatus({ ...source, process: null }).text, "采集进程未运行") }) + +test("gap details retain reasons, exhausted state and legacy records while collecting", () => { + const source = { + collection_stats: { round_running: 1 }, + history_cursor: { + bill_retries: { + "https://api.congress.gov/v3/bill/119/hr/1": { document: { url: "https://api.congress.gov/v3/bill/119/hr/1" }, attempts: 3, next_at: 1 } + }, + attachment_retries: { + photo: { document: { url: "https://www.justice.gov/api/release", canonical: "https://www.justice.gov/release" }, attempts: 2, next_at: 1793854270, error: "官方图片附件尚未提取文字，请核对原文" } + }, + document_retries: { + body: { document: { url: "https://www.justice.gov/api/body" }, attempts: 1, next_at: 1793853098, error: "HTTP 503" } + } + } + } satisfies Partial + const gaps = collectionGaps(source) + assert.equal(gaps.length, 3) + assert.equal(gaps[0].error, "HTTP 503") + assert.equal(gaps[0].nextAt, new Date(1793853098 * 1000).toISOString()) + assert.equal(gaps[1].url, "https://www.justice.gov/release") + assert.match(gaps[1].error, /图片/) + assert.equal(gaps[1].exhausted, false) + assert.equal(gaps[2].exhausted, true) + assert.equal(gaps[2].url, "https://www.congress.gov/bill/119th-congress/house-bill/1") + assert.match(gaps[2].error, /历史记录未保存具体原因/) + assert.equal(source.history_cursor.bill_retries["https://api.congress.gov/v3/bill/119/hr/1"].attempts, 3) + assert.deepEqual(collectionGaps({}), []) +}) + +test("Congress gap links use public bill pages for all supported bill types", () => { + const types = { + hr: "house-bill", + s: "senate-bill", + hjres: "house-joint-resolution", + sjres: "senate-joint-resolution", + hconres: "house-concurrent-resolution", + sconres: "senate-concurrent-resolution", + hres: "house-resolution", + sres: "senate-resolution" + } + for (const [kind, label] of Object.entries(types)) { + const url = `https://api.congress.gov/v3/bill/119/${kind}/1071?format=json` + for (const group of ["bill_retries", "attachment_retries", "document_retries"] as const) { + const source = { history_cursor: { [group]: { [url]: { document: { url }, attempts: 3, next_at: 1 } } } } + assert.equal(collectionGaps(source)[0].url, `https://www.congress.gov/bill/119th-congress/${label}/1071`) + assert.equal(source.history_cursor[group][url].document.url, url) + } + } +}) + +test("gap links preserve canonical pages and do not rewrite unrelated URLs", () => { + for (const url of [ + "https://www.congress.gov/bill/119th-congress/house-bill/1", + "https://www.justice.gov/release", + "https://api.congress.gov/v3/committee-meeting/119/house/1", + "https://api.congress.gov.example.com/v3/bill/119/hr/1" + ]) { + const source = { history_cursor: { bill_retries: { old: { document: { url: "https://api.congress.gov/v3/bill/119/hr/1", canonical: url }, attempts: 3, next_at: 1 } } } } + assert.equal(collectionGaps(source)[0].url, url) + } +})