#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Generate /data/www/files/thhweekly/weekly_opportunity_20261009.html from verified projects. 铁律0: a project with no valid screenshot (>=100KB) or no text snapshot is DROPPED.""" import io, json, os, re, datetime DATE = "20261106" BASE = "/data/www/files" SHOT = os.path.join(BASE, "screenshots", DATE) SNAP = os.path.join(BASE, "snapshots", DATE) TODAY = datetime.date(2026, 11, 6) CYCLE_START, CYCLE_END = "2026-10-06", "2026-11-06" TITLE = "2026年11月 第1周 商业机会资讯" projs = json.load(io.open(os.path.join(BASE, "projects_%s.json" % DATE), encoding="utf-8")) CAPTCHA_MARK = ["Access Verification", "Please slide to verify", "人机验证", "暂无数据", "访问验证", "滑动验证", "Just a moment", "Security Verification"] def snip_text(i): p = os.path.join(SNAP, "s%02d.txt" % i) return io.open(p, encoding="utf-8", errors="ignore").read() if os.path.exists(p) else "" def shot_ok(i): p = os.path.join(SHOT, "p%02d.png" % i) return os.path.exists(p) and os.path.getsize(p) >= 100000 def parse_date(s): m = re.search(r"(20\d{2})\s*[-年/.]\s*(\d{1,2})\s*[-月/.]\s*(\d{1,2})", s or "") if not m: return None try: return datetime.date(int(m.group(1)), int(m.group(2)), int(m.group(3))) except ValueError: return None DATE_RE = r"(\d{4}\s*[-年/.]\s*\d{1,2}\s*[-月/.]\s*\d{1,2})" DL_PATS = [ r"提交(?:投标|响应)文件截止时间[\s\S]{0,90}?" + DATE_RE, r"(?:投标|响应文件|响应|报价|申请)截止(?:时间|日期)[\s\S]{0,50}?" + DATE_RE, r"截止(?:时间|日期)[\s\S]{0,50}?" + DATE_RE, r"并于[\s\S]{0,14}?" + DATE_RE + r"[\s\S]{0,34}?前(?:递交|提交|上传)", r"(?:开标|开启)(?:时间|日期)[\s\S]{0,50}?" + DATE_RE, ] def extract(text): dl, bud = None, None for pat in DL_PATS: m = re.search(pat, text) if m: d = parse_date(m.group(1)) if d: dl = d break for pat in [r"预算(?:总)?金额[^0-9]{0,14}([\d,]+(?:\.\d+)?)\s*万元", r"预算(?:总)?金额[^0-9]{0,14}([\d,]+(?:\.\d+)?)\s*元", r"最高限价[^0-9]{0,14}([\d,]+(?:\.\d+)?)\s*万元", r"最高限价[^0-9]{0,14}([\d,]+(?:\.\d+)?)\s*元", # 青海风格: 预算金额（元）：797200 (unit BEFORE the number; first hit = project total) r"预算(?:总)?金额[（(]元[)）][：:\s]{0,4}([\d,]+(?:\.\d+)?)"]: m = re.search(pat, text) if m: v = float(m.group(1).replace(",", "")) if "万" not in pat: v = v / 10000.0 if v > 0: bud = round(v, 2) break if bud is None: # award tables: 中标（成交）金额 header, amount may sit 200+ chars later in the row for pat, div in [ (r"中标[（(]成交[)）]金额[\s\S]{0,240}?([\d,]+(?:\.\d+)?)\s*万元", 1), (r"中标[（(]成交[)）]金额[\s\S]{0,240}?([\d,]+(?:\.\d+)?)\s*元", 10000.0), (r"(?:中标|成交)金额[：:]?\s*([\d,]+(?:\.\d+)?)\s*万元", 1), (r"(?:中标|成交)金额[：:]?\s*([\d,]+(?:\.\d+)?)\s*元", 10000.0), ]: m = re.search(pat, text) if m: v = float(m.group(1).replace(",", "")) / div if 0.01 <= v <= 200000: bud = round(v, 2) break if bud is None: m = re.search(r"中标(?:（成交）|\(成交\))?(?:金额|价)?[^0-9]{0,14}([\d,]+(?:\.\d+)?)\s*万元", text) if m: bud = float(m.group(1).replace(",", "")) if bud is None: m = re.search(r"(?:中标|成交)(?:（成交）|\(成交\))?[^0-9]{0,20}([\d,]+(?:\.\d+)?)\s*元", text) if m: v = float(m.group(1).replace(",", "")) if v > 100000: bud = round(v / 10000, 2) return dl, bud def award_amount(text): """中标（成交）金额 from the award notice body, in 万元. Skips 代理服务费 noise.""" pats = [ (r"中标[（(]成交[)）]金额[\s\S]{0,60}?([\d,]+(?:\.\d+)?)\s*万元", 1), (r"中标[（(]成交[)）]金额[\s\S]{0,60}?([\d,]+(?:\.\d+)?)\s*元", 10000.0), (r"中标[（(]成交[)）]金额[\s\S]{0,320}?([\d,]+\.\d+)\s*元", 10000.0), (r"(?:中标|成交)(?:价|金额)[：:]?\s*([\d,]+(?:\.\d+)?)\s*(万元|元)", None), ] for pat, div in pats: for m in re.finditer(pat, text): pre = text[max(0, m.start() - 40):m.start()] if "服务费" in pre: continue try: v = float(m.group(1).replace(",", "")) except Exception: continue if div is None: unit = m.group(2) or "" v = v if "万" in unit else v / 10000.0 else: v = v / div if 0.5 <= v <= 200000: return round(v, 2) return None BUYER_PATS = [ r"采购人信息[\s\S]{0,60}?名\s*称[：:]\s*([^\n]{2,40})", r"采购人名称[：:]\s*([^\n]{2,40})", r"采购单位名称[：:]\s*([^\n]{2,40})", r"采购人[：:]\s*([^\n]{2,40})", r"采购单位[：:]\s*([^\n]{2,40})", r"招标人[：:]\s*([^\n]{2,40})", r"受\s*([^\n，,]{4,30}?)\s*的?委托", ] def extract_buyer(text, fallback): for pat in BUYER_PATS: m = re.search(pat, text) if m: v = m.group(1).strip().strip("：: 　") v = re.sub(r"^(名称|地址)[：:]?", "", v).strip() if 3 <= len(v) <= 40 and not v.startswith("项目"): return v[:40] if fallback and fallback != "未披露": return fallback return "未披露" FIT_ORDER = [ (lambda t, b: ("党校" in t or "干部学院" in t) and any(k in t for k in ("信息化", "智慧", "平台", "系统", "媒体", "课程", "实训", "直播", "教室")), "我司有党校信息化/媒体制作成熟产品与课程开发能力，可直接复用"), (lambda t, b: ("党校" in t or "干部学院" in t or "行政学院" in t) and any(k in t for k in ("教学", "录播", "媒体", "研讨", "课程", "课件", "培训", "质量提升", "展陈", "教育基地", "设备")), "党校教学/媒体制作设施与我司录播、课件制作业务高度契合"), (lambda t, b: "组织部" in t and any(k in t for k in ("党建", "系统", "平台", "数字", "培训", "远程教育", "信息化")), "我司已有党建平台建设与干部培训内容案例，经验丰富"), (lambda t, b: "党员教育" in t and any(k in t for k in ("课件", "视频", "课程", "培训", "平台", "系统", "网络")), "我司具备党员教育内容制作与平台建设能力"), (lambda t, b: ("党建" in t) and any(k in t for k in ("平台", "数智", "数字化", "信息化", "智慧", "应用", "阵地", "示范")), "智慧党建平台/数智应用与党建阵地建设是我司重点方向，建议重点跟进"), (lambda t, b: any(k in t for k in ("展陈", "展览", "布展", "陈列", "展厅", "策展")) and any(k in t for k in ("红色", "党建", "廉政", "廉洁", "革命", "烈士", "纪念馆", "文史", "党性", "党的", "党校史")), "我司有红色/党性主题展陈与布展内容制作经验"), (lambda t, b: any(k in t for k in ("展陈", "展览", "布展", "展厅", "策展")), "展陈内容制作与我司内容业务相关，可关注合作"), (lambda t, b: "融媒体" in t or "融媒体中心" in b, "融媒体内容制作与传播是我司可承接业务"), (lambda t, b: any(k in t for k in ("宣传片", "专题片", "纪录片", "宣传视频", "拍摄", "摄制", "系列视频", "视频制作", "电视片", "宣传推广", "宣传阵地", "宣传推介", "文化交流")), "宣传视频/宣传推广内容制作与我司内容制作主业相关，可关注合作"), (lambda t, b: "思政" in t and any(k in t for k in ("国企", "数字化")), "国企思政数字化是我司重点拓展方向"), (lambda t, b: any(k in t for k in ("系统建设", "运维", "平台建设", "平台采购", "云资源", "远程教育")), "我司有系统建设与运维全流程服务能力"), (lambda t, b: "培训" in t or "能力提升" in t, "培训服务与内容开发是我司核心业务方向"), (lambda t, b: "智慧校园" in t or "教育信息化" in t or "智慧教育" in t, "教育信息化是我司可承接方向"), (lambda t, b: "课程" in t or "课件" in t, "课程资源开发制作是我司核心业务方向"), ] def field_weight(t, b): if any(k in t for k in ("智慧党建", "智慧党校")): return 50 if any(k in t for k in ("党校", "干部学院", "干部教育培训", "干部教育", "干部培训", "行政学院")): return 50 if "党员教育" in t and any(k in t for k in ("平台", "系统", "网络", "综合")): return 50 if "党建" in t and any(k in t for k in ("平台", "系统", "数智", "数字化", "信息化", "智慧", "应用")): return 50 if any(k in t for k in ("融媒体", "信息化", "政务", "智慧校园", "数字化", "智慧教育")): return 40 if "组织部" in t and any(k in t for k in ("平台", "系统")): return 40 if any(k in t for k in ("展陈", "展览", "布展", "改陈布展", "体验馆", "展厅", "策展")): return 40 if any(k in t for k in ("党群服务中心", "党建阵地", "党群服务站", "阵地建设")): return 40 if "党建" in t: return 40 if any(k in t for k in ("培训", "能力提升", "职业教育", "高等教育", "教育质量", "教育提升", "开放大学", "教师发展")): return 40 if any(k in t for k in ("宣传片", "专题片", "纪录片", "微短剧", "视频拍摄", "视频制作", "拍摄", "微课", "课程", "摄制", "宣传视频", "系列视频", "宣传推广", "宣传阵地", "宣传推介", "文化交流")): return 30 return 10 KW = ["智慧", "云平台", "数字化", "信息化", "融媒体", "网络培训", "培训", "能力提升", "视频拍摄", "党员教育片", "课程制作", "直播", "视频", "课件", "系统建设", "运维"] HARD_NOISE = ["家具", "空调", "餐饮", "食堂", "物业", "书籍", "图书", "桌椅", "易耗品", "宿舍", "后勤", "食材", "印刷", "广告", "慰问品", "租赁", "出版", "房屋"] SOFT_NOISE = ["设备采购", "设备购置"] TEACH = ["教学", "培训", "教育", "录播", "媒体", "信息", "智慧", "系统", "平台", "课件", "课程", "展陈", "演播"] def score(t, b, prov): s = field_weight(t.replace("经济和信息化", "经信"), b) t2 = t.replace("卫视频道", "").replace("经济和信息化", "经信") for k in KW: if k in t2: s += 5 if prov in ("四川", "青海", "重庆", "云南", "贵州"): s += 20 if any(n in t for n in HARD_NOISE): s = min(s, 30) elif any(n in t for n in SOFT_NOISE) and not any(k in t for k in TEACH): s = min(s, 30) return s def fit_text(t, b, m): for fn, txt in FIT_ORDER: try: if fn(t, b): return txt except Exception: pass if m >= 80: return "与我司核心业务高度契合" if m >= 60: return "与我司业务有一定关联，可探索合作" return "业务关联度一般，可关注" def make_note(t, ty): core = [k for k in ("智慧党建", "党校", "组织部", "党员教育", "融媒体", "展陈", "宣传片", "信息化", "智慧校园", "培训", "平台", "课程") if k in t][:3] return ("命中：" + "/".join(core) if core else "行业拓展") + "｜" + ty kept, dropped = [], [] for i, p in enumerate(projs, 1): raw = snip_text(i) ok_shot, ok_snap = shot_ok(i), len(raw.strip()) > 80 if not (ok_shot and ok_snap): sz = os.path.getsize(os.path.join(SHOT, "p%02d.png" % i)) if os.path.exists(os.path.join(SHOT, "p%02d.png" % i)) else 0 dropped.append((i, p["title"][:36], "shot=%s(%dB) snap=%s" % (ok_shot, sz, ok_snap))) continue body = raw.split("公告正文：", 1)[1] if "公告正文：" in raw else raw body = body.split("\n发布日期：", 1)[0] if any(m.lower() in body.lower() for m in CAPTCHA_MARK): dropped.append((i, p["title"][:36], "页面为人机验证/空壳页，无法取证")) continue dl, bud = extract(body) ty = p["type"] buyer = extract_buyer(body, p.get("buyer", "")) budget = p.get("budget") if str(budget) in ("未披露", "None", ""): budget = bud if bud is not None else budget if ty == "中标公告": aw = award_amount(body) if aw: budget = aw try: budget_v = round(float(str(budget).replace(",", "").replace("元", "").split("（")[0]), 2) except Exception: budget_v = "未披露" if ty == "中标公告": status = "已中标" elif ty in ("意向公开", "单一来源公示"): status = "意向预告" else: status = "可参与" if dl and dl < TODAY: dropped.append((i, p["title"][:36], "截止日期 %s 已过期" % dl)) continue m = score(p["title"], buyer, p["province"]) lvl = "高" if m >= 70 else ("中" if m >= 40 else "低") kept.append({ "idx": 0, "title": p["title"], "buyer": buyer, "province": p["province"], "budget": budget_v, "deadline": dl.isoformat() if dl else "未披露", "pub_date": p["pub_date"], "info_type": ty, "source": p["source"], "url": p["url"], "note": make_note(p["title"], ty), "fit": fit_text(p["title"], buyer, m), "match": m, "match_level": lvl, "status": status, "screenshot": "screenshots/%s/p%02d.png" % (DATE, i), "_src_idx": i, }) for n, p in enumerate(kept, 1): p["idx"] = n # rewrite snapshot headers with the verified final values (body untouched) for p in kept: i = p["_src_idx"] fp = os.path.join(SNAP, "s%02d.txt" % i) if not os.path.exists(fp): continue raw = io.open(fp, encoding="utf-8", errors="ignore").read() body = raw.split("公告正文：", 1)[1] if "公告正文：" in raw else raw body = body.split("\n发布日期：", 1)[0] head = ("项目名称：%s\n采购单位：%s\n预算金额：%s万元\n公告正文：\n%s\n发布日期：%s\n来源URL：%s\n" % (p["title"], p["buyer"], p["budget"], body.strip(), p["pub_date"], p["url"])) io.open(fp, "w", encoding="utf-8").write(head) # orphan cleanup -> 截图数 == 项目数 used = set(p["_src_idx"] for p in kept) orphans = [] for i in range(1, len(projs) + 1): if i in used: continue for f in (os.path.join(SHOT, "p%02d.png" % i), os.path.join(SNAP, "s%02d.txt" % i)): if os.path.exists(f): os.remove(f) orphans.append(os.path.basename(f)) ns = len(kept) n_shot = sum(1 for p in kept if os.path.exists(os.path.join(BASE, p["screenshot"]))) n_snap = sum(1 for p in kept if len(snip_text(p["_src_idx"]).strip()) > 80) print("kept=%d dropped=%d" % (len(kept), len(dropped))) for d in dropped: print(" DROP #%d %s | %s" % d) print("orphans removed:", orphans) print("shot=%d snap=%d" % (n_shot, n_snap)) html = io.open(os.path.join(BASE, "report_template.html"), encoding="utf-8").read() repl = { "__PROJECTS_JSON__": json.dumps(kept, ensure_ascii=False), "__TITLE__": TITLE, "__CYCLE_START__": CYCLE_START, "__CYCLE_END__": CYCLE_END, "__GEN_DATE__": TODAY.isoformat(), "__DATE_DIR__": DATE, "__V_RATE__": "100%% (%d/%d)" % (ns, ns), "__V_SHOTS__": "%d/%d条" % (n_shot, ns), "__V_SNAPS__": "%d/%d条" % (n_snap, ns), } for k, v in repl.items(): html = html.replace(k, v) out = os.path.join(BASE, "thhweekly", "weekly_opportunity_%s.html" % DATE) io.open(out, "w", encoding="utf-8").write(html) json.dump(kept, io.open(os.path.join(BASE, "final_projects_%s.json" % DATE), "w", encoding="utf-8"), ensure_ascii=False, indent=1) print("written:", out, len(html), "bytes") print("high=%d mid=%d low=%d" % (sum(1 for p in kept if p["match_level"] == "高"), sum(1 for p in kept if p["match_level"] == "中"), sum(1 for p in kept if p["match_level"] == "低"))) from collections import Counter print(Counter(p["province"] for p in kept))