afac18e843
- W26: NestJS API + Next.js Web + Prisma + SQLite scaffold - W26: Lead 模型 + Crawler + 询价发布助手 (线索池 + 一键发布) - W29.x: MCP 集成 (Integration 模型 + connectMcp + publish_rfq 5 工具端到端跑通) - W30: 3 字段补充 (notes / quotationType / quotationDemand) - W31 P1: 自动填 enterpriseId / userId (Prisma companyUserId 字段 + get_app_identity tool + lead.service.ts publishArgs) - W31: detail 详情页 + admin UI 集成管理 (OpenAppListView + AuditLogListView) 关联: ~/Documents/02-平台开发/产品开发/F002-W31-交接文档.md GIT-AUDIT-REPORT-2026-06-24.md
372 lines
12 KiB
Python
Executable File
372 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
中核集团电子采购平台 公告抓取工具
|
||
数据源：https://www.cnncecp.com/xzbgg/index.jhtml
|
||
平台：中核集团电子采购平台（CNNC）
|
||
|
||
抓取类型：
|
||
- 招标公告（/xzbgg/）
|
||
- 资格预审公告（/zhzzsgg/）
|
||
- 中标候选人公示（/zhzzbhxr/）
|
||
- 中标结果公示（/zhzbjggs/）
|
||
- 终止公告（/zhzzzgg/）
|
||
|
||
注意：详情页有滑块验证码，无法自动获取正文内容。
|
||
本脚本从列表页提取：标题、发布日期、文件下载截止时间、状态、详情链接。
|
||
|
||
用法：
|
||
python3 fetch-cnncecp.py # 差量更新（默认抓取最近7天招标公告）
|
||
python3 fetch-cnncecp.py --all-types # 抓取所有公告类型
|
||
python3 fetch-cnncecp.py --from 2026-05-01 # 指定起始日期
|
||
python3 fetch-cnncecp.py --force # 强制重新抓取（不跳过已下载）
|
||
python3 fetch-cnncecp.py --keyword 防火门 # 按关键词过滤
|
||
python3 fetch-cnncecp.py --max-pages 5 # 限制最大翻页数
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from datetime import datetime, timedelta, date
|
||
|
||
try:
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
except ImportError:
|
||
print("缺少依赖，请安装: pip3 install requests beautifulsoup4")
|
||
sys.exit(1)
|
||
|
||
# ============ 配置区 ============
|
||
BASE_URL = "https://www.cnncecp.com"
|
||
PAGE_SIZE = 16 # 每页显示条数（网站固定）
|
||
|
||
# 公告类型映射：路径 -> 名称
|
||
NOTICE_TYPES = {
|
||
"xzbgg": "招标公告",
|
||
"zhzzsgg": "资格预审公告",
|
||
"zhzzbhxr": "中标候选人公示",
|
||
"zhzbjggs": "中标结果公示",
|
||
"zhzzzgg": "终止公告",
|
||
}
|
||
DEFAULT_TYPES = ["xzbgg"] # 默认只抓取招标公告
|
||
|
||
OUTPUT_BASE = os.path.expanduser("~/desktop/招标公告")
|
||
|
||
# 移动端 UA 绕过 WAF
|
||
HEADERS = {
|
||
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1",
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||
}
|
||
|
||
# ============ 核心函数 ============
|
||
|
||
|
||
def fetch_list_page(type_path, page=1):
|
||
"""获取列表页 HTML"""
|
||
if page == 1:
|
||
url = f"{BASE_URL}/{type_path}/index.jhtml"
|
||
else:
|
||
url = f"{BASE_URL}/{type_path}/index_{page}.jhtml"
|
||
|
||
resp = requests.get(url, headers=HEADERS, timeout=30)
|
||
resp.raise_for_status()
|
||
resp.encoding = "utf-8"
|
||
return resp.text
|
||
|
||
|
||
def parse_list_page(html, type_label):
|
||
"""解析列表页，提取公告列表"""
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
notices = []
|
||
|
||
# 查找公告列表区域
|
||
list_div = soup.find("div", class_="List1")
|
||
if not list_div:
|
||
return notices, 0
|
||
|
||
# 提取总记录数
|
||
total = 0
|
||
page_info = soup.find("div", class_="page")
|
||
if page_info:
|
||
total_match = re.search(r"共(\d+)条记录", page_info.get_text())
|
||
if total_match:
|
||
total = int(total_match.group(1))
|
||
|
||
# 解析每条公告
|
||
for li in list_div.find_all("li"):
|
||
try:
|
||
spans = li.find_all("span")
|
||
a_tag = li.find("a")
|
||
if not a_tag:
|
||
continue
|
||
|
||
notice = {
|
||
"title": a_tag.get_text(strip=True),
|
||
"detail_url": BASE_URL + a_tag.get("href", ""),
|
||
"notice_id": re.search(r"/(\d+)\.jhtml", a_tag.get("href", "")).group(1) if re.search(r"/(\d+)\.jhtml", a_tag.get("href", "")) else "",
|
||
"type": type_label,
|
||
"publish_date": "",
|
||
"deadline": "",
|
||
"status": "",
|
||
}
|
||
|
||
# 解析各字段
|
||
for span in spans:
|
||
text = span.get_text(strip=True)
|
||
cls = span.get("class", [])
|
||
|
||
if "Gray" in cls:
|
||
# 发布日期
|
||
if re.match(r"\d{4}-\d{2}-\d{2}", text):
|
||
notice["publish_date"] = text
|
||
elif "Green" in cls:
|
||
# 正在报名
|
||
notice["status"] = text.strip("[]")
|
||
elif "Blue" in cls:
|
||
# 变更公告
|
||
notice["status"] = text.strip("[]")
|
||
|
||
# 检查 span 内的 em 标签（文件下载截止在 <em class="Red"> 中）
|
||
em_tag = span.find("em", class_="Red")
|
||
if em_tag:
|
||
em_text = em_tag.get_text(strip=True)
|
||
dl_match = re.search(r"文件下载截止[：:](.+)", em_text)
|
||
if dl_match:
|
||
notice["deadline"] = dl_match.group(1).strip()
|
||
|
||
notices.append(notice)
|
||
except Exception as e:
|
||
print(f" ⚠️ 解析异常: {e}")
|
||
continue
|
||
|
||
return notices, total
|
||
|
||
|
||
def match_date_range(publish_date_str, date_from, date_to):
|
||
"""按发布日期过滤"""
|
||
if not publish_date_str:
|
||
return True
|
||
try:
|
||
dt = datetime.strptime(publish_date_str[:10], "%Y-%m-%d").date()
|
||
return date_from <= dt <= date_to
|
||
except ValueError:
|
||
return True
|
||
|
||
|
||
def load_state(state_path):
|
||
"""加载已抓取记录"""
|
||
if os.path.exists(state_path):
|
||
try:
|
||
with open(state_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
|
||
def save_state(state_path, state):
|
||
"""保存已抓取记录"""
|
||
os.makedirs(os.path.dirname(state_path), exist_ok=True)
|
||
with open(state_path, "w", encoding="utf-8") as f:
|
||
json.dump(state, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def generate_markdown(notices, output_dir, date_range, keyword=""):
|
||
"""生成公告清单 Markdown"""
|
||
lines = [
|
||
f"# 中核集团电子采购平台 公告清单\n",
|
||
f"**抓取时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')} | **日期范围**: {date_range}",
|
||
f"**公告总数**: {len(notices)} 条",
|
||
]
|
||
if keyword:
|
||
lines.append(f"**关键词**: {keyword}")
|
||
lines.extend([
|
||
f"",
|
||
f"> 数据来源：{BASE_URL} | 平台：中核集团电子采购平台",
|
||
f"> ⚠️ 详情页需滑块验证码，无法自动获取正文。如需详情请点击链接手动查看。",
|
||
f"",
|
||
"---\n",
|
||
])
|
||
|
||
for i, n in enumerate(notices, 1):
|
||
status_display = f"[{n['status']}]" if n['status'] else ""
|
||
deadline_display = n['deadline'] if n['deadline'] else '-'
|
||
lines.extend([
|
||
f"## {i}. {n['title']}",
|
||
f"- **发布日期**: {n['publish_date']}",
|
||
f"- **公告类型**: {n['type']}",
|
||
f"- **状态**: {status_display}",
|
||
f"- **文件下载截止**: {deadline_display}",
|
||
f"- **公告ID**: {n['notice_id']}",
|
||
f"- **详情链接**: [查看]({n['detail_url']})",
|
||
"",
|
||
"---\n",
|
||
])
|
||
|
||
content = "\n".join(lines)
|
||
date_str = date_range.replace("~", "-")
|
||
kw_suffix = f"_{keyword}" if keyword else ""
|
||
filename = f"公告清单_中核集团{kw_suffix}_{date_str}.md"
|
||
filepath = os.path.join(output_dir, filename)
|
||
|
||
with open(filepath, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
|
||
print(f" 📄 公告清单已生成: {filepath}")
|
||
return filepath
|
||
|
||
|
||
def run():
|
||
parser = argparse.ArgumentParser(description="中核集团电子采购平台 公告抓取")
|
||
parser.add_argument("--from", dest="from_date", help="起始日期 YYYY-MM-DD")
|
||
parser.add_argument("--to", dest="to_date", help="截止日期 YYYY-MM-DD")
|
||
parser.add_argument("--all-types", action="store_true", help="抓取所有公告类型")
|
||
parser.add_argument("--force", action="store_true", help="强制重新抓取")
|
||
parser.add_argument("--keyword", default="", help="关键词过滤（匹配标题）")
|
||
parser.add_argument("--max-pages", type=int, default=0, help="最大翻页数（0=不限制）")
|
||
args = parser.parse_args()
|
||
|
||
today = date.today()
|
||
|
||
if args.from_date:
|
||
date_from = datetime.strptime(args.from_date, "%Y-%m-%d").date()
|
||
else:
|
||
date_from = today - timedelta(days=7)
|
||
|
||
if args.to_date:
|
||
date_to = datetime.strptime(args.to_date, "%Y-%m-%d").date()
|
||
else:
|
||
date_to = today
|
||
|
||
date_range = f"{date_from}~{date_to}"
|
||
days_span = (date_to - date_from).days + 1
|
||
|
||
# 公告类型
|
||
notice_types = list(NOTICE_TYPES.keys()) if args.all_types else DEFAULT_TYPES
|
||
|
||
# 输出目录
|
||
output_dir = os.path.join(OUTPUT_BASE, today.strftime("%Y-%m-%d"))
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
# 状态文件
|
||
state_file = os.path.join(output_dir, "cnncecp_downloaded.json")
|
||
state = load_state(state_file)
|
||
|
||
all_notices = []
|
||
new_count = 0
|
||
skip_count = 0
|
||
filtered_count = 0
|
||
|
||
print(f"===== 中核集团电子采购平台 公告抓取 =====")
|
||
print(f"日期范围: {date_from} ~ {date_to} ({days_span}天)")
|
||
print(f"公告类型: {', '.join([NOTICE_TYPES.get(t, t) for t in notice_types])}")
|
||
if args.keyword:
|
||
print(f"关键词: {args.keyword}")
|
||
print()
|
||
|
||
for ntype in notice_types:
|
||
label = NOTICE_TYPES.get(ntype, ntype)
|
||
print(f"[{label}] 正在获取...")
|
||
|
||
page = 1
|
||
type_total = 0
|
||
type_new = 0
|
||
type_skip = 0
|
||
type_filtered = 0
|
||
has_more = True
|
||
|
||
while has_more:
|
||
# 限制翻页数
|
||
if args.max_pages > 0 and page > args.max_pages:
|
||
print(f" ⏩ 达到最大翻页限制 ({args.max_pages})")
|
||
break
|
||
|
||
try:
|
||
html = fetch_list_page(ntype, page)
|
||
except Exception as e:
|
||
print(f" ❌ 第 {page} 页请求失败: {e}")
|
||
break
|
||
|
||
notices, total = parse_list_page(html, label)
|
||
|
||
if page == 1:
|
||
if total > 0:
|
||
total_pages = (total + PAGE_SIZE - 1) // PAGE_SIZE
|
||
print(f" 📊 共 {total} 条记录 / {total_pages} 页")
|
||
else:
|
||
print(f" ⚠️ 未找到记录")
|
||
break
|
||
|
||
if not notices:
|
||
break
|
||
|
||
for notice in notices:
|
||
type_total += 1
|
||
|
||
# 日期过滤
|
||
if not match_date_range(notice["publish_date"], date_from, date_to):
|
||
type_filtered += 1
|
||
filtered_count += 1
|
||
continue
|
||
|
||
# 关键词过滤
|
||
if args.keyword and args.keyword not in notice["title"]:
|
||
type_filtered += 1
|
||
filtered_count += 1
|
||
continue
|
||
|
||
notice_id = notice["notice_id"]
|
||
|
||
# 去重
|
||
if not args.force and notice_id in state:
|
||
type_skip += 1
|
||
skip_count += 1
|
||
continue
|
||
|
||
all_notices.append(notice)
|
||
state[notice_id] = {
|
||
"title": notice["title"],
|
||
"type": notice["type"],
|
||
"publish_date": notice["publish_date"],
|
||
"deadline": notice["deadline"],
|
||
"status": notice["status"],
|
||
"detail_url": notice["detail_url"],
|
||
"downloaded_at": datetime.now().isoformat(),
|
||
}
|
||
type_new += 1
|
||
new_count += 1
|
||
|
||
print(f" 第 {page} 页: {len(notices)} 条 | 新增: {type_new} | 跳过: {type_skip} | 过滤: {type_filtered}")
|
||
|
||
# 判断是否还有下一页
|
||
if len(notices) < PAGE_SIZE:
|
||
has_more = False
|
||
else:
|
||
page += 1
|
||
# 请求间隔，避免被限流
|
||
import time
|
||
time.sleep(0.5)
|
||
|
||
print(f" ✅ {label} 完成: 新增 {type_new} | 跳过 {type_skip} | 过滤 {type_filtered}")
|
||
print()
|
||
|
||
# 保存状态
|
||
save_state(state_file, state)
|
||
|
||
# 生成清单
|
||
if all_notices:
|
||
generate_markdown(all_notices, output_dir, date_range, args.keyword)
|
||
|
||
print(f"\n========== 完成 ==========")
|
||
print(f"✅ 新增: {new_count} | ⏩ 跳过: {skip_count} | 🔍 过滤: {filtered_count}")
|
||
print(f"📁 保存目录: {output_dir}")
|
||
print(f"📋 状态文件: {state_file}")
|
||
if all_notices:
|
||
print(f"📄 公告清单已生成")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run()
|