残留
link_m = re.search(
r']*title="([^"]+)"', row
)
if not link_m:
continue
tag_m = re.search(r'([^<]+)', row)
date_m = re.search(r"| \s*(20\d{2}-\d{2}-\d{2})\s* | ", row)
tds = re.findall(r"([^<]*) | ", row)
items.append(
{
"id": link_m.group(2),
"url_path": html_lib.unescape(link_m.group(1)),
"title": html_lib.unescape(link_m.group(3)).strip(),
"tag": tag_m.group(1).strip() if tag_m else "",
"publish_date": date_m.group(1) if date_m else "",
"org": tds[-1].strip() if tds else "",
}
)
return items
def fetch_all_notices(category, date_from, max_pages):
"""翻页抓取指定板块的公告列表（按发布时间倒序，翻到早于 date_from 即停）"""
all_items = []
for page_no in range(max_pages):
html_text = fetch_list_page(category, page_no)
items = parse_list_html(html_text)
if not items:
break
all_items.extend(items)
print(f" 第 {page_no + 1} 页: {len(items)} 条")
# 倒序排列：本页最旧一条都早于起始日期，无需继续翻页
dates = [it["publish_date"] for it in items if it["publish_date"]]
if dates and max(dates) < str(date_from):
break
return all_items
def fetch_detail(notice_id, timeout=30):
"""抓取详情页，返回 (html_text, body_text, attach_list)
body_text 为去标签后的正文纯文本（供金额/截止/地区提取），
attach_list 为详情页附件直链列表（PDF/Word/Excel）。
"""
url = f"{BASE_URL}/front/index/v_notice_view?id={notice_id}"
resp = requests.get(url, headers=HEADERS, timeout=timeout)
resp.raise_for_status()
html_text = resp.text
cleaned = re.sub(r"|", " ", html_text)
body_text = html_lib.unescape(re.sub(r"<[^>]+>", " ", cleaned))
body_text = re.sub(r"[ \t\r]+", " ", body_text)
attach_list = re.findall(r'href="(https?://[^"]+\.(?:pdf|docx?|xlsx?)[^"]*)"', cleaned)
return html_text, body_text, attach_list
def download_attachment(url, save_path, timeout=60):
"""下载附件文件（PDF/Word/Excel 直链）"""
resp = requests.get(url, headers={"User-Agent": HEADERS["User-Agent"]}, timeout=timeout, stream=True)
resp.raise_for_status()
with open(save_path, "wb") as f:
for chunk in resp.iter_content(chunk_size=65536):
f.write(chunk)
# ============ 业务过滤与字段提取（沿用旧版） ============
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['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 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 process_details(notices, output_dir, html_dir, pdf_dir, state, state_file):
"""对新公告做详情页增强：存 HTML 源码 + 提取金额/截止/地区 + 下载附件 PDF
新站详情页是纯 SSR HTML，requests 直接可取（不再依赖 Playwright）。
单条失败仅告警，不阻断。
"""
if not notices:
return
print(f"\n[详情增强] {len(notices)} 条新公告，提取金额/截止/地区 + 存档...")
success = 0
for i, notice in enumerate(notices):
notice_id = notice["id"]
print(f" [{i+1}/{len(notices)}] {notice['title'][:40]}...", end=" ", flush=True)
try:
html_text, body_text, attach_list = fetch_detail(notice_id)
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"
with open(os.path.join(html_dir, html_name), "w", encoding="utf-8") as f:
f.write(html_text)
state[notice_id]["html"] = f"html/{html_name}"
# 字段提取
amount = extract_amount_from_text(body_text)
deadline = extract_deadline_from_text(body_text)
region = extract_region_from_text(body_text, notice.get("org", "")) or state[notice_id].get("region", "")
region_detail = extract_region_detail_from_text(body_text)
if amount:
state[notice_id]["amount"] = amount
if deadline:
state[notice_id]["deadline"] = deadline
if region:
state[notice_id]["region"] = region
if region_detail:
state[notice_id]["region_detail"] = region_detail
# 附件下载（取第一个 PDF/文档）
if attach_list:
ext = ".pdf" if ".pdf" in attach_list[0].lower() else os.path.splitext(attach_list[0])[1] or ".pdf"
pdf_name = f"{base_name}{ext}"
try:
download_attachment(attach_list[0], os.path.join(pdf_dir, pdf_name))
state[notice_id]["pdf"] = f"pdf/{pdf_name}"
except Exception as e:
print(f"⚠️ 附件下载失败({e})", end=" ")
print(f"✅ amount={amount or '-'} deadline={deadline or '-'} region={region or '-'}")
success += 1
except Exception as e:
print(f"❌ {e}")
# 边处理边落盘，中断不丢已处理的记录
with open(state_file, "w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False, indent=2)
if success:
print(f" 📦 详情增强完成: {success}/{len(notices)} 条")
def run():
parser = argparse.ArgumentParser(description="新矿集团供应链平台 公告抓取（新站 SSR 版）")
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", dest="max_pages", type=int, default=MAX_PAGES_DEFAULT,
help=f"每板块最大翻页数（默认 {MAX_PAGES_DEFAULT}）")
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
# 允许的类型集合：默认商机类+流标类，--all-types 全部
allowed_tags = BUSINESS_TAGS | FLOWN_TAGS
if args.all_types:
allowed_tags = allowed_tags | RESULT_TAGS
# 输出目录
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"===== 新矿集团供应链平台 公告抓取（新站 {BASE_URL}） =====")
print(f"日期范围: {date_from} ~ {date_to} ({days_span}天)")
print(f"类型范围: {'全部' if args.all_types else '商机类+流标类'}")
print()
for category in CATEGORIES:
print(f"[{category}] 正在获取...")
try:
items = fetch_all_notices(category, date_from, args.max_pages)
except Exception as e:
print(f" ❌ 板块 {category} 抓取失败: {e}")
continue
print(f" ✅ 列表解析 {len(items)} 条")
type_new = 0
type_skip = 0
for item in items:
tag = item["tag"]
# 类型过滤
if tag and tag not in allowed_tags:
filtered_count += 1
continue
# 日期过滤
if not match_date_range(item["publish_date"], date_from, date_to):
filtered_count += 1
continue
# 关键词过滤
if args.keyword:
if args.keyword not in item["title"] and args.keyword not in item["org"]:
filtered_count += 1
continue
# 多次询价标题过滤 (二次/三次/四次...)
if not RETRY_PATTERN.search(item["title"]):
filtered_count += 1
continue
# 发布组织白名单（新站"发布单位"列常为代理机构，采购人名多在标题里，故标题+单位联合匹配）
org_text = item["org"] + item["title"]
if not any(org_name in org_text for org_name in ORG_WHITELIST):
filtered_count += 1
continue
notice_id = item["id"]
# 去重
if not args.force and notice_id in state:
skip_count += 1
type_skip += 1
continue
notice = {
"id": notice_id,
"title": item["title"],
"org": item["org"],
"type": tag or "公告",
"status": "",
"publish_time": item["publish_date"],
"detail_url": f"{BASE_URL}{item['url_path']}",
"mobile_url": "",
"source": "新矿集团",
}
# 从标题/组织名初步提取地区（详情页提取时会补充/覆盖）
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)
# 生成清单
if all_notices:
generate_markdown(all_notices, output_dir, date_range)
# 详情页增强：HTML 存档 + 金额/截止/地区提取 + 附件 PDF 转存
process_details(all_notices, output_dir, html_dir, pdf_dir, state, state_file)
print(f"\n========== 完成 ==========")
print(f"✅ 新增: {new_count} | ⏩ 跳过: {skip_count} | 🔍 日期过滤: {filtered_count}")
print(f"📁 保存目录: {output_dir}")
if __name__ == "__main__":
run()