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