bd213eb5ec
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
729 lines
27 KiB
Python
Executable File
729 lines
27 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 load_accumulated_state(output_base, state_filename, today_str):
|
||
"""加载历史状态实现跨天去重：合并所有历史状态文件中的已下载记录"""
|
||
accumulated = {}
|
||
last_run_date = None
|
||
state_dir = os.path.join(output_base, today_str)
|
||
# 先加载今日状态
|
||
today_state = load_state(os.path.join(state_dir, state_filename))
|
||
if today_state:
|
||
for k, v in today_state.items():
|
||
if k != "lastRun":
|
||
accumulated[k] = v
|
||
last_run = today_state.get("lastRun")
|
||
if last_run:
|
||
last_run_date = last_run[:10]
|
||
# 扫描历史目录（最近 30 天），合并之前下载的记录
|
||
try:
|
||
for i in range(1, 31):
|
||
d = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d")
|
||
path = os.path.join(output_base, d, state_filename)
|
||
if os.path.exists(path):
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
prev = json.load(f)
|
||
except:
|
||
continue
|
||
for k, v in prev.items():
|
||
if k != "lastRun" and k not in accumulated:
|
||
accumulated[k] = v
|
||
if k == "lastRun" and not last_run_date:
|
||
last_run_date = str(v)[:10]
|
||
except:
|
||
pass
|
||
return accumulated, last_run_date
|
||
|
||
|
||
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()
|
||
|
||
# 加载历史已下载记录（跨天去重）
|
||
today_str = today.strftime("%Y-%m-%d")
|
||
accumulated_state, last_run_date = load_accumulated_state(OUTPUT_BASE, "xkgroup_downloaded.json", today_str)
|
||
|
||
if args.from_date:
|
||
date_from = datetime.strptime(args.from_date, "%Y-%m-%d").date()
|
||
elif not args.force and last_run_date:
|
||
date_from = datetime.strptime(last_run_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_str)
|
||
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 = accumulated_state
|
||
|
||
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
|
||
|
||
# 从标题/组织名初步提取地区（详情页提取时会补充/覆盖）
|
||
title_region = extract_region_from_text(notice["title"], notice["org"])
|
||
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(),
|
||
"region": title_region,
|
||
}
|
||
new_count += 1
|
||
type_new += 1
|
||
|
||
print(f" 新增: {type_new} | 跳过: {type_skip}")
|
||
print()
|
||
|
||
# 保存状态（含 lastRun 时间戳，用于下次增量爬取）
|
||
state["lastRun"] = datetime.now().isoformat()
|
||
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 list(state.items()):
|
||
if not isinstance(info, dict):
|
||
continue
|
||
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_deadline_from_text(text):
|
||
"""从公告正文中提取截止日期（采购文件获取截止时间）"""
|
||
if not text:
|
||
return ""
|
||
# 模式1: "起至2026年6月22日19:00（北京时间）止" → 取截止日期
|
||
m = re.search(r"起至(\d{4})年(\d{1,2})月(\d{1,2})日(\d{1,2}):(\d{2})", text)
|
||
if m:
|
||
return f"{m.group(1)}-{m.group(2).zfill(2)}-{m.group(3).zfill(2)}"
|
||
# 模式2: "获取时间：...至2026-06-22 19:00"
|
||
m = re.search(r"至\s*(\d{4})-(\d{1,2})-(\d{1,2})\s+\d{1,2}:\d{2}", text)
|
||
if m:
|
||
return f"{m.group(1)}-{m.group(2).zfill(2)}-{m.group(3).zfill(2)}"
|
||
# 模式3: "截止时间：2026年6月22日" 或 "截止日期：2026-06-22"
|
||
for pat in [
|
||
r"截止[时间日期]*[：:]\s*(\d{4})年(\d{1,2})月(\d{1,2})日",
|
||
r"截止[时间日期]*[：:]\s*(\d{4})-(\d{1,2})-(\d{1,2})",
|
||
r"递交[文件]*截止[时间日期]*[：:]\s*(\d{4})年(\d{1,2})月(\d{1,2})日",
|
||
r"递交[文件]*截止[时间日期]*[：:]\s*(\d{4})-(\d{1,2})-(\d{1,2})",
|
||
]:
|
||
m = re.search(pat, text)
|
||
if m:
|
||
return f"{m.group(1)}-{m.group(2).zfill(2)}-{m.group(3).zfill(2)}"
|
||
return ""
|
||
|
||
|
||
# 全国省份/直辖市/自治区
|
||
PROVINCES = [
|
||
'北京', '天津', '上海', '重庆',
|
||
'河北', '山西', '辽宁', '吉林', '黑龙江',
|
||
'江苏', '浙江', '安徽', '福建', '江西', '山东',
|
||
'河南', '湖北', '湖南', '广东', '海南',
|
||
'四川', '贵州', '云南', '陕西', '甘肃', '青海',
|
||
'台湾', '内蒙古', '广西', '西藏', '宁夏', '新疆',
|
||
]
|
||
TWO_CHAR_PROVS = [p for p in PROVINCES if len(p) == 2]
|
||
SPECIAL_PROVS = ['内蒙古', '广西', '新疆', '西藏', '宁夏']
|
||
|
||
|
||
def extract_province_best(text):
|
||
"""从任意文本中提取最可能的省份（先关键词匹配，避免正则在多字省份上截断）"""
|
||
if not text:
|
||
return ''
|
||
text_clean = text.replace(' ', '')
|
||
# 1. 特殊多字省份（优先，避免"新疆维吾尔自治区"被截）
|
||
for sp in SPECIAL_PROVS:
|
||
if sp in text_clean:
|
||
return sp
|
||
# 2. 直辖市
|
||
for dc in ['北京', '天津', '上海', '重庆']:
|
||
if dc in text_clean:
|
||
return dc
|
||
# 3. "XX省"（如"陕西省"、"广东省"）
|
||
m = re.search(r'([\u4e00-\u9fff]{2}省)', text)
|
||
if m:
|
||
return m.group(1)[:2]
|
||
# 4. 两字省份关键词
|
||
for p in TWO_CHAR_PROVS:
|
||
if p in text_clean:
|
||
return p
|
||
return ''
|
||
|
||
|
||
def extract_city_from_text(text):
|
||
"""从文本中提取城市名（XX市）"""
|
||
if not text:
|
||
return ''
|
||
m = re.search(r'([\u4e00-\u9fff]{2,4}(?:市))', text)
|
||
if m:
|
||
city = m.group(1)
|
||
if city not in ['北京市', '天津市', '上海市', '重庆市']:
|
||
return city
|
||
return ''
|
||
|
||
|
||
def extract_region_from_text(text, org=""):
|
||
"""从公告正文或公司名中提取地区（省份）"""
|
||
if not text and not org:
|
||
return ""
|
||
# 1. 从正文找项目/交货/施工地点
|
||
if text:
|
||
for pat in [
|
||
r"(?:项目|交货|送货|施工|到货|供货|工程|实施|服务)[地址地点]*[：:]\s*([\u4e00-\u9fff]{2,8}(?:省|市|自治区))",
|
||
r"(?:地[址点]|所在地区)[：:]\s*([\u4e00-\u9fff]{2,8}(?:省|市|自治区))",
|
||
r"位\s*于\s*([\u4e00-\u9fff]{2,4}(?:省|市|自治区))",
|
||
r"(?:项目|工程)\s*(?:位于|所在地|所在)[：:：\s]*([\u4e00-\u9fff]{2,4}(?:省|市|自治区))",
|
||
]:
|
||
m = re.search(pat, text)
|
||
if m:
|
||
region = m.group(1)
|
||
for sp in SPECIAL_PROVS:
|
||
if sp in region:
|
||
return sp
|
||
return region[:2] # 取前两个字
|
||
# 2. 从全文扫描省份关键词
|
||
prov = extract_province_best(text)
|
||
if prov:
|
||
return prov
|
||
# 3. 从公司名推断
|
||
if org:
|
||
org_prov = extract_province_best(org)
|
||
if org_prov:
|
||
return org_prov
|
||
return ""
|
||
|
||
|
||
def extract_region_detail_from_text(text):
|
||
"""从公告正文中提取详细收货/交货地址"""
|
||
if not text:
|
||
return ""
|
||
for pat in [
|
||
r"(?:交货|收货|送货|项目|施工|到货|供货|工程|实施|服务)[地址地点]*[：:]\s*([\u4e00-\u9fff\d\-—\s]{6,60}(?:路|街|道|巷|号|村|镇|乡|区|县|市|大厦|楼|园|区|厂|公司|仓库))(?:\n|$)",
|
||
r"(?:地[址点])[：:]\s*([\u4e00-\u9fff\d\-—\s]{6,60}(?:路|街|道|巷|号|村|镇|乡|区|县|市|大厦|楼|园|区|厂|公司|仓库))(?:\n|$)",
|
||
r"建设地点[：:]\s*([\u4e00-\u9fff\d\-—\s]{6,60}?)(?:\n|$)",
|
||
r"项目地点[：:]\s*([\u4e00-\u9fff\d\-—\s]{6,60}?)(?:\n|$)",
|
||
]:
|
||
m = re.search(pat, text)
|
||
if m:
|
||
addr = m.group(1).strip()
|
||
if len(addr) >= 6:
|
||
return addr
|
||
return ""
|
||
|
||
|
||
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 not isinstance(info, dict):
|
||
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)
|
||
deadline = extract_deadline_from_text(text)
|
||
region = extract_region_from_text(text, info.get("org", ""))
|
||
region_detail = extract_region_detail_from_text(text)
|
||
updated = False
|
||
if amount:
|
||
state[notice_id]["amount"] = amount
|
||
updated = True
|
||
if deadline:
|
||
state[notice_id]["deadline"] = deadline
|
||
updated = True
|
||
if region:
|
||
state[notice_id]["region"] = region
|
||
updated = True
|
||
if region_detail:
|
||
state[notice_id]["region_detail"] = region_detail
|
||
updated = True
|
||
if updated:
|
||
success += 1
|
||
print(f"✅ amount={amount or '-'} deadline={deadline or '-'} region={region or '-'} addr={region_detail or '-'}")
|
||
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
|
||
deadline = extract_deadline_from_text(page_text)
|
||
if deadline and state is not None and notice_id in state:
|
||
state[notice_id]["deadline"] = deadline
|
||
region = extract_region_from_text(page_text, notice.get("org", ""))
|
||
if region and state is not None and notice_id in state:
|
||
state[notice_id]["region"] = region
|
||
region_detail = extract_region_detail_from_text(page_text)
|
||
if region_detail and state is not None and notice_id in state:
|
||
state[notice_id]["region_detail"] = region_detail
|
||
|
||
# 保存 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()
|