#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Select + score + curl-verify the 20261030 verification batch.""" import json, io, re, os, subprocess, datetime from collections import Counter, defaultdict BASE = "/data/www/files" cands = json.load(io.open(BASE + "/cand_pool_1030.json", encoding="utf-8")) EXTRA_NOISE = ["四位一体", "抓党建促乡村振兴", "租赁办公办学用房", "租赁权拍卖", "拍卖", "易耗品", "清洁用品", "废标", "终止公告", "流标", "重新招标失败", "食堂", "食材", "车辆", "加油", "保险", "印刷", "物业", "保洁", "公开招聘", "遴选公告", "人事考试", "体检", "医疗", "药品", "疫苗", "蔬菜", "肉类", "米面", "粮油", "服装", "制服", "门禁", "监控", "电梯", "消防", "空调", "家具", "绿化", "道路", "硬化", "管网", "水库", "河道", "桥梁", "安置", "老旧小区", "拆除", "土建", # 2nd pass: construction / upkeep / logistics noise "食宿", "厨具", "供热", "供暖", "电力增容", "配电", "修缮", "整修", "维修", "附属工程", "公租房", "合同变更", "合同公告", "运行管理服务", "办公楼", "装修", "电缆安装", "干道", "危房", "加固", "外墙", "消防设施", "电梯采购", "物业服务", "保安服务", "餐饮服务", "食材配送", "苗木", "沥青", "边坡", "挡墙", "围墙", "路面", # medical / archive / prison / housing / telecom noise "医院", "卫生院", "中医", "磁共振", "X射线", "X线", "DR(", "巡诊车", "果园", "监狱", "矫治", "保障性住房", "线路租用", "已撤销", "档案数字化", "文物数字化", "馆藏文物", "中医药", "警务", "看守所", "土地承包", "互联互通四级", "增材", "赛事配套"] KW = ["智慧", "云平台", "数字化", "信息化", "融媒体", "网络培训", "培训", "能力提升", "视频拍摄", "党员教育片", "课程制作", "直播", "视频", "课件", "系统建设", "运维"] HARD_NOISE = ["家具", "空调", "餐饮", "食堂", "物业", "书籍", "图书", "桌椅", "易耗品", "宿舍", "后勤", "食材", "印刷", "广告", "慰问品", "租赁", "出版", "房屋"] SOFT_NOISE = ["设备采购", "设备购置"] TEACH = ["教学", "培训", "教育", "录播", "媒体", "信息", "智慧", "系统", "平台", "课件", "课程", "展陈", "演播"] def field_weight(t): 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 def score(t, prov): t2 = t.replace("卫视频道", "").replace("经济和信息化", "经信") s = field_weight(t2) 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 # ---- dedupe + noise ---- AWARD_MARK = ("中标（成交）结果", "中标(成交)结果", "中标结果", "成交结果", "结果公告", "结果公示", "中标公告", "成交公告", "中标（成交）公告", "中标候选人") INTENT_MARK = ("意向公开", "采购意向", "需求公示", "单一来源公示", "单一来源采购方式") def fix_type(t, cur): if any(k in t for k in AWARD_MARK): return "中标公告" if any(k in t for k in INTENT_MARK): return "意向公开" if cur in ("中标公告", "意向公开"): return cur return "招标公告" pool, seen = [], set() for c in cands: t = c["title"] if c["source"].startswith("全国公共资源交易平台"): continue # ggzy.gov.cn 详情页本期被 IP 限流 (code:800)，无法截图 -> 不收录 if t.endswith(".ZBJ") or t.endswith(".zbj"): continue if any(n in t for n in EXTRA_NOISE): continue if t.endswith("合同") or "项目合同" in t: continue key = re.sub(r"[\s（）()【】\-—·]+", "", t)[:30] if key in seen: continue seen.add(key) c["score"] = score(t, c["province"]) c["type"] = fix_type(t, c.get("type") or "") pool.append(c) print("after dedupe/noise:", len(pool), Counter(p["province"] for p in pool)) # ---- select ---- pool.sort(key=lambda x: (-x["score"], x["pub_date"])) sel = [] # Sichuan: up to 42, prefer score >= 40 sc = [p for p in pool if p["province"] == "四川"] sel += sc[:46] chosen = set(id(p) for p in sel) # Others: capped per province byp = defaultdict(list) for p in pool: if p["province"] in ("四川", "青海"): continue byp[p["province"]].append(p) CAP = {"甘肃": 20, "重庆": 16} for prov, lst in sorted(byp.items(), key=lambda kv: -len(kv[1])): sel += lst[:CAP.get(prov, 3)] # top up to 86 total with best remaining (exclude Qinghai) if len(sel) < 86: taken = set(id(p) for p in sel) rest = [p for p in pool if p["province"] != "青海" and id(p) not in taken] sel += rest[:86 - len(sel)] # hard dedupe (same dict object added twice) + stable order uniq, ids = [], set() for p in sel: if id(p) in ids: continue ids.add(id(p)) uniq.append(p) sel = uniq # sort final: Sichuan first, then others by province/score sel.sort(key=lambda x: (0 if x["province"] == "四川" else 1, x["province"], -x["score"])) for i, p in enumerate(sel, 1): p["idx"] = i print("selected:", len(sel), Counter(p["province"] for p in sel)) print("score buckets:", Counter("高" if p["score"] >= 70 else ("中" if p["score"] >= 40 else "低") for p in sel)) # ---- curl liveness pre-check ---- UA = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/149.0.0.0 Safari/537.36" def head(p): try: r = subprocess.run(["curl", "-s", "-m", "25", "-o", "/dev/null", "-w", "%{http_code}", "-L", "-A", UA, p["url"]], capture_output=True, text=True, timeout=35) return r.stdout.strip() except Exception: return "ERR" import concurrent.futures as cf with cf.ThreadPoolExecutor(max_workers=8) as ex: codes = list(ex.map(head, sel)) dead = [] for p, c in zip(sel, codes): p["http"] = c if c not in ("200",): dead.append((p["idx"], c, p["title"][:40])) print("dead links:", len(dead)) for d in dead: print(" DEAD", d) json.dump(sel, io.open(BASE + "/projects_20261030.json", "w", encoding="utf-8"), ensure_ascii=False, indent=1) print("\n=== SELECTED ===") for p in sel: print("%2d %s %-5s %-4s %3d %s %s" % (p["idx"], p["http"], p["province"], p["type"][:4], p["score"], p["pub_date"], p["title"][:62]))