#!/usr/bin/env python3 """Fill missing win-bid units from public Yifangbao detail pages. The script does not use browser credentials. It searches public Yifangbao result pages by project title when necessary, but primarily uses the source contentId to open mobile public detail pages and extract supplier fields such as "履约供应商名称". """ from __future__ import annotations import argparse import html import json import re import subprocess import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from datetime import datetime from difflib import SequenceMatcher from pathlib import Path from typing import Any from openpyxl import load_workbook from openpyxl.styles import PatternFill ROOT = Path(__file__).resolve().parents[1] DEFAULT_INPUT = ROOT / "docs/final-results/20260813-yifangbao-phone-legal-enrichment/7月线索导入模板-0715-0812-电话法人已补全-可靠来源二次补查.xlsx" DEFAULT_OUTPUT = ROOT / "docs/final-results/20260813-yifangbao-phone-legal-enrichment/7月线索导入模板-0715-0812-电话法人已补全-中标单位全量补查.xlsx" DEFAULT_AUDIT = ROOT / "docs/final-results/20260813-yifangbao-phone-legal-enrichment/中标单位全量补查审计.json" BASE = "https://www.yfbzb.com" MOBILE_BASE = "https://m.yfbzb.com" SEARCH = BASE + "/search/winBidSearch?defaultSearch=true&keyword={keyword}" UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Mobile/15E148 Safari/604.1" YELLOW = "FFF2CC" RED = "F4CCCC" FIELD_PATTERNS = [ ("履约供应商名称", re.compile(r"(?:履约供应商名称|履约供应商名称\s*)[:：]\s*([^。；;\n\r<]+)")), ("成交供应商名称", re.compile(r"(?:成交供应商名称|成交供应商|成交人|中选人)[:：]\s*([^。；;\n\r<]+)")), ("中标供应商名称", re.compile(r"(?:中标供应商名称|中标供应商|中标单位名称|中标单位|中标人)[:：]\s*([^。；;\n\r<]+)")), ("第一中标候选人", re.compile(r"(?:第一中标候选人|第一成交候选人|第一中选候选人|第一候选人)[:：]\s*([^。；;\n\r<]+)")), ("供应商名称", re.compile(r"(?:供应商名称)[:：]\s*([^。；;\n\r<]+)")), ] COMPANY_SUFFIX = r"(?:集团股份有限公司|股份有限公司|有限责任公司|集团有限公司|有限公司|集团|中心|银行|分行|合作社|经营部|商行|公司|厂|院|所|社|店)" COMPANY_NAME = rf"([\u4e00-\u9fa5A-Za-z0-9（）()·\-]{{2,70}}?{COMPANY_SUFFIX})" CANDIDATE_CONTEXT_PATTERNS = [ ("第一候选人单位名称", re.compile(rf"第一(?:中标|成交|中选)?候选人.{{0,120}}?单位名称[:：]\s*{COMPANY_NAME}")), ("第一候选人", re.compile(rf"第一(?:中标|成交|中选)?候选人[:：]\s*{COMPANY_NAME}")), ("候选人第1名", re.compile(rf"(?:中标|成交|中选)?候选人第\s*1\s*名[:：]\s*{COMPANY_NAME}")), ("推荐第一候选人", re.compile(rf"中标候选人\(({COMPANY_NAME[1:-1]})\).*?推荐其为第一中标候选人")), ("拟中标人", re.compile(rf"拟中标(?:候选人|人)?\s*[:：]?\s*{COMPANY_NAME}")), ("第一名拟入选供应商", re.compile(rf"第一名拟入选供应商[:：]\s*{COMPANY_NAME}")), ("候选人字段", re.compile(rf"(?:中标候选人|成交候选人|中选候选人)[:：]\s*{COMPANY_NAME}")), ("候选人如下", re.compile(rf"(?:中选|成交|中标)候选人如下[:：]\s*(?:\d+[、.])?\s*{COMPANY_NAME}")), ("智慧食堂选包候选人", re.compile(rf"{COMPANY_NAME}\s+选包124-智慧食堂")), ] MANUAL_PUBLIC_CONFIRMED = { "617990634": { "supplier": "南京小牛智能科技有限公司", "field": "公开来源第一中标候选人", "source_url": "https://www.cntcitc.com.cn/news/show-15236.html", "evidence": "中招国际招标有限公司官网公开公示：智慧食堂结算设备项目，第一中标候选人为南京小牛智能科技有限公司。", } } @dataclass class Candidate: href: str title: str score: float def clean(value: Any) -> str: if value is None: return "" text = str(value).strip() return "" if text in {"None", "nan", "--", "暂未公布"} else text def strip_tags(text: str) -> str: text = re.sub(r")<[^<]*)*", " ", text, flags=re.I) text = re.sub(r")<[^<]*)*", " ", text, flags=re.I) text = re.sub(r"<[^>]+>", " ", text) return re.sub(r"\s+", " ", html.unescape(text)).strip() def norm(text: str) -> str: text = html.unescape(re.sub(r"<[^>]+>", "", text)) return re.sub(r"[\s，,。；;：:（）()【】\\[\\]<>《》\"'“”‘’\-—_]+", "", text) def row_date_key(value: Any) -> str: if isinstance(value, datetime): return value.strftime("%Y%m%d") text = clean(value) m = re.match(r"(\d{4})[/-](\d{1,2})[/-](\d{1,2})", text) if not m: return "" return f"{int(m.group(1)):04d}{int(m.group(2)):02d}{int(m.group(3)):02d}" def parse_qiye_url(url: str) -> dict[str, str]: match = re.search(r"/infoDetail/(\d+)/(\d+)/([^/?#]+)", clean(url)) if not match: return {} return {"contentId": match.group(1), "areaId": match.group(2), "pageFrom": match.group(3)} def mobile_detail_url(content_id: str, date_key: str) -> str: return f"{MOBILE_BASE}/winbid/detail/{date_key}_{content_id}.html" def request_text(url: str, retries: int = 2) -> str: last = None for attempt in range(retries + 1): try: completed = subprocess.run( [ "curl", "-L", "--connect-timeout", "12", "--max-time", "30", "-A", UA, "-H", "Accept-Language: zh-CN,zh;q=0.9", "-H", f"Referer: {MOBILE_BASE}/", url, ], check=True, capture_output=True, ) text = completed.stdout.decode("utf-8", errors="ignore") if "denied by UA ACL" in text or "

403 Forbidden

" in text: raise RuntimeError("detail page blocked by 403/UA ACL") return text except (subprocess.CalledProcessError, urllib.error.URLError, TimeoutError) as exc: last = exc if attempt < retries: time.sleep(0.6 * (attempt + 1)) raise RuntimeError(f"request failed: {url}: {last}") def extract_candidates(search_html: str, source_title: str, date_key: str) -> list[Candidate]: out: list[Candidate] = [] source_norm = norm(source_title) for href, title_html in re.findall(r']+class="firstTdAAA"[^>]+href="([^"]+)"[^>]+title="([^"]+)"', search_html): title = strip_tags(title_html) target_norm = norm(title) if not target_norm: continue score = SequenceMatcher(None, source_norm, target_norm).ratio() if source_norm and (source_norm in target_norm or target_norm in source_norm): score = max(score, 0.96) if date_key and f"/detail/{date_key}_" in href: score += 0.03 out.append(Candidate(urllib.parse.urljoin(BASE, href), title, min(score, 1.0))) out.sort(key=lambda item: item.score, reverse=True) return out def search_candidates(title: str, date_key: str) -> list[Candidate]: queries = [] base_title = re.sub(r"合同履约验收公告$", "", title).strip() if base_title: queries.append(base_title) if "的网上超市采购项目" in title: queries.append(title.split("的网上超市采购项目", 1)[0]) queries.append(title) seen: dict[str, Candidate] = {} for query in dict.fromkeys(q for q in queries if q): url = SEARCH.format(keyword=urllib.parse.quote(query)) page = request_text(url) for candidate in extract_candidates(page, title, date_key): if candidate.href not in seen or candidate.score > seen[candidate.href].score: seen[candidate.href] = candidate if seen: break time.sleep(0.15) return sorted(seen.values(), key=lambda item: item.score, reverse=True) def reject_non_award(title: str, text: str) -> str: combined = re.split(r"关注 乙方宝服务号|相关中标信息推荐", f"{title} {text[:2000]}")[0] if re.search(r"(废标|流标|终止|失败|采购失败|比选失败|询比失败|响应供应商不足|报名单位数量不足|有效供应商不足|不足三家|直接采购公示|方式公示|其他公告|录取结果公告|预留项目执行情况公告|面向中小企业预留项目执行情况|招标公告\[?变更公告\]?)", combined): return "公告类型未形成可确认中标单位" return "" def clean_supplier_name(value: str) -> str: supplier = re.sub(r"\s+", "", html.unescape(value)).strip("* ：:，,。；;") supplier = re.sub(r"(三、|四、|五、|六、|七、|八、|九、|十、|十一、|一、|二、).*$", "", supplier).strip() supplier = re.sub(r"(为中标人|为第一中标候选人|为成交供应商|为成交人|为中选人|排名第?一|综合排名第一)$", "", supplier).strip() supplier = re.sub(r"^(名称|单位名称|供应商名称|中标候选人[:：]?|成交候选人[:：]?|中选候选人[:：]?|中标候选人第\d+名|第一名拟入选供应商)", "", supplier).strip("：:") return supplier def looks_like_company(text: str) -> bool: if not text or len(text) < 4 or len(text) > 80: return False if any(token in text for token in ["点击登录查看", "****", "报价", "排名", "金额", "项目", "公告", "采购人", "代理", "联系方式"]): return False return bool(re.search(r"(公司|集团|中心|银行|分行|合作社|经营部|商行|厂|院|所|社|店)$", text)) def extract_first_rank_from_tables(detail_html: str) -> tuple[str, str]: for table in re.findall(r"", detail_html, flags=re.I | re.S): rows = re.findall(r"", table, flags=re.I | re.S) if len(rows) < 2: continue header = strip_tags(rows[0]) if not re.search(r"(中标候选人|成交候选人|中选候选人|供应商|中标人|成交人)", header): continue for row in rows[1:4]: cells = [strip_tags(cell) for cell in re.findall(r"", row, flags=re.I | re.S)] cells = [clean_supplier_name(cell) for cell in cells if clean_supplier_name(cell)] if not cells: continue if any(cell in {"1", "第一名", "第1名"} for cell in cells) or "排名" in header or "中标候选人" in header: for cell in cells: if looks_like_company(cell): return cell, "表格第一候选人" return "", "" def extract_candidate_context(text: str) -> tuple[str, str]: for field, pattern in CANDIDATE_CONTEXT_PATTERNS: match = pattern.search(text) if not match: continue supplier = clean_supplier_name(match.group(1)) if supplier and looks_like_company(supplier): return supplier, field match = re.search(rf"中标候选人名称[:：]\s*{COMPANY_NAME}", text) if match: supplier = clean_supplier_name(match.group(1)) if supplier and looks_like_company(supplier): return supplier, "中标候选人名称" match = re.search(rf"投标单位名称.{{0,160}}?1\s+{COMPANY_NAME}", text) if match: supplier = clean_supplier_name(match.group(1)) if supplier and looks_like_company(supplier): return supplier, "表格第一投标单位" return "", "" def extract_supplier(detail_html: str, title: str = "") -> tuple[str, str, str]: blocks = [] meta = re.search(r'= 20: script_blocks.append(plain) for key in ("description", "content"): for raw in re.findall(rf"{key}:`([^`]+)`", detail_html): try: script_blocks.append(strip_tags(raw.encode("utf-8").decode("unicode_escape"))) except UnicodeDecodeError: script_blocks.append(strip_tags(raw)) blocks.extend(sorted(set(script_blocks), key=len, reverse=True)[:5]) blocks.append(strip_tags(detail_html)) text = "\n".join(blocks) text = re.sub(r"\s+", " ", text) reject_reason = reject_non_award(title, text) for field, pattern in FIELD_PATTERNS: match = pattern.search(text) if not match: continue supplier = clean_supplier_name(match.group(1)) if supplier and looks_like_company(supplier): return supplier, field, reject_reason candidate_supplier, candidate_field = extract_candidate_context(text) if candidate_supplier: return candidate_supplier, candidate_field, reject_reason table_supplier, table_field = extract_first_rank_from_tables(detail_html) if table_supplier: return table_supplier, table_field, reject_reason return "", "", reject_reason def fill(args: argparse.Namespace) -> dict[str, Any]: wb = load_workbook(args.input) ws = wb.active headers = [clean(ws.cell(2, col).value) for col in range(1, ws.max_column + 1)] col = {header: idx + 1 for idx, header in enumerate(headers) if header and header not in {"单位名称"}} # The import template repeats "单位名称": col 10 is bidder, col 13 is win-bid unit. winbid_unit_col = 13 if headers[12] == "单位名称" and headers[13] == "中标单位联系人" else None required = ["项目名称", "信息发布时间", "官网查看地址", "本次来源类型", "本次置信度", "本次来源URL", "证据摘要", "缺口/说明"] missing = [name for name in required if name not in col] if missing or not winbid_unit_col: raise RuntimeError(f"missing columns: {missing}; winbid_unit_col={winbid_unit_col}") yellow = PatternFill("solid", fgColor=YELLOW) red = PatternFill("solid", fgColor=RED) audit: list[dict[str, Any]] = [] filled = 0 questionable = 0 failed = 0 for row in range(3, ws.max_row + 1): if clean(ws.cell(row, winbid_unit_col).value): continue title = clean(ws.cell(row, col["项目名称"]).value) if not title: continue date_key = row_date_key(ws.cell(row, col["信息发布时间"]).value) params = parse_qiye_url(clean(ws.cell(row, col["官网查看地址"]).value)) record: dict[str, Any] = {"row": row, "title": title, "date_key": date_key, **params} try: if params.get("contentId") and date_key: detail_url = mobile_detail_url(params["contentId"], date_key) detail = request_text(detail_url) score = 1.0 match_title = title else: candidates = search_candidates(title, date_key) record["candidates"] = [{"href": c.href, "title": c.title, "score": round(c.score, 4)} for c in candidates[:3]] if not candidates or candidates[0].score < 0.72: failed += 1 record["status"] = "未定位可靠详情" audit.append(record) continue best = candidates[0] detail_url = best.href.replace("https://www.yfbzb.com", "https://m.yfbzb.com") detail = request_text(detail_url) score = best.score match_title = best.title supplier, field, reject_reason = extract_supplier(detail, title) manual = MANUAL_PUBLIC_CONFIRMED.get(params.get("contentId", "")) if not supplier and manual: supplier = manual["supplier"] field = manual["field"] detail_url = manual["source_url"] reject_reason = "候选人公示，需复核最终中标结果" record.update({"detail_url": detail_url, "match_title": match_title, "score": round(score, 4), "field": field, "supplier": supplier, "reject_reason": reject_reason}) if not supplier: failed += 1 record["status"] = reject_reason or "未解析到供应商字段" note = reject_reason or "乙方宝公开详情未展示中标单位字段；需登录乙方宝详情人工复核" ws.cell(row, col["缺口/说明"], note).fill = red ws.cell(row, col["本次来源类型"], "乙方宝公开详情").fill = red ws.cell(row, col["本次置信度"], "待复核").fill = red ws.cell(row, col["本次来源URL"], detail_url).fill = red audit.append(record) continue is_high = field in {"履约供应商名称", "成交供应商名称", "中标供应商名称", "供应商名称"} and score >= 0.9 and not reject_reason fill_color = yellow if is_high else red ws.cell(row, winbid_unit_col, supplier).fill = fill_color ws.cell(row, col["本次来源类型"], "乙方宝公开详情").fill = fill_color ws.cell(row, col["本次置信度"], "高" if is_high else "中").fill = fill_color ws.cell(row, col["本次来源URL"], detail_url).fill = fill_color evidence = manual["evidence"] if manual else f"乙方宝公开详情字段：{field}：{supplier}" ws.cell(row, col["证据摘要"], evidence).fill = fill_color note = "" if is_high else f"请复核：解析字段 {field}" + (f"；{reject_reason}" if reject_reason else "") ws.cell(row, col["缺口/说明"], note).fill = fill_color filled += 1 questionable += 0 if is_high else 1 record["status"] = "已写回-高置信" if is_high else "已写回-待复核" except Exception as exc: # keep row-level audit and continue failed += 1 record["status"] = "异常" record["error"] = str(exc) audit.append(record) time.sleep(args.delay) args.output.parent.mkdir(parents=True, exist_ok=True) wb.save(args.output) args.audit.write_text(json.dumps(audit, ensure_ascii=False, indent=2), encoding="utf-8") return { "input": str(args.input), "output": str(args.output), "audit": str(args.audit), "filled": filled, "questionable": questionable, "failed": failed, "audited_rows": len(audit), } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Fill missing win-bid units from public Yifangbao pages.") parser.add_argument("--input", type=Path, default=DEFAULT_INPUT) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) parser.add_argument("--audit", type=Path, default=DEFAULT_AUDIT) parser.add_argument("--delay", type=float, default=0.25) return parser.parse_args() def main() -> int: print(json.dumps(fill(parse_args()), ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())