Files
wiki/.claude/skills/policy-radar/tools/scrapers.py
T
推理链专家 e492965ba1 为政策雷达 skill 添加11个政府网站的 Python 抓取工具
覆盖全部11个数据源的可复用 scraper 模块 tools/scrapers.py:
- gov.cn: 国务院政策文件库(HTML 解析)
- xinhuanet (news.cn): 新华社受权发布(HTML 解析)
- scio (pbc镜像): 国新办新闻发布会文字实录
- pbc: 央行货币政策/公告
- mof: 财政部政策
- ndrc: 发改委产业政策
- csrc: 证监会公告(HTML 解析)
- mofcom: 商务部政策(JSON API)
- people: 人民日报社论(HTML 解析)
- cctv: 央视新闻联播(JSONP API)
- ce: 经济日报政策解读

更新 SKILL.md 引用新工具,统一入口 scrape_list()/scrape_article()。

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: multica-agent <github@multica.ai>
2026-07-21 06:52:54 +00:00

633 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
政策雷达 - 政府网站抓取工具集
每个函数返回标准化格式: {"title": str, "date": str, "url": str, "content": str}
列表函数返回: [{"title": str, "date": str, "url": str}, ...]
"""
import re
import json
import urllib.request
import urllib.error
import ssl
from html import unescape
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
TIMEOUT = 15
# 忽略 SSL 证书验证(部分政府站点使用自签名证书)
_SSL_CTX = ssl.create_default_context()
_SSL_CTX.check_hostname = False
_SSL_CTX.verify_mode = ssl.CERT_NONE
def _fetch(url, headers=None):
"""通用 HTTP GET,返回解码后的 HTML 文本"""
hdrs = {"User-Agent": USER_AGENT}
if headers:
hdrs.update(headers)
req = urllib.request.Request(url, headers=hdrs)
with urllib.request.urlopen(req, timeout=TIMEOUT, context=_SSL_CTX) as resp:
raw = resp.read()
for enc in ("utf-8", "gbk", "gb2312"):
try:
return raw.decode(enc)
except UnicodeDecodeError:
continue
return raw.decode("utf-8", errors="replace")
def _extract_paragraphs(html):
"""从 HTML 中提取正文段落(通用)"""
paragraphs = re.findall(r"<p[^>]*>(.*?)</p>", html, re.DOTALL)
result = []
for p in paragraphs:
text = re.sub(r"<[^>]+>", "", p).strip()
text = unescape(text)
if text and len(text) > 10:
result.append(text)
return result
# ============================================================
# 优先级 1: 中央政府门户
# ============================================================
def gov_search(keyword, page=1):
"""
中国政府网 - 搜索 API(可搜索新闻发布会等)
URL: https://sousuo.www.gov.cn/search-gov/data?t=zhengcelibrary&q=...
返回 JSON,包含 title/url/summary
注意: 索引有 1-2 周延迟,最新内容可能尚未收录
"""
import urllib.parse
q = urllib.parse.quote(keyword)
api_url = (
f"https://sousuo.www.gov.cn/search-gov/data?t=zhengcelibrary"
f"&q={q}&p={page}&n=20&timetype=timeqb&mintime=&maxtime="
)
html = _fetch(api_url)
try:
data = json.loads(html)
except json.JSONDecodeError:
return []
results = []
search_result = data.get("searchVO", {}).get("catMap", {}).get("zhengcelibrary", {})
items = search_result.get("listVO", []) if isinstance(search_result, dict) else []
for item in items:
title = re.sub(r"<[^>]+>", "", item.get("title", "")).strip()
url = item.get("url", "")
date = item.get("pubtime", "")[:10] if item.get("pubtime") else ""
if title and url:
results.append({"title": title, "date": date, "url": url})
return results
def gov_zhengce_list():
"""
中国政府网 - 国务院政策文件库
URL: https://www.gov.cn/zhengce/
结构: 服务端渲染,<li><a href="...">标题</a><span>日期</span></li>
"""
html = _fetch("https://www.gov.cn/zhengce/")
items = re.findall(
r'<li>\s*<a href="([^"]+)"[^>]*>\s*([^<]+)\s*</a>\s*<span>\s*([\d-]+)\s*</span>',
html,
)
results = []
for url, title, date in items:
title = title.strip()
if not title:
continue
if url.startswith("./"):
url = "https://www.gov.cn/zhengce/" + url[2:]
results.append({"title": title, "date": date, "url": url})
return results
def gov_article(url):
"""
中国政府网 - 政策文章正文
正文在 <div class="pages_content"> 内的 <p> 标签中
"""
html = _fetch(url)
# 提取标题
title_match = re.search(r'<title>([^<]+)</title>', html)
title = title_match.group(1).strip() if title_match else ""
# 提取发布日期
date_match = re.search(r"发布时间[:]\s*([\d年\-./]+)", html)
date = date_match.group(1).strip() if date_match else ""
# 提取正文
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": date,
"url": url,
"content": "\n\n".join(content_paras),
}
def scio_press_list():
"""
国新办 - 新闻发布会文字实录
注意: scio.gov.cn 有反爬保护,此处使用 PBC 站点作为镜像
PBC 的 "国新办举行新闻发布会" 系列完整保留了国新办原文
URL: https://www.pbc.gov.cn/goutongjiaoliu/113456/113469/index.html
"""
html = _fetch(
"https://www.pbc.gov.cn/goutongjiaoliu/113456/113469/index.html"
)
items = re.findall(
r'<a[^>]*href="(/goutongjiaoliu/[^"]+)"[^>]*>([^<]{10,})</a>',
html,
)
results = []
seen = set()
for href, title in items:
title = title.strip()
if title and href not in seen and "国新办" in title:
seen.add(href)
results.append(
{
"title": title,
"date": "",
"url": "https://www.pbc.gov.cn" + href,
}
)
return results
def xinhuanet_list():
"""
新华社 - 受权发布/时政
注意: 使用 www.news.cn(新域名),旧 xinhuanet.com 只有 2021 年旧数据
URL: https://www.news.cn/
结构: 服务端渲染,标题在 <div class="tit"><a href="URL">标题</a></div>
文章链接格式: /{channel}/{YYYYMMDD}/{32-hex}/c.html
"""
html = _fetch("https://www.news.cn/")
pattern = r'<div class="tit"[^>]*><a[^>]*href=[\'"](https?://www\.news\.cn/[^"\']+/20\d{6}/[a-f0-9]{32}/c\.html)[\'"][^>]*>([^<]+)</a>'
items = re.findall(pattern, html)
results = []
seen = set()
for url, title in items:
title = title.strip()
if title and url not in seen:
seen.add(url)
date_match = re.search(r"/(20\d{6})/", url)
date = ""
if date_match:
d = date_match.group(1)
date = f"{d[:4]}-{d[4:6]}-{d[6:8]}"
results.append({"title": title, "date": date, "url": url})
return results
def xinhuanet_article(url):
"""
新华社 - 文章正文
标题: <h1><span class="title">...</span></h1>
正文: <div id="detailContent"> 内的 <p> 标签
"""
html = _fetch(url)
title_match = re.search(r'<h1[^>]*>.*?<span class="title"[^>]*>([^<]+)</span>', html, re.DOTALL)
if not title_match:
title_match = re.search(r'<title>([^<]+)</title>', html)
title = title_match.group(1).strip() if title_match else ""
# 提取正文
content_match = re.search(r'id="detailContent"[^>]*>(.*?)</div>\s*(?:<div|</)', html, re.DOTALL)
if content_match:
content_html = content_match.group(1)
content_paras = _extract_paragraphs(content_html)
else:
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": "",
"url": url,
"content": "\n\n".join(content_paras),
}
# ============================================================
# 优先级 2: 部委官网
# ============================================================
def pbc_list():
"""
中国人民银行 - 货币政策/公告/沟通
URL: https://www.pbc.gov.cn/goutongjiaoliu/113456/113469/index.html
结构: 服务端渲染,<a href="/goutongjiaoliu/...">标题</a>
"""
html = _fetch(
"https://www.pbc.gov.cn/goutongjiaoliu/113456/113469/index.html"
)
items = re.findall(
r'<a[^>]*href="(/goutongjiaoliu/[^"]+)"[^>]*>([^<]{10,})</a>',
html,
)
results = []
seen = set()
for href, title in items:
title = title.strip()
if title and href not in seen and len(title) > 5:
seen.add(href)
# 从 URL 提取日期 (格式: 2026071518015458230)
date_match = re.search(r"/(20\d{6})\d{11}/", href)
date = ""
if date_match:
d = date_match.group(1)
date = f"{d[:4]}-{d[4:6]}-{d[6:8]}"
results.append(
{
"title": title,
"date": date,
"url": "https://www.pbc.gov.cn" + href,
}
)
return results
def pbc_article(url):
"""中国人民银行 - 文章正文(正文在 <p> 标签中)"""
html = _fetch(url)
title_match = re.search(r"<title>([^<]+)</title>", html)
title = title_match.group(1).strip() if title_match else ""
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": "",
"url": url,
"content": "\n\n".join(content_paras),
}
def mof_list():
"""
财政部 - 财政新闻/政策
URL: https://www.mof.gov.cn/zhengwuxinxi/caizhengxinwen/
结构: 服务端渲染,<a href="t202607...">标题</a> + 日期
"""
html = _fetch("https://www.mof.gov.cn/zhengwuxinxi/caizhengxinwen/")
items = re.findall(
r'href="([^"]*t2026[^"]+\.htm)"[^>]*>\s*([^<]{10,80})\s*</a>\s*(?:.*?(\d{4}[-/]\d{2}[-/]\d{2}))?',
html,
re.DOTALL,
)
results = []
seen = set()
for url, title, date in items:
title = title.strip()
if title and url not in seen:
seen.add(url)
if url.startswith("./") or url.startswith("t20"):
url = "https://www.mof.gov.cn/zhengwuxinxi/caizhengxinwen/" + url.replace("./", "")
results.append({"title": title, "date": date or "", "url": url})
return results
def mof_article(url):
"""财政部 - 文章正文"""
html = _fetch(url)
title_match = re.search(r"<title>([^<]+)</title>", html)
title = title_match.group(1).strip() if title_match else ""
date_match = re.search(r"发布日期[:]\s*([\d年\-./]+)", html)
date = date_match.group(1).strip() if date_match else ""
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": date,
"url": url,
"content": "\n\n".join(content_paras),
}
def ndrc_list(page=1):
"""
发改委 - 产业政策/规划
URL: https://www.ndrc.gov.cn/xxgk/
结构: 服务端渲染,<ul class="u-list"><li><a>标题</a><span>日期(YYYY/MM/DD)</span></li></ul>
分页: index_{page-1}.html
"""
if page == 1:
list_url = "https://www.ndrc.gov.cn/xxgk/"
else:
list_url = f"https://www.ndrc.gov.cn/xxgk/index_{page - 1}.html"
html = _fetch(list_url)
items = re.findall(
r'<a[^>]*href="([^"]+)"[^>]*>\s*([^<]{10,})\s*</a>\s*<span>\s*([\d/]+)\s*</span>',
html,
)
results = []
for url, title, date in items:
title = title.strip()
if not title:
continue
if url.startswith("./"):
url = "https://www.ndrc.gov.cn" + url[1:]
results.append({"title": title, "date": date, "url": url})
return results
def ndrc_article(url):
"""发改委 - 文章正文(正文在 div.TRS_Editor 内)"""
html = _fetch(url)
title_match = re.search(r"<title>([^<]+)</title>", html)
title = title_match.group(1).strip() if title_match else ""
editor_match = re.search(r'class="TRS_Editor"[^>]*>(.*?)</div>\s*<', html, re.DOTALL)
if editor_match:
content_paras = _extract_paragraphs(editor_match.group(1))
else:
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": "",
"url": url,
"content": "\n\n".join(content_paras),
}
def csrc_list(page=1):
"""
证监会 - 主席讲话/政策公告
URL: https://www.csrc.gov.cn/csrc/c100028/
结构: HTML 页面(API 返回空结果,需用 common/searchList 获取 HTML
HTML URL: /common/searchList/{channelid}?_isAgg=true&_isJson=true&_pageSize=18&_template=index&page={N}
"""
channel_id = "c100028"
url = f"https://www.csrc.gov.cn/common/searchList/{channel_id}?_isAgg=true&_isJson=true&_pageSize=18&_template=index&page={page}"
html = _fetch(url)
items = re.findall(
r'<a[^>]*href="(/csrc/c100028/c\d+/content\.shtml)"[^>]*>([^<]{10,})</a>',
html,
)
results = []
seen = set()
for path, title in items:
title = title.strip()
if title and path not in seen:
seen.add(path)
results.append(
{
"title": title,
"date": "",
"url": "https://www.csrc.gov.cn" + path,
}
)
return results
def csrc_article(url):
"""证监会 - 文章正文"""
html = _fetch(url)
title_match = re.search(r"<title>([^<]+)</title>", html)
title = title_match.group(1).strip() if title_match else ""
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": "",
"url": url,
"content": "\n\n".join(content_paras),
}
def mofcom_list():
"""
商务部 - 外贸/消费政策
URL: https://www.mofcom.gov.cn/
结构: 政策列表页使用 JSON API
API: /api-gateway/jpaas-publish-server/front/page/build/unit
"""
html = _fetch("https://www.mofcom.gov.cn/")
# 首页新闻是 SSR
items = re.findall(
r'<a[^>]*href="([^"]+)"[^>]*>\s*([^<]{10,80})\s*</a>',
html,
)
results = []
seen = set()
for url, title in items:
title = title.strip()
if title and url not in seen and len(title) > 8:
seen.add(url)
if url.startswith("/"):
url = "https://www.mofcom.gov.cn" + url
results.append({"title": title, "date": "", "url": url})
return results
def mofcom_article(url):
"""商务部 - 文章正文(正文在 div.wms-con 内)"""
html = _fetch(url)
title_match = re.search(r"<title>([^<]+)</title>", html)
title = title_match.group(1).strip() if title_match else ""
content_match = re.search(r'class="wms-con"[^>]*>(.*?)</div>\s*<', html, re.DOTALL)
if content_match:
content_paras = _extract_paragraphs(content_match.group(1))
else:
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": "",
"url": url,
"content": "\n\n".join(content_paras),
}
# ============================================================
# 优先级 3: 权威媒体转载
# ============================================================
def people_editorial_list():
"""
人民日报 - 社论/观点
URL: http://opinion.people.com.cn/GB/8213/49160/index.html
结构: 主页面有多个子栏目,每个子栏目页面有 <li> 文章列表
文章URL格式: /n1/2026/0623/c223228-40745676.html (相对) 或完整URL
"""
base = "http://opinion.people.com.cn/GB/8213/49160/"
html = _fetch(base + "index.html")
# 获取所有子栏目链接(完整URL
sub_urls = re.findall(r'href="(http://opinion\.people\.com\.cn/GB/8213/49160/\d+/index\.html)"', html)
sub_urls = list(dict.fromkeys(sub_urls))
results = []
seen = set()
for sub in sub_urls:
try:
sub_html = _fetch(sub)
# 解析 <li> 项目(注意:人民网的链接使用单引号 href='...'
lis = re.findall(r"<li[^>]*>(.*?)</li>", sub_html, re.DOTALL)
for li in lis:
# 匹配单引号或双引号的 href
m = re.search(r"href=['\"](/n1/\d{4}/\d{4}/c[\d-]+\.html)['\"][^>]*>([^<]{10,})</a>", li)
if m:
url_path, title = m.groups()
title = title.strip()
url = "http://opinion.people.com.cn" + url_path
if title and url not in seen:
seen.add(url)
date_match = re.search(r"/n1/(20\d{2})/(\d{4})/", url)
date = ""
if date_match:
year, md = date_match.groups()
date = f"{year}-{md[:2]}-{md[2:]}"
results.append({"title": title, "date": date, "url": url})
except Exception:
continue
return results
def people_article(url):
"""人民日报 - 文章正文(正文在 <p> 标签中,class=rm_txt_zw 的容器内)"""
html = _fetch(url)
title_match = re.search(r"<title>([^<]+)</title>", html)
title = title_match.group(1).strip() if title_match else ""
# 正文在 p 标签中
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": "",
"url": url,
"content": "\n\n".join(content_paras),
}
def cctv_list(page=1):
"""
央视新闻 - 新闻联播文字版
URL: https://news.cctv.com/lbj/
结构: JSONP API
API: https://news.cctv.com/2019/07/gaiban/cmsdatainterface/page/china_{page}.jsonp?callback=china
注意: 需要设置 Referer 头
"""
api_url = f"https://news.cctv.com/2019/07/gaiban/cmsdatainterface/page/china_{page}.jsonp?callback=china"
raw = _fetch(api_url, headers={"Referer": "https://news.cctv.com/"})
# 解析 JSONP: china({...})
json_match = re.search(r"[^(]+\((\{.*\})\)", raw, re.DOTALL)
if not json_match:
return []
try:
data = json.loads(json_match.group(1))
except json.JSONDecodeError:
return []
items = data.get("data", {}).get("list", []) if isinstance(data.get("data"), dict) else []
results = []
for item in items:
title = item.get("title", "").strip()
url = item.get("url", item.get("link", ""))
date = item.get("focus_date", item.get("date", ""))
if title and url:
results.append(
{
"title": title,
"date": str(date)[:10],
"url": url,
}
)
return results
def cctv_article(url):
"""央视新闻 - 文章正文(正文在 div.content_area 内)"""
html = _fetch(url)
title_match = re.search(r"<title>([^<]+)</title>", html)
title = title_match.group(1).strip() if title_match else ""
content_match = re.search(r'class="content_area"[^>]*>(.*?)</div>\s*<', html, re.DOTALL)
if content_match:
content_paras = _extract_paragraphs(content_match.group(1))
else:
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": "",
"url": url,
"content": "\n\n".join(content_paras),
}
def ce_list():
"""
经济日报 - 部委政策解读
URL: http://www.ce.cn/
结构: 服务端渲染,文章链接格式: /xwzx/gnsz/gdxw/202607/t20260721_xxxxxxx.shtml
政策相关文章在 /gnsz/ (国内时政) 和 /xwzx/ (新闻) 栏目下
"""
html = _fetch("http://www.ce.cn/")
items = re.findall(
r'<a[^>]*href="([^"]*t2026[^"]+\.shtml)"[^>]*>([^<]{10,})</a>',
html,
)
results = []
seen = set()
for url, title in items:
title = title.strip()
# 过滤政策相关文章(国内时政/新闻栏目)
if "/gnsz/" not in url and "/xwzx/" not in url and "/syj/" not in url:
continue
if title and url not in seen:
seen.add(url)
if url.startswith("/"):
url = "http://www.ce.cn" + url
# 从 URL 提取日期
date_match = re.search(r"/(20\d{6})/", url)
date = ""
if date_match:
d = date_match.group(1)
date = f"{d[:4]}-{d[4:6]}-{d[6:8]}"
results.append({"title": title, "date": date, "url": url})
return results
def ce_article(url):
"""经济日报 - 文章正文(正文在 div.content.clearfix 内)"""
html = _fetch(url)
title_match = re.search(r'id="NewsArticleTitle"[^>]*>([^<]+)', html)
if not title_match:
title_match = re.search(r"<title>([^<]+)</title>", html)
title = title_match.group(1).strip() if title_match else ""
content_match = re.search(r'class="content clearfix"[^>]*>(.*?)</div>\s*<', html, re.DOTALL)
if content_match:
content_paras = _extract_paragraphs(content_match.group(1))
else:
content_paras = _extract_paragraphs(html)
return {
"title": title,
"date": "",
"url": url,
"content": "\n\n".join(content_paras),
}
# ============================================================
# 统一入口
# ============================================================
SCRAPERS = {
"gov.cn": {"list": gov_zhengce_list, "article": gov_article, "name": "中国政府网"},
"xinhuanet": {"list": xinhuanet_list, "article": xinhuanet_article, "name": "新华社"},
"pbc": {"list": pbc_list, "article": pbc_article, "name": "中国人民银行"},
"mof": {"list": mof_list, "article": mof_article, "name": "财政部"},
"ndrc": {"list": ndrc_list, "article": ndrc_article, "name": "发改委"},
"csrc": {"list": csrc_list, "article": csrc_article, "name": "证监会"},
"mofcom": {"list": mofcom_list, "article": mofcom_article, "name": "商务部"},
"people": {"list": people_editorial_list, "article": people_article, "name": "人民日报"},
"cctv": {"list": cctv_list, "article": cctv_article, "name": "央视新闻"},
"ce": {"list": ce_list, "article": ce_article, "name": "经济日报"},
"scio": {"list": scio_press_list, "article": pbc_article, "name": "国新办(PBC镜像)"},
}
def scrape_list(site_key, **kwargs):
"""统一列表抓取入口"""
scraper = SCRAPERS.get(site_key)
if not scraper:
return []
return scraper["list"](**kwargs)
def scrape_article(site_key, url):
"""统一文章抓取入口"""
scraper = SCRAPERS.get(site_key)
if not scraper:
return {"title": "", "date": "", "url": url, "content": ""}
return scraper["article"](url)