diff --git a/backend/src/alphapulse/api/crawler.py b/backend/src/alphapulse/api/crawler.py
index b38222a..d9a9017 100644
--- a/backend/src/alphapulse/api/crawler.py
+++ b/backend/src/alphapulse/api/crawler.py
@@ -18,6 +18,7 @@ from alphapulse.core.db import get_session
from alphapulse.core.redis import KEY_PREFIX, redis
from alphapulse.core.official_sources import REGIONS
from alphapulse.calendar.catalog import FAMILIES
+from alphapulse.crawler.content_policy import ignored_retry
from alphapulse.crawler.events import CONTROL_CHANNEL, HISTORY_SOURCES, LOG_CHANNEL, STATUS_KEY
from alphapulse.crawler.spiders.us_official import configuration_error
from alphapulse.models import CrawlLog, CrawlRun, EconomicRelease, NewsItem, Source
@@ -196,13 +197,16 @@ async def list_sources(session: Annotated[AsyncSession, Depends(get_session)]) -
db_size = await session.scalar(select(func.pg_database_size(func.current_database())))
attachment_stats = {}
for source in sources:
- retries = (source.history_cursor or {}).get('attachment_retries', {})
+ groups = {key: {url: value for url, value in (source.history_cursor or {}).get(key, {}).items() if not ignored_retry(key, value)} for key in ('attachment_retries', 'document_retries', 'bill_retries')}
+ ignored = sum(ignored_retry(key, value) for key in groups for value in (source.history_cursor or {}).get(key, {}).values())
+ retries = groups['attachment_retries']
pending = sum(value['attempts'] < 3 for value in retries.values())
- documents = (source.history_cursor or {}).get('document_retries', {})
+ documents = groups['document_retries']
document_pending = sum(value['attempts'] < 3 for value in documents.values())
- bills = (source.history_cursor or {}).get('bill_retries', {})
+ bills = groups['bill_retries']
bill_pending = sum(value['attempts'] < 3 for value in bills.values())
attachment_stats[source.code] = {
+ 'policy_ignored': ignored,
'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),
diff --git a/backend/src/alphapulse/core/public_http.py b/backend/src/alphapulse/core/public_http.py
index d2c5c82..7f95247 100644
--- a/backend/src/alphapulse/core/public_http.py
+++ b/backend/src/alphapulse/core/public_http.py
@@ -19,6 +19,9 @@ async def public_address(url):
async def fetch_public_response(url):
async with httpx.AsyncClient(timeout=20, trust_env=False) as client:
for _ in range(4):
+ from alphapulse.crawler.content_policy import file_url
+ if file_url(url):
+ raise ValueError('订阅地址不支持 PDF 或附件')
parsed, address = await public_address(url)
# Pin the checked address while retaining TLS hostname verification.
target = httpx.URL(url).copy_with(host=address)
@@ -27,11 +30,18 @@ async def fetch_public_response(url):
location = response.headers.get('location')
if not location: raise ValueError('重定向缺少地址')
url = str(httpx.URL(url).join(location)); continue
- chunks, size = [], 0
+ from scrapy.http import Headers
+ from alphapulse.crawler.content_policy import binary_headers
+ if binary_headers(Headers(response.headers.multi_items())):
+ raise ValueError('订阅返回不支持的文件类型')
+ chunks, size, prefix = [], 0, b''
async for chunk in response.aiter_bytes():
size += len(chunk)
if size > 2*1024*1024: raise ValueError('响应超过2MiB限制')
chunks.append(chunk)
+ prefix = (prefix + chunk)[:8]
+ if prefix.lstrip().startswith(b'%PDF'):
+ raise ValueError('订阅返回 PDF 文件')
body = b''.join(chunks)
# Bytes are already decompressed; Scrapy must not decode them again.
headers = [(k, v) for k, v in response.headers.multi_items()
diff --git a/backend/src/alphapulse/core/us_news.py b/backend/src/alphapulse/core/us_news.py
index bcac5d4..9d17067 100644
--- a/backend/src/alphapulse/core/us_news.py
+++ b/backend/src/alphapulse/core/us_news.py
@@ -32,7 +32,7 @@ SOURCES = {
'doj': OfficialSource('DOJ·加密机构执法', ('https://www.justice.gov/api/v1/press_releases.json?pagesize=50&sort=date&direction=DESC&fields=uuid,date',), r'/(?:opa|usao[^/]*|archives/opa)/(?:pr|speech)/[^/]+', '.field--name-body, .field--name-field-pr-body', ('www.justice.gov',)),
'fed': OfficialSource('美联储·政策与讲话', tuple(f'https://www.federalreserve.gov/newsevents/{kind}/{year}-{suffix}.htm' for year in range(datetime.now(UTC).year, 2024, -1) for kind, suffix in (('pressreleases', 'press'), ('speech', 'speeches'), ('testimony', 'testimony'))) + ('https://www.federalreserve.gov/monetarypolicy/fomccalendars.htm', 'https://www.federalreserve.gov/releases/h41/releaseDates.json'), r'/(?:newsevents/(?:pressreleases|speech|testimony)/[^/]+\.htm|monetarypolicy/(?:fomcminutes|fomcprojtabl)\d+\.htm|releases/h41/\d+/)', '#article, .col-xs-12.col-sm-8.col-md-8, #content', ('www.federalreserve.gov',), 60),
}
-RULE_VERSION = 'us-policy-v2'
+RULE_VERSION = 'us-policy-v3-text-images'
CRYPTO = re.compile(r'\b(?:crypto(?:currency|currencies|[ -]assets?)?|digital[ -]assets?|virtual[ -]currenc(?:y|ies)|stablecoins?|bitcoin|ethereum|blockchain|tokeniz(?:ed|ation))\b', re.I)
POLICY = re.compile(r'\b(?:regulat\w*|rules?|guidance|legislation|acts?|sanctions?|licenses?|executive order|monetary policy|federal funds|interest rates?|balance sheet|liquidity|quantitative|refunding|borrowing|debt limit|tariffs?|fiscal|budget|appropriation|tax reform)\b', re.I)
INSTITUTION = re.compile(r'\b(?:exchanges?|banks?|financial institutions?|payment processors?|custodians?|stablecoin issuers?|binance|coinbase|tether|circle|kraken|bybit|bitfinex)\b', re.I)
@@ -88,7 +88,7 @@ def published(value: str) -> tuple[datetime, str]:
raise ValueError('官方材料缺少可识别的发布时间')
-def validate_body(content: str) -> None:
+def validate_body(content: str, *, allow_images: bool = False) -> None:
"""Reject error documents without requiring full-page HTML around fragments."""
if content.lstrip().startswith(' None:
raise ValueError('官方正文返回验证或登录页面')
if re.search(r'enable javascript and cookies to continue', text, re.I):
raise ValueError('官方正文返回验证页面')
- if len(text) < 40:
+ if len(text) < 40 and not (allow_images and sel.css('img[src], img[data-src]')):
raise ValueError('官方正文为空或过短')
+class UnsupportedDocument(ValueError):
+ """官方页面只提供文件，没有支持的网页正文。"""
+
+
def article(source: str, html: str, url: str, *, release_date: str | None = None) -> dict:
sel = Selector(html)
- validate_body(html)
+ from alphapulse.crawler.content_policy import SKIP_REASON
+ file_entry = (source == 'sec' and sel.css('article.node--type-regulation a[href*=".pdf"]')) or (source == 'fincen' and sel.css('main .field--name-field-advisory-file, main a[href*="/system/files/"]'))
+ if file_entry and not sel.css(SOURCES[source].body):
+ raise UnsupportedDocument(SKIP_REASON)
+ validate_body(html, allow_images=True)
selectors = {
'fed': ('#article h3.title', '#article h3', '#article h2', '#content .heading h3', '#content .col-xs-12.col-sm-8.col-md-8 h3'),
'treasury': ('h2.uswds-page-title',),
@@ -158,9 +166,10 @@ def article(source: str, html: str, url: str, *, release_date: str | None = None
if parent is not None:
parent.remove(node.root)
body = sorted(body, key=lambda node: len(node.xpath('string(.)').get()), reverse=True)
- content = body[0].get()
+ from alphapulse.crawler.content_policy import clean_html
+ content = clean_html(body[0].get(), url)
text = ' '.join(body[0].xpath('.//text()').getall()).strip()
- if len(text) < 40:
+ if len(text) < 40 and not body[0].css('img[src], img[data-src]'):
raise ValueError('官方正文为空或过短')
attachments = []
for link in body[0].css('a[href]'):
diff --git a/backend/src/alphapulse/crawler/content_policy.py b/backend/src/alphapulse/crawler/content_policy.py
new file mode 100644
index 0000000..a29e13d
--- /dev/null
+++ b/backend/src/alphapulse/crawler/content_policy.py
@@ -0,0 +1,64 @@
+"""新闻只采集网页文字与图片引用；经济日历不使用此策略。"""
+import re
+from urllib.parse import urljoin, urlsplit
+
+from parsel import Selector
+
+POLICY = 'web_text_images_v1'
+SKIP_REASON = '按采集规则不下载 PDF 和附件'
+FILE_PATH = re.compile(r'\.(?:pdf|docx?|xlsx?|pptx?|zip|gz|rar|7z|rtf|odt|ods|odp|exe|bin)(?:$|/)', re.I)
+
+
+def file_url(url):
+ parts = urlsplit(url)
+ return bool(FILE_PATH.search(parts.path) or re.search(r'\.pdf(?:[?&#]|$)', parts.query, re.I))
+
+
+def binary_headers(headers):
+ kind = headers.get(b'Content-Type', b'').split(b';')[0].strip().lower()
+ if not kind:
+ return False # 无类型时由首段 PDF 签名和后续正文解析继续校验。
+ return not (kind.startswith(b'text/') or kind in (
+ b'application/json', b'application/xml', b'application/rss+xml',
+ b'application/atom+xml', b'application/xhtml+xml',
+ ) or kind.endswith((b'+json', b'+xml')))
+
+
+
+def clean_html(html, base=''):
+ """剔除执行/嵌入节点和已确认的动态分享编码，保留正文及图片。"""
+ sel = Selector(html.replace('\x00', ''))
+ for node in sel.css('script, style, iframe, object, embed, video, audio, .shareDL'):
+ parent = node.root.getparent()
+ if parent is not None:
+ previous, tail = node.root.getprevious(), node.root.tail
+ parent.remove(node.root)
+ if tail:
+ if previous is None:
+ parent.text = (parent.text or '') + tail
+ else:
+ previous.tail = (previous.tail or '') + tail
+ for node in sel.css('[href*="/cdn-cgi/l/email-protection"]'):
+ node.root.attrib.pop('href', None)
+ for node in sel.css('[data-cfemail]'):
+ node.root.attrib.pop('data-cfemail', None)
+ for node in sel.css('img, picture source'):
+ if node.root.tag == 'img':
+ src = node.attrib.get('data-src') or node.attrib.get('src')
+ if src:
+ node.root.set('src', urljoin(base, src))
+ for attr in ('srcset', 'data-srcset'):
+ if value := node.attrib.get(attr):
+ candidates = []
+ for part in value.split(','):
+ fields = part.strip().split()
+ if fields:
+ candidates.append(' '.join([urljoin(base, fields[0]), *fields[1:]]))
+ node.root.set('srcset', ', '.join(candidates))
+ # Selector adds an html/body wrapper; keep the original fragment contract.
+ body = sel.css('body')
+ return ''.join(body[0].xpath('node()').getall()) if body else sel.get()
+
+
+def ignored_retry(group, value):
+ return group == 'attachment_retries' or bool(value.get('skipped_policy')) or file_url(value.get('document', {}).get('url', ''))
diff --git a/backend/src/alphapulse/crawler/middlewares.py b/backend/src/alphapulse/crawler/middlewares.py
index 7268c3a..e221eca 100644
--- a/backend/src/alphapulse/crawler/middlewares.py
+++ b/backend/src/alphapulse/crawler/middlewares.py
@@ -46,6 +46,12 @@ class RawResponseMiddleware:
return cls(crawler)
async def process_response(self, request: Request, response: Response) -> Response:
+ if response.body.lstrip().startswith(b'%PDF'):
+ if not 200 <= response.status < 300:
+ from scrapy.spidermiddlewares.httperror import HttpError
+ raise HttpError(response)
+ request.meta['policy_skipped'] = True
+ raise IgnoreRequest('按采集规则不保存 PDF 响应')
sha256 = hashlib.sha256(response.body).hexdigest()
headers = {k.decode(): b", ".join(v).decode("latin-1") for k, v in response.headers.items()}
async with SessionLocal() as session:
@@ -66,3 +72,38 @@ class RawResponseMiddleware:
await session.commit()
request.meta["raw_response_id"] = raw_id
return response
+
+
+class NewsContentPolicyMiddleware:
+ """在下载前及响应头阶段阻止文件；此下载器仅由新闻采集使用。"""
+
+ @classmethod
+ def from_crawler(cls, crawler):
+ from scrapy import signals
+ obj = cls()
+ crawler.signals.connect(obj.headers_received, signal=signals.headers_received)
+ crawler.signals.connect(obj.bytes_received, signal=signals.bytes_received)
+ return obj
+
+ def process_request(self, request):
+ from alphapulse.crawler.content_policy import file_url, SKIP_REASON
+ if file_url(request.url) or request.meta.get('parsed'):
+ request.meta['policy_skipped'] = True
+ raise IgnoreRequest(SKIP_REASON)
+
+ def headers_received(self, headers, request, **kwargs):
+ from scrapy.exceptions import StopDownload
+ from alphapulse.crawler.content_policy import binary_headers
+ if binary_headers(headers):
+ request.meta['policy_skipped'] = True
+ raise StopDownload()
+
+ def bytes_received(self, data, request, **kwargs):
+ from scrapy.exceptions import StopDownload
+ prefix = request.meta.get('_content_prefix', b'')
+ if len(prefix) < 8:
+ prefix = (prefix + data)[:8]
+ request.meta['_content_prefix'] = prefix
+ if prefix.lstrip().startswith(b'%PDF'):
+ request.meta['policy_skipped'] = True
+ raise StopDownload()
diff --git a/backend/src/alphapulse/crawler/pipelines.py b/backend/src/alphapulse/crawler/pipelines.py
index 2787ba5..9c50b92 100644
--- a/backend/src/alphapulse/crawler/pipelines.py
+++ b/backend/src/alphapulse/crawler/pipelines.py
@@ -6,7 +6,9 @@ from scrapy.crawler import Crawler
from sqlalchemy import func, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
+from alphapulse.crawler.content_policy import clean_html
from alphapulse.core.db import SessionLocal
+from alphapulse.crawler.spiders.base import html_to_text
from alphapulse.core.redis import redis
from alphapulse.crawler.events import NEWS_CHANNEL
from alphapulse.models import NewsItem, NewsRevision
@@ -31,6 +33,11 @@ class NewsPipeline:
async def process_item(self, item: dict[str, Any]) -> dict[str, Any]:
spider = self.crawler.spider
+ for field in ('title', 'summary', 'author'):
+ if isinstance(item.get(field), str):
+ item[field] = item[field].replace('\x00', '')
+ item['content_html'] = clean_html(item['content_html'], item['url'])
+ item['content_text'] = html_to_text(item['content_html'])
digest = content_hash(item)
raw_response_id = item.pop("raw_response_id")
locked = (
@@ -56,11 +63,14 @@ class NewsPipeline:
if inserted_id is not None:
kind = "new"
else:
+ same_content = (news.title.replace('\x00', '') == item['title']
+ and news.summary.replace('\x00', '') == item['summary']
+ and clean_html(news.content_html, news.url) == item['content_html'])
# 主记录保存最新元数据；正文修订仍只由内容哈希变化决定。
for field in UPDATABLE_FIELDS:
setattr(news, field, item[field])
news.last_seen_at = func.now()
- if news.content_hash != digest:
+ if not same_content and news.content_hash != digest:
kind = "revised"
news.content_hash = digest
news.revision_no += 1
diff --git a/backend/src/alphapulse/crawler/settings.py b/backend/src/alphapulse/crawler/settings.py
index b60ea9f..32c212c 100644
--- a/backend/src/alphapulse/crawler/settings.py
+++ b/backend/src/alphapulse/crawler/settings.py
@@ -25,6 +25,7 @@ RETRY_TIMES = 2
RETRY_HTTP_CODES = [500, 502, 503, 504, 522, 524, 408]
DOWNLOADER_MIDDLEWARES = {
+ "alphapulse.crawler.middlewares.NewsContentPolicyMiddleware": 90,
"alphapulse.crawler.middlewares.RedisRateLimitMiddleware": 100,
# process_response 按优先级从大到小调用：排在 HttpCompression(590) 之后拿到解压后的响应体，
# 排在 Retry(550) 之前，被重试的每次响应也都会记录
diff --git a/backend/src/alphapulse/crawler/spiders/rss.py b/backend/src/alphapulse/crawler/spiders/rss.py
index 1c30e75..c100ca1 100644
--- a/backend/src/alphapulse/crawler/spiders/rss.py
+++ b/backend/src/alphapulse/crawler/spiders/rss.py
@@ -12,7 +12,7 @@ from zoneinfo import ZoneInfo
import httpx
from scrapy.crawler import Crawler
-from scrapy.http import Request, Response
+from scrapy.http import Request, Response, Headers
from scrapy.settings.default_settings import RETRY_EXCEPTIONS
from alphapulse.crawler.spiders.base import NewsSpider, html_to_text
@@ -164,7 +164,16 @@ class BlsDownloadHandler:
async with httpx.AsyncClient(verify=self.ssl_context, timeout=request.meta.get("download_timeout", self.timeout)) as client:
async with client.stream("GET", request.url, headers=headers) as response:
# 保留原始压缩体，避免 httpx 与 Scrapy 的 HttpCompressionMiddleware 重复解压。
- body = b"".join([chunk async for chunk in response.aiter_raw()])
+ from alphapulse.crawler.content_policy import binary_headers
+ if binary_headers(Headers(response.headers.multi_items())):
+ raise ValueError('订阅返回不支持的文件类型')
+ chunks, prefix = [], b''
+ async for chunk in response.aiter_raw():
+ chunks.append(chunk)
+ prefix = (prefix + chunk)[:8]
+ if prefix.lstrip().startswith(b'%PDF'):
+ raise ValueError('订阅返回 PDF 文件')
+ body = b''.join(chunks)
return Response(
str(response.url), status=response.status_code, headers=response.headers.multi_items(),
body=body, request=request,
diff --git a/backend/src/alphapulse/crawler/spiders/us_official.py b/backend/src/alphapulse/crawler/spiders/us_official.py
index 872fc0e..2bb54aa 100644
--- a/backend/src/alphapulse/crawler/spiders/us_official.py
+++ b/backend/src/alphapulse/crawler/spiders/us_official.py
@@ -4,24 +4,24 @@ import json
import re
import time
import uuid
-from contextlib import nullcontext
from datetime import UTC, datetime, timedelta
from html import escape, unescape
from email.utils import parsedate_to_datetime
from lxml import etree
-from pypdf import apply_configuration
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from scrapy import Request
+from scrapy.exceptions import StopDownload
from scrapy.http import Response
from scrapy.spidermiddlewares.httperror import HttpError
from sqlalchemy import select
+from twisted.python.failure import Failure
-from alphapulse.calendar.umich import pdf_text
+from alphapulse.crawler.content_policy import POLICY, SKIP_REASON, file_url
from alphapulse.core.config import get_settings
from alphapulse.core.db import SessionLocal
-from alphapulse.core.us_news import RULE_VERSION, SOURCES, article, identity, official_url, published, selection, validate_body
+from alphapulse.core.us_news import RULE_VERSION, SOURCES, UnsupportedDocument, article, identity, official_url, published, selection, validate_body
from alphapulse.crawler.spiders.base import HISTORY_RUN_SECONDS, NewsSpider, html_to_text
from alphapulse.models import NewsItem, Source
@@ -143,17 +143,11 @@ class USOfficialSpider(NewsSpider):
self.cursor[lane]['pending'] = [d for d in self.cursor[lane]['pending'] if d['url'] not in indexes]
state = self.cursor.setdefault('realtime', {'entry': 0})
retries = self.cursor.get('document_retries', {})
- due_documents = [{**value['document'], 'document_retry': True} for value in retries.values() if value['attempts'] < 3 and value['next_at'] <= self.run.started_at.timestamp()]
+ rechecks = [{**v['document'], 'document_retry': True, 'policy_recheck': True} for v in retries.values() if v['attempts'] >= 3 and not v.get('policy_checked') and not v.get('skipped_policy') and not file_url(v['document']['url'])][:3]
+ state.setdefault('pending', []).extend(d for d in rechecks if d['url'] not in {p['url'] for p in state['pending']})
+ due_documents = [{**value['document'], 'document_retry': True} for value in retries.values() if not value.get('skipped_policy') and value['attempts'] < 3 and value['next_at'] <= self.run.started_at.timestamp()]
if due_documents:
state.setdefault('pending', []).extend(copy.deepcopy(d) for d in due_documents if d['url'] not in {p['url'] for p in state['pending']})
- due = [value['document'] for value in self.cursor.get('attachment_retries', {}).values() if value['attempts'] < 3 and value['next_at'] <= self.run.started_at.timestamp()]
- for doc in due:
- pending = state.setdefault('pending', [])
- existing = next((value for value in pending if value['url'] == doc['url']), None)
- if existing is None:
- pending.append({**copy.deepcopy(doc), 'attachment_retry': True})
- else:
- existing['attachment_retry'] = True
bill_due = [{**v['document'], 'bill_retry': True} for v in self.cursor.get('bill_retries', {}).values() if v['attempts'] < 3 and v['next_at'] <= self.run.started_at.timestamp()]
if bill_due:
state.setdefault('pending', []).extend(d for d in bill_due if d['url'] not in {p['url'] for p in state['pending']})
@@ -182,10 +176,15 @@ class USOfficialSpider(NewsSpider):
return []
if state.get('pending'):
doc = state['pending'][0]
+ if file_url(doc['url']) or doc.get('format') == 'pdf':
+ return self.skip_document(lane, doc)
gap = self.cursor.get('document_retries', {}).get(doc['url'])
- if doc['url'] in self.run.document_retries_started or (gap and (gap['attempts'] >= 3 or gap['next_at'] > self.run.started_at.timestamp())):
+ recheck = doc.get('policy_recheck') and gap and not gap.get('policy_checked')
+ if doc['url'] in self.run.document_retries_started or (not recheck and gap and (gap.get('skipped_policy') or gap['attempts'] >= 3 or gap['next_at'] > self.run.started_at.timestamp())):
state['pending'].pop(0)
return self.advance(lane)
+ if recheck:
+ gap['policy_checked'] = POLICY
if gap:
doc['document_retry'] = True
self.run.document_retries_started.add(doc['url'])
@@ -344,6 +343,9 @@ class USOfficialSpider(NewsSpider):
continue
fincen_row = a.xpath('ancestor::tr[ancestor::main][1]') if self.name == 'fincen' and response.url.split('?')[0] in SOURCES['fincen'].entries[1:] else []
fincen_pdf = bool(fincen_row and urlsplit(url).path.lower().endswith('.pdf'))
+ if file_url(url) and re.search(pattern, urlsplit(url).path) and not fincen_pdf:
+ docs.append({'url': url, 'kind': 'article', 'format': 'pdf'})
+ continue
if fincen_pdf or (re.search(pattern, urlsplit(url).path) and url not in SOURCES[self.name].entries and not re.search(r'/\d{4}-(?:press|speeches|testimony)\.htm$', urlsplit(url).path) and not url.lower().endswith('.pdf')):
doc = {'url': url, 'kind': 'article'}
if fincen_row:
@@ -352,11 +354,13 @@ class USOfficialSpider(NewsSpider):
begin = self.run.history_target if lane == 'history' else self.run.cutoff
if re.fullmatch(r'(?:0?[1-9]|1[0-2])/\d{4}', date) and int(date[-4:]) < begin.year:
continue
- doc['listing_date'] = published(date)[0].date().isoformat()
+ try:
+ doc['listing_date'] = published(date)[0].date().isoformat()
+ except ValueError:
+ if not fincen_pdf:
+ raise
if fincen_pdf:
title = fincen_row.xpath('normalize-space(td[3])').get() if response.url.split('?')[0] == SOURCES['fincen'].entries[2] else a.xpath('normalize-space(.)').get()
- if not title:
- raise ValueError('FinCEN PDF 列表缺少标题')
doc.update(format='pdf', title=title, listing_url=response.url)
elif self.name in ('sec', 'treasury', 'cftc', 'fincen', 'whitehouse'):
container = {
@@ -410,7 +414,7 @@ class USOfficialSpider(NewsSpider):
self.cursor['checked'] = {k: v for k, v in checked.items() if v >= threshold}
docs = [d for d in docs if self.cursor['checked'].get(d['url'], 0) < threshold]
retries = self.cursor.get('document_retries', {})
- docs = [d for d in docs if d['url'] not in retries or (retries[d['url']]['attempts'] < 3 and retries[d['url']]['next_at'] <= self.run.started_at.timestamp())]
+ docs = [d for d in docs if d['url'] not in retries or (not retries[d['url']].get('skipped_policy') and retries[d['url']]['attempts'] < 3 and retries[d['url']]['next_at'] <= self.run.started_at.timestamp())]
docs = [{**d, 'document_retry': True} if d['url'] in retries else d for d in docs]
if next_url and official_url(self.name, next_url, response.url) == response.url:
raise ValueError('官方列表翻页重复当前页面，未推进断点')
@@ -440,7 +444,7 @@ class USOfficialSpider(NewsSpider):
now = datetime.now(UTC)
if not previous or previous['next_at'] <= now.timestamp():
retries[doc['url']] = {'document': copy.deepcopy(doc), 'attempts': previous.get('attempts', 0) + 1,
- 'next_at': (now + timedelta(hours=1)).timestamp(), 'error': error}
+ 'next_at': (now + timedelta(hours=1)).timestamp(), 'error': error, **({'policy_checked': POLICY} if previous.get('policy_checked') else {})}
self.run.document_retries_started.add(doc['url'])
self.run.errors.append(error)
self.logger.error('文档缺口：%s；%s', doc['url'], error)
@@ -449,7 +453,15 @@ class USOfficialSpider(NewsSpider):
return self.advance(request.meta['lane'])
def on_request_error(self, failure):
+ # 文件响应可在响应头阶段被截断，但 HTTP 故障仍须保留退避和重试语义。
+ if failure.check(StopDownload) and not 200 <= failure.value.response.status < 300:
+ response, request = failure.value.response, failure.request
+ request.meta.pop('policy_skipped', None)
+ failure = Failure(HttpError(response))
+ failure.request = request
status = failure.value.response.status if failure.check(HttpError) else None
+ if failure.request.meta.get('parsed'):
+ return list(self.attach_or_finish(failure.request.meta['parsed'], failure.request.meta['lane']))
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))
@@ -461,10 +473,10 @@ class USOfficialSpider(NewsSpider):
self.block_until(until)
except (ValueError, TypeError, OverflowError):
self.logger.warning('忽略无效 Retry-After 响应头')
- if self.name in ('sec', 'doj', 'congress') and (parsed := failure.request.meta.get('parsed')):
- attachment = next(a for a in parsed['extra']['attachments'] if a['status'] == 'pending')
- attachment.update(status='unavailable', error=f'请求失败：{status or repr(failure.value)}')
- return list(self.attach_or_finish(parsed, failure.request.meta['lane']))
+ if failure.request.meta.get('policy_skipped'):
+ if doc := failure.request.meta.get('document'):
+ return self.skip_document(failure.request.meta['lane'], doc)
+ return super().on_request_error(failure)
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:
@@ -475,6 +487,9 @@ class USOfficialSpider(NewsSpider):
try:
async for result in self._parse_document(response):
yield result
+ except UnsupportedDocument:
+ for request in self.skip_document(response.meta['lane'], response.meta['document']):
+ yield request
except (ValueError, KeyError, TypeError) as exc:
if '验证' in str(exc) or '登录页面' in str(exc):
self.block_until(datetime.now(UTC) + timedelta(minutes=15))
@@ -487,6 +502,10 @@ class USOfficialSpider(NewsSpider):
async def _parse_document(self, response):
lane, doc = response.meta['lane'], response.meta['document']
begin = self.run.history_target if lane == 'history' else self.run.cutoff
+ if file_url(doc['url']) or doc.get('format') == 'pdf':
+ for request in self.skip_document(lane, doc):
+ yield request
+ return
if not (doc.get('document_retry') or doc.get('attachment_retry')) and doc.get('listing_date') and published(doc['listing_date'])[0].date() < begin.date():
self.run.stats['outside_range'] = self.run.stats.get('outside_range', 0) + 1
for result in self.complete(lane):
@@ -572,12 +591,16 @@ class USOfficialSpider(NewsSpider):
yield item
else:
for version in data['textVersions']:
+ if version.get('formats') and all(f['type'] == 'PDF' for f in version['formats']):
+ self.cursor[lane]['bill']['policy_skipped'] = True
+ self.run.stats['policy_skipped'] = self.run.stats.get('policy_skipped', 0) + 1
+ continue
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)
+ 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 None
if not chosen:
self.cursor[lane]['bill']['missing_material'] = True
self.cursor[lane]['bill'].setdefault('material_errors', []).append(f"{version.get('type', '正文版本')} 缺少支持的正文格式")
@@ -628,7 +651,7 @@ class USOfficialSpider(NewsSpider):
attachment_html = ''
elif not isinstance(attachment_html, str):
raise ValueError('DOJ API 附件结构未识别')
- html = f'
{escape(title)}
{record["body"]}{attachment_html}
'
+ html = f'
{escape(title)}
{record["body"]}{record.get("image") or ""}{attachment_html}
'
validate_body(record['body'])
parsed = article(self.name, html, canonical)
known = {a['url'] for a in parsed['extra']['attachments']}
@@ -640,20 +663,13 @@ class USOfficialSpider(NewsSpider):
parsed['source_updated_at'] = published(times[0])[0]
parsed['extra']['official_uuid'] = record['uuid']
doc = {**doc, 'canonical': canonical}
- elif doc['kind'] == 'text' or doc.get('format') == 'pdf':
- at, precision = published(doc['listing_date'] if doc.get('format') == 'pdf' else doc['at'])
- if response.body.startswith(b'%PDF'):
- content = '
' + escape(self.pdf_text(response.body)) + '
'
- else:
- content = response.text
- validate_body(content)
- if doc.get('format') == 'pdf':
- raise ValueError('官方 PDF 返回非 PDF 正文')
- if '<' not in content:
- content = '
' + escape(content) + '
'
+ elif doc['kind'] == 'text':
+ at, precision = published(doc['at'])
+ content = response.text
+ validate_body(content)
+ if '<' not in content:
+ content = '
' + escape(content) + '
'
parsed = {'title': doc['title'], 'content_html': content, 'published_at': at, 'extra': {'content_scope': 'article', 'language': 'en', 'published_at_precision': precision, 'published_date': at.date().isoformat(), 'rule_version': RULE_VERSION, 'official_document': doc, 'attachments': []}}
- if doc.get('format') == 'pdf':
- parsed['extra'].update(published_at_basis='official_listing', published_at_source=doc['listing_url'])
if doc.get('pdf'):
parsed['extra']['attachments'].append({'url': official_url(self.name, doc['pdf']), 'title': 'Official PDF', 'status': 'pending'})
else:
@@ -698,7 +714,7 @@ class USOfficialSpider(NewsSpider):
def finish_bill(self, lane):
context = self.cursor[lane].pop('bill')
retries = self.cursor.setdefault('bill_retries', {})
- if not context['bodies'] or context.get('missing_material'):
+ if (not context['bodies'] and not context.get('policy_skipped')) or context.get('missing_material'):
previous = retries.get(context['url'], {})
errors = list(context.get('material_errors', []))
if not context['bodies']:
@@ -714,66 +730,38 @@ class USOfficialSpider(NewsSpider):
if lane == 'realtime':
self.cursor.setdefault('checked', {})[context['document_url']] = self.run.started_at.timestamp()
+ def skip_document(self, lane, doc):
+ if doc.get('bill'):
+ self.cursor[lane]['bill']['policy_skipped'] = True
+ previous = self.cursor.get('document_retries', {}).get(doc['url'], {})
+ self.cursor.setdefault('document_retries', {})[doc['url']] = {
+ **previous, 'document': doc, 'attempts': previous.get('attempts', 0),
+ 'next_at': 0, 'skipped_policy': POLICY, 'error': SKIP_REASON,
+ 'previous_error': previous.get('previous_error') or previous.get('error'),
+ }
+ self.run.stats['policy_skipped'] = self.run.stats.get('policy_skipped', 0) + 1
+ self.cursor[lane]['pending'].pop(0)
+ return self.advance(lane)
+
def attach_or_finish(self, parsed, lane):
- retired = self.cursor.get('attachment_retries', {}).get(parsed['url'], {})
- if retired.get('retired_at'):
- for attachment in parsed['extra']['attachments']:
- if attachment['status'] == 'pending':
- attachment.update(status='unavailable', error=retired['error'])
- 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()
+ for attachment in parsed['extra'].get('attachments', []):
+ attachment.update(status='skipped_policy', error=SKIP_REASON)
+ parsed['extra']['collection_policy'] = POLICY
+ parsed['extra'].setdefault('content_scope', 'article')
+ parsed.pop('retry_document', None)
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'])
- retry_doc = parsed.pop('retry_document', None)
- retries = self.cursor.setdefault('attachment_retries', {})
- if incomplete and retry_doc and not retired.get('retired_at'):
- previous = retries.get(parsed['url'], {})
- 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, 'failed_urls': [a['url'] for a in parsed['extra']['attachments'] if a['status'] != 'complete']}
- 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'
- self.run.stats['partial'] = self.run.stats.get('partial', 0) + int(incomplete)
raw_id = parsed.pop('raw_response_id')
- item = self.item(Response(parsed['url'], request=Request(parsed['url'], meta={'raw_response_id': raw_id})), summary='', **parsed)
- yield item
+ yield self.item(Response(parsed['url'], request=Request(parsed['url'], meta={'raw_response_id': raw_id})), summary='', **parsed)
else:
self.run.stats['excluded'] = self.run.stats.get('excluded', 0) + 1
- if incomplete:
- self.run.stats['unclassified_attachment_gaps'] = self.run.stats.get('unclassified_attachment_gaps', 0) + 1
yield from self.complete(lane)
- def pdf_text(self, body):
- # Bound the known complex FinCEN report without changing other collectors' limits.
- context = apply_configuration(stream_parsing_work_maximum_count=4_000_000) if self.name == 'fincen' else nullcontext()
- with context:
- return pdf_text(body)
-
def parse_attachment(self, response):
- parsed = response.meta['parsed']
- attachment = next(a for a in parsed['extra']['attachments'] if a['url'] == response.url or a['status'] == 'pending')
- attachment['raw_response_id'] = response.meta['raw_response_id']
- try:
- if response.status != 200:
- raise ValueError(f'HTTP {response.status}')
- if response.headers.get('Content-Type', b'').lower().startswith(b'image/') and response.body.startswith((b'\x89PNG\r\n\x1a\n', b'\xff\xd8\xff', b'GIF87a', b'GIF89a')):
- raise ValueError('官方图片附件尚未提取文字，请核对原文')
- text = self.pdf_text(response.body)
- attachment['status'] = 'complete'
- parsed['content_html'] += f"
图表说明'
+ value = clean_html(html, 'https://official.example/news/1')
+ assert '政策正文' in value and '3%' in value and '图表说明' in value
+ assert 'https://official.example/chart.png' in value
+ assert 'https://official.example/a.png 1x' in value
+ assert 'iframe' not in value and 'script' not in value and '\x00' not in value
+ assert clean_html(value, 'https://official.example/news/1') == value
+ assert clean_html(html.replace('3%', '4%')) != clean_html(html)
+
+
+def test_email_share_encoding_is_not_news_content():
+ html = '
The server returned an HTML document instead of the expected official PDF.
', lane='realtime', document=doc))]
- assert not asyncio.run(collect())
- assert doc['url'] not in obj.cursor.get('checked', {})
- assert obj.cursor['document_retries'][doc['url']]['attempts'] == 1
- assert '非 PDF' in obj.run.errors[0]
+ return [v async for v in obj.parse_document(response(doc['url'], '
PDF or error document
', lane='history', document=doc))]
+ assert asyncio.run(collect()) == []
+ assert obj.cursor['document_retries'][doc['url']]['skipped_policy']
+ assert not obj.run.errors and not obj.cursor['history']['pending']
+
def test_fincen_pdf_table_missing_date_fails_explicitly(trading):
obj = spider(trading, 'fincen')
body = '
