500 lines
18 KiB
Python
500 lines
18 KiB
Python
"""cninfo 巨潮资讯网爬虫 v2.0。
|
||
|
||
从 watchlist.yaml 的 code + orgId 拼接 URL,通过 Crawl4AI(Playwright)渲染 SPA 页面提取数据。
|
||
三种数据类型:
|
||
1. 公司最新公告 → https://www.cninfo.com.cn/new/disclosure/stock?stockCode={code}&orgId={orgId}#latestAnnouncement
|
||
2. 投资者调研 → 同上, #research
|
||
3. 互动易问答 → https://irm.cninfo.com.cn/ircs/search?keyword={code}
|
||
|
||
策略:
|
||
- 公告/调研: Playwright SPA 渲染 → DOM 提取(A 方案,REST API 的 stock 参数不可靠)
|
||
- 互动易: requests 优先,失败回退 Playwright
|
||
- 增量: ann_id 去重,首次从 2026-01-01 全量,后续增量 7 天
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
from datetime import date, datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
from loguru import logger
|
||
|
||
from .models import CninfoItem
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 常量
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
SOURCE_ID = "cninfo"
|
||
CNINFO_PDF_BASE = os.environ.get("CNINFO_PDF_BASE", "http://static.cninfo.com.cn")
|
||
|
||
# 日期范围: 从 2026-01-01 开始
|
||
DATE_START = "2026-01-01"
|
||
|
||
# 翻页限制
|
||
MAX_ANNOUNCE_PAGES = 5
|
||
MAX_RESEARCH_PAGES = 1
|
||
MAX_IRM_ITEMS = 20
|
||
|
||
# 请求间隔(秒)
|
||
REQUEST_DELAY = 0.5
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 工具函数
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def _url_hash(url: str) -> str:
|
||
return hashlib.sha1(url.encode("utf-8")).hexdigest()[:16]
|
||
|
||
|
||
def _today_str() -> str:
|
||
return date.today().strftime("%Y%m%d")
|
||
|
||
|
||
def _load_watchlist() -> list[dict]:
|
||
import yaml
|
||
try:
|
||
with open("configs/watchlist.yaml", encoding="utf-8") as f:
|
||
data = yaml.safe_load(f) or {}
|
||
return list(data.get("watchlist") or [])
|
||
except Exception:
|
||
logger.exception("加载 watchlist.yaml 失败")
|
||
return []
|
||
|
||
|
||
def _make_api_headers() -> dict[str, str]:
|
||
return {
|
||
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36",
|
||
"Accept": "text/html,application/xhtml+xml",
|
||
}
|
||
|
||
|
||
def _build_stock_url(code: str, org_id: str) -> str:
|
||
"""拼接 cninfo 个股公告页 URL。"""
|
||
return f"https://www.cninfo.com.cn/new/disclosure/stock?stockCode={code}&orgId={org_id}"
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Crawl4AI SPA 渲染
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
async def _render_page(url: str, timeout_ms: int = 60000,
|
||
delay_ms: int = 20) -> str:
|
||
"""用 Crawl4AI 渲染 SPA 页面,返回 HTML 字符串。"""
|
||
from crawl4ai import AsyncWebCrawler, BrowserConfig, CacheMode, CrawlerRunConfig
|
||
|
||
bconf = BrowserConfig(headless=True, verbose=False)
|
||
rconf = CrawlerRunConfig(
|
||
cache_mode=CacheMode.BYPASS,
|
||
page_timeout=timeout_ms,
|
||
delay_before_return_html=delay_ms,
|
||
)
|
||
async with AsyncWebCrawler(config=bconf) as c:
|
||
result = await c.arun(url=url, config=rconf)
|
||
return getattr(result, "html", "") or ""
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 公告 + 调研: 从渲染后的 SPA 页面 DOM 提取
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def _parse_announcement_list(html: str, code: str,
|
||
item_type: str = "announcement") -> list[CninfoItem]:
|
||
"""从 cninfo 个股页面的渲染 HTML 中提取公告/调研列表。
|
||
|
||
页面结构: 公告列表以 <a> 标签呈现,关键属性:
|
||
- data-seccode: 股票代码
|
||
- data-id: 公告 ID
|
||
- href: 包含 announcementTime / announcementType 等参数
|
||
"""
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
items: list[CninfoItem] = []
|
||
seen: set[str] = set()
|
||
|
||
# 查找所有带 data-seccode=code 的公告链接
|
||
for el in soup.find_all(attrs={"data-seccode": code}):
|
||
ann_id = (el.get("data-id") or "").strip()
|
||
if not ann_id or ann_id in seen:
|
||
continue
|
||
seen.add(ann_id)
|
||
|
||
title = el.get_text(strip=True)
|
||
if len(title) < 5:
|
||
continue
|
||
|
||
href = el.get("href", "")
|
||
|
||
# 从 href 提取发布时间
|
||
pub_time = ""
|
||
date_match = re.search(r"announcementTime=(\d{4}-\d{2}-\d{2})", href)
|
||
if date_match:
|
||
pub_time = date_match.group(1)
|
||
|
||
# 公告类型
|
||
ann_type = ""
|
||
type_match = re.search(r"announcementType=(\w+)", href)
|
||
if type_match:
|
||
ann_type = type_match.group(1)
|
||
|
||
# 板块代码
|
||
plate = ""
|
||
plate_match = re.search(r"plate=(\w+)", href)
|
||
if plate_match:
|
||
plate = plate_match.group(1)
|
||
|
||
# PDF URL
|
||
pdf_url = ""
|
||
if pub_time and ann_id:
|
||
pdf_url = f"{CNINFO_PDF_BASE}/finalpage/{pub_time}/{ann_id}.PDF"
|
||
|
||
items.append(CninfoItem(
|
||
stock_code=code,
|
||
stock_name="",
|
||
title=title,
|
||
content="",
|
||
publish_time=pub_time,
|
||
item_type=item_type,
|
||
url=pdf_url,
|
||
ann_id=ann_id,
|
||
extra={"announcement_type": ann_type, "plate": plate},
|
||
))
|
||
|
||
return items
|
||
|
||
|
||
def _parse_irm_text(text: str, code: str) -> list[CninfoItem]:
|
||
"""从互动易页面文本中提取问答。
|
||
|
||
注意: cninfo 互动易搜索页是 Vue SPA，问答数据通过需认证的 API 加载。
|
||
公开访问时页面显示"暂无数据"，此时应返回空列表。
|
||
"""
|
||
items: list[CninfoItem] = []
|
||
url = f"https://irm.cninfo.com.cn/ircs/search?keyword={code}"
|
||
|
||
# 检测是否为无数据页面
|
||
if "暂无数据" in text:
|
||
logger.debug(" {} 互动易页面显示'暂无数据'(需登录认证)", code)
|
||
return items
|
||
|
||
# 检测是否为 SPA 空壳（无 JS 渲染时只有导航文本）
|
||
text_stripped = text.strip()
|
||
if len(text_stripped) < 200 and code in text_stripped:
|
||
logger.debug(" {} 互动易页面内容过短(SPA 空壳)", code)
|
||
return items
|
||
|
||
# 按常见分隔模式拆分问答块
|
||
blocks = re.split(r"\n(?=\d+\.|\b[问答][：:])", text)
|
||
for i, block in enumerate(blocks[:MAX_IRM_ITEMS]):
|
||
block = block.strip()
|
||
if len(block) > 30 and code in block:
|
||
items.append(CninfoItem(
|
||
stock_code=code,
|
||
stock_name="",
|
||
title=f"{code} 互动问答 #{i + 1}",
|
||
content=block[:5000],
|
||
publish_time="",
|
||
item_type="irm",
|
||
url=url,
|
||
ann_id=_url_hash(f"{url}#{i}"),
|
||
))
|
||
return items
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 互动易
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
async def _fetch_irm_playwright(code: str) -> list[CninfoItem]:
|
||
"""Playwright 渲染互动易搜索页。"""
|
||
url = f"https://irm.cninfo.com.cn/ircs/search?keyword={code}"
|
||
try:
|
||
html = await _render_page(url, timeout_ms=30000, delay_ms=8)
|
||
except Exception as e:
|
||
logger.warning(" {} 互动易 Playwright 失败: {}", code, e)
|
||
return []
|
||
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
body_text = soup.get_text()
|
||
return _parse_irm_text(body_text, code)
|
||
|
||
|
||
def _fetch_irm_requests(code: str) -> list[CninfoItem]:
|
||
"""requests 获取互动易搜索页。"""
|
||
url = f"https://irm.cninfo.com.cn/ircs/search?keyword={code}"
|
||
headers = _make_api_headers()
|
||
headers["Referer"] = "https://irm.cninfo.com.cn/"
|
||
|
||
try:
|
||
r = requests.get(url, headers=headers, timeout=15)
|
||
r.raise_for_status()
|
||
except Exception as e:
|
||
logger.debug(" {} 互动易 requests 失败: {}", code, e)
|
||
return []
|
||
|
||
soup = BeautifulSoup(r.text, "html.parser")
|
||
body_text = soup.get_text()
|
||
return _parse_irm_text(body_text, code)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 单股票抓取
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
async def _crawl_one_stock_async(stock: dict,
|
||
start_date: str = DATE_START,
|
||
end_date: str | None = None) -> list[CninfoItem]:
|
||
"""异步抓取单个公司的公告 + 调研 + 互动易。"""
|
||
code = stock["code"]
|
||
name = stock["name"]
|
||
org_id = stock.get("orgId", "").strip()
|
||
|
||
if not org_id:
|
||
logger.warning("{} ({}) 未配置 orgId,跳过", code, name)
|
||
return []
|
||
|
||
end_date = end_date or date.today().strftime("%Y-%m-%d")
|
||
base_url = _build_stock_url(code, org_id)
|
||
results: list[CninfoItem] = []
|
||
|
||
# --- 1) 公告 (#latestAnnouncement) ---
|
||
announce_url = f"{base_url}#latestAnnouncement"
|
||
logger.info("抓取 {} ({}) 公告: {}", code, name, announce_url[:80])
|
||
try:
|
||
html = await _render_page(announce_url, timeout_ms=60000, delay_ms=20)
|
||
# 日期过滤: 只保留 start_date 之后的
|
||
items = _parse_announcement_list(html, code, item_type="announcement")
|
||
filtered = [it for it in items if it.publish_time >= start_date]
|
||
# 限制页数: 每个页面约 30 条, 5 页 ≈ 150 条(SPA 一次加载可能超过一页)
|
||
filtered = filtered[:MAX_ANNOUNCE_PAGES * 30]
|
||
for it in filtered:
|
||
it.stock_name = name
|
||
results.extend(filtered)
|
||
logger.info(" {} 公告: {} 条 (过滤后)", code, len(filtered))
|
||
except Exception as e:
|
||
logger.error(" {} 公告抓取失败: {}", code, e)
|
||
|
||
# --- 2) 调研 (#research) ---
|
||
research_url = f"{base_url}#research"
|
||
logger.info("抓取 {} ({}) 调研: {}", code, name, research_url[:80])
|
||
try:
|
||
html = await _render_page(research_url, timeout_ms=60000, delay_ms=20)
|
||
items = _parse_announcement_list(html, code, item_type="research")
|
||
filtered = [it for it in items if it.publish_time >= start_date]
|
||
filtered = filtered[:MAX_RESEARCH_PAGES * 30]
|
||
for it in filtered:
|
||
it.stock_name = name
|
||
results.extend(filtered)
|
||
logger.info(" {} 调研: {} 条 (过滤后)", code, len(filtered))
|
||
except Exception as e:
|
||
logger.warning(" {} 调研抓取失败(可能无调研页面): {}", code, e)
|
||
|
||
# --- 3) 互动易 ---
|
||
logger.info("抓取 {} ({}) 互动易", code, name)
|
||
try:
|
||
irm_items = _fetch_irm_requests(code)
|
||
if not irm_items:
|
||
logger.info(" {} 互动易 requests 无结果,回退 Playwright...", code)
|
||
irm_items = await _fetch_irm_playwright(code)
|
||
for it in irm_items:
|
||
it.stock_name = name
|
||
results.extend(irm_items)
|
||
logger.info(" {} 互动易: {} 条", code, len(irm_items))
|
||
except Exception as e:
|
||
logger.error(" {} 互动易抓取失败: {}", code, e)
|
||
|
||
return results
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 增量保存
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def _load_seen_ids(out_dir: Path) -> set[str]:
|
||
"""从 index.jsonl 加载已保存的公告 ID 集合。"""
|
||
seen: set[str] = set()
|
||
index_path = out_dir / "index.jsonl"
|
||
if index_path.is_file():
|
||
with index_path.open("r", encoding="utf-8") as f:
|
||
for line in f:
|
||
try:
|
||
rec = json.loads(line.strip())
|
||
aid = rec.get("ann_id", "")
|
||
if aid:
|
||
seen.add(aid)
|
||
except (json.JSONDecodeError, KeyError):
|
||
continue
|
||
return seen
|
||
|
||
|
||
def _save_items(items: list[CninfoItem], out_dir: Path) -> int:
|
||
"""增量保存 CninfoItem 列表,跳过已存在的 ann_id。返回新增条数。"""
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
seen = _load_seen_ids(out_dir)
|
||
logger.info("cninfo 增量模式: 已有 {} 条历史记录", len(seen))
|
||
|
||
index_path = out_dir / "index.jsonl"
|
||
new_count = 0
|
||
|
||
with index_path.open("a", encoding="utf-8") as index_f:
|
||
for item in items:
|
||
if item.ann_id and item.ann_id in seen:
|
||
continue
|
||
seen.add(item.ann_id)
|
||
new_count += 1
|
||
|
||
fname = _url_hash(item.ann_id or item.title + item.stock_code)
|
||
json_path = out_dir / f"{fname}.json"
|
||
json_path.write_text(
|
||
item.model_dump_json(indent=2, ensure_ascii=False),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
meta = item.model_dump(mode="json")
|
||
meta["source_id"] = SOURCE_ID
|
||
meta["json_file"] = f"{fname}.json"
|
||
index_f.write(json.dumps(meta, ensure_ascii=False) + "\n")
|
||
|
||
logger.info("cninfo 已保存: 新增 {} 条 (总计 {} 条) -> {}", new_count, len(seen), out_dir)
|
||
return new_count
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 主入口
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def crawl_watchlist(*, save: bool = True) -> list[CninfoItem]:
|
||
"""从 watchlist 抓取所有公司的公告+调研+互动易。
|
||
|
||
首次运行从 2026-01-01 开始,后续增量运行(最近 7 天)。
|
||
通过判定 data/raw/cninfo/ 下是否有历史 index.jsonl 来区分首次/增量。
|
||
"""
|
||
watchlist = _load_watchlist()
|
||
if not watchlist:
|
||
logger.warning("关注列表为空")
|
||
return []
|
||
|
||
out_dir = Path("data/raw") / SOURCE_ID / _today_str()
|
||
|
||
# 判断首次还是增量
|
||
has_history = False
|
||
raw_root = Path("data/raw") / SOURCE_ID
|
||
if raw_root.is_dir():
|
||
for d in sorted(raw_root.glob("*"), reverse=True):
|
||
idx = d / "index.jsonl"
|
||
if idx.is_file() and idx.stat().st_size > 0:
|
||
has_history = True
|
||
break
|
||
|
||
if has_history:
|
||
start = (date.today() - timedelta(days=7)).strftime("%Y-%m-%d")
|
||
logger.info("cninfo 增量模式: 日期范围 {} ~ 今天", start)
|
||
else:
|
||
start = DATE_START
|
||
logger.info("cninfo 首次全量: 日期范围 {} ~ 今天", start)
|
||
|
||
end = date.today().strftime("%Y-%m-%d")
|
||
|
||
# 并发抓取所有股票
|
||
async def _run_all():
|
||
tasks = [_crawl_one_stock_async(s, start_date=start, end_date=end)
|
||
for s in watchlist]
|
||
all_items: list[CninfoItem] = []
|
||
for coro in asyncio.as_completed(tasks):
|
||
try:
|
||
items = await coro
|
||
all_items.extend(items)
|
||
except Exception as e:
|
||
logger.error("某股票抓取异常: {}", e)
|
||
return all_items
|
||
|
||
all_items = asyncio.run(_run_all())
|
||
|
||
# 统计
|
||
ann_count = sum(1 for it in all_items if it.item_type == "announcement")
|
||
res_count = sum(1 for it in all_items if it.item_type == "research")
|
||
irm_count = sum(1 for it in all_items if it.item_type == "irm")
|
||
logger.info(
|
||
"cninfo 抓取完成: 公告 {} / 调研 {} / 互动易 {} ({} 家公司,共 {} 条)",
|
||
ann_count, res_count, irm_count, len(watchlist), len(all_items),
|
||
)
|
||
|
||
if save and all_items:
|
||
new_count = _save_items(all_items, out_dir)
|
||
logger.info("cninfo 本轮新增 {} 条", new_count)
|
||
|
||
return all_items
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# PDF 正文提取
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def enrich_articles_with_pdf(day_str: str | None = None, *, limit: int = 0) -> int:
|
||
"""下载 PDF 并用 MarkItDown 提取正文补充到 Article.content。"""
|
||
day_str = day_str or _today_str()
|
||
proc_dir = Path("data/processed") / SOURCE_ID / day_str
|
||
if not proc_dir.is_dir():
|
||
return 0
|
||
|
||
from markitdown import MarkItDown
|
||
|
||
enriched = 0
|
||
for fp in sorted(proc_dir.glob("*.json")):
|
||
if limit and enriched >= limit:
|
||
break
|
||
try:
|
||
article_data = json.loads(fp.read_text(encoding="utf-8"))
|
||
except (json.JSONDecodeError, OSError):
|
||
continue
|
||
|
||
content = article_data.get("content", "")
|
||
pdf_url = ""
|
||
|
||
# 多种方式提取 PDF URL
|
||
pdf_match = re.search(r"原文链接:\s*(https?://\S+\.pdf)", content, re.IGNORECASE)
|
||
if pdf_match:
|
||
pdf_url = pdf_match.group(1)
|
||
else:
|
||
pdf_match = re.search(r"PDF链接:\s*(https?://\S+\.pdf)", content, re.IGNORECASE)
|
||
if pdf_match:
|
||
pdf_url = pdf_match.group(1)
|
||
else:
|
||
url = article_data.get("url", "")
|
||
if url.lower().endswith(".pdf"):
|
||
pdf_url = url
|
||
|
||
if not pdf_url:
|
||
continue
|
||
|
||
try:
|
||
r = requests.get(pdf_url, headers=_make_api_headers(), timeout=30)
|
||
r.raise_for_status()
|
||
tmp_path = Path("/tmp") / f"cninfo_pdf_{_url_hash(pdf_url)}.pdf"
|
||
tmp_path.write_bytes(r.content)
|
||
md = MarkItDown()
|
||
result = md.convert(str(tmp_path))
|
||
text = (result.text_content or "").strip()[:20000]
|
||
if tmp_path.exists():
|
||
tmp_path.unlink()
|
||
if text and len(text) >= 50:
|
||
article_data["content"] = text
|
||
article_data["word_count"] = len(text)
|
||
fp.write_text(json.dumps(article_data, ensure_ascii=False, indent=2),
|
||
encoding="utf-8")
|
||
enriched += 1
|
||
except Exception as e:
|
||
logger.warning("PDF 提取失败 {}: {}", pdf_url, e)
|
||
|
||
return enriched
|