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
533 lines
19 KiB
Python
Executable File
533 lines
19 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
新矿集团供应链平台 公告抓取工具
|
||
数据源：https://xkgroup.kdcloud.com/isv/q5m9/dist/srmportal.html
|
||
API：getPortalNoticeData (REST, 无需登录即可访问 guest 数据)
|
||
|
||
抓取类型：
|
||
C - 寻源公告（核心采购公告）
|
||
D - 流标公告
|
||
B - 结果公示
|
||
|
||
用法：
|
||
python3 fetch-xkgroup.py # 差量更新（默认抓取昨天的寻源公告）
|
||
python3 fetch-xkgroup.py --all-types # 抓取所有公告类型
|
||
python3 fetch-xkgroup.py --from 2026-05-01 # 指定起始日期
|
||
python3 fetch-xkgroup.py --force # 强制重新抓取（不跳过已下载）
|
||
python3 fetch-xkgroup.py --keyword 水泥 # 按关键词过滤
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from datetime import datetime, timedelta, date
|
||
|
||
try:
|
||
import requests
|
||
except ImportError:
|
||
print("缺少 requests 库，请安装: pip3 install requests")
|
||
sys.exit(1)
|
||
|
||
# ============ 配置区 ============
|
||
BASE_URL = "https://xkgroup.kdcloud.com/kapi/app/srm/srmwebapi"
|
||
ACCOUNT_ID = "2066845847806639104"
|
||
PAGE_SIZE = 50
|
||
|
||
# 公告类型映射
|
||
NOTICE_TYPES = {
|
||
"C": "寻源公告",
|
||
"D": "流标公告",
|
||
"B": "结果公示",
|
||
}
|
||
DEFAULT_TYPES = ["C"] # 默认只抓取寻源公告
|
||
|
||
OUTPUT_BASE = os.path.expanduser("~/desktop/招标公告")
|
||
|
||
HEADERS = {
|
||
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Referer": "https://xkgroup.kdcloud.com/isv/q5m9/dist/srmportal.html",
|
||
}
|
||
|
||
# ============ 核心函数 ============
|
||
|
||
|
||
def fetch_notices(notice_type, page=1, size=PAGE_SIZE):
|
||
"""调用 API 获取公告列表"""
|
||
params = {
|
||
"page": page,
|
||
"size": size,
|
||
"type": notice_type,
|
||
"apiname": "getPortalNoticeData",
|
||
"methodname": "getPortalNoticeData",
|
||
"accountId": ACCOUNT_ID,
|
||
}
|
||
resp = requests.get(BASE_URL, params=params, headers=HEADERS, timeout=30)
|
||
resp.raise_for_status()
|
||
return resp.json()
|
||
|
||
|
||
def fetch_all_notices(notice_type):
|
||
"""翻页获取指定类型的所有公告"""
|
||
all_items = []
|
||
page = 1
|
||
|
||
while True:
|
||
data = fetch_notices(notice_type, page=page)
|
||
if not data.get("success"):
|
||
print(f" ❌ API 请求失败: {data.get('errorCode', 'unknown')}")
|
||
break
|
||
|
||
items = data.get("data", {}).get("noticedata", [])
|
||
total = data.get("data", {}).get("total", 0)
|
||
|
||
all_items.extend(items)
|
||
|
||
total_page = (total + PAGE_SIZE - 1) // PAGE_SIZE
|
||
print(f" 第 {page}/{total_page} 页: {len(items)} 条")
|
||
|
||
if len(items) < PAGE_SIZE or page >= total_page:
|
||
break
|
||
|
||
page += 1
|
||
|
||
return all_items
|
||
|
||
|
||
def normalize_notice(item, notice_type_label):
|
||
"""统一公告数据格式"""
|
||
notice_id = item.get("noticeId", "")
|
||
pub_time = item.get("publishtime", "")
|
||
|
||
# 解析状态
|
||
status = item.get("status", "")
|
||
norm_status = status if status else "未知"
|
||
|
||
return {
|
||
"id": notice_id,
|
||
"title": item.get("noticename", ""),
|
||
"org": item.get("orgname", ""),
|
||
"org_number": item.get("orgnumber", ""),
|
||
"type": notice_type_label,
|
||
"type_code": item.get("type", ""),
|
||
"status": norm_status,
|
||
"publish_time": pub_time,
|
||
"detail_url": item.get("url", ""),
|
||
"mobile_url": item.get("mobileurl", ""),
|
||
"source": "新矿集团",
|
||
}
|
||
|
||
|
||
def match_date_range(publish_time_str, date_from, date_to):
|
||
"""按发布时间过滤日期范围"""
|
||
if not publish_time_str:
|
||
return True # 无时间就不过滤
|
||
try:
|
||
dt = datetime.strptime(publish_time_str[:10], "%Y-%m-%d")
|
||
dt_date = dt.date()
|
||
return date_from <= dt_date <= 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:
|
||
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):
|
||
"""生成公告清单 Markdown"""
|
||
lines = [
|
||
f"# 新矿集团供应链平台 公告清单\n",
|
||
f"**抓取时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')} | **日期范围**: {date_range}",
|
||
f"**公告总数**: {len(notices)} 条\n",
|
||
f"> 数据来源：{BASE_URL} | 平台：新矿集团供应链平台\n",
|
||
"---\n",
|
||
]
|
||
|
||
for i, n in enumerate(notices, 1):
|
||
lines.extend([
|
||
f"## {i}. {n['title']}",
|
||
f"- **发布组织**: {n['org']}",
|
||
f"- **类型**: {n['type']}",
|
||
f"- **状态**: {n['status']}",
|
||
f"- **发布时间**: {n['publish_time']}",
|
||
f"- **详情**: [查看]({n['detail_url']})",
|
||
"",
|
||
"---\n",
|
||
])
|
||
|
||
content = "\n".join(lines)
|
||
date_str = date_range.replace("~", "-")
|
||
filename = f"公告清单_新矿集团_{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="关键词过滤")
|
||
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) # 默认抓取最近 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)
|
||
|
||
# 子目录：html（源码） + pdf（转存）
|
||
html_dir = os.path.join(output_dir, "html")
|
||
pdf_dir = os.path.join(output_dir, "pdf")
|
||
os.makedirs(html_dir, exist_ok=True)
|
||
os.makedirs(pdf_dir, exist_ok=True)
|
||
|
||
# 状态文件
|
||
state_file = os.path.join(output_dir, "xkgroup_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])}")
|
||
print()
|
||
|
||
for ntype in notice_types:
|
||
label = NOTICE_TYPES.get(ntype, ntype)
|
||
print(f"[{label}] 正在获取...")
|
||
|
||
items = fetch_all_notices(ntype)
|
||
print(f" ✅ API 返回 {len(items)} 条")
|
||
|
||
type_new = 0
|
||
type_skip = 0
|
||
for item in items:
|
||
notice = normalize_notice(item, label)
|
||
|
||
# 日期过滤
|
||
if not match_date_range(notice["publish_time"], date_from, date_to):
|
||
filtered_count += 1
|
||
continue
|
||
|
||
# 关键词过滤
|
||
if args.keyword:
|
||
title = notice["title"]
|
||
org = notice["org"]
|
||
if args.keyword not in title and args.keyword not in org:
|
||
filtered_count += 1
|
||
continue
|
||
|
||
notice_id = notice["id"]
|
||
|
||
# 去重
|
||
if not args.force and notice_id in state:
|
||
skip_count += 1
|
||
type_skip += 1
|
||
continue
|
||
|
||
all_notices.append(notice)
|
||
state[notice_id] = {
|
||
"title": notice["title"],
|
||
"org": notice["org"],
|
||
"publish_time": notice["publish_time"],
|
||
"type": notice["type"],
|
||
"detail_url": notice["detail_url"],
|
||
"downloaded_at": datetime.now().isoformat(),
|
||
}
|
||
new_count += 1
|
||
type_new += 1
|
||
|
||
print(f" 新增: {type_new} | 跳过: {type_skip}")
|
||
print()
|
||
|
||
# 保存状态
|
||
save_state(state_file, state)
|
||
|
||
# 为缺少金额的记录从详情页提取
|
||
extract_amounts_for_existing(state, output_dir)
|
||
|
||
# 生成清单
|
||
if all_notices:
|
||
generate_markdown(all_notices, output_dir, date_range)
|
||
|
||
# 详情页转存 PDF + HTML（新公告 + 已有状态中缺文件的）
|
||
all_pending = list(all_notices)
|
||
if not args.force:
|
||
for notice_id, info in state.items():
|
||
pdf_key = info.get("pdf", "")
|
||
html_key = info.get("html", "")
|
||
pdf_ok = pdf_key and os.path.exists(os.path.join(output_dir, pdf_key))
|
||
html_ok = html_key and os.path.exists(os.path.join(output_dir, html_key))
|
||
if pdf_ok and html_ok:
|
||
continue
|
||
all_pending.append({
|
||
"id": notice_id,
|
||
"title": info.get("title", ""),
|
||
"org": info.get("org", ""),
|
||
"publish_time": info.get("publish_time", ""),
|
||
"type": info.get("type", ""),
|
||
"mobile_url": f"https://xkgroup.kdcloud.com/mobile.html?form=mobsp_quonotice_view&billid={notice_id}&userId=guest&accountId=2066845847806639104",
|
||
"detail_url": f"https://xkgroup.kdcloud.com/index.html?formId=quo_notice¬iceId={notice_id}&userId=guest&accountId=2066845847806639104",
|
||
})
|
||
|
||
if all_pending:
|
||
download_pdfs(all_pending, output_dir, html_dir, pdf_dir, state, state_file, force=args.force)
|
||
else:
|
||
print(" ℹ️ 指定日期范围内无新公告")
|
||
|
||
print(f"\n========== 完成 ==========")
|
||
print(f"✅ 新增: {new_count} | ⏩ 跳过: {skip_count} | 🔍 日期过滤: {filtered_count}")
|
||
print(f"📁 保存目录: {output_dir}")
|
||
|
||
|
||
def extract_amount_from_text(text):
|
||
"""从公告正文中提取金额（万元）"""
|
||
if not text:
|
||
return ""
|
||
# 优先匹配：预算金额、控制价、限价、概算、投资估算（必须带"万元"）
|
||
for pattern in [
|
||
r"(?:预算|控制|限价|概算|投资估算|项目投资|合同估算)[金额价款]*[：:]\s*([\d,.]+)\s*万元",
|
||
r"(?:预算|控制|限价|概算|投资估算|项目投资|合同估算)[金额价款]*[：:]\s*([\d,.]+)\s*(?:万|万元)",
|
||
]:
|
||
m = re.search(pattern, text)
|
||
if m:
|
||
val = m.group(1).replace(",", "")
|
||
try:
|
||
num = float(val)
|
||
if num > 0:
|
||
return f"{num}万元"
|
||
except ValueError:
|
||
pass
|
||
# 备选：文中出现 "XXX万元" 的模式（取第一个合理值，排除明显非金额的）
|
||
for m in re.finditer(r"([\d,.]+)\s*万元", text):
|
||
val = m.group(1).replace(",", "")
|
||
try:
|
||
num = float(val)
|
||
if 1 <= num <= 999999:
|
||
return f"{num}万元"
|
||
except ValueError:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def extract_amounts_for_existing(state, output_dir):
|
||
"""为已抓取但缺少金额的公告，访问详情页提取金额"""
|
||
try:
|
||
import asyncio
|
||
from playwright.async_api import async_playwright
|
||
except ImportError:
|
||
print(" ⚠️ Playwright 未安装，跳过金额提取")
|
||
return
|
||
|
||
# 筛选缺少金额的记录
|
||
pending = []
|
||
for notice_id, info in state.items():
|
||
if notice_id == "lastRun":
|
||
continue
|
||
if info.get("amount"):
|
||
continue
|
||
pending.append(notice_id)
|
||
|
||
if not pending:
|
||
print(" ✅ 所有记录已有金额信息")
|
||
return
|
||
|
||
print(f"\n[金额提取] {len(pending)} 条记录缺少金额，正在从详情页提取...")
|
||
|
||
async def _run():
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.launch(headless=True)
|
||
success = 0
|
||
|
||
for i, notice_id in enumerate(pending):
|
||
info = state[notice_id]
|
||
url = f"https://xkgroup.kdcloud.com/mobile.html?form=mobsp_quonotice_view&billid={notice_id}&userId=guest&accountId=2066845847806639104"
|
||
print(f" [{i+1}/{len(pending)}] {info.get('title', '')[:40]}...", end=" ", flush=True)
|
||
|
||
try:
|
||
page = await browser.new_page()
|
||
page.set_default_timeout(20000)
|
||
await page.goto(url, wait_until="domcontentloaded", timeout=20000)
|
||
await page.wait_for_timeout(2000)
|
||
text = await page.evaluate("() => document.body.innerText")
|
||
await page.close()
|
||
|
||
amount = extract_amount_from_text(text)
|
||
if amount:
|
||
state[notice_id]["amount"] = amount
|
||
success += 1
|
||
print(f"✅ {amount}")
|
||
else:
|
||
print("⏭️ 未找到金额")
|
||
except Exception as e:
|
||
print(f"❌ {e}")
|
||
|
||
await browser.close()
|
||
return success
|
||
|
||
import asyncio
|
||
try:
|
||
success = asyncio.run(_run())
|
||
except RuntimeError:
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
success = loop.run_until_complete(_run())
|
||
loop.close()
|
||
|
||
if success:
|
||
save_state(os.path.join(output_dir, "xkgroup_downloaded.json"), state)
|
||
print(f" 💰 金额提取完成: {success} 条")
|
||
|
||
|
||
def download_pdfs(notices, output_dir, html_dir, pdf_dir, state=None, state_file=None, force=False):
|
||
"""将新矿集团公告详情页转存为 PDF"""
|
||
try:
|
||
import asyncio
|
||
from playwright.async_api import async_playwright
|
||
except ImportError:
|
||
print(" ⚠️ Playwright 未安装，跳过 PDF 下载")
|
||
return
|
||
|
||
print("\n[PDF] 正在下载详情页...")
|
||
|
||
async def _run():
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.launch(headless=True)
|
||
success = 0
|
||
fail = 0
|
||
|
||
for i, notice in enumerate(notices):
|
||
notice_id = notice.get("id", "")
|
||
# 手机版页面含完整内容，桌面版只有摘要
|
||
pdf_url = notice.get("mobile_url", "") or notice.get("detail_url", "")
|
||
|
||
if not notice_id or not pdf_url:
|
||
continue
|
||
|
||
pub_date = notice.get("publish_time", "")[:10]
|
||
safe_title = re.sub(r"[\\/:*?\"<>|]", "_", notice["title"][:40])
|
||
base_name = f"{pub_date}_{notice_id}_{safe_title}"
|
||
html_name = f"{base_name}.html"
|
||
pdf_name = f"{base_name}.pdf"
|
||
html_path = os.path.join(html_dir, html_name)
|
||
pdf_path = os.path.join(pdf_dir, pdf_name)
|
||
rel_pdf = f"pdf/{pdf_name}"
|
||
|
||
if not force and os.path.exists(pdf_path):
|
||
continue
|
||
|
||
print(f" [{i+1}/{len(notices)}] {notice['title'][:40]}...", end=" ", flush=True)
|
||
|
||
try:
|
||
page = await browser.new_page()
|
||
page.set_default_timeout(30000)
|
||
await page.goto(pdf_url, wait_until="domcontentloaded", timeout=25000)
|
||
await page.wait_for_timeout(2000)
|
||
|
||
# 提取页面文本，解析金额
|
||
page_text = await page.evaluate("() => document.body.innerText")
|
||
amount = extract_amount_from_text(page_text)
|
||
if amount and state is not None and notice_id in state:
|
||
state[notice_id]["amount"] = amount
|
||
|
||
# 保存 HTML 源码到 html/ 目录（供 AI 阅读）
|
||
html_source = await page.content()
|
||
with open(html_path, "w", encoding="utf-8") as f:
|
||
f.write(html_source)
|
||
|
||
# 获取实际内容高度（SPA 可能 overflow:hidden，scrollHeight 不准确）
|
||
content_height = await page.evaluate(
|
||
"() => Math.max("
|
||
" document.documentElement.scrollHeight,"
|
||
" document.body.scrollHeight,"
|
||
" ...Array.from(document.querySelectorAll('body *'))"
|
||
" .map(el => el.getBoundingClientRect().bottom)"
|
||
" .filter(v => v < 50000),"
|
||
" 1000"
|
||
")"
|
||
)
|
||
await page.pdf(
|
||
path=pdf_path,
|
||
width="1440px",
|
||
height=f"{content_height + 50}px",
|
||
print_background=True,
|
||
)
|
||
await page.close()
|
||
html_size = os.path.getsize(html_path)
|
||
pdf_size = os.path.getsize(pdf_path)
|
||
print(f"✅ PDF:{pdf_size//1024}KB HTML:{html_size//1024}KB")
|
||
success += 1
|
||
|
||
# 更新状态文件，记录路径
|
||
if state is not None and state_file and notice_id in state:
|
||
state[notice_id]["pdf"] = rel_pdf
|
||
state[notice_id]["html"] = f"html/{html_name}"
|
||
with open(state_file, "w", encoding="utf-8") as f:
|
||
json.dump(state, f, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
print(f"❌ {e}")
|
||
fail += 1
|
||
|
||
await browser.close()
|
||
return success, fail
|
||
|
||
import asyncio
|
||
try:
|
||
success, fail = asyncio.run(_run())
|
||
if success:
|
||
print(f" 📄 PDF 下载完成: {success} 成功" + (f", {fail} 失败" if fail else ""))
|
||
except RuntimeError:
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
success, fail = loop.run_until_complete(_run())
|
||
loop.close()
|
||
if success:
|
||
print(f" 📄 PDF 下载完成: {success} 成功" + (f", {fail} 失败" if fail else ""))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run()
|