Commit 60e25568 by zhenhuajiang

Initial commit

parents
# Dify API
# Example: http://localhost or https://your-dify.example.com
DIFY_BASE_URL=
DIFY_API_KEY=
# Either set dataset id directly, or let the agent auto-create one by name.
DIFY_DATASET_ID=
DIFY_DATASET_NAME=湖北人社政策知识库
DIFY_INDEXING_TECHNIQUE=high_quality
DIFY_BATCH_SIZE=50
# Local crawler
POLICY_REQUEST_TIMEOUT=20
POLICY_USER_AGENT=
# Optional local overrides
# POLICY_DB_PATH=
# POLICY_SITES_PATH=
.env
.venv/
data/
logs/
__pycache__/
*.pyc
*.pyo
*.pyd
.pytest_cache/
.mypy_cache/
.coverage
htmlcov/
.vscode/
.claude/
*.yml
# 湖北人社政策采集代理
这个仓库默认采集湖北省内各级人力资源和社会保障部门官网的政务公开政策栏目,并同步到 Dify 知识库。
当前默认站点配置文件是 `policy_agent/sources/hubei_hrss_sites.json`,已内置 18 个湖北省内人社局/人社厅政策入口;站点清单见 `hubei_hrss_zwgk_links.md``hubei_hrss_zwgk_links.csv`
## 目录
- `policy_agent/sources/hubei_hrss_sites.json`
湖北省人社系统站点配置
- `policy_agent/crawler.py`
通用抓取与正文抽取,包含宜昌统一公开平台 API 适配
- `policy_agent/storage.py`
SQLite 去重、版本管理、同步状态记录
- `policy_agent/dify_client.py`
Dify Dataset API 封装
- `policy_agent/cli.py`
命令行入口
- `policy_agent/scripts/setup_local.ps1`
一键初始化脚本
- `policy_agent/scripts/run_daily_update.ps1`
单次执行脚本
- `policy_agent/scripts/register_midnight_task.ps1`
每天 0 点计划任务注册脚本
## 运行要求
- Windows PowerShell
- Python 3.10 及以上
- 可访问 Dify API
- 如需 Playwright 渲染,允许下载 Chromium
## 快速开始
### 1. 克隆仓库
```powershell
git clone <your-repo-url>
cd "Policy Collection Agent"
```
### 2. 一键初始化
```powershell
powershell -ExecutionPolicy Bypass -File .\policy_agent\scripts\setup_local.ps1
```
脚本会自动:
- 创建仓库内 `.venv`
- 安装 `requirements.txt`
- 安装 Playwright Chromium
- 若不存在 `.env`,从 `.env.example` 复制一份
### 3. 配置 `.env`
至少填写:
```env
DIFY_BASE_URL=
DIFY_API_KEY=
```
可选配置:
```env
DIFY_DATASET_ID=
DIFY_DATASET_NAME=湖北人社政策知识库
DIFY_INDEXING_TECHNIQUE=high_quality
DIFY_BATCH_SIZE=50
POLICY_REQUEST_TIMEOUT=20
POLICY_USER_AGENT=
```
说明:
- 优先填写 `DIFY_DATASET_ID`
- 如果未填 `DIFY_DATASET_ID`,程序会按 `DIFY_DATASET_NAME` 查找或自动创建知识库
- 如果要临时切回别的站点文件,可用 `POLICY_SITES_PATH` 覆盖默认来源
### 4. 检查配置
```powershell
.\.venv\Scripts\python.exe -m policy_agent.cli --project-root . check-config
```
### 5. 预览单站点
```powershell
.\.venv\Scripts\python.exe -m policy_agent.cli --project-root . preview --site "武汉市人力资源和社会保障局" --limit 5 --max-pages 20
```
### 6. 执行一次全量同步
```powershell
.\.venv\Scripts\python.exe -m policy_agent.cli --project-root . run --max-pages 20
```
## 常用命令
```powershell
.\.venv\Scripts\python.exe -m policy_agent.cli --project-root . check-config
.\.venv\Scripts\python.exe -m policy_agent.cli --project-root . list-sites
.\.venv\Scripts\python.exe -m policy_agent.cli --project-root . preview --site "湖北省人力资源和社会保障厅" --limit 10 --max-pages 20
.\.venv\Scripts\python.exe -m policy_agent.cli --project-root . run --site "宜昌市人力资源和社会保障局" --max-pages 20
.\.venv\Scripts\python.exe -m policy_agent.cli --project-root . run --max-pages 300 --max-depth 3
```
## 每天 0 点自动运行
使用管理员 PowerShell:
```powershell
Set-Location "D:\Policy Collection Agent"
.\policy_agent\scripts\register_midnight_task.ps1
```
默认任务名是 `HubeiPolicyDailyUpdate`,触发时间为每天 `00:00`
## 去重与版本规则
- 同一 URL 且正文哈希不变:跳过
- 同一 URL 但正文发生变化:更新现有记录
- 不同 URL 但正文哈希相同:视为重复
- 同一 `policy_key` 出现新版本时:旧版本标记为 `revised`,新版本标记为 `effective`
- Dify 中旧 `document_id` 失效时:自动重建文档并回写新 ID
## 本地文件位置
- SQLite 数据文件默认在 `data/policy_agent.db`
- 日志文件默认在 `logs/policy_agent.log`
这些文件已经加入 `.gitignore`,不会污染仓库。
## 当前说明
- 默认配置聚焦“湖北人社系统政策公开栏目”,不再默认抓取综合政府门户站
- 宜昌市人社局走统一公开平台 API,`deptid` 已改为可配置,便于后续迁移到其他宜昌部门
- 如果 Dify 返回 `Could not connect to Weaviate`,说明是知识库向量库异常,不是采集代码本身的问题
from __future__ import annotations
import csv
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable
from urllib.parse import urljoin, urlparse
import requests
from bs4 import BeautifulSoup
HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
),
"Accept-Language": "zh-CN,zh;q=0.9",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
}
OUTPUT_MD = Path("hubei_government_zwgk_links.md")
OUTPUT_CSV = Path("hubei_government_zwgk_links.csv")
KEYWORDS = ("政务公开", "政府信息公开", "信息公开")
FALLBACK_PATHS = (
"/zwgk/",
"/zwgk/index.shtml",
"/xxgk/",
"/zfxxgk/",
"/gk/",
)
@dataclass(frozen=True)
class Region:
name: str
level: str
candidates: tuple[str, ...]
REGIONS: tuple[Region, ...] = (
Region("湖北省人民政府", "省级", ("https://www.hubei.gov.cn/",)),
Region("武汉市人民政府", "副省级市", ("https://www.wuhan.gov.cn/",)),
Region("黄石市人民政府", "地级市", ("https://www.huangshi.gov.cn/",)),
Region("十堰市人民政府", "地级市", ("https://www.shiyan.gov.cn/",)),
Region("宜昌市人民政府", "地级市", ("https://www.yichang.gov.cn/",)),
Region("襄阳市人民政府", "地级市", ("https://www.xiangyang.gov.cn/",)),
Region("鄂州市人民政府", "地级市", ("https://www.ezhou.gov.cn/",)),
Region("荆门市人民政府", "地级市", ("https://www.jingmen.gov.cn/",)),
Region("孝感市人民政府", "地级市", ("https://www.xiaogan.gov.cn/",)),
Region("荆州市人民政府", "地级市", ("https://www.jingzhou.gov.cn/",)),
Region("黄冈市人民政府", "地级市", ("https://www.huanggang.gov.cn/", "https://www.hg.gov.cn/")),
Region("咸宁市人民政府", "地级市", ("https://www.xianning.gov.cn/",)),
Region("随州市人民政府", "地级市", ("https://www.suizhou.gov.cn/",)),
Region("恩施土家族苗族自治州人民政府", "自治州", ("https://www.enshi.gov.cn/",)),
Region("仙桃市人民政府", "省直辖县级市", ("https://www.xiantao.gov.cn/",)),
Region("潜江市人民政府", "省直辖县级市", ("https://www.hbqj.gov.cn/",)),
Region("天门市人民政府", "省直辖县级市", ("https://www.tianmen.gov.cn/",)),
Region("神农架林区人民政府", "林区", ("https://www.snj.gov.cn/",)),
)
def fetch(url: str) -> requests.Response | None:
try:
response = requests.get(url, headers=HEADERS, timeout=20, verify=False)
response.raise_for_status()
if not response.encoding or response.encoding.lower() == "iso-8859-1":
response.encoding = response.apparent_encoding or "utf-8"
return response
except requests.RequestException:
return None
def simplify_text(text: str) -> str:
return "".join(text.split())
def iter_candidate_links(soup: BeautifulSoup, base_url: str) -> Iterable[tuple[str, str]]:
for anchor in soup.find_all("a", href=True):
text = simplify_text(anchor.get_text(" ", strip=True))
href = anchor["href"].strip()
if not href or href.lower().startswith("javascript:"):
continue
yield text, urljoin(base_url, href)
def find_zwgk_link(home_url: str, html: str) -> tuple[str | None, str]:
soup = BeautifulSoup(html, "html.parser")
candidates: list[tuple[int, str, str]] = []
for text, absolute_url in iter_candidate_links(soup, home_url):
if not any(keyword in text for keyword in KEYWORDS):
continue
score = 0
if "政务公开" in text:
score += 5
if "政府信息公开" in text:
score += 4
if "信息公开目录" in text:
score += 1
parsed = urlparse(absolute_url)
if parsed.netloc == urlparse(home_url).netloc:
score += 2
if any(token in absolute_url.lower() for token in ("/zwgk", "/xxgk", "/zfxxgk", "channelid")):
score += 2
candidates.append((score, text, absolute_url))
if candidates:
score, text, link = max(candidates, key=lambda item: item[0])
return link, text
return None, ""
def verify_public_link(homepage: str, public_link: str) -> tuple[str | None, str]:
response = fetch(public_link)
if not response:
return None, "政务公开入口无法访问"
text = simplify_text(response.text[:3000])
title = simplify_text(BeautifulSoup(response.text, "html.parser").title.get_text()) if BeautifulSoup(response.text, "html.parser").title else ""
if any(keyword in text or keyword in title for keyword in KEYWORDS):
return response.url, "已验证"
return response.url, "入口可访问,页面关键词较弱"
def probe_region(region: Region) -> dict[str, str]:
result = {
"地区": region.name,
"级别": region.level,
"官网首页": "",
"政务公开链接": "",
"状态": "官网不可访问",
"说明": "",
}
homepage_response = None
for candidate in region.candidates:
homepage_response = fetch(candidate)
if homepage_response:
result["官网首页"] = homepage_response.url
break
if not homepage_response:
return result
public_link, label = find_zwgk_link(homepage_response.url, homepage_response.text)
if public_link:
verified_link, note = verify_public_link(homepage_response.url, public_link)
if verified_link:
result["政务公开链接"] = verified_link
result["状态"] = "可访问"
result["说明"] = f"首页栏目文案:{label};{note}"
return result
for path in FALLBACK_PATHS:
fallback_url = urljoin(homepage_response.url, path)
verified_link, note = verify_public_link(homepage_response.url, fallback_url)
if verified_link:
result["政务公开链接"] = verified_link
result["状态"] = "可访问"
if label:
result["说明"] = f"首页栏目文案:{label};回退路径命中;{note}"
else:
result["说明"] = f"首页未直接识别栏目,回退路径命中;{note}"
return result
result["状态"] = "官网可访问,但未识别政务公开入口"
if label:
result["说明"] = f"首页发现相关栏目文案:{label},但链接验证失败"
else:
result["说明"] = "首页与常见公开路径均未命中"
return result
def write_markdown(rows: list[dict[str, str]]) -> None:
lines = [
"# 湖北省政府官网政务公开入口清单",
"",
"说明:本清单按“湖北省人民政府 + 各市州及省直辖县级行政区政府门户”整理,重点记录官网首页及其可访问的“政务公开/政府信息公开”入口。",
"",
"| 地区 | 级别 | 官网首页 | 政务公开链接 | 状态 | 说明 |",
"| --- | --- | --- | --- | --- | --- |",
]
for row in rows:
home = row["官网首页"] or "-"
public = row["政务公开链接"] or "-"
note = row["说明"].replace("|", "\\|")
lines.append(
f"| {row['地区']} | {row['级别']} | {home} | {public} | {row['状态']} | {note or '-'} |"
)
OUTPUT_MD.write_text("\n".join(lines) + "\n", encoding="utf-8")
def write_csv(rows: list[dict[str, str]]) -> None:
fieldnames = ["地区", "级别", "官网首页", "政务公开链接", "状态", "说明"]
with OUTPUT_CSV.open("w", newline="", encoding="utf-8-sig") as file:
writer = csv.DictWriter(file, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
def main() -> None:
requests.packages.urllib3.disable_warnings() # type: ignore[attr-defined]
rows = [probe_region(region) for region in REGIONS]
write_markdown(rows)
write_csv(rows)
for row in rows:
print(
f"{row['地区']}\t{row['状态']}\t{row['官网首页']}\t{row['政务公开链接']}\t{row['说明']}"
)
if __name__ == "__main__":
main()
# -*- coding: utf-8 -*-
"""调试:打印页面关键 HTML 结构"""
import sys, io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
import requests
from bs4 import BeautifulSoup
import re
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Accept-Language": "zh-CN,zh;q=0.9",
}
URL = "https://www.wuhan.gov.cn/zwgk/xxgk/zfgz_new/202602/t20260211_2728640.shtml"
resp = requests.get(URL, headers=HEADERS, timeout=15)
resp.encoding = "utf-8"
soup = BeautifulSoup(resp.text, "html.parser")
print("=== 所有 h1/h2 标签 ===")
for tag in soup.find_all(["h1","h2"]):
print(f" <{tag.name} class='{tag.get('class','')}' id='{tag.get('id','')}'>")
print(f" 文字: {tag.get_text(strip=True)[:80]}")
print("\n=== 包含'政府规章'/'决定' 文字的元素 ===")
for el in soup.find_all(string=re.compile(r'关于修改|政府规章|政府令')):
p = el.parent
print(f" <{p.name} class='{p.get('class','')}'>: {str(el).strip()[:80]}")
print("\n=== 文号模式(包含年号的文本)===")
for el in soup.find_all(string=re.compile(r'第\d+号|〔\d{4}〕|\(\d{4}\)')):
p = el.parent
print(f" <{p.name} class='{p.get('class','')}'>: {str(el).strip()[:100]}")
print("\n=== class 含 'title'/'tit'/'head' 的 div/p/span ===")
for el in soup.find_all(class_=re.compile(r'title|tit|head|subject', re.I)):
print(f" <{el.name} class='{el.get('class','')}'>: {el.get_text(strip=True)[:80]}")
# -*- coding: utf-8 -*-
"""
find_igs_api.py
暴力探测 IGS CMS(泰豪软件)常见 API 端点
武汉市政府网站使用 IGS CMS 系统
"""
import sys, io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
import requests
import json
import re
session = requests.Session()
session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Accept-Language": "zh-CN,zh;q=0.9",
"Referer": "https://www.wuhan.gov.cn/zwgk/?channelid=26164",
})
BASE = "https://www.wuhan.gov.cn"
CHANNEL_ID = "26164"
# 先访问一次列表页,获取 Cookie
session.get(f"{BASE}/zwgk/?channelid={CHANNEL_ID}", timeout=10)
# IGS CMS 所有已知 API 变体(根据泰豪 IGS 产品文档及社区逆向整理)
CANDIDATES = [
# 标准 IGS CMS 文章列表接口
("/igs/front/siteContent/getSiteContentList.jhtml", {"channelId": CHANNEL_ID, "pageIndex": 1, "pageSize": 15}),
("/igs/front/article/queryArticleList.jhtml", {"channelId": CHANNEL_ID, "pageIndex": 1, "pageSize": 15}),
("/igs/front/article/getArticleList.jhtml", {"channelId": CHANNEL_ID, "pageIndex": 1, "pageSize": 15}),
("/igs/front/content/getContentList.jhtml", {"channelId": CHANNEL_ID, "pageIndex": 1, "pageSize": 15}),
("/igs/front/channel/getChildList.jhtml", {"channelId": CHANNEL_ID, "pageIndex": 1, "pageSize": 15}),
# IGS 新版 REST 格式
("/igs/front/article/list", {"channelId": CHANNEL_ID, "page": 1, "size": 15}),
("/igs/front/v1/article/list", {"channelId": CHANNEL_ID, "page": 1, "size": 15}),
# 武汉政府网特有路径(基于 URL 模式推断)
("/zwgk/api/list", {"channelId": CHANNEL_ID, "pageIndex": 1}),
("/zwgk/ssi/queryList.jhtml", {"channelId": CHANNEL_ID, "pageIndex": 1}),
# 站群系统常见格式
("/ssi/front/queryArticleList.jhtml", {"channelId": CHANNEL_ID, "pageIndex": 1, "pageSize": 15}),
("/ssi/front/article/queryList.jhtml", {"channelId": CHANNEL_ID, "pageIndex": 1, "pageSize": 15}),
# 尝试 channelid(小写)
("/igs/front/article/queryArticleList.jhtml", {"channelid": CHANNEL_ID, "pageIndex": 1, "pageSize": 15}),
("/igs/front/siteContent/getSiteContentList.jhtml", {"channelid": CHANNEL_ID, "pageIndex": 1, "pageSize": 15}),
]
print("=" * 70)
print(f"探测 IGS CMS API 端点(channelId={CHANNEL_ID})")
print("=" * 70)
found = []
for path, params in CANDIDATES:
url = BASE + path
# 尝试 POST
try:
r = session.post(url, data=params, timeout=8)
ct = r.headers.get("Content-Type", "")
body = r.text[:400]
is_json = "json" in ct or (body.strip().startswith("{") or body.strip().startswith("["))
status_str = f"POST [{r.status_code}]"
if r.status_code in (200, 201) and len(r.content) > 100 and r.status_code != 404:
note = "★ 可能有效" if is_json else "HTML响应"
print(f"{status_str} {path} {note}")
print(f" CT={ct} size={len(r.content)}")
print(f" {body[:200]}")
if is_json:
found.append((url, "POST", params, body))
elif r.status_code not in (404, 301, 302, 403):
print(f"{status_str} {path} CT={ct}")
except Exception as e:
pass
# 尝试 GET(带参数)
try:
r = session.get(url, params=params, timeout=8)
ct = r.headers.get("Content-Type", "")
body = r.text[:400]
is_json = "json" in ct or (body.strip().startswith("{") or body.strip().startswith("["))
status_str = f"GET [{r.status_code}]"
if r.status_code in (200, 201) and len(r.content) > 100 and r.status_code != 404:
note = "★ 可能有效" if is_json else "HTML响应"
print(f"{status_str} {path} {note}")
print(f" CT={ct} size={len(r.content)}")
print(f" {body[:200]}")
if is_json:
found.append((url, "GET", params, body))
elif r.status_code not in (404, 301, 302, 403):
print(f"{status_str} {path} CT={ct}")
except Exception as e:
pass
print("\n" + "=" * 70)
print(f"发现 {len(found)} 个有效 JSON 接口")
print("=" * 70)
for url, method, params, body in found:
print(f" [{method}] {url} params={params}")
print(f" 响应: {body[:300]}")
# -*- coding: utf-8 -*-
"""
find_list_api.py
专门寻找武汉政府网政策列表的真实 API 端点
通过 Playwright 拦截浏览器的所有网络请求
"""
import sys, io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
from playwright.sync_api import sync_playwright
import json, re
LIST_URL = "https://www.wuhan.gov.cn/zwgk/?channelid=26164"
def find_list_api():
api_requests = []
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
context = browser.new_context(
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
locale="zh-CN",
)
page = context.new_page()
# 拦截所有请求
def on_request(request):
url = request.url
method = request.method
# 过滤掉图片、CSS、字体等静态资源
if not any(url.endswith(ext) for ext in ['.png', '.jpg', '.gif', '.css', '.woff', '.woff2', '.ico']):
api_requests.append({
"method": method,
"url": url,
"headers": dict(request.headers),
"post_data": request.post_data,
})
# 拦截所有响应,特别关注 JSON 格式的
json_responses = []
def on_response(response):
url = response.url
ct = response.headers.get("content-type", "")
if "json" in ct or "jhtml" in url:
try:
body = response.text()
if body and len(body) > 50:
json_responses.append({
"url": url,
"status": response.status,
"content_type": ct,
"body_preview": body[:500],
})
except:
pass
page.on("request", on_request)
page.on("response", on_response)
print(f"正在访问: {LIST_URL}")
page.goto(LIST_URL, wait_until="networkidle", timeout=30000)
page.wait_for_timeout(3000)
print(f"\n共拦截到 {len(api_requests)} 个非静态请求:")
print("-" * 60)
for req in api_requests:
print(f"[{req['method']}] {req['url']}")
if req['post_data']:
print(f" POST DATA: {req['post_data'][:200]}")
print(f"\n共拦截到 {len(json_responses)} 个 JSON 响应:")
print("-" * 60)
for resp in json_responses:
print(f"[{resp['status']}] {resp['url']}")
print(f" CT: {resp['content_type']}")
print(f" BODY: {resp['body_preview']}")
print()
# 获取渲染后的文章列表
print("\n渲染后页面的文章链接:")
print("-" * 60)
links = page.query_selector_all("a[href]")
articles = []
for link in links:
href = link.get_attribute("href") or ""
text = link.inner_text().strip()
if len(text) > 8 and ("/zwgk/" in href or re.search(r'\d{6}/t\d', href)):
articles.append((text, href))
for title, href in articles[:20]:
print(f" [{title[:50]}] -> {href}")
browser.close()
return api_requests, json_responses
if __name__ == "__main__":
find_list_api()
地区,级别,官网首页,政务公开链接,状态,备注
湖北省人民政府,省级,https://www.hubei.gov.cn/,https://www.hubei.gov.cn/xxgk/,当前环境受限,官网首页当前环境返回 412,xxgk 目录可从站内公开内容路径确认,建议浏览器直接访问核对
武汉市人民政府,副省级市,https://www.wuhan.gov.cn/,https://www.wuhan.gov.cn/zwgk/?channelid=26164,可访问,参考你提供的武汉示例链接
黄石市人民政府,地级市,https://www.huangshi.gov.cn/,https://www.huangshi.gov.cn/xxxgk/,可访问,首页政务公开入口
十堰市人民政府,地级市,https://www.shiyan.gov.cn/,http://www.shiyan.gov.cn/xxgk/,可访问,首页政府信息公开入口
宜昌市人民政府,地级市,http://www.yichang.gov.cn/,http://www.yichang.gov.cn/zfxxgk/,可访问,首页政府信息公开入口
襄阳市人民政府,地级市,http://www.xiangyang.gov.cn/wzsy/,http://xxgk.xiangyang.gov.cn/,可访问,首页政府信息公开跳转到独立公开子站
鄂州市人民政府,地级市,https://www.ezhou.gov.cn/sy/,https://www.ezhou.gov.cn/gk/,可访问,首页政府信息公开入口
荆门市人民政府,地级市,https://www.jingmen.gov.cn/,https://www.jingmen.gov.cn/col/col16474/index.html,可访问,首页政府信息公开目录入口
孝感市人民政府,地级市,https://www.xiaogan.gov.cn/,https://www.xiaogan.gov.cn/c/www/zc.jhtml,可访问,首页政府信息公开入口
荆州市人民政府,地级市,https://www.jingzhou.gov.cn/,http://zwgk.jingzhou.gov.cn/list_children.shtml?column_id=54272,可访问,首页政府信息公开目录入口
黄冈市人民政府,地级市,https://www.hg.gov.cn/,https://www.hg.gov.cn/zwgk/public/column/6636765?type=4&action=list&nav=0&id=7025468,可访问,首页政务公开入口
咸宁市人民政府,地级市,http://www.xianning.gov.cn/,http://www.xianning.gov.cn/xxgk/,可访问,首页政府信息公开入口
随州市人民政府,地级市,http://www.suizhou.gov.cn/,http://www.suizhou.gov.cn/zwgk/,可访问,首页政府信息公开入口
恩施土家族苗族自治州人民政府,自治州,http://www.enshi.gov.cn/,http://www.enshi.gov.cn/zc/,可访问,首页政府信息公开入口
仙桃市人民政府,省直辖县级市,https://www.xiantao.gov.cn/,https://www.xiantao.gov.cn/zfxxgk/,可访问,首页政府信息公开入口
潜江市人民政府,省直辖县级市,https://www.hbqj.gov.cn/,https://www.hbqj.gov.cn/xxgk/,可访问,首页含政府信息公开模块,公开内容集中在 xxgk 目录
天门市人民政府,省直辖县级市,http://www.tianmen.gov.cn/,http://www.tianmen.gov.cn/zwgk/,可访问,首页政府信息公开入口
神农架林区人民政府,林区,http://www.snj.gov.cn/,http://www.snj.gov.cn/zwgk/,可访问,首页政府信息公开入口
\ No newline at end of file
# 湖北省政府官网政务公开入口清单
说明:本清单按“省级门户 + 湖北省各市州、省直辖县级行政区政府门户”整理,重点记录官网首页及其“政务公开 / 政府信息公开”入口。
校验时间:2026-04-09
备注:湖北省人民政府门户在当前环境下直接请求返回 `412 Precondition Failed`,已单独标注;其余市州和省直辖县级行政区入口已逐个核到可访问页面。
| 地区 | 级别 | 官网首页 | 政务公开链接 | 状态 | 备注 |
| --- | --- | --- | --- | --- | --- |
| 湖北省人民政府 | 省级 | https://www.hubei.gov.cn/ | https://www.hubei.gov.cn/xxgk/ | 当前环境受限 | 官网首页当前环境返回 412,`xxgk` 目录可从站内公开内容路径确认,建议浏览器直接访问核对 |
| 武汉市人民政府 | 副省级市 | https://www.wuhan.gov.cn/ | https://www.wuhan.gov.cn/zwgk/?channelid=26164 | 可访问 | 参考你提供的武汉示例链接 |
| 黄石市人民政府 | 地级市 | https://www.huangshi.gov.cn/ | https://www.huangshi.gov.cn/xxxgk/ | 可访问 | 首页“政务公开”入口 |
| 十堰市人民政府 | 地级市 | https://www.shiyan.gov.cn/ | http://www.shiyan.gov.cn/xxgk/ | 可访问 | 首页“政府信息公开”入口 |
| 宜昌市人民政府 | 地级市 | http://www.yichang.gov.cn/ | http://www.yichang.gov.cn/zfxxgk/ | 可访问 | 首页“政府信息公开”入口 |
| 襄阳市人民政府 | 地级市 | http://www.xiangyang.gov.cn/wzsy/ | http://xxgk.xiangyang.gov.cn/ | 可访问 | 首页“政府信息公开”跳转到独立公开子站 |
| 鄂州市人民政府 | 地级市 | https://www.ezhou.gov.cn/sy/ | https://www.ezhou.gov.cn/gk/ | 可访问 | 首页“政府信息公开”入口 |
| 荆门市人民政府 | 地级市 | https://www.jingmen.gov.cn/ | https://www.jingmen.gov.cn/col/col16474/index.html | 可访问 | 首页“政府信息公开目录”入口 |
| 孝感市人民政府 | 地级市 | https://www.xiaogan.gov.cn/ | https://www.xiaogan.gov.cn/c/www/zc.jhtml | 可访问 | 首页“政府信息公开”入口 |
| 荆州市人民政府 | 地级市 | https://www.jingzhou.gov.cn/ | http://zwgk.jingzhou.gov.cn/list_children.shtml?column_id=54272 | 可访问 | 首页“政府信息公开目录”入口 |
| 黄冈市人民政府 | 地级市 | https://www.hg.gov.cn/ | https://www.hg.gov.cn/zwgk/public/column/6636765?type=4&action=list&nav=0&id=7025468 | 可访问 | 首页“政务公开”入口 |
| 咸宁市人民政府 | 地级市 | http://www.xianning.gov.cn/ | http://www.xianning.gov.cn/xxgk/ | 可访问 | 首页“政府信息公开”入口 |
| 随州市人民政府 | 地级市 | http://www.suizhou.gov.cn/ | http://www.suizhou.gov.cn/zwgk/ | 可访问 | 首页“政府信息公开”入口 |
| 恩施土家族苗族自治州人民政府 | 自治州 | http://www.enshi.gov.cn/ | http://www.enshi.gov.cn/zc/ | 可访问 | 首页“政府信息公开”入口 |
| 仙桃市人民政府 | 省直辖县级市 | https://www.xiantao.gov.cn/ | https://www.xiantao.gov.cn/zfxxgk/ | 可访问 | 首页“政府信息公开”入口 |
| 潜江市人民政府 | 省直辖县级市 | https://www.hbqj.gov.cn/ | https://www.hbqj.gov.cn/xxgk/ | 可访问 | 首页含“政府信息公开”模块,公开内容集中在 `xxgk` 目录 |
| 天门市人民政府 | 省直辖县级市 | http://www.tianmen.gov.cn/ | http://www.tianmen.gov.cn/zwgk/ | 可访问 | 首页“政府信息公开”入口 |
| 神农架林区人民政府 | 林区 | http://www.snj.gov.cn/ | http://www.snj.gov.cn/zwgk/ | 可访问 | 首页“政府信息公开”入口 |
name,homepage,public_entry,notes
湖北省人力资源和社会保障厅,https://rst.hubei.gov.cn/,https://rst.hubei.gov.cn/zfxxgk/zc/gfxwj/,省厅规范性文件
武汉市人力资源和社会保障局,https://rsj.wuhan.gov.cn/,https://rsj.wuhan.gov.cn/zwgk_17/zc/gfxwj/zwgk_list.html,示例站点
黄石市人力资源和社会保障局,https://rsj.huangshi.gov.cn/,https://rsj.huangshi.gov.cn/xxgk/zc/gfxwj/,含政策解读栏目
十堰市人力资源和社会保障局,http://rsj.shiyan.gov.cn/,http://rsj.shiyan.gov.cn/srlzyhshbzj/zc/gfxwj_new/,原zc页面会跳到gfxwj_new
宜昌市人力资源和社会保障局,http://rsj.yichang.gov.cn/,http://www.yichang.gov.cn/zfxxgk/list.html?depid=858&catid=522&t=4,统一公开平台并使用官方API
襄阳市人力资源和社会保障局,http://rsj.xiangyang.gov.cn/,http://rsj.xiangyang.gov.cn/zwgk/zc/zcfg/,含政策解读栏目
鄂州市人力资源和社会保障局,https://rsj.ezhou.gov.cn/,https://rsj.ezhou.gov.cn/xxgk/zc/bmgfxwj/,部门规范性文件
荆门市人力资源和社会保障局,http://rsj.jingmen.gov.cn/,http://rsj.jingmen.gov.cn/col/col10845/index.html,另补政策解读与参保政策入口
孝感市人力资源和社会保障局,https://rsj.xiaogan.gov.cn/,https://rsj.xiaogan.gov.cn/c/xgsrlzyhshbzj/zc.jhtml,http会自动跳转到https
荆州市人力资源和社会保障局,http://rsj.jingzhou.gov.cn/,http://jzrsj.zwgk.jingzhou.gov.cn/list.shtml?column_id=35910,政策文件在独立公开域名
黄冈市人力资源和社会保障局,https://rsj.hg.gov.cn/,https://rsj.hg.gov.cn/zwgk/public/column/6636189?type=4&catId=7026842&action=list&nav=0,公开平台列表示例
咸宁市人力资源和社会保障局,http://rsj.xianning.gov.cn/,http://rsj.xianning.gov.cn/xxgk/zc/wjzl/,规范性文件资料库
随州市人力资源和社会保障局,http://rsj.suizhou.gov.cn/,http://rsj.suizhou.gov.cn/fbjd_16/zwgk/xxgkml/zcfg/,另补通知公告入口
恩施州人力资源和社会保障局,http://rsj.enshi.gov.cn/,http://rsj.enshi.gov.cn/xxgk/zc/gfxwj/,同时补充政策解读和其他主动公开
仙桃市人力资源和社会保障局,https://www.xiantao.gov.cn/bmxxgk/srsj/,https://www.xiantao.gov.cn/bmxxgk/srsj/zfxxgk/zc/gfxwj/,由部门公开页逐级跳转进入
潜江市人力资源和社会保障局,https://www.hbqj.gov.cn/srlzyhshbzj/,https://www.hbqj.gov.cn/srlzyhshbzj/zfxxgk/zc/gfxwj/,当前仅确认规范性文件入口稳定
天门市人力资源和社会保障局,https://www.tianmen.gov.cn/zwgk/bmhxzxxgkml/bm/srlzyhshbzj/,https://www.tianmen.gov.cn/zwgk/bmhxzxxgkml/bm/srlzyhshbzj/zfxxgk/zc/bmwj/,同时补充政策解读入口
神农架林区人力资源和社会保障局,http://rsj.snj.gov.cn/,http://rsj.snj.gov.cn/zc_38462/,林区入口
# 湖北省人社局政务公开政策入口
当前项目默认采集的是湖北省内人力资源和社会保障部门官网的政策公开栏目,不再使用综合政府门户站。
| 名称 | 官网 | 政策入口 | 备注 |
| --- | --- | --- | --- |
| 湖北省人力资源和社会保障厅 | https://rst.hubei.gov.cn/ | https://rst.hubei.gov.cn/zfxxgk/zc/gfxwj/ | 省厅规范性文件 |
| 武汉市人力资源和社会保障局 | https://rsj.wuhan.gov.cn/ | https://rsj.wuhan.gov.cn/zwgk_17/zc/gfxwj/zwgk_list.html | 示例站点 |
| 黄石市人力资源和社会保障局 | https://rsj.huangshi.gov.cn/ | https://rsj.huangshi.gov.cn/xxgk/zc/gfxwj/ | 含政策解读栏目 |
| 十堰市人力资源和社会保障局 | http://rsj.shiyan.gov.cn/ | http://rsj.shiyan.gov.cn/srlzyhshbzj/zc/gfxwj_new/ | 原 `zc/` 页面会跳到 `gfxwj_new/` |
| 宜昌市人力资源和社会保障局 | http://rsj.yichang.gov.cn/ | http://www.yichang.gov.cn/zfxxgk/list.html?depid=858&catid=522&t=4 | 统一公开平台,列表由官方 API 提供 |
| 襄阳市人力资源和社会保障局 | http://rsj.xiangyang.gov.cn/ | http://rsj.xiangyang.gov.cn/zwgk/zc/zcfg/ | 含政策解读栏目 |
| 鄂州市人力资源和社会保障局 | https://rsj.ezhou.gov.cn/ | https://rsj.ezhou.gov.cn/xxgk/zc/bmgfxwj/ | 部门规范性文件 |
| 荆门市人力资源和社会保障局 | http://rsj.jingmen.gov.cn/ | http://rsj.jingmen.gov.cn/col/col10845/index.html | 另补了政策解读与参保政策入口 |
| 孝感市人力资源和社会保障局 | https://rsj.xiaogan.gov.cn/ | https://rsj.xiaogan.gov.cn/c/xgsrlzyhshbzj/zc.jhtml | `http` 会自动跳转到 `https` |
| 荆州市人力资源和社会保障局 | http://rsj.jingzhou.gov.cn/ | http://jzrsj.zwgk.jingzhou.gov.cn/list.shtml?column_id=35910 | 政策文件在独立公开域名 |
| 黄冈市人力资源和社会保障局 | https://rsj.hg.gov.cn/ | https://rsj.hg.gov.cn/zwgk/public/column/6636189?type=4&catId=7026842&action=list&nav=0 | 公开平台列表示例 |
| 咸宁市人力资源和社会保障局 | http://rsj.xianning.gov.cn/ | http://rsj.xianning.gov.cn/xxgk/zc/wjzl/ | 规范性文件资料库 |
| 随州市人力资源和社会保障局 | http://rsj.suizhou.gov.cn/ | http://rsj.suizhou.gov.cn/fbjd_16/zwgk/xxgkml/zcfg/ | 另补通知公告入口 |
| 恩施州人力资源和社会保障局 | http://rsj.enshi.gov.cn/ | http://rsj.enshi.gov.cn/xxgk/zc/gfxwj/ | 同时补充政策解读、其他主动公开 |
| 仙桃市人力资源和社会保障局 | https://www.xiantao.gov.cn/bmxxgk/srsj/ | https://www.xiantao.gov.cn/bmxxgk/srsj/zfxxgk/zc/gfxwj/ | 由部门公开页逐级跳转进入 |
| 潜江市人力资源和社会保障局 | https://www.hbqj.gov.cn/srlzyhshbzj/ | https://www.hbqj.gov.cn/srlzyhshbzj/zfxxgk/zc/gfxwj/ | 当前仅确认规范性文件入口稳定 |
| 天门市人力资源和社会保障局 | https://www.tianmen.gov.cn/zwgk/bmhxzxxgkml/bm/srlzyhshbzj/ | https://www.tianmen.gov.cn/zwgk/bmhxzxxgkml/bm/srlzyhshbzj/zfxxgk/zc/bmwj/ | 同时补充政策解读入口 |
| 神农架林区人力资源和社会保障局 | http://rsj.snj.gov.cn/ | http://rsj.snj.gov.cn/zc_38462/ | 林区入口 |
"""Hubei policy collection agent."""
from .pipeline import PipelineResult, PolicyPipeline
__all__ = ["PipelineResult", "PolicyPipeline"]
from __future__ import annotations
import argparse
import sys
from pathlib import Path
from .config import load_config, load_sites
from .pipeline import PolicyPipeline
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Hubei policy collection agent")
parser.add_argument(
"--project-root",
type=Path,
default=Path(__file__).resolve().parent.parent,
help="Project root path",
)
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("check-config", help="Validate local config")
subparsers.add_parser("list-sites", help="List configured sites")
preview_parser = subparsers.add_parser("preview", help="Preview crawl results without Dify sync")
preview_parser.add_argument("--site", required=True, help="Site name")
preview_parser.add_argument("--limit", type=int, default=10, help="Preview article count")
preview_parser.add_argument("--max-pages", type=int, help="Override max pages for this run")
preview_parser.add_argument("--max-depth", type=int, help="Override max depth for this run")
run_parser = subparsers.add_parser("run", help="Run crawler and Dify sync once")
run_parser.add_argument("--site", action="append", help="Optional site name filter")
run_parser.add_argument("--max-pages", type=int, help="Override max pages for this run")
run_parser.add_argument("--max-depth", type=int, help="Override max depth for this run")
return parser
def main() -> None:
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
if hasattr(sys.stderr, "reconfigure"):
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
parser = build_parser()
args = parser.parse_args()
config = load_config(args.project_root)
if args.command == "check-config":
errors = config.validate()
if errors:
raise SystemExit("\n".join(errors))
print("配置检查通过")
return
if args.command == "list-sites":
for site in load_sites(config):
print(f"{site.name}\t{site.public_entry}")
return
if args.command == "preview":
from .crawler import PolicyCrawler
sites = [site for site in load_sites(config) if site.name == args.site]
if not sites:
raise SystemExit(f"未找到站点: {args.site}")
if args.max_pages is not None:
from dataclasses import replace
sites = [replace(sites[0], max_pages=args.max_pages)]
if args.max_depth is not None:
from dataclasses import replace
sites = [replace(sites[0], max_depth=args.max_depth)]
crawler = PolicyCrawler(config)
outcome = crawler.crawl_site(sites[0])
print(f"pages={outcome.fetched_pages}\tfailed={outcome.failed_pages}\tarticles={len(outcome.articles)}")
for item in outcome.articles[: args.limit]:
print(f"{item.publish_date}\t{item.category}\t{item.title}\t{item.source_url}")
return
if args.command == "run":
pipeline = PolicyPipeline(config)
try:
results = pipeline.run(site_names=args.site, max_pages=args.max_pages, max_depth=args.max_depth)
finally:
pipeline.close()
for result in results:
print(
f"{result.site_name}\tpages={result.fetched_pages}\tarticles={result.discovered_articles}"
f"\tinserted={result.inserted}\tupdated={result.updated}\tdeleted={result.deleted}"
f"\tduplicates={result.duplicates}"
f"\tfailed={result.failed_pages}"
)
if __name__ == "__main__":
main()
from __future__ import annotations
import json
import os
from dataclasses import dataclass
from pathlib import Path
from .env import load_dotenv
from .models import SiteConfig
@dataclass(slots=True)
class AppConfig:
project_root: Path
data_dir: Path
logs_dir: Path
db_path: Path
sites_path: Path
request_timeout: int
user_agent: str
dify_base_url: str
dify_api_key: str
dify_dataset_id: str
dify_dataset_name: str
dify_indexing_technique: str
dify_batch_size: int
def validate(self) -> list[str]:
errors: list[str] = []
if not self.sites_path.exists():
errors.append(f"站点配置不存在: {self.sites_path}")
if not self.dify_base_url:
errors.append("缺少 DIFY_BASE_URL")
if not self.dify_api_key:
errors.append("缺少 DIFY_API_KEY")
if not self.dify_dataset_id and not self.dify_dataset_name:
errors.append("至少配置 DIFY_DATASET_ID 或 DIFY_DATASET_NAME")
return errors
def load_config(project_root: Path | None = None) -> AppConfig:
root = project_root or Path(__file__).resolve().parent.parent
load_dotenv(root / ".env")
data_dir = root / "data"
logs_dir = root / "logs"
data_dir.mkdir(exist_ok=True)
logs_dir.mkdir(exist_ok=True)
return AppConfig(
project_root=root,
data_dir=data_dir,
logs_dir=logs_dir,
db_path=Path(os.getenv("POLICY_DB_PATH", data_dir / "policy_agent.db")),
sites_path=Path(os.getenv("POLICY_SITES_PATH", root / "policy_agent" / "sources" / "hubei_hrss_sites.json")),
request_timeout=int(os.getenv("POLICY_REQUEST_TIMEOUT", "20")),
user_agent=os.getenv(
"POLICY_USER_AGENT",
(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
),
),
dify_base_url=os.getenv("DIFY_BASE_URL", "").rstrip("/"),
dify_api_key=os.getenv("DIFY_API_KEY", "").strip(),
dify_dataset_id=os.getenv("DIFY_DATASET_ID", "").strip(),
dify_dataset_name=os.getenv("DIFY_DATASET_NAME", "湖北人社政策知识库").strip(),
dify_indexing_technique=os.getenv("DIFY_INDEXING_TECHNIQUE", "high_quality").strip(),
dify_batch_size=int(os.getenv("DIFY_BATCH_SIZE", "50")),
)
def load_sites(config: AppConfig) -> list[SiteConfig]:
rows = json.loads(config.sites_path.read_text(encoding="utf-8"))
return [SiteConfig(**row) for row in rows]
from __future__ import annotations
import json
import re
from collections import deque
from dataclasses import dataclass
from typing import Iterable
from urllib.parse import urljoin, urlparse
import requests
from bs4 import BeautifulSoup
from .config import AppConfig
from .models import PolicyRecord, SiteConfig
from .utils import (
compact_text,
dedupe_preserve_order,
derive_policy_key,
extract_doc_no,
dumps_json,
extract_date,
guess_category,
has_non_policy_title_hint,
is_probable_policy_article,
sha256_text,
simplify_text,
summarize_for_embedding,
today_local,
)
def _fetch_rendered_html(url: str, user_agent: str, timeout: int) -> str | None:
"""用 Playwright 渲染页面,返回完整 HTML。Playwright 未安装时返回 None。"""
try:
from playwright.sync_api import sync_playwright, TimeoutError as PwTimeout
except ImportError:
return None
try:
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
ctx = browser.new_context(user_agent=user_agent, locale="zh-CN")
page = ctx.new_page()
page.goto(url, wait_until="networkidle", timeout=timeout * 1000)
page.wait_for_timeout(2000)
html = page.content()
browser.close()
return html
except PwTimeout:
return None
except Exception:
return None
ARTICLE_URL_PATTERN = re.compile(r"/(20\d{2})(\d{2})?/t?20\d{6}[_\d]*\.(?:s?html?)$", re.I)
POLICY_PATH_HINTS = (
"/zfwj/",
"/gfxwj/",
"/zcfg/",
"/zcjd/",
"/zcwj/",
"/qtwj/",
"/gsgg/",
"/tzgg/",
"/dczj/",
"/gfxwj_1/",
"/gfwj/",
"/qtzdgkwj/",
"/xxgk_zcfg/",
)
GENERIC_SECTION_TITLES = (
"公示公告",
"通知公告",
"政策解读",
"政府信息公开",
"信息公开指南",
"信息公开制度",
"规范性文件统一发布平台",
)
NEWS_PATH_HINTS = (
"/xwdt/",
"/ywdt/",
"/sytt/",
"/whyw/",
"/tt/",
"/focus/",
"/tpxw/",
"/xgyw/",
"/esxw/",
"/syyw/",
)
DEFAULT_CONTENT_SELECTORS = (
"div.article",
"div.article_content",
"div.article-content",
"div.articleContent",
"div.article_con",
"div.article_main",
"div#article",
"div.pages_content",
"div.detail",
"div.detail_content",
"div.wp_articlecontent",
"div.view",
"div.content",
"div.Content",
"div.TRS_Editor",
"div.zwxlb_nr",
)
TITLE_SELECTORS = (
"h1.doctitle",
"h3.doctitle",
"h1.article_title",
"h2.article_title",
"h1.article-title",
"h2.article-title",
"div.articleTitle",
"div.content_title",
"h1.title",
"h2.title",
"h1",
"h2",
"div.articleTitle",
)
DATE_SELECTORS = (
"div.article_info",
"div.article-time",
"p.info",
"span.time",
"div.info span",
"div.info",
"div.pages-date",
"span.pubtime",
"span.date",
)
GENERIC_INDEX_TITLES = ("信息公开", "政府信息公开", "政务公开", "首页")
BUREAU_INDEX_SUFFIXES = (
"规范性文件",
"政策解读",
"公示公告",
"通知公告",
"其他主动公开文件",
"政务公开",
)
YICHANG_API_BASE = "https://xxgkapi.yichang.gov.cn/"
YICHANG_API_ROUTE_TEMPLATES = (
("other/governmentdoc", "582"),
("other/governmentdoc", "583"),
("other/governmentdoc", "584"),
("other/govpretation", None),
)
HUBEI_POLICY_RESULT_PREFIXES = (
"http://www.hubei.gov.cn/zfwj/ezbf/",
"https://www.hubei.gov.cn/zfwj/ezbf/",
"http://www.hubei.gov.cn/zfwj/ezf/",
"https://www.hubei.gov.cn/zfwj/ezf/",
"http://www.hubei.gov.cn/zfwj/ezbd/",
"https://www.hubei.gov.cn/zfwj/ezbd/",
"http://www.hubei.gov.cn/zfwj/ezbh/",
"https://www.hubei.gov.cn/zfwj/ezbh/",
"http://www.hubei.gov.cn/zfwj/szfl/",
"https://www.hubei.gov.cn/zfwj/szfl/",
"http://www.hubei.gov.cn/xxgk/gz/",
"https://www.hubei.gov.cn/xxgk/gz/",
"http://www.hubei.gov.cn/xxgk/hbzcjd/zcjdjgh/",
"https://www.hubei.gov.cn/xxgk/hbzcjd/zcjdjgh/",
)
HUBEI_SOGOU_QUERIES = (
"site:www.hubei.gov.cn/zfwj/ezbf 通知",
"site:www.hubei.gov.cn/zfwj/ezbf 意见",
"site:www.hubei.gov.cn/zfwj/ezf 通知",
"site:www.hubei.gov.cn/zfwj/ezf 意见",
"site:www.hubei.gov.cn/zfwj/ezbd 通知",
"site:www.hubei.gov.cn/zfwj/ezbh 通知",
"site:www.hubei.gov.cn/zfwj/szfl 办法",
"site:www.hubei.gov.cn/xxgk/gz 办法",
"site:www.hubei.gov.cn/xxgk/hbzcjd/zcjdjgh 解读",
)
HUBEI_MARKDOWN_SKIP_LINES = {
"×",
"简 繁",
"拼音",
"辅助浏览",
"轻松阅读",
"读屏",
"视图",
"放大",
"缩小",
"配色",
"白底黑字蓝链接",
"蓝底黄字白链接",
"黄底黑字蓝链接",
"黑底黄字白链接",
"页面原始配色",
"大鼠标",
"辅助线",
"显示屏",
"声音",
"指读",
"连读",
"减速",
"加速",
"音量",
"增加音量",
"减小音量",
"后退",
"前进",
"全屏",
"帮助",
"换肤",
"搜索",
"首页",
"政务公开",
"政务服务",
"政民互动",
"知音湖北",
}
JSONP_WRAPPER_PATTERN = re.compile(r"^[^(]+\((.*)\)\s*;?\s*$", re.S)
@dataclass(slots=True)
class CrawlOutcome:
fetched_pages: int
failed_pages: int
articles: list[PolicyRecord]
class PolicyCrawler:
def __init__(self, config: AppConfig) -> None:
self.config = config
requests.packages.urllib3.disable_warnings() # type: ignore[attr-defined]
self.session = requests.Session()
self.session.headers.update(
{
"User-Agent": config.user_agent,
"Accept-Language": "zh-CN,zh;q=0.9",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
}
)
def fetch(self, url: str) -> requests.Response | None:
try:
response = self.session.get(url, timeout=self.config.request_timeout, verify=False)
response.raise_for_status()
if not response.encoding or response.encoding.lower() == "iso-8859-1":
response.encoding = response.apparent_encoding or "utf-8"
return response
except requests.RequestException:
return None
def _request_json(self, url: str, params: dict[str, object] | None = None, headers: dict[str, str] | None = None) -> dict[str, object] | None:
try:
response = self.session.get(
url,
params=params,
headers=headers,
timeout=self.config.request_timeout,
verify=False,
)
response.raise_for_status()
return response.json()
except (ValueError, requests.RequestException):
return None
def _request_json_payload(
self,
url: str,
params: dict[str, object] | None = None,
headers: dict[str, str] | None = None,
) -> object | None:
try:
response = self.session.get(
url,
params=params,
headers=headers,
timeout=self.config.request_timeout,
verify=False,
)
response.raise_for_status()
except requests.RequestException:
return None
try:
return response.json()
except ValueError:
text = response.text.strip()
match = JSONP_WRAPPER_PATTERN.match(text)
if not match:
return None
try:
payload_text = re.sub(r",(\s*[\]}])", r"\1", match.group(1))
return json.loads(payload_text)
except ValueError:
return None
def _fetch_jina_markdown(self, url: str) -> str | None:
proxy_url = f"https://r.jina.ai/http://{url}"
try:
response = self.session.get(proxy_url, timeout=max(60, self.config.request_timeout), verify=False)
response.raise_for_status()
return response.text
except requests.RequestException:
return None
def _crawl_candidate_urls(self, site: SiteConfig, candidate_urls: Iterable[str]) -> CrawlOutcome:
articles: list[PolicyRecord] = []
article_urls: set[str] = set()
fetched_pages = 0
failed_pages = 0
for current_url in candidate_urls:
response = self.fetch(current_url)
if not response:
failed_pages += 1
continue
fetched_pages += 1
soup = BeautifulSoup(response.text, "html.parser")
article = self.parse_article(site, response.url, soup)
if article and article.source_url not in article_urls:
article_urls.add(article.source_url)
articles.append(article)
return CrawlOutcome(fetched_pages=fetched_pages, failed_pages=failed_pages, articles=articles)
def _crawl_yichang(self, site: SiteConfig) -> CrawlOutcome:
requests.packages.urllib3.disable_warnings() # type: ignore[attr-defined]
api_pages = max(1, min(4, site.max_pages // 80 + 1))
candidate_urls: list[str] = []
deptid = (site.api_deptid or "").strip() or "846"
for endpoint, catid in YICHANG_API_ROUTE_TEMPLATES:
for page in range(1, api_pages + 1):
params: dict[str, object] = {
"areaid": "1",
"deptid": deptid,
"page": page,
"limit": 20,
"keywords": "",
}
if catid:
params["catid"] = catid
payload = self._request_json(
f"{YICHANG_API_BASE}{endpoint}",
params=params,
headers={"Referer": site.public_entry},
)
if not payload:
continue
data = payload.get("data")
if not isinstance(data, dict):
continue
items = data.get("lists")
if not isinstance(items, list) or not items:
break
for item in items:
if not isinstance(item, dict):
continue
release_url = item.get("releaseaddress")
if isinstance(release_url, str) and release_url.strip():
candidate_urls.append(release_url.strip())
if len(items) < 20:
break
return self._crawl_candidate_urls(site, dedupe_preserve_order(candidate_urls)[: site.max_pages])
def _search_sogou_urls(self, query: str, page: int) -> list[str]:
params: dict[str, object] = {"query": query}
if page > 1:
params["page"] = page
params["ie"] = "utf8"
try:
response = requests.get(
"https://www.sogou.com/web",
params=params,
headers={
"User-Agent": self.config.user_agent,
"Accept-Language": "zh-CN,zh;q=0.9",
"Referer": "https://www.sogou.com/",
},
timeout=self.config.request_timeout,
verify=False,
)
response.raise_for_status()
if not response.encoding or response.encoding.lower() == "iso-8859-1":
response.encoding = response.apparent_encoding or "utf-8"
except requests.RequestException:
return []
urls: list[str] = []
for match in re.finditer(r'https?://www\.hubei\.gov\.cn[^"\'\s<>]+', response.text):
url = match.group(0).strip()
if "..." in url:
continue
urls.append(url)
return dedupe_preserve_order(urls)
def _is_hubei_policy_result(self, url: str) -> bool:
return any(url.startswith(prefix) for prefix in HUBEI_POLICY_RESULT_PREFIXES)
def _clean_hubei_markdown_body(self, title: str, markdown: str) -> str:
body = markdown.split("Markdown Content:", 1)[-1].replace("\r\n", "\n").strip()
if title:
last_title_pos = body.rfind(title)
if last_title_pos != -1:
body = body[last_title_pos + len(title) :].strip()
lines: list[str] = []
for raw_line in body.splitlines():
line = simplify_text(raw_line)
if not line:
continue
if line in HUBEI_MARKDOWN_SKIP_LINES:
continue
if line.startswith("当前位置:"):
continue
if line.startswith("本站PC版"):
continue
if line.startswith("编辑:") or line.startswith("责编:") or line.startswith("审核:"):
break
if line.startswith("扫一扫在手机上查看当前页面"):
break
if line.startswith("* [") or line.startswith("* ["):
continue
lines.append(line)
return "\n".join(lines).strip()
def _parse_hubei_markdown_article(self, site: SiteConfig, url: str, markdown: str) -> PolicyRecord | None:
title_match = re.search(r"^Title:\s*(.+)$", markdown, flags=re.MULTILINE)
title = simplify_text(title_match.group(1)) if title_match else ""
for sep in (" -- ", " - ", "-", "_", "|", "—"):
if sep in title:
title = title.split(sep, 1)[0].strip()
break
if not title or self.is_generic_title(title) or has_non_policy_title_hint(title):
return None
content = self._clean_hubei_markdown_body(title, markdown)
if len(compact_text(content)) < 180:
return None
if not is_probable_policy_article(title, url, content):
return None
publish_date = extract_date(content[:2000]) or extract_date(url)
doc_no = extract_doc_no(content[:4000]) or extract_doc_no(title)
category = guess_category(title, url)
payload_for_hash = json.dumps(
{"title": title, "content": content, "publish_date": publish_date, "doc_no": doc_no},
ensure_ascii=False,
sort_keys=True,
)
source_site = urlparse(url).netloc
policy_key = derive_policy_key(site.name, title, category, url, doc_no)
return PolicyRecord(
site_name=site.name,
region=site.region,
level=site.level,
source_site=source_site,
source_url=url,
title=title,
content=content,
publish_date=publish_date,
crawl_date=today_local(),
category=category,
doc_no=doc_no,
effective_date=publish_date,
status="effective",
policy_key=policy_key,
version_no=1,
content_hash=sha256_text(payload_for_hash),
summary_text=summarize_for_embedding(title, publish_date, doc_no, category, content),
attachments_json=dumps_json([]),
)
def _crawl_hubei_province(self, site: SiteConfig) -> CrawlOutcome:
requests.packages.urllib3.disable_warnings() # type: ignore[attr-defined]
search_pages = max(1, min(4, site.max_pages // 80 + 1))
candidate_urls: list[str] = []
for query in HUBEI_SOGOU_QUERIES:
for page in range(1, search_pages + 1):
candidate_urls.extend(self._search_sogou_urls(query, page))
candidate_urls = [
url for url in dedupe_preserve_order(candidate_urls) if self._is_hubei_policy_result(url)
]
articles: list[PolicyRecord] = []
article_urls: set[str] = set()
fetched_pages = 0
failed_pages = 0
for current_url in candidate_urls[: site.max_pages]:
markdown = self._fetch_jina_markdown(current_url)
if not markdown:
failed_pages += 1
continue
fetched_pages += 1
article = self._parse_hubei_markdown_article(site, current_url, markdown)
if article and article.source_url not in article_urls:
article_urls.add(article.source_url)
articles.append(article)
return CrawlOutcome(fetched_pages=fetched_pages, failed_pages=failed_pages, articles=articles)
def _crawl_wuhan_jsonp(self, site: SiteConfig) -> CrawlOutcome:
candidate_urls: list[str] = []
if site.api_list_url:
payload = self._request_json_payload(site.api_list_url)
if isinstance(payload, list):
for item in payload:
if not isinstance(item, dict):
continue
url = item.get("url")
if isinstance(url, str) and url.strip():
candidate_urls.append(url.strip())
jsonp_outcome = self._crawl_candidate_urls(site, dedupe_preserve_order(candidate_urls)[: site.max_pages])
default_outcome = self._crawl_default(site)
merged_articles: list[PolicyRecord] = []
seen_urls: set[str] = set()
for outcome in (jsonp_outcome, default_outcome):
for article in outcome.articles:
if article.source_url in seen_urls:
continue
seen_urls.add(article.source_url)
merged_articles.append(article)
return CrawlOutcome(
fetched_pages=jsonp_outcome.fetched_pages + default_outcome.fetched_pages,
failed_pages=jsonp_outcome.failed_pages + default_outcome.failed_pages,
articles=merged_articles,
)
def _crawl_rendered_entry(self, site: SiteConfig) -> CrawlOutcome:
entry_urls = dedupe_preserve_order([site.public_entry, *site.seed_urls])
candidate_urls: list[str] = []
rendered_pages = 0
failed_pages = 0
for current_url in entry_urls:
rendered = _fetch_rendered_html(current_url, self.config.user_agent, self.config.request_timeout)
if not rendered:
failed_pages += 1
continue
rendered_pages += 1
soup = BeautifulSoup(rendered, "html.parser")
candidate_urls.extend(self.extract_candidate_links(site, current_url, soup))
candidate_urls.extend(self.extract_candidate_links_from_html(site, current_url, rendered))
article_outcome = self._crawl_candidate_urls(site, dedupe_preserve_order(candidate_urls)[: site.max_pages])
return CrawlOutcome(
fetched_pages=rendered_pages + article_outcome.fetched_pages,
failed_pages=failed_pages + article_outcome.failed_pages,
articles=article_outcome.articles,
)
def _crawl_default(self, site: SiteConfig) -> CrawlOutcome:
requests.packages.urllib3.disable_warnings() # type: ignore[attr-defined]
queue: deque[tuple[str, int]] = deque()
queued_urls = dedupe_preserve_order([site.homepage, site.public_entry, *site.seed_urls])
for url in queued_urls:
queue.append((url, 0))
visited: set[str] = set()
article_urls: set[str] = set()
articles: list[PolicyRecord] = []
fetched_pages = 0
failed_pages = 0
while queue and fetched_pages < site.max_pages:
current_url, depth = queue.popleft()
if current_url in visited:
continue
visited.add(current_url)
# 判断当前 URL 是否是文章详情页(通过 URL 模式判断)
is_article_url = bool(ARTICLE_URL_PATTERN.search(current_url))
# 文章详情页始终用 requests 直接抓(渲染开销大且不必要)
# 列表/目录页若站点配置了 use_playwright,则用 Playwright 渲染
use_playwright_for_this = site.use_playwright and not is_article_url
if use_playwright_for_this:
rendered = _fetch_rendered_html(current_url, self.config.user_agent, self.config.request_timeout)
if rendered:
fetched_pages += 1
soup = BeautifulSoup(rendered, "html.parser")
for link in self.extract_candidate_links(site, current_url, soup):
if link not in visited:
queue.append((link, depth + 1))
continue
response = self.fetch(current_url)
if not response:
failed_pages += 1
continue
fetched_pages += 1
soup = BeautifulSoup(response.text, "html.parser")
if self.is_article_page(response.url, soup):
article = self.parse_article(site, response.url, soup)
if article and article.source_url not in article_urls:
article_urls.add(article.source_url)
articles.append(article)
continue
if depth >= site.max_depth:
continue
links = dedupe_preserve_order(
[
*self.extract_candidate_links(site, response.url, soup),
*self.extract_candidate_links_from_html(site, response.url, response.text),
]
)
for link in links:
if link not in visited:
queue.append((link, depth + 1))
return CrawlOutcome(fetched_pages=fetched_pages, failed_pages=failed_pages, articles=articles)
def crawl_site(self, site: SiteConfig) -> CrawlOutcome:
if site.crawl_mode == "yichang_api":
return self._crawl_yichang(site)
if site.crawl_mode == "wuhan_jsonp":
return self._crawl_wuhan_jsonp(site)
if site.crawl_mode == "render_entry":
return self._crawl_rendered_entry(site)
if site.crawl_mode == "hubei_search" or site.name == "湖北省人民政府":
return self._crawl_hubei_province(site)
return self._crawl_default(site)
def score_candidate_link(self, url: str, text: str) -> int:
lower_url = url.lower()
score = 0
if ARTICLE_URL_PATTERN.search(lower_url):
score += 100
if any(token in lower_url for token in POLICY_PATH_HINTS):
score += 70
if any(token in lower_url for token in ("/zwgk/", "/xxgk/", "/zfxxgk/", "/zc/", "/gk/")):
score += 30
if any(token in text for token in ("政策", "通知", "公告", "意见", "办法", "解读", "规范性文件", "征求意见")):
score += 25
if any(token in lower_url for token in NEWS_PATH_HINTS):
score -= 60
if any(token in text for token in ("图片", "视频", "要闻", "动态", "新闻")):
score -= 20
return score
def extract_candidate_links(self, site: SiteConfig, base_url: str, soup: BeautifulSoup) -> list[str]:
allowed_hosts = set(site.allowed_hosts)
include_keywords = [keyword.lower() for keyword in site.include_url_keywords]
exclude_keywords = [keyword.lower() for keyword in site.exclude_url_keywords]
base_lower = base_url.lower()
candidates: dict[str, int] = {}
for anchor in soup.find_all("a", href=True):
href = anchor["href"].strip()
if not href or href.lower().startswith("javascript:"):
continue
if any(href.lower().endswith(ext) for ext in (".pdf", ".doc", ".docx", ".xls", ".xlsx", ".zip")):
continue
absolute = urljoin(base_url, href)
parsed = urlparse(absolute)
if parsed.scheme not in ("http", "https"):
continue
if parsed.netloc not in allowed_hosts:
continue
text = compact_text(anchor.get_text(" ", strip=True))
lower_url = absolute.lower()
if exclude_keywords and any(keyword in lower_url for keyword in exclude_keywords):
continue
if include_keywords and not (
any(keyword in lower_url for keyword in include_keywords)
or (any(keyword in base_lower for keyword in include_keywords) and ARTICLE_URL_PATTERN.search(lower_url))
):
continue
score = self.score_candidate_link(absolute, text)
if ARTICLE_URL_PATTERN.search(lower_url):
candidates[absolute] = max(candidates.get(absolute, -10_000), score)
continue
if any(token in lower_url for token in ("/zwgk/", "/xxgk/", "/zfxxgk/", "/zc/", "/gk/")):
candidates[absolute] = max(candidates.get(absolute, -10_000), score)
continue
if any(token in text for token in ("政策", "通知", "公告", "公开", "解读", "规范性文件")):
candidates[absolute] = max(candidates.get(absolute, -10_000), score)
return [url for url, _ in sorted(candidates.items(), key=lambda item: item[1], reverse=True)]
def extract_candidate_links_from_html(self, site: SiteConfig, base_url: str, html: str) -> list[str]:
allowed_hosts = set(site.allowed_hosts)
include_keywords = [keyword.lower() for keyword in site.include_url_keywords]
exclude_keywords = [keyword.lower() for keyword in site.exclude_url_keywords]
base_lower = base_url.lower()
candidates: dict[str, int] = {}
for raw_url in re.findall(r'(?:"|\')((?:https?://|/)[^"\'<>\s]+)(?:"|\')', html):
absolute = urljoin(base_url, raw_url.strip())
parsed = urlparse(absolute)
if parsed.scheme not in ("http", "https"):
continue
if parsed.netloc not in allowed_hosts:
continue
lower_url = absolute.lower()
if exclude_keywords and any(keyword in lower_url for keyword in exclude_keywords):
continue
if include_keywords and not (
any(keyword in lower_url for keyword in include_keywords)
or (any(keyword in base_lower for keyword in include_keywords) and ARTICLE_URL_PATTERN.search(lower_url))
):
continue
if any(lower_url.endswith(ext) for ext in (".pdf", ".doc", ".docx", ".xls", ".xlsx", ".zip")):
continue
if not (
ARTICLE_URL_PATTERN.search(lower_url)
or any(token in lower_url for token in POLICY_PATH_HINTS)
or any(token in lower_url for token in ("/zwgk/", "/xxgk/", "/zfxxgk/", "/zc/", "/gk/", "/art/"))
):
continue
score = self.score_candidate_link(absolute, "")
if ARTICLE_URL_PATTERN.search(lower_url):
score += 15
candidates[absolute] = max(candidates.get(absolute, -10_000), score)
return [url for url, _ in sorted(candidates.items(), key=lambda item: item[1], reverse=True)]
def is_generic_title(self, title: str) -> bool:
compact_title = compact_text(title)
if not compact_title:
return True
if compact_title in GENERIC_INDEX_TITLES or compact_title in GENERIC_SECTION_TITLES:
return True
if compact_title.startswith("信息公开-"):
return True
if (
"人民政府" in compact_title
and not any(token in compact_title for token in ("通知", "公告", "意见", "办法", "决定", "解读", "条例", "细则"))
and len(compact_title) <= 30
):
return True
if "门户网站" in compact_title and len(compact_title) <= 20:
return True
if "政府信息公开" in compact_title and len(compact_title) <= 20:
return True
if "标准化规范化建设" in compact_title:
return True
if (
any(token in compact_title for token in ("人力资源和社会保障局", "人力资源和社会保障厅", "人社局"))
and any(compact_title.endswith(suffix) for suffix in BUREAU_INDEX_SUFFIXES)
and len(compact_title) <= 30
):
return True
return False
def is_article_page(self, url: str, soup: BeautifulSoup) -> bool:
parsed = urlparse(url)
lower_path = parsed.path.lower()
if "channelid=" in parsed.query and not lower_path.endswith((".shtml", ".html", ".htm")):
return False
if lower_path.endswith("/") and not ARTICLE_URL_PATTERN.search(url):
return False
title = self.extract_title(soup)
if not title:
return False
if self.is_generic_title(title):
return False
if has_non_policy_title_hint(title):
return False
if len(soup.find_all("a", href=True)) > 120 and not ARTICLE_URL_PATTERN.search(url):
return False
content = self.extract_content(soup)
if not content or len(compact_text(content)) < 180:
return False
if ARTICLE_URL_PATTERN.search(url):
return True
if not is_probable_policy_article(title, url, content):
return False
if lower_path.endswith((".shtml", ".html", ".htm")):
return True
return bool(extract_doc_no(content[:1200]) or extract_date(content[:1000]))
def extract_title(self, soup: BeautifulSoup) -> str:
fallback = ""
for selector in TITLE_SELECTORS:
for element in soup.select(selector):
title = simplify_text(element.get_text(" ", strip=True))
if title:
if not self.is_generic_title(title):
return title
if not fallback:
fallback = title
if soup.title and soup.title.string:
title_tag = simplify_text(soup.title.string)
for sep in (" -- ", " - ", "-", "_", "|", "—"):
if sep in title_tag:
title_tag = title_tag.split(sep, 1)[0].strip()
break
if title_tag and not self.is_generic_title(title_tag):
return title_tag
if not fallback:
fallback = title_tag
return fallback
def extract_publish_date(self, url: str, soup: BeautifulSoup) -> str:
for selector in (
'meta[name="PubDate"]',
'meta[name="pubdate"]',
'meta[name="publishdate"]',
'meta[property="article:published_time"]',
'meta[name="historytime"]',
):
element = soup.select_one(selector)
if element and element.get("content"):
date_value = extract_date(element["content"])
if date_value:
return date_value
for selector in DATE_SELECTORS:
element = soup.select_one(selector)
if element:
date_value = extract_date(element.get_text(" ", strip=True))
if date_value:
return date_value
page_text = soup.get_text("\n", strip=True)[:1500]
return extract_date(url) or extract_date(page_text)
def extract_content(self, soup: BeautifulSoup) -> str:
for selector in DEFAULT_CONTENT_SELECTORS:
element = soup.select_one(selector)
if element:
text = simplify_text(element.get_text("\n", strip=True))
if len(compact_text(text)) >= 120:
return text
text = simplify_text(soup.get_text("\n", strip=True))
return text[:12000]
def extract_attachments(self, base_url: str, soup: BeautifulSoup) -> list[dict[str, str]]:
attachments: list[dict[str, str]] = []
for anchor in soup.find_all("a", href=True):
href = anchor["href"].strip()
if not href:
continue
if not any(href.lower().endswith(ext) for ext in (".pdf", ".doc", ".docx", ".xls", ".xlsx", ".zip")):
continue
attachments.append(
{
"name": simplify_text(anchor.get_text(" ", strip=True)) or href.split("/")[-1],
"url": urljoin(base_url, href),
}
)
return attachments
def parse_article(self, site: SiteConfig, url: str, soup: BeautifulSoup) -> PolicyRecord | None:
title = self.extract_title(soup)
content = self.extract_content(soup)
if not title or not content:
return None
if self.is_generic_title(title):
return None
if has_non_policy_title_hint(title):
return None
if not is_probable_policy_article(title, url, content):
return None
publish_date = self.extract_publish_date(url, soup)
doc_no = extract_doc_no(content[:1200]) or extract_doc_no(title)
category = guess_category(title, url)
attachments = self.extract_attachments(url, soup)
payload_for_hash = json.dumps(
{"title": title, "content": content, "publish_date": publish_date, "doc_no": doc_no},
ensure_ascii=False,
sort_keys=True,
)
source_site = urlparse(url).netloc
policy_key = derive_policy_key(site.name, title, category, url, doc_no)
return PolicyRecord(
site_name=site.name,
region=site.region,
level=site.level,
source_site=source_site,
source_url=url,
title=title,
content=content,
publish_date=publish_date,
crawl_date=today_local(),
category=category,
doc_no=doc_no,
effective_date=publish_date,
status="effective",
policy_key=policy_key,
version_no=1,
content_hash=sha256_text(payload_for_hash),
summary_text=summarize_for_embedding(title, publish_date, doc_no, category, content),
attachments_json=dumps_json(attachments),
)
from __future__ import annotations
from typing import Any
import requests
from .config import AppConfig
class DifyClient:
def __init__(self, config: AppConfig) -> None:
self.config = config
self.session = requests.Session()
self.session.headers.update(
{
"Authorization": f"Bearer {config.dify_api_key}",
"Content-Type": "application/json",
}
)
def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
response = self.session.request(
method,
f"{self.config.dify_base_url}{path}",
timeout=self.config.request_timeout,
**kwargs,
)
response.raise_for_status()
return response.json() if response.text.strip() else {}
def list_datasets(self) -> list[dict[str, Any]]:
page = 1
datasets: list[dict[str, Any]] = []
while True:
payload = self._request("GET", f"/v1/datasets?page={page}&limit=100")
items = payload.get("data", [])
datasets.extend(items)
if len(items) < 100:
return datasets
page += 1
def create_dataset(self, name: str) -> str:
payload = self._request(
"POST",
"/v1/datasets",
json={
"name": name,
"indexing_technique": self.config.dify_indexing_technique,
"permission": "only_me",
},
)
return str(payload["id"])
def ensure_dataset(self) -> str:
if self.config.dify_dataset_id:
return self.config.dify_dataset_id
for dataset in self.list_datasets():
if dataset.get("name") == self.config.dify_dataset_name:
return str(dataset["id"])
dataset_id = self.create_dataset(self.config.dify_dataset_name)
self.config.dify_dataset_id = dataset_id
return dataset_id
def create_document_by_text(self, dataset_id: str, name: str, text: str) -> str:
payload = self._request(
"POST",
f"/v1/datasets/{dataset_id}/document/create-by-text",
json={
"name": name,
"text": text,
"indexing_technique": self.config.dify_indexing_technique,
"process_rule": {"mode": "automatic"},
},
)
document = payload.get("document") or payload
return str(document["id"])
def update_document_by_text(self, dataset_id: str, document_id: str, name: str, text: str) -> None:
self._request(
"POST",
f"/v1/datasets/{dataset_id}/documents/{document_id}/update_by_text",
json={
"name": name,
"text": text,
"process_rule": {"mode": "automatic"},
},
)
def delete_document(self, dataset_id: str, document_id: str) -> None:
self._request(
"DELETE",
f"/v1/datasets/{dataset_id}/documents/{document_id}",
)
from __future__ import annotations
from pathlib import Path
def load_dotenv(dotenv_path: Path) -> None:
"""Load a very small subset of .env files into process env."""
import os
if not dotenv_path.exists():
return
for raw_line in dotenv_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
os.environ.setdefault(key, value)
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass(slots=True)
class SiteConfig:
name: str
region: str
level: str
homepage: str
public_entry: str
allowed_hosts: list[str]
seed_urls: list[str] = field(default_factory=list)
include_url_keywords: list[str] = field(default_factory=list)
exclude_url_keywords: list[str] = field(default_factory=list)
max_depth: int = 2
max_pages: int = 80
use_playwright: bool = False
crawl_mode: str = "default"
api_deptid: str = ""
api_list_url: str = ""
@dataclass(slots=True)
class PolicyRecord:
site_name: str
region: str
level: str
source_site: str
source_url: str
title: str
content: str
publish_date: str
crawl_date: str
category: str
doc_no: str
effective_date: str
status: str
policy_key: str
version_no: int
content_hash: str
summary_text: str
attachments_json: str
@dataclass(slots=True)
class PipelineResult:
site_name: str
fetched_pages: int = 0
discovered_articles: int = 0
inserted: int = 0
updated: int = 0
deleted: int = 0
duplicates: int = 0
failed_pages: int = 0
from __future__ import annotations
import logging
import sys
from dataclasses import replace
from pathlib import Path
import requests
from .config import AppConfig, load_config, load_sites
from .crawler import PolicyCrawler
from .dify_client import DifyClient
from .models import PipelineResult, SiteConfig
from .storage import PolicyStore
def _build_logger(log_path: Path) -> logging.Logger:
logger = logging.getLogger("policy_agent")
logger.setLevel(logging.INFO)
logger.handlers.clear()
formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
file_handler = logging.FileHandler(log_path, encoding="utf-8")
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
return logger
class PolicyPipeline:
def __init__(self, config: AppConfig | None = None) -> None:
self.config = config or load_config()
self.store = PolicyStore(self.config.db_path)
self.crawler = PolicyCrawler(self.config)
self.dify = DifyClient(self.config)
self.logger = _build_logger(self.config.logs_dir / "policy_agent.log")
def close(self) -> None:
self.store.close()
def run(
self,
site_names: list[str] | None = None,
max_pages: int | None = None,
max_depth: int | None = None,
) -> list[PipelineResult]:
errors = self.config.validate()
if errors:
raise ValueError("配置不完整: " + ";".join(errors))
dataset_id = self.dify.ensure_dataset()
self.logger.info("Using Dify dataset: %s", dataset_id)
sites = load_sites(self.config)
if site_names:
wanted = {name.strip() for name in site_names}
sites = [site for site in sites if site.name in wanted]
if max_pages is not None:
sites = [replace(site, max_pages=max_pages) for site in sites]
if max_depth is not None:
sites = [replace(site, max_depth=max_depth) for site in sites]
results: list[PipelineResult] = []
for site in sites:
results.append(self._run_site(site, dataset_id))
return results
def _run_site(self, site: SiteConfig, dataset_id: str) -> PipelineResult:
run_id = self.store.begin_run(site.name)
self.logger.info("Start crawling: %s", site.name)
result = PipelineResult(site_name=site.name)
notes = ""
try:
outcome = self.crawler.crawl_site(site)
result.fetched_pages = outcome.fetched_pages
result.failed_pages = outcome.failed_pages
result.discovered_articles = len(outcome.articles)
for article in outcome.articles:
action, row = self.store.upsert_policy(article)
if action == "duplicate" and row["dify_document_id"]:
result.duplicates += 1
elif row["dify_document_id"]:
try:
self.dify.update_document_by_text(
dataset_id=dataset_id,
document_id=row["dify_document_id"],
name=article.title[:180],
text=article.summary_text,
)
result.updated += 1
except requests.HTTPError as exc:
response_text = exc.response.text if exc.response is not None else ""
status_code = exc.response.status_code if exc.response is not None else 0
if status_code == 400 and "Document is not available" in response_text:
document_id = self.dify.create_document_by_text(
dataset_id=dataset_id,
name=article.title[:180],
text=article.summary_text,
)
self.store.update_dify_document_id(str(row["source_url"]), document_id)
result.updated += 1
else:
raise
else:
document_id = self.dify.create_document_by_text(
dataset_id=dataset_id,
name=article.title[:180],
text=article.summary_text,
)
self.store.update_dify_document_id(str(row["source_url"]), document_id)
if action == "updated":
result.updated += 1
elif action == "duplicate":
result.duplicates += 1
else:
result.inserted += 1
result.deleted += self._delete_revised_documents(
dataset_id=dataset_id,
policy_key=str(row["policy_key"]),
current_source_url=str(row["source_url"]),
)
notes = f"articles={result.discovered_articles};deleted={result.deleted}"
self.logger.info(
"Finished %s: pages=%s articles=%s inserted=%s updated=%s deleted=%s duplicates=%s failed=%s",
site.name,
result.fetched_pages,
result.discovered_articles,
result.inserted,
result.updated,
result.deleted,
result.duplicates,
result.failed_pages,
)
return result
except Exception as exc:
notes = f"error={exc}"
self.logger.exception("Pipeline failed for %s", site.name)
raise
finally:
self.store.finish_run(run_id, result, notes=notes)
def _delete_revised_documents(self, dataset_id: str, policy_key: str, current_source_url: str) -> int:
deleted = 0
revised_rows = self.store.get_outdated_policies(policy_key, current_source_url)
for revised_row in revised_rows:
document_id = revised_row["dify_document_id"]
if not document_id:
continue
try:
self.dify.delete_document(dataset_id=dataset_id, document_id=str(document_id))
except requests.HTTPError as exc:
response_text = exc.response.text if exc.response is not None else ""
status_code = exc.response.status_code if exc.response is not None else 0
if status_code == 404:
pass
elif status_code == 400 and "Document is not available" in response_text:
pass
else:
raise
self.store.clear_dify_document_id(int(revised_row["id"]))
deleted += 1
return deleted
param(
[string]$TaskName = "HubeiPolicyDailyUpdate",
[string]$ProjectRoot = ""
)
$ErrorActionPreference = "Stop"
if ([string]::IsNullOrWhiteSpace($ProjectRoot)) {
$ProjectRoot = (Resolve-Path (Join-Path $PSScriptRoot "..\..")).Path
} else {
$ProjectRoot = (Resolve-Path $ProjectRoot).Path
}
$RunScript = Join-Path $ProjectRoot "policy_agent\scripts\run_daily_update.ps1"
$PowerShellExe = "$env:SystemRoot\System32\WindowsPowerShell\v1.0\powershell.exe"
if (-not (Test-Path $RunScript)) {
throw "Run script not found: $RunScript"
}
$Action = New-ScheduledTaskAction `
-Execute $PowerShellExe `
-Argument "-NoProfile -ExecutionPolicy Bypass -File `"$RunScript`" -ProjectRoot `"$ProjectRoot`""
$Trigger = New-ScheduledTaskTrigger -Daily -At 00:00
$Settings = New-ScheduledTaskSettingsSet -StartWhenAvailable -MultipleInstances IgnoreNew
Register-ScheduledTask `
-TaskName $TaskName `
-Action $Action `
-Trigger $Trigger `
-Settings $Settings `
-Description "Daily Hubei policy sync to Dify knowledge base" `
-Force | Out-Null
Write-Host "Scheduled task registered: $TaskName"
Write-Host "Project root: $ProjectRoot"
param(
[string]$ProjectRoot = "",
[string]$PythonExe = "",
[int]$MaxPages = 0,
[int]$MaxDepth = 0,
[string[]]$Site = @()
)
$ErrorActionPreference = "Stop"
if ([string]::IsNullOrWhiteSpace($ProjectRoot)) {
$ProjectRoot = (Resolve-Path (Join-Path $PSScriptRoot "..\..")).Path
} else {
$ProjectRoot = (Resolve-Path $ProjectRoot).Path
}
function Resolve-PythonExe {
param(
[string]$PreferredPythonExe,
[string]$ResolvedProjectRoot
)
$candidates = @()
if (-not [string]::IsNullOrWhiteSpace($PreferredPythonExe)) {
$candidates += $PreferredPythonExe
}
if (-not [string]::IsNullOrWhiteSpace($env:POLICY_PYTHON)) {
$candidates += $env:POLICY_PYTHON
}
$venvPython = Join-Path $ResolvedProjectRoot ".venv\Scripts\python.exe"
if (Test-Path $venvPython) {
$candidates += $venvPython
}
foreach ($candidate in $candidates) {
if (Test-Path $candidate) {
return (Resolve-Path $candidate).Path
}
}
$pythonCommand = Get-Command python -ErrorAction SilentlyContinue
if ($pythonCommand) {
return $pythonCommand.Source
}
throw "Python executable not found. Use -PythonExe, set POLICY_PYTHON, or run setup_local.ps1 first."
}
$PythonExe = Resolve-PythonExe -PreferredPythonExe $PythonExe -ResolvedProjectRoot $ProjectRoot
$EnvFile = Join-Path $ProjectRoot ".env"
if (-not (Test-Path $EnvFile)) {
throw ".env not found: $EnvFile. Copy .env.example to .env and fill in Dify configuration first."
}
$arguments = @(
"-m", "policy_agent.cli",
"--project-root", $ProjectRoot,
"run"
)
if ($MaxPages -gt 0) {
$arguments += @("--max-pages", $MaxPages)
}
if ($MaxDepth -gt 0) {
$arguments += @("--max-depth", $MaxDepth)
}
foreach ($siteName in $Site) {
if (-not [string]::IsNullOrWhiteSpace($siteName)) {
$arguments += @("--site", $siteName)
}
}
Set-Location $ProjectRoot
& $PythonExe @arguments
exit $LASTEXITCODE
param(
[string]$ProjectRoot = "",
[string]$PythonExe = "",
[switch]$SkipPlaywrightInstall
)
$ErrorActionPreference = "Stop"
if ([string]::IsNullOrWhiteSpace($ProjectRoot)) {
$ProjectRoot = (Resolve-Path (Join-Path $PSScriptRoot "..\..")).Path
} else {
$ProjectRoot = (Resolve-Path $ProjectRoot).Path
}
function Resolve-BootstrapPython {
param([string]$PreferredPythonExe)
if (-not [string]::IsNullOrWhiteSpace($PreferredPythonExe)) {
if (-not (Test-Path $PreferredPythonExe)) {
throw "Python executable not found: $PreferredPythonExe"
}
return (Resolve-Path $PreferredPythonExe).Path
}
$pythonCommand = Get-Command python -ErrorAction SilentlyContinue
if ($pythonCommand) {
return $pythonCommand.Source
}
$pyLauncher = Get-Command py -ErrorAction SilentlyContinue
if ($pyLauncher) {
return $pyLauncher.Source
}
throw "Python executable not found. Install Python 3.10+ first or pass -PythonExe."
}
$BootstrapPython = Resolve-BootstrapPython -PreferredPythonExe $PythonExe
$VenvDir = Join-Path $ProjectRoot ".venv"
$VenvPython = Join-Path $VenvDir "Scripts\python.exe"
$RequirementsPath = Join-Path $ProjectRoot "requirements.txt"
$EnvExamplePath = Join-Path $ProjectRoot ".env.example"
$EnvPath = Join-Path $ProjectRoot ".env"
if (-not (Test-Path $VenvPython)) {
Write-Host "Creating local virtual environment: $VenvDir"
if ($BootstrapPython -like "*\\py.exe") {
& $BootstrapPython -3 -m venv $VenvDir
} else {
& $BootstrapPython -m venv $VenvDir
}
}
Write-Host "Installing Python dependencies..."
& $VenvPython -m pip install --upgrade pip
& $VenvPython -m pip install -r $RequirementsPath
if (-not $SkipPlaywrightInstall) {
Write-Host "Installing Playwright Chromium browser..."
& $VenvPython -m playwright install chromium
}
if ((Test-Path $EnvExamplePath) -and -not (Test-Path $EnvPath)) {
Copy-Item $EnvExamplePath $EnvPath
Write-Host "Created .env from .env.example. Fill in DIFY_BASE_URL and DIFY_API_KEY before running."
}
Write-Host ""
Write-Host "Setup complete."
Write-Host "Virtualenv Python: $VenvPython"
Write-Host "Next step:"
Write-Host " 1. Edit $EnvPath"
Write-Host " 2. Run: $VenvPython -m policy_agent.cli --project-root `"$ProjectRoot`" check-config"
[
{
"name": "湖北省人力资源和社会保障厅",
"region": "湖北省",
"level": "省级",
"homepage": "https://rst.hubei.gov.cn/",
"public_entry": "https://rst.hubei.gov.cn/zfxxgk/zc/gfxwj/",
"allowed_hosts": ["rst.hubei.gov.cn"],
"seed_urls": [
"https://rst.hubei.gov.cn/zfxxgk/zc/gfxwj/",
"https://rst.hubei.gov.cn/zfxxgk/zc/zcjd/"
],
"include_url_keywords": [
"/zfxxgk/zc/gfxwj/",
"/zfxxgk/zc/zcjd/"
],
"exclude_url_keywords": [
"/bmdt/",
"/dtyw/",
"/tzgg/",
"index.shtml?name="
],
"max_depth": 2,
"max_pages": 80,
"crawl_mode": "render_entry"
},
{
"name": "武汉市人力资源和社会保障局",
"region": "武汉市",
"level": "副省级市",
"homepage": "https://rsj.wuhan.gov.cn/",
"public_entry": "https://rsj.wuhan.gov.cn/zwgk_17/zc/gfxwj/zwgk_list.html",
"allowed_hosts": ["rsj.wuhan.gov.cn"],
"seed_urls": [
"https://rsj.wuhan.gov.cn/zwgk_17/zc/gfxwj/zwgk_list.html",
"https://rsj.wuhan.gov.cn/zwgk_17/zc/zcjd_1/zwgk_list.html"
],
"include_url_keywords": [
"/zwgk_17/zc/gfxwj/",
"/zwgk_17/zc/zcjd_1/"
],
"exclude_url_keywords": [
"/qtzdgkwj/"
],
"max_depth": 2,
"max_pages": 80,
"crawl_mode": "wuhan_jsonp",
"api_list_url": "https://rsj.wuhan.gov.cn/cslm/index.jsonp"
},
{
"name": "黄石市人力资源和社会保障局",
"region": "黄石市",
"level": "地级市",
"homepage": "https://rsj.huangshi.gov.cn/",
"public_entry": "https://rsj.huangshi.gov.cn/xxgk/zc/gfxwj/",
"allowed_hosts": ["rsj.huangshi.gov.cn"],
"seed_urls": [
"https://rsj.huangshi.gov.cn/xxgk/zc/gfxwj/",
"https://rsj.huangshi.gov.cn/xxgk/zc/zcjd/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "十堰市人力资源和社会保障局",
"region": "十堰市",
"level": "地级市",
"homepage": "http://rsj.shiyan.gov.cn/",
"public_entry": "http://rsj.shiyan.gov.cn/srlzyhshbzj/zc/gfxwj_new/",
"allowed_hosts": ["rsj.shiyan.gov.cn"],
"seed_urls": [
"http://rsj.shiyan.gov.cn/srlzyhshbzj/zc/gfxwj_new/",
"http://rsj.shiyan.gov.cn/srlzyhshbzj/zc/zcjd/"
],
"include_url_keywords": [
"/srlzyhshbzj/zc/gfxwj_new/",
"/srlzyhshbzj/zc/zcjd/"
],
"exclude_url_keywords": [
"/qtzdgkwj/",
"/gzdt/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "宜昌市人力资源和社会保障局",
"region": "宜昌市",
"level": "地级市",
"homepage": "http://rsj.yichang.gov.cn/",
"public_entry": "http://www.yichang.gov.cn/zfxxgk/list.html?depid=858&catid=522&t=4",
"allowed_hosts": ["www.yichang.gov.cn", "rsj.yichang.gov.cn"],
"seed_urls": [
"http://www.yichang.gov.cn/zfxxgk/list.html?depid=858&catid=522&t=4"
],
"max_depth": 2,
"max_pages": 80,
"crawl_mode": "yichang_api",
"api_deptid": "858"
},
{
"name": "襄阳市人力资源和社会保障局",
"region": "襄阳市",
"level": "地级市",
"homepage": "http://rsj.xiangyang.gov.cn/",
"public_entry": "http://rsj.xiangyang.gov.cn/zwgk/zc/zcfg/",
"allowed_hosts": ["rsj.xiangyang.gov.cn"],
"seed_urls": [
"http://rsj.xiangyang.gov.cn/zwgk/zc/zcfg/",
"http://rsj.xiangyang.gov.cn/zwgk/zc/zcjd/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "鄂州市人力资源和社会保障局",
"region": "鄂州市",
"level": "地级市",
"homepage": "https://rsj.ezhou.gov.cn/",
"public_entry": "https://rsj.ezhou.gov.cn/xxgk/zc/bmgfxwj/",
"allowed_hosts": ["rsj.ezhou.gov.cn"],
"seed_urls": [
"https://rsj.ezhou.gov.cn/xxgk/zc/bmgfxwj/",
"https://rsj.ezhou.gov.cn/xxgk/zc/zcjd/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "荆门市人力资源和社会保障局",
"region": "荆门市",
"level": "地级市",
"homepage": "http://rsj.jingmen.gov.cn/",
"public_entry": "http://rsj.jingmen.gov.cn/col/col10845/index.html",
"allowed_hosts": ["rsj.jingmen.gov.cn"],
"seed_urls": [
"http://rsj.jingmen.gov.cn/col/col10845/index.html",
"http://rsj.jingmen.gov.cn/col/col10846/index.html",
"http://rsj.jingmen.gov.cn/col/col10881/index.html"
],
"include_url_keywords": [
"/col/col10845/",
"/col/col10846/",
"/col/col10881/",
"/art_10845_",
"/art_10846_",
"/art_10881_"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "孝感市人力资源和社会保障局",
"region": "孝感市",
"level": "地级市",
"homepage": "https://rsj.xiaogan.gov.cn/",
"public_entry": "https://rsj.xiaogan.gov.cn/c/xgsrlzyhshbzj/zc.jhtml",
"allowed_hosts": ["rsj.xiaogan.gov.cn"],
"seed_urls": [
"https://rsj.xiaogan.gov.cn/c/xgsrlzyhshbzj/zc.jhtml"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "荆州市人力资源和社会保障局",
"region": "荆州市",
"level": "地级市",
"homepage": "http://rsj.jingzhou.gov.cn/",
"public_entry": "http://jzrsj.zwgk.jingzhou.gov.cn/list.shtml?column_id=35910",
"allowed_hosts": ["rsj.jingzhou.gov.cn", "jzrsj.zwgk.jingzhou.gov.cn"],
"seed_urls": [
"http://jzrsj.zwgk.jingzhou.gov.cn/list.shtml?column_id=35910",
"http://jzrsj.zwgk.jingzhou.gov.cn/list.shtml?column_id=35912"
],
"include_url_keywords": [
"column_id=35910",
"column_id=35912",
"/35910/",
"/35912/"
],
"max_depth": 2,
"max_pages": 80,
"crawl_mode": "render_entry"
},
{
"name": "黄冈市人力资源和社会保障局",
"region": "黄冈市",
"level": "地级市",
"homepage": "https://rsj.hg.gov.cn/",
"public_entry": "https://rsj.hg.gov.cn/zwgk/public/column/6636189?type=4&catId=7026842&action=list&nav=0",
"allowed_hosts": ["rsj.hg.gov.cn"],
"seed_urls": [
"https://rsj.hg.gov.cn/zwgk/public/column/6636189?type=4&catId=7026842&action=list&nav=0",
"https://rsj.hg.gov.cn/zwgk/public/column/6636189?type=4&catId=7025468&action=list&nav=0"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "咸宁市人力资源和社会保障局",
"region": "咸宁市",
"level": "地级市",
"homepage": "http://rsj.xianning.gov.cn/",
"public_entry": "http://rsj.xianning.gov.cn/xxgk/zc/wjzl/",
"allowed_hosts": ["rsj.xianning.gov.cn"],
"seed_urls": [
"http://rsj.xianning.gov.cn/xxgk/zc/wjzl/",
"http://rsj.xianning.gov.cn/xxgk/zc/zcjd/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "随州市人力资源和社会保障局",
"region": "随州市",
"level": "地级市",
"homepage": "http://rsj.suizhou.gov.cn/",
"public_entry": "http://rsj.suizhou.gov.cn/fbjd_16/zwgk/xxgkml/zcfg/",
"allowed_hosts": ["rsj.suizhou.gov.cn"],
"seed_urls": [
"http://rsj.suizhou.gov.cn/fbjd_16/zwgk/xxgkml/zcfg/",
"http://rsj.suizhou.gov.cn/fbjd_16/zwgk/zc/qtzdgkwj/tzgg/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "恩施州人力资源和社会保障局",
"region": "恩施土家族苗族自治州",
"level": "自治州",
"homepage": "http://rsj.enshi.gov.cn/",
"public_entry": "http://rsj.enshi.gov.cn/xxgk/zc/gfxwj/",
"allowed_hosts": ["rsj.enshi.gov.cn"],
"seed_urls": [
"http://rsj.enshi.gov.cn/xxgk/zc/gfxwj/",
"http://rsj.enshi.gov.cn/xxgk/zc/zcjd/",
"http://rsj.enshi.gov.cn/xxgk/zc/qtzdgk/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "仙桃市人力资源和社会保障局",
"region": "仙桃市",
"level": "省直辖县级市",
"homepage": "https://www.xiantao.gov.cn/bmxxgk/srsj/",
"public_entry": "https://www.xiantao.gov.cn/bmxxgk/srsj/zfxxgk/zc/gfxwj/",
"allowed_hosts": ["www.xiantao.gov.cn"],
"seed_urls": [
"https://www.xiantao.gov.cn/bmxxgk/srsj/zfxxgk/zc/gfxwj/",
"https://www.xiantao.gov.cn/bmxxgk/srsj/zfxxgk/zc/zcjd/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "潜江市人力资源和社会保障局",
"region": "潜江市",
"level": "省直辖县级市",
"homepage": "https://www.hbqj.gov.cn/srlzyhshbzj/",
"public_entry": "https://www.hbqj.gov.cn/srlzyhshbzj/zfxxgk/zc/gfxwj/",
"allowed_hosts": ["www.hbqj.gov.cn"],
"seed_urls": [
"https://www.hbqj.gov.cn/srlzyhshbzj/zfxxgk/zc/gfxwj/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "天门市人力资源和社会保障局",
"region": "天门市",
"level": "省直辖县级市",
"homepage": "https://www.tianmen.gov.cn/zwgk/bmhxzxxgkml/bm/srlzyhshbzj/",
"public_entry": "https://www.tianmen.gov.cn/zwgk/bmhxzxxgkml/bm/srlzyhshbzj/zfxxgk/zc/bmwj/",
"allowed_hosts": ["www.tianmen.gov.cn"],
"seed_urls": [
"https://www.tianmen.gov.cn/zwgk/bmhxzxxgkml/bm/srlzyhshbzj/zfxxgk/zc/bmwj/",
"https://www.tianmen.gov.cn/zwgk/bmhxzxxgkml/bm/srlzyhshbzj/zfxxgk/zc/zcjd/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "神农架林区人力资源和社会保障局",
"region": "神农架林区",
"level": "林区",
"homepage": "http://rsj.snj.gov.cn/",
"public_entry": "http://rsj.snj.gov.cn/zc_38462/",
"allowed_hosts": ["rsj.snj.gov.cn", "www.snj.gov.cn"],
"seed_urls": [
"http://rsj.snj.gov.cn/zc_38462/"
],
"include_url_keywords": [
"/zc_38462/"
],
"max_depth": 2,
"max_pages": 80,
"crawl_mode": "render_entry"
}
]
[
{
"name": "湖北省人民政府",
"region": "湖北省",
"level": "省级",
"homepage": "https://www.hubei.gov.cn/",
"public_entry": "https://www.hubei.gov.cn/xxgk/",
"allowed_hosts": ["www.hubei.gov.cn"],
"seed_urls": ["https://www.hubei.gov.cn/xxgk/zc/", "https://www.hubei.gov.cn/xxgk/zcjd/"],
"max_depth": 2,
"max_pages": 80
},
{
"name": "武汉市人民政府",
"region": "武汉市",
"level": "副省级市",
"homepage": "https://www.wuhan.gov.cn/",
"public_entry": "https://www.wuhan.gov.cn/zwgk/?channelid=26164",
"allowed_hosts": ["www.wuhan.gov.cn"],
"seed_urls": [
"https://www.wuhan.gov.cn/zwgk/tzgg/",
"https://www.wuhan.gov.cn/hdjl/dczj/",
"https://www.wuhan.gov.cn/gfxwj/index.shtml"
],
"max_depth": 2,
"max_pages": 80,
"use_playwright": false
},
{
"name": "黄石市人民政府",
"region": "黄石市",
"level": "地级市",
"homepage": "https://www.huangshi.gov.cn/",
"public_entry": "https://www.huangshi.gov.cn/xxxgk/",
"allowed_hosts": ["www.huangshi.gov.cn"],
"seed_urls": [
"https://www.huangshi.gov.cn/xxxgk/2020_gfxwj/",
"https://www.huangshi.gov.cn/xxxgk/qtzdgkwj/",
"https://www.huangshi.gov.cn/xxxgk/zcjd/"
],
"max_depth": 2,
"max_pages": 80,
"use_playwright": false
},
{
"name": "十堰市人民政府",
"region": "十堰市",
"level": "地级市",
"homepage": "https://www.shiyan.gov.cn/",
"public_entry": "http://www.shiyan.gov.cn/xxgk/",
"allowed_hosts": ["www.shiyan.gov.cn"],
"seed_urls": [
"http://www.shiyan.gov.cn/xxgk/zc_67263/xxgk_tzgg/",
"http://www.shiyan.gov.cn/zcjd/",
"http://www.shiyan.gov.cn/xxgk/xxgk_fdgk/qtzdgknr/data/"
],
"max_depth": 2,
"max_pages": 80,
"use_playwright": true
},
{
"name": "宜昌市人民政府",
"region": "宜昌市",
"level": "地级市",
"homepage": "http://www.yichang.gov.cn/",
"public_entry": "http://www.yichang.gov.cn/zfxxgk/",
"allowed_hosts": ["www.yichang.gov.cn"],
"seed_urls": [],
"max_depth": 2,
"max_pages": 80
},
{
"name": "襄阳市人民政府",
"region": "襄阳市",
"level": "地级市",
"homepage": "http://www.xiangyang.gov.cn/wzsy/",
"public_entry": "http://xxgk.xiangyang.gov.cn/",
"allowed_hosts": ["www.xiangyang.gov.cn", "xxgk.xiangyang.gov.cn"],
"seed_urls": [],
"max_depth": 2,
"max_pages": 80
},
{
"name": "鄂州市人民政府",
"region": "鄂州市",
"level": "地级市",
"homepage": "https://www.ezhou.gov.cn/sy/",
"public_entry": "https://www.ezhou.gov.cn/gk/",
"allowed_hosts": ["www.ezhou.gov.cn"],
"seed_urls": [
"https://www.ezhou.gov.cn/gk/gz1/",
"https://www.ezhou.gov.cn/gk/gfxwj/",
"https://www.ezhou.gov.cn/gk/zcjd/",
"https://www.ezhou.gov.cn/gk/gsgg_1/"
],
"max_depth": 2,
"max_pages": 80,
"use_playwright": true
},
{
"name": "荆门市人民政府",
"region": "荆门市",
"level": "地级市",
"homepage": "https://www.jingmen.gov.cn/",
"public_entry": "https://www.jingmen.gov.cn/col/col16474/index.html",
"allowed_hosts": ["www.jingmen.gov.cn"],
"seed_urls": [],
"max_depth": 2,
"max_pages": 80
},
{
"name": "孝感市人民政府",
"region": "孝感市",
"level": "地级市",
"homepage": "https://www.xiaogan.gov.cn/",
"public_entry": "https://www.xiaogan.gov.cn/c/www/zc.jhtml",
"allowed_hosts": ["www.xiaogan.gov.cn"],
"seed_urls": [
"https://www.xiaogan.gov.cn/c/www/gfxwj.jhtml",
"https://www.xiaogan.gov.cn/c/www/zcjd.jhtml",
"https://www.xiaogan.gov.cn/c/www/gsgg.jhtml"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "荆州市人民政府",
"region": "荆州市",
"level": "地级市",
"homepage": "https://www.jingzhou.gov.cn/",
"public_entry": "http://zwgk.jingzhou.gov.cn/list_children.shtml?column_id=54272",
"allowed_hosts": ["www.jingzhou.gov.cn", "zwgk.jingzhou.gov.cn"],
"seed_urls": [],
"max_depth": 2,
"max_pages": 80
},
{
"name": "黄冈市人民政府",
"region": "黄冈市",
"level": "地级市",
"homepage": "https://www.hg.gov.cn/",
"public_entry": "https://www.hg.gov.cn/zwgk/public/column/6636765?type=4&action=list&nav=0&id=7025468",
"allowed_hosts": ["www.hg.gov.cn"],
"seed_urls": [],
"max_depth": 2,
"max_pages": 80,
"use_playwright": false
},
{
"name": "咸宁市人民政府",
"region": "咸宁市",
"level": "地级市",
"homepage": "http://www.xianning.gov.cn/",
"public_entry": "http://www.xianning.gov.cn/xxgk/",
"allowed_hosts": ["www.xianning.gov.cn", "xtzyk.xianning.gov.cn"],
"seed_urls": [
"http://www.xianning.gov.cn/xxgk/zc/zfwj/xzbf/",
"http://www.xianning.gov.cn/xxgk/zc/zfwj/xzbh/",
"http://www.xianning.gov.cn/xxgk/zc/zcjd/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "随州市人民政府",
"region": "随州市",
"level": "地级市",
"homepage": "http://www.suizhou.gov.cn/",
"public_entry": "http://www.suizhou.gov.cn/zwgk/",
"allowed_hosts": ["www.suizhou.gov.cn"],
"seed_urls": [
"http://www.suizhou.gov.cn/zwgk/zfwj/szbf/szf/",
"http://www.suizhou.gov.cn/zwgk/zfwj/szbf/szbf/",
"http://www.suizhou.gov.cn/zwgk/zfwj/qt_5744/tzgg/",
"http://www.suizhou.gov.cn/zwgk/zfwj/zcjd/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "恩施土家族苗族自治州人民政府",
"region": "恩施州",
"level": "自治州",
"homepage": "http://www.enshi.gov.cn/",
"public_entry": "http://www.enshi.gov.cn/zc/",
"allowed_hosts": ["www.enshi.gov.cn"],
"seed_urls": [
"http://www.enshi.gov.cn/zc/zc/gfxwj_1/",
"http://www.enshi.gov.cn/zc/zc/zcwj/",
"http://www.enshi.gov.cn/zc/zc/zcjd/"
],
"max_depth": 2,
"max_pages": 80,
"use_playwright": true
},
{
"name": "仙桃市人民政府",
"region": "仙桃市",
"level": "省直辖县级市",
"homepage": "https://www.xiantao.gov.cn/",
"public_entry": "https://www.xiantao.gov.cn/zfxxgk/",
"allowed_hosts": ["www.xiantao.gov.cn"],
"seed_urls": [
"https://www.xiantao.gov.cn/zfxxgk/zfwjk/gfwj/",
"https://www.xiantao.gov.cn/zfxxgk/zfwjk/qtwj/",
"https://www.xiantao.gov.cn/zfxxgk/zfwjk/zcjd/"
],
"max_depth": 2,
"max_pages": 80,
"use_playwright": true
},
{
"name": "潜江市人民政府",
"region": "潜江市",
"level": "省直辖县级市",
"homepage": "https://www.hbqj.gov.cn/",
"public_entry": "https://www.hbqj.gov.cn/xxgk/",
"allowed_hosts": ["www.hbqj.gov.cn"],
"seed_urls": [
"https://www.hbqj.gov.cn/xxgk/zc/gfxwj/",
"https://www.hbqj.gov.cn/xxgk/zc/qtwj/",
"https://www.hbqj.gov.cn/xxgk/zc/zcjzcjd/"
],
"max_depth": 2,
"max_pages": 80,
"use_playwright": true
},
{
"name": "天门市人民政府",
"region": "天门市",
"level": "省直辖县级市",
"homepage": "http://www.tianmen.gov.cn/",
"public_entry": "http://www.tianmen.gov.cn/zwgk/",
"allowed_hosts": ["www.tianmen.gov.cn"],
"seed_urls": [
"http://www.tianmen.gov.cn/zwgk/zc/zcfg/",
"http://www.tianmen.gov.cn/zwgk/zc/zcjd/",
"http://www.tianmen.gov.cn/zwgk/zc/tzgg/"
],
"max_depth": 2,
"max_pages": 80
},
{
"name": "神农架林区人民政府",
"region": "神农架林区",
"level": "林区",
"homepage": "http://www.snj.gov.cn/",
"public_entry": "http://www.snj.gov.cn/zwgk/",
"allowed_hosts": ["www.snj.gov.cn"],
"seed_urls": [
"http://www.snj.gov.cn/zwgk/zc/gfxwj/",
"http://www.snj.gov.cn/zwgk/zc/zfbwj/",
"http://www.snj.gov.cn/zwgk/zc/zcjd/"
],
"max_depth": 2,
"max_pages": 80,
"use_playwright": true
}
]
from __future__ import annotations
import sqlite3
from dataclasses import asdict
from pathlib import Path
from .models import PipelineResult, PolicyRecord
from .utils import derive_policy_key, utc_now_iso
class PolicyStore:
def __init__(self, db_path: Path) -> None:
self.db_path = db_path
self.conn = sqlite3.connect(str(db_path))
self.conn.row_factory = sqlite3.Row
self._init_schema()
self._rebuild_policy_lineage()
def close(self) -> None:
self.conn.close()
def _init_schema(self) -> None:
self.conn.executescript(
"""
CREATE TABLE IF NOT EXISTS policies (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_name TEXT NOT NULL,
region TEXT NOT NULL,
level TEXT NOT NULL,
source_site TEXT NOT NULL,
source_url TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
content TEXT NOT NULL,
publish_date TEXT,
crawl_date TEXT NOT NULL,
category TEXT NOT NULL,
doc_no TEXT,
effective_date TEXT,
status TEXT NOT NULL,
policy_key TEXT NOT NULL,
version_no INTEGER NOT NULL DEFAULT 1,
is_latest INTEGER NOT NULL DEFAULT 1,
content_hash TEXT NOT NULL,
summary_text TEXT NOT NULL,
attachments_json TEXT NOT NULL,
dify_document_id TEXT,
replaced_by_url TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_policies_policy_key ON policies(policy_key);
CREATE INDEX IF NOT EXISTS idx_policies_publish_date ON policies(publish_date);
CREATE INDEX IF NOT EXISTS idx_policies_region ON policies(region);
CREATE INDEX IF NOT EXISTS idx_policies_latest ON policies(policy_key, is_latest);
CREATE TABLE IF NOT EXISTS crawl_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_name TEXT NOT NULL,
started_at TEXT NOT NULL,
finished_at TEXT,
fetched_pages INTEGER NOT NULL DEFAULT 0,
discovered_articles INTEGER NOT NULL DEFAULT 0,
inserted_count INTEGER NOT NULL DEFAULT 0,
updated_count INTEGER NOT NULL DEFAULT 0,
duplicate_count INTEGER NOT NULL DEFAULT 0,
failed_count INTEGER NOT NULL DEFAULT 0,
notes TEXT
);
"""
)
self.conn.commit()
def _rebuild_policy_lineage(self) -> None:
rows = self.conn.execute(
"""
SELECT id, site_name, title, category, publish_date, effective_date, created_at,
updated_at, source_url, doc_no, status
FROM policies
ORDER BY id
"""
).fetchall()
if not rows:
return
grouped: dict[str, list[sqlite3.Row]] = {}
for row in rows:
policy_key = derive_policy_key(
str(row["site_name"]),
str(row["title"]),
str(row["category"]),
str(row["source_url"]),
str(row["doc_no"] or ""),
)
grouped.setdefault(policy_key, []).append(row)
updates: list[tuple[str, int, int, str | None, str, int]] = []
for policy_key, group_rows in grouped.items():
ordered_rows = sorted(group_rows, key=self._policy_sort_key)
for index, row in enumerate(ordered_rows):
is_latest = 1 if index == len(ordered_rows) - 1 else 0
replaced_by_url = None if is_latest else str(ordered_rows[index + 1]["source_url"])
status = str(row["status"]) if is_latest and str(row["status"]) != "revised" else "effective"
if not is_latest:
status = "revised"
updates.append((policy_key, index + 1, is_latest, replaced_by_url, status, int(row["id"])))
self.conn.executemany(
"""
UPDATE policies
SET policy_key = ?, version_no = ?, is_latest = ?, replaced_by_url = ?, status = ?
WHERE id = ?
""",
updates,
)
self.conn.commit()
def _policy_sort_key(self, row: sqlite3.Row) -> tuple[str, str, str, int]:
sequence_date = (
str(row["publish_date"] or "")
or str(row["effective_date"] or "")
or str(row["created_at"] or "")
or str(row["updated_at"] or "")
)
return (
sequence_date,
str(row["created_at"] or ""),
str(row["updated_at"] or ""),
int(row["id"]),
)
def begin_run(self, site_name: str) -> int:
cursor = self.conn.execute(
"INSERT INTO crawl_runs(site_name, started_at) VALUES(?, ?)",
(site_name, utc_now_iso()),
)
self.conn.commit()
return int(cursor.lastrowid)
def finish_run(self, run_id: int, result: PipelineResult, notes: str = "") -> None:
self.conn.execute(
"""
UPDATE crawl_runs
SET finished_at = ?, fetched_pages = ?, discovered_articles = ?, inserted_count = ?,
updated_count = ?, duplicate_count = ?, failed_count = ?, notes = ?
WHERE id = ?
""",
(
utc_now_iso(),
result.fetched_pages,
result.discovered_articles,
result.inserted,
result.updated,
result.duplicates,
result.failed_pages,
notes,
run_id,
),
)
self.conn.commit()
def get_policy_by_url(self, source_url: str) -> sqlite3.Row | None:
return self.conn.execute(
"SELECT * FROM policies WHERE source_url = ?",
(source_url,),
).fetchone()
def get_latest_policy_by_key(self, policy_key: str) -> sqlite3.Row | None:
return self.conn.execute(
"""
SELECT * FROM policies
WHERE policy_key = ? AND is_latest = 1
ORDER BY version_no DESC, publish_date DESC
LIMIT 1
""",
(policy_key,),
).fetchone()
def get_policy_by_hash(self, content_hash: str) -> sqlite3.Row | None:
return self.conn.execute(
"SELECT * FROM policies WHERE content_hash = ? LIMIT 1",
(content_hash,),
).fetchone()
def upsert_policy(self, record: PolicyRecord) -> tuple[str, sqlite3.Row]:
now = utc_now_iso()
existing = self.get_policy_by_url(record.source_url)
if existing:
if existing["content_hash"] == record.content_hash:
return "duplicate", existing
self.conn.execute(
"""
UPDATE policies
SET title = ?, content = ?, publish_date = ?, crawl_date = ?, category = ?, doc_no = ?,
effective_date = ?, status = ?, content_hash = ?, summary_text = ?,
attachments_json = ?, updated_at = ?
WHERE source_url = ?
""",
(
record.title,
record.content,
record.publish_date,
record.crawl_date,
record.category,
record.doc_no,
record.effective_date,
record.status,
record.content_hash,
record.summary_text,
record.attachments_json,
now,
record.source_url,
),
)
self.conn.commit()
refreshed = self.get_policy_by_url(record.source_url)
assert refreshed is not None
return "updated", refreshed
duplicate_hash = self.get_policy_by_hash(record.content_hash)
if duplicate_hash:
return "duplicate", duplicate_hash
latest = self.get_latest_policy_by_key(record.policy_key)
version_no = 1
if latest:
version_no = int(latest["version_no"]) + 1
self.conn.execute(
"""
UPDATE policies
SET is_latest = 0, status = 'revised', replaced_by_url = ?, updated_at = ?
WHERE id = ?
""",
(record.source_url, now, latest["id"]),
)
payload = asdict(record)
payload["version_no"] = version_no
self.conn.execute(
"""
INSERT INTO policies(
site_name, region, level, source_site, source_url, title, content, publish_date,
crawl_date, category, doc_no, effective_date, status, policy_key, version_no,
is_latest, content_hash, summary_text, attachments_json, created_at, updated_at
)
VALUES(
:site_name, :region, :level, :source_site, :source_url, :title, :content, :publish_date,
:crawl_date, :category, :doc_no, :effective_date, :status, :policy_key, :version_no,
1, :content_hash, :summary_text, :attachments_json, :created_at, :updated_at
)
""",
{
**payload,
"created_at": now,
"updated_at": now,
},
)
self.conn.commit()
inserted = self.get_policy_by_url(record.source_url)
assert inserted is not None
return "inserted", inserted
def update_dify_document_id(self, source_url: str, document_id: str) -> None:
self.conn.execute(
"UPDATE policies SET dify_document_id = ?, updated_at = ? WHERE source_url = ?",
(document_id, utc_now_iso(), source_url),
)
self.conn.commit()
def clear_dify_document_id(self, policy_id: int) -> None:
self.conn.execute(
"UPDATE policies SET dify_document_id = NULL, updated_at = ? WHERE id = ?",
(utc_now_iso(), policy_id),
)
self.conn.commit()
def get_outdated_policies(self, policy_key: str, current_source_url: str) -> list[sqlite3.Row]:
cursor = self.conn.execute(
"""
SELECT * FROM policies
WHERE policy_key = ? AND source_url <> ? AND is_latest = 0 AND dify_document_id IS NOT NULL
ORDER BY version_no DESC, updated_at DESC
""",
(policy_key, current_source_url),
)
return cursor.fetchall()
from __future__ import annotations
import hashlib
import json
import re
from datetime import datetime, timezone
from typing import Iterable
from urllib.parse import urlparse
DATE_PATTERNS = (
re.compile(r"(20\d{2})[-/年.](\d{1,2})[-/月.](\d{1,2})"),
re.compile(r"(20\d{2})(\d{2})(\d{2})"),
)
DOC_NO_PATTERN = re.compile(r"([^\s,。,;;()()]{1,20}[〔\[]?\d{4}[〕\]]?\d+号)")
QUOTED_TITLE_PATTERN = re.compile(r"[《〈](.+?)[》〉]")
POLICY_TITLE_KEYWORDS = (
"通知",
"通告",
"公告",
"意见",
"办法",
"细则",
"方案",
"规定",
"决定",
"规则",
"规范",
"解读",
"公开征求意见",
"公示",
"暂行",
"实施",
)
POLICY_URL_HINTS = (
"/zwgk/",
"/xxgk/",
"/zfxxgk/",
"/zc/",
"/zcfg/",
"/gfxwj/",
"/zcjd/",
"/tzgg/",
"/gsgg/",
"/gk/",
)
NON_POLICY_TITLE_HINTS = (
"工作动态",
"图片新闻",
"视频新闻",
"要闻",
"门户网站",
"领导活动",
"部门动态",
"县市区动态",
"新闻发布会",
"政府常务会议",
"统一发布平台",
"放假",
"连休",
)
def sha256_text(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def simplify_text(text: str) -> str:
cleaned = (text or "").replace("\u200b", "").replace("\ufeff", "").replace("\xa0", " ")
return re.sub(r"\s+", " ", cleaned).strip()
def compact_text(text: str) -> str:
return "".join((text or "").split())
def extract_date(text: str) -> str:
source = text or ""
for pattern in DATE_PATTERNS:
match = pattern.search(source)
if match:
year, month, day = match.groups()
try:
parsed = datetime(int(year), int(month), int(day))
except ValueError:
continue
return parsed.strftime("%Y-%m-%d")
return ""
def extract_doc_no(text: str) -> str:
match = DOC_NO_PATTERN.search(text or "")
return match.group(1) if match else ""
def utc_now_iso() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def today_local() -> str:
return datetime.now().strftime("%Y-%m-%d")
def summarize_for_embedding(title: str, publish_date: str, doc_no: str, category: str, content: str) -> str:
body = simplify_text(content)
if len(body) > 4000:
body = body[:4000]
header = [
f"标题:{title}",
f"发布日期:{publish_date or '未知'}",
f"文号:{doc_no or '未知'}",
f"分类:{category or '未分类'}",
"",
]
return "\n".join(header) + body
def dumps_json(data: object) -> str:
return json.dumps(data, ensure_ascii=False, sort_keys=True)
def _normalize_title_legacy(title: str) -> str:
title = compact_text(title)
title = re.sub(r"^(解读|政策解读)[::]", "", title)
title = re.sub(r"^(关于修改|关于废止|关于公布|关于印发)", "", title)
return title
def _derive_policy_key_legacy(title: str, doc_no: str) -> str:
title = _normalize_title_legacy(title)
quoted = QUOTED_TITLE_PATTERN.search(title)
key_source = quoted.group(1) if quoted else title
key_source = re.sub(r"(的通知|的公告|的通告|的意见|的决定|的办法|的实施细则|政策解读.*)$", "", key_source)
key_source = re.sub(r"[^\w\u4e00-\u9fff]+", "", key_source)
if doc_no:
return f"{key_source}:{compact_text(doc_no)}"
return key_source[:80]
def normalize_title(title: str) -> str:
title = compact_text(title)
title = re.sub(r"^【[^】]+】", "", title)
title = re.sub(r"^(解读|政策解读)[::]", "", title)
return title
def extract_policy_subject(title: str) -> str:
normalized = normalize_title(title)
quoted = QUOTED_TITLE_PATTERN.search(normalized)
if quoted:
subject = quoted.group(1)
else:
subject = normalized
if "关于" in subject:
subject = subject[subject.find("关于") :]
subject = re.sub(
r"^关于(印发|发布|公布|实施|施行|修订|修改|废止|转发|做好|进一步做好|公开征求|公开征集|征求)",
"",
subject,
)
subject = re.sub(
r"^(印发|发布|公布|实施|施行|修订|修改|废止|转发|公开征求|公开征集|征求)",
"",
subject,
)
subject = re.sub(r"(的通知|的通告|的公告|的意见|的决定|的办法|政策解读.*|解读)$", "", subject)
subject = re.sub(r"[^\w\u4e00-\u9fff]+", "", subject)
return subject[:120] or normalized[:120]
def normalize_policy_category(category: str) -> str:
compact_category = compact_text(category)
if "解读" in compact_category:
return "interpretation"
if any(token in compact_category for token in ("公告", "通知")):
return "notice"
return "policy"
def is_generic_policy_subject(site_name: str, subject: str) -> bool:
compact_subject = compact_text(subject)
compact_site_name = compact_text(site_name)
generic_subjects = {
"政策法规",
"通知公告",
"政策解读",
"政策文件",
"政务公开",
"信息公开",
}
if not compact_subject or len(compact_subject) <= 4:
return True
if compact_subject in generic_subjects:
return True
if compact_subject == compact_site_name:
return True
if compact_subject.endswith(compact_site_name) and len(compact_subject) <= len(compact_site_name) + 6:
return True
return False
def derive_policy_key(site_name: str, title: str, category: str, source_url: str = "", doc_no: str = "") -> str:
scope = compact_text(site_name)[:40]
category_key = normalize_policy_category(category)
subject = extract_policy_subject(title)
if is_generic_policy_subject(site_name, subject):
fallback = ""
if source_url:
parsed = urlparse(source_url)
fallback = re.sub(r"[^\w\u4e00-\u9fff]+", "", f"{parsed.netloc}{parsed.path}")[-80:]
if not fallback:
fallback = compact_text(doc_no)
if fallback:
return f"{scope}:{category_key}:{subject}:{fallback}"
return f"{scope}:{category_key}:{subject}"
def normalize_title(title: str) -> str:
title = compact_text(title)
title = re.sub(r"^\u3010[^\u3011]+\u3011", "", title)
title = re.sub(r"^(?:\u89e3\u8bfb|\u653f\u7b56\u89e3\u8bfb)[:\uff1a]", "", title)
return title
def extract_policy_subject(title: str) -> str:
normalized = normalize_title(title)
quoted = QUOTED_TITLE_PATTERN.search(normalized)
if quoted:
subject = quoted.group(1)
else:
subject = normalized
if "\u5173\u4e8e" in subject:
subject = subject[subject.find("\u5173\u4e8e") :]
subject = re.sub(
r"^\u5173\u4e8e(\u5370\u53d1|\u53d1\u5e03|\u516c\u5e03|\u5b9e\u65bd|\u65bd\u884c|\u4fee\u8ba2|\u4fee\u6539|\u5e9f\u6b62|\u8f6c\u53d1|\u505a\u597d|\u8fdb\u4e00\u6b65\u505a\u597d|\u516c\u5f00\u5f81\u6c42|\u516c\u5f00\u5f81\u96c6|\u5f81\u6c42)",
"",
subject,
)
subject = re.sub(
r"^(\u5370\u53d1|\u53d1\u5e03|\u516c\u5e03|\u5b9e\u65bd|\u65bd\u884c|\u4fee\u8ba2|\u4fee\u6539|\u5e9f\u6b62|\u8f6c\u53d1|\u516c\u5f00\u5f81\u6c42|\u516c\u5f00\u5f81\u96c6|\u5f81\u6c42)",
"",
subject,
)
subject = re.sub(
r"(\u7684\u901a\u77e5|\u7684\u901a\u544a|\u7684\u516c\u544a|\u7684\u610f\u89c1|\u7684\u51b3\u5b9a|\u7684\u529e\u6cd5|\u653f\u7b56\u89e3\u8bfb.*|\u89e3\u8bfb)$",
"",
subject,
)
subject = re.sub(r"[^\w\u4e00-\u9fff]+", "", subject)
return subject[:120] or normalized[:120]
def normalize_policy_category(category: str) -> str:
compact_category = compact_text(category)
if "\u89e3\u8bfb" in compact_category:
return "interpretation"
if any(token in compact_category for token in ("\u516c\u544a", "\u901a\u77e5")):
return "notice"
return "policy"
def is_generic_policy_subject(site_name: str, subject: str) -> bool:
compact_subject = compact_text(subject)
compact_site_name = compact_text(site_name)
generic_subjects = {
"\u653f\u7b56\u6cd5\u89c4",
"\u901a\u77e5\u516c\u544a",
"\u653f\u7b56\u89e3\u8bfb",
"\u653f\u7b56\u6587\u4ef6",
"\u653f\u52a1\u516c\u5f00",
"\u4fe1\u606f\u516c\u5f00",
}
if not compact_subject or len(compact_subject) <= 4:
return True
if compact_subject in generic_subjects:
return True
if compact_subject == compact_site_name:
return True
if compact_subject.endswith(compact_site_name) and len(compact_subject) <= len(compact_site_name) + 6:
return True
return False
def derive_policy_key(site_name: str, title: str, category: str, source_url: str = "", doc_no: str = "") -> str:
scope = compact_text(site_name)[:40]
category_key = normalize_policy_category(category)
subject = extract_policy_subject(title)
if is_generic_policy_subject(site_name, subject):
fallback = ""
if source_url:
parsed = urlparse(source_url)
fallback = re.sub(r"[^\w\u4e00-\u9fff]+", "", f"{parsed.netloc}{parsed.path}")[-80:]
if not fallback:
fallback = compact_text(doc_no)
if fallback:
return f"{scope}:{category_key}:{subject}:{fallback}"
return f"{scope}:{category_key}:{subject}"
def guess_category(title: str, url: str) -> str:
compact_title = compact_text(title)
lower_url = (url or "").lower()
if "解读" in compact_title or "/zcjd/" in lower_url:
return "政策解读"
if any(token in lower_url for token in ("/tzgg/", "/gsgg/", "公告")):
return "通知公告"
if any(token in compact_title for token in ("征求意见", "公示")):
return "通知公告"
return "政策文件"
def looks_like_policy_title(title: str) -> bool:
compact_title = compact_text(title)
if not compact_title:
return False
if any(token in compact_title for token in NON_POLICY_TITLE_HINTS):
return False
return any(token in compact_title for token in POLICY_TITLE_KEYWORDS)
def has_non_policy_title_hint(title: str) -> bool:
compact_title = compact_text(title)
return any(token in compact_title for token in NON_POLICY_TITLE_HINTS)
def looks_like_policy_url(url: str) -> bool:
lower_url = (url or "").lower()
return any(token in lower_url for token in POLICY_URL_HINTS)
def is_probable_policy_article(title: str, url: str, content: str) -> bool:
if looks_like_policy_title(title):
return True
if looks_like_policy_url(url) and any(token in compact_text(content)[:300] for token in POLICY_TITLE_KEYWORDS):
return True
return False
def dedupe_preserve_order(items: Iterable[str]) -> list[str]:
seen: set[str] = set()
output: list[str] = []
for item in items:
if item in seen:
continue
seen.add(item)
output.append(item)
return output
requests>=2.31,<3
beautifulsoup4>=4.12,<5
lxml>=5,<6
playwright>=1.48,<2
# -*- coding: utf-8 -*-
"""
test_wuhan_scraper.py
武汉政府网爬取可行性测试 - 完整版
运行方式: conda activate policy && python test_wuhan_scraper.py
"""
import sys, io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
import requests
from bs4 import BeautifulSoup
import re, json
session = requests.Session()
session.verify = False
session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Accept-Language": "zh-CN,zh;q=0.9",
"Referer": "https://www.wuhan.gov.cn/",
})
import urllib3
urllib3.disable_warnings()
BASE = "https://www.wuhan.gov.cn"
CHANNEL_ID = "26164"
LIST_URL = f"{BASE}/zwgk/?channelid={CHANNEL_ID}"
# ──────────────────────────────────────────────────────────
# PART 1: 探测列表页的真实 API 接口
# ──────────────────────────────────────────────────────────
print("=" * 65)
print("PART 1: 探测列表 API 接口")
print("=" * 65)
# 先访问列表页获取 Cookie
session.get(LIST_URL, timeout=10)
IGS_APIS = [
# IGS CMS 标准接口
f"{BASE}/igs/front/siteContent/getSiteContentList.jhtml",
f"{BASE}/igs/front/article/queryArticleList.jhtml",
f"{BASE}/igs/front/article/getArticleList.jhtml",
f"{BASE}/igs/front/content/getContentList.jhtml",
# 武汉网站特有前缀
f"{BASE}/zwgk/api/articleList",
f"{BASE}/zwgk/list.json",
f"{BASE}/ssi/front/queryArticleList.jhtml",
]
PARAM_SETS = [
{"channelId": CHANNEL_ID, "pageIndex": "1", "pageSize": "15"},
{"channelid": CHANNEL_ID, "pageIndex": "1", "pageSize": "15"},
{"channelId": CHANNEL_ID, "page": "1", "limit": "15"},
]
found_api = None
for url in IGS_APIS:
for params in PARAM_SETS:
try:
r = session.post(url, data=params, timeout=8)
ct = r.headers.get("Content-Type", "")
if r.status_code == 200 and len(r.content) > 200:
body = r.text[:400]
is_json = "json" in ct or body.strip().startswith(("{", "["))
if is_json:
print(f"★ 发现接口!POST {url}")
print(f" 参数: {params}")
print(f" 响应: {body}")
found_api = (url, "POST", params)
break
except Exception as e:
pass
try:
r = session.get(url, params=params, timeout=8)
ct = r.headers.get("Content-Type", "")
if r.status_code == 200 and len(r.content) > 200:
body = r.text[:400]
is_json = "json" in ct or body.strip().startswith(("{", "["))
if is_json:
print(f"★ 发现接口!GET {url}")
print(f" 参数: {params}")
print(f" 响应: {body}")
found_api = (url, "GET", params)
break
except Exception as e:
pass
if found_api:
break
if not found_api:
print("未发现 JSON API 接口,需要用 Playwright 渲染")
# ──────────────────────────────────────────────────────────
# PART 2: 分析文章详情页 HTML 结构(无需 API)
# ──────────────────────────────────────────────────────────
print("\n" + "=" * 65)
print("PART 2: 抓取文章详情页")
print("=" * 65)
ARTICLE_URL = f"{BASE}/zwgk/xxgk/zfgz_new/202602/t20260211_2728640.shtml"
print(f"测试 URL: {ARTICLE_URL}")
r = session.get(ARTICLE_URL, timeout=15)
r.encoding = "utf-8"
soup = BeautifulSoup(r.text, "html.parser")
# 提取字段
title_el = soup.select_one("h3.doctitle")
title = title_el.get_text(strip=True) if title_el else "未找到"
# 日期:在 URL 里提取或从页面
date_match = re.search(r'/(\d{4})(\d{2})/t', ARTICLE_URL)
date = f"{date_match.group(1)}-{date_match.group(2)}" if date_match else "未知"
# 文号
docrn_el = soup.select_one("div.docrn")
doc_num = docrn_el.get_text(strip=True) if docrn_el else "无文号"
# 正文
article_head = soup.select_one("div.articleHead")
content = article_head.get_text(separator="\n", strip=True) if article_head else ""
print(f"标题: {title}")
print(f"日期: {date}")
print(f"文号: {doc_num}")
print(f"正文前200字: {content[:200]}")
print(f"正文总长: {len(content)} 字")
# 附件
attachments = []
for a in soup.find_all("a", href=True):
href = a["href"]
if any(href.endswith(ext) for ext in [".pdf", ".doc", ".docx", ".xls", ".xlsx"]):
attachments.append((a.get_text(strip=True), href))
print(f"附件: {attachments if attachments else '无'}")
# ──────────────────────────────────────────────────────────
# PART 3: 尝试 Playwright(需要已安装 Chromium)
# ──────────────────────────────────────────────────────────
print("\n" + "=" * 65)
print("PART 3: Playwright 拦截列表页 API")
print("=" * 65)
try:
from playwright.sync_api import sync_playwright
api_calls = []
json_responses = []
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
ctx = browser.new_context(locale="zh-CN")
page = ctx.new_page()
def on_response(response):
url = response.url
ct = response.headers.get("content-type", "")
if "json" in ct or "jhtml" in url:
try:
body = response.text()
json_responses.append({"url": url, "ct": ct, "body": body[:600]})
except:
pass
# 记录所有非静态请求
if not re.search(r'\.(png|jpg|gif|css|woff|woff2|ico|svg)(\?|$)', url):
api_calls.append(f"[{response.status}] {response.request.method} {url}")
page.on("response", on_response)
page.goto(LIST_URL, wait_until="networkidle", timeout=30000)
page.wait_for_timeout(3000)
print(f"拦截到 {len(api_calls)} 个请求")
for c in api_calls:
print(f" {c}")
print(f"\n其中 JSON 响应 {len(json_responses)} 个:")
for jr in json_responses:
print(f" {jr['url']}")
print(f" {jr['body'][:200]}")
# 渲染后的文章链接
links = page.eval_on_selector_all("a[href]", "els => els.map(e => ({text: e.innerText.trim(), href: e.href}))")
articles = [l for l in links if len(l['text']) > 8 and ('/zwgk/' in l['href'] or re.search(r'\d{6}/t\d', l['href']))]
print(f"\n渲染后文章链接 {len(articles)} 篇:")
for a in articles[:15]:
print(f" [{a['text'][:50]}] -> {a['href']}")
browser.close()
except ImportError:
print("Playwright 未安装,跳过")
except Exception as e:
print(f"Playwright 错误: {e}")
print("请先运行: playwright install chromium")
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment