#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ OPC 开放数据集 · 导出器 =========================================================== 把仓库内的源数据导出为「机器可读、可引用、可复现」的开放数据集。 输入（仓库内，只读）： data/opc_cities_all.json 103 城八维全量（引擎输出） data/opc_rank.json 103 城带名次榜单（引擎输出） data/opc_cities_31.json 31 城权威原始数据（白皮书原文） cms/public/js/city_metrics.json 逐城逐指标官方溯源（v3.5） scripts/opc_rank.py 八维评分引擎（复现用） 输出（本目录）： opc_rank_103.csv 榜单（含名次） opc_cities_all.csv 103 城全量（不含名次） *.json 上述 JSON 原样副本（机器可读） opc_rank.py 评分引擎副本（复现包） MANIFEST.txt 文件清单 + MD5 + 数据版本 + 生成时间 用法： python3 export_open_data.py # 在仓库 OPC官网/ 根目录或本目录运行均可 python3 export_open_data.py --root /path/to/OPC官网 说明：本脚本不修改任何源数据，只读取并导出。 机构名铁律：输出文件与文档中的运营主体一律使用全称 「北京创业科创科技中心院士专家科技服务站」。 """ import argparse import csv import hashlib import json import os import shutil import sys from datetime import datetime, timezone, timedelta OPERATOR = "北京创业科创科技中心院士专家科技服务站" SITE = "https://opc.bjgov.cn" CN_TZ = timezone(timedelta(hours=8)) # CSV 表头（中文，便于 Excel 直接打开） COLS = [ ("rank", "排名"), ("city", "城市"), ("province", "省份/地区"), ("region", "区域"), ("gdp", "GDP(亿元)"), ("income", "居民人均可支配收入(元/年)"), ("d1", "D1 市场空间"), ("d2", "D2 成本压力"), ("d3", "D3 产业适配"), ("d4", "D4 政策友好"), ("d5", "D5 人才可用"), ("d6", "D6 生活可负担"), ("d7", "D7 数字基建算力"), ("d8", "D8 融资创投"), ("total", "总分(100分制)"), ("grade", "评级"), # 口径基准列：使「混合口径」在机器可读层面显式可见、可筛选、可审计。 # 内地 100 城 = 八维 v4.0 加权；港澳台 3 城总分仍为六维 v3.x 权重口径（未按八维重算）。 ("basis", "口径基准"), ] def basis_of(city, region=None): """判定某城市总分所依据的权重口径。 依据（已逐城复算验证）： - 港澳台三城总分与六维 v3.x 权重（25/22/20/13/10/10）完全吻合， 且其 D1–D6 存于大写键、d7/d8 为 null； - 其余 100 城总分与八维 v4.0 权重（22/20/18/12/8/7/8/5）一致。 """ if region == "港澳台" or city in ("香港", "澳门", "台湾省", "台湾"): return "六维v3.x(港澳台·未按八维重算)" return "八维v4.0" def find_root(explicit=None): """定位 OPC官网/ 仓库根目录。 2026-09-13 修正：原判据只认 `data/opc_cities_all.json` 存在，但 `cms/public/data/` 下有一个同名文件，向上回溯到 `cms/public` 时即被误判为仓库根， 产物会被写到 `cms/public/cms/public/data/`。现改为**双条件**判据 （同时存在 `data/opc_cities_all.json` 与 `cms/` 子目录）， 并把当前工作目录及其上级纳入候选且优先命中。 """ if explicit: return os.path.abspath(explicit) cands = [] cur = os.getcwd() for _ in range(5): cands.append(cur); cur = os.path.dirname(cur) cur = os.path.dirname(os.path.abspath(__file__)) for _ in range(5): cands.append(cur); cur = os.path.dirname(cur) for c in cands: if (os.path.isfile(os.path.join(c, "data", "opc_cities_all.json")) and os.path.isdir(os.path.join(c, "cms"))): return c return None def md5_of(path): h = hashlib.md5() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def write_csv(path, rows, cols, with_rank): use = [c for c in cols if with_rank or c[0] != "rank"] with open(path, "w", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) w.writerow([c[1] for c in use]) for r in rows: out = [] for key, _ in use: v = r.get(key) out.append("" if v is None else v) w.writerow(out) def snap_and_index(out_dir, items, version, asof, now): """把当前数据集固化为不可变快照，并登记到 versions.json。 快照目录：snapshots// 版本索引：versions.json（按数据截至日升序，含每期文件 MD5 与校验锚点） slug 规则：数据版本号中的空格与分隔符归一化 + 数据截至日， 例如 v4.0_103city_8dim + 2026-09-01 -> v4.0-103city-8dim_2026-09-01 """ slug_src = version.replace("·", "-").replace("(", "-").replace(")", "").replace(" ", "") slug_src = slug_src.replace("，", "-").replace("+", "-") base_slug = f"{slug_src}_{asof}" if asof else slug_src # 读版本索引（先读，才能判定是否需要生成新的导出修订号） idx_file = os.path.join(out_dir, "versions.json") idx = {"dataset": "OPC 城市选址八维评分数据集", "operator": OPERATOR, "site": SITE, "versions": []} if os.path.isfile(idx_file): try: with open(idx_file, encoding="utf-8") as f: old = json.load(f) if isinstance(old.get("versions"), list): idx["versions"] = old["versions"] except Exception: pass # ---- 先算当前导出的文件指纹（内容键），再决定 slug ---- cur_md5 = {} for name in sorted(items): src = os.path.join(out_dir, name) if os.path.isfile(src): cur_md5[name] = md5_of(src) content_key = hashlib.md5( "|".join(f"{k}:{v}" for k, v in sorted(cur_md5.items())).encode("utf-8") ).hexdigest() same_version = [v for v in idx["versions"] if v.get("id", "").startswith(base_slug)] if any(v.get("content_key") == content_key for v in same_version): # 内容与既有快照一致 → 复用该快照，不新增条目（幂等） slug = next(v["id"] for v in same_version if v.get("content_key") == content_key) else: # 内容有变化（或为首次导出）→ 若同版本已有快照，追加导出修订号 slug = base_slug if not same_version else f"{base_slug}-r{len(same_version) + 1}" snap_dir = os.path.join(out_dir, "snapshots", slug) os.makedirs(snap_dir, exist_ok=True) files_meta = {} for name in sorted(items): src = os.path.join(out_dir, name) if not os.path.isfile(src): continue dst = os.path.join(snap_dir, name) # 快照一经写下即视为不可变：同版本同内容重复导出时保留最早一份，不覆盖 if not os.path.isfile(dst): shutil.copy2(src, dst) files_meta[name] = md5_of(dst) # 快照自带一份 MANIFEST，便于单目录自证 mf_src = os.path.join(out_dir, "MANIFEST.txt") if os.path.isfile(mf_src): shutil.copy2(mf_src, os.path.join(snap_dir, "MANIFEST.txt")) # 校验锚点（榜单前三名，便于跨期比对） anchors = {} rank_file = os.path.join(out_dir, "opc_rank.json") if os.path.isfile(rank_file): with open(rank_file, encoding="utf-8") as f: rc = json.load(f).get("cities", []) rc = sorted(rc, key=lambda c: (c.get("rank") is None, c.get("rank") or 0)) for c in rc[:3]: if c.get("city"): anchors[c["city"]] = c.get("total") idx_file = os.path.join(out_dir, "versions.json") idx = {"dataset": "OPC 城市选址八维评分数据集", "operator": OPERATOR, "site": SITE, "versions": []} if os.path.isfile(idx_file): try: with open(idx_file, encoding="utf-8") as f: old = json.load(f) if isinstance(old.get("versions"), list): idx["versions"] = old["versions"] except Exception: pass entry = { "id": slug, "version": version, "data_asof": asof, "exported_at": now.strftime("%Y-%m-%d %H:%M:%S %z"), "snapshot_path": f"public/data/snapshots/{slug}/", "files": files_meta, "content_key": content_key, "anchors": anchors, "note": "本快照写入后不再修改。同一数据版本内如导出内容发生变化（字段增补、口径注释等），" "会追加导出修订号（-r2、-r3…）另存新快照，历史快照保留以便回溯。", } idx["versions"] = [v for v in idx["versions"] if v.get("id") != slug] + [entry] idx["versions"].sort(key=lambda v: (v.get("data_asof") or "", v.get("exported_at") or "")) with open(idx_file, "w", encoding="utf-8") as f: json.dump(idx, f, ensure_ascii=False, indent=2) f.write("\n") return idx["versions"] def main(): ap = argparse.ArgumentParser() ap.add_argument("--root", default=None, help="OPC官网/ 仓库根目录") ap.add_argument("--no-snapshot", action="store_true", help="只导出，不写快照与版本索引") args = ap.parse_args() root = find_root(args.root) if not root: print("× 找不到仓库根目录（需包含 data/opc_cities_all.json）", file=sys.stderr) return 1 out_dir = os.path.join(root, "cms", "public", "data") os.makedirs(out_dir, exist_ok=True) now = datetime.now(CN_TZ) # ---------- 读取源数据 ---------- with open(os.path.join(root, "data", "opc_cities_all.json"), encoding="utf-8") as f: all_data = json.load(f) with open(os.path.join(root, "data", "opc_rank.json"), encoding="utf-8") as f: rank_data = json.load(f) cities = all_data["cities"] rank_cities = rank_data["cities"] meta = all_data.get("meta", {}) data_version = meta.get("version", "") data_asof = meta.get("updated", "") # 榜单按 rank 升序排列（缺失时按总分降序兜底） rank_cities = sorted( rank_cities, key=lambda c: (c.get("rank") is None, c.get("rank") or 0, -(c.get("total") or 0)), ) # ---------- 口径基准（派生字段，使混合口径可被机器识别与审计） ---------- for c in cities: c["basis"] = basis_of(c.get("city"), c.get("region")) for c in rank_cities: c["basis"] = basis_of(c.get("city"), c.get("region")) # ---------- 导出 CSV ---------- csv_rank = os.path.join(out_dir, "opc_rank_103.csv") csv_all = os.path.join(out_dir, "opc_cities_all.csv") write_csv(csv_rank, rank_cities, COLS, with_rank=True) write_csv(csv_all, cities, COLS, with_rank=False) # ---------- JSON（机器可读；含派生字段 basis） ---------- # data/opc_rank.json 与 data/opc_cities_all.json 导出时注入 "basis"（口径基准）， # 使「港澳台 3 城为六维口径、其余 100 城为八维口径」这一事实在数据层面显式可见。 copied = [] for name, payload in (("opc_rank.json", rank_data), ("opc_cities_all.json", all_data)): with open(os.path.join(out_dir, name), "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) copied.append(name) json_pairs = [ ("data/opc_cities_31.json", "opc_cities_31.json"), ("cms/public/js/city_metrics.json", "city_metrics.json"), ] for src_rel, dst_name in json_pairs: src = os.path.join(root, src_rel) if not os.path.isfile(src): print("! 跳过缺失源文件:", src_rel, file=sys.stderr) continue shutil.copy2(src, os.path.join(out_dir, dst_name)) copied.append(dst_name) # ---------- 复现包：评分引擎 ---------- engine_src = os.path.join(root, "scripts", "opc_rank.py") if os.path.isfile(engine_src): shutil.copy2(engine_src, os.path.join(out_dir, "opc_rank.py")) copied.append("opc_rank.py") # ---------- MANIFEST ---------- manifest = os.path.join(out_dir, "MANIFEST.txt") items = sorted(set(copied) | {"opc_rank_103.csv", "opc_cities_all.csv"}) lines = [ "OPC 开放数据集 · 文件清单（MANIFEST）", "=" * 60, "运营主体：" + OPERATOR, "站点：" + SITE, "数据版本：" + str(data_version), "数据截至日：" + str(data_asof), "生成时间：" + now.strftime("%Y-%m-%d %H:%M:%S %z"), "生成脚本：export_open_data.py", "=" * 60, "", f"{'文件':<24}{'字节':>10} MD5", ] for name in items: p = os.path.join(out_dir, name) if os.path.isfile(p): lines.append(f"{name:<24}{os.path.getsize(p):>10} {md5_of(p)}") lines += [ "", "=" * 60, "引用本数据集请注明：", f"{OPERATOR}．AI一人公司创业·OPC城市Top100排名（数据版 {data_version}）[EB/OL]．", f"（{data_asof}）[引用日期]．{SITE}/?r=data．", "", "字段口径与来源见 https://opc.bjgov.cn/?r=data-dict", "评分算法见 opc_rank.py（本目录）", "", "【口径基准说明 · 重要】", " CSV 与 JSON 均含派生字段 basis（口径基准）：", " 八维v4.0 —— 内地 100 城，权重 22/20/18/12/8/7/8/5", " 六维v3.x(港澳台·未按八维重算) —— 香港特别行政区/澳门特别行政区/台湾省 3 城", " 港澳台三城 D1–D6 已完成评估（存于大写键 D1..D6），D4 记 0、D7/D8 记 null；", " 其总分沿用六维权重（25/22/20/13/10/10），尚未按八维 v4.0 重算，", " 按八维换算（D7/D8 计 0）应为 香港 48.8 / 澳门 49.6 / 台湾省 53.5。", " 该值为「最保守下界」：把「未评估」一律当作「零分」，八维口径下合计 25% 的权重", " （D4 政策 12% + D7 算力 8% + D8 融资 5%）被强制置零，不代表三城的实际水平。", " 三城的可比水平须待 D4/D7/D8 完成同口径评估后确定；在此之前本站不以该值参与排序或对外比较。", " 三城在榜单中与内地城市同表排序，该名次不代表八维口径下的相对位置，仅供参考。", " 此为现行版本已知口径差异，修订将随数据版本升级执行并在更新日志留痕。", " 详见 https://opc.bjgov.cn/?r=quality", ] with open(manifest, "w", encoding="utf-8") as f: f.write("\n".join(lines) + "\n") # ---------- P2-3 数据集版本号 + 历史快照 ---------- # 每次导出都把当前版本固化为一份不可变快照，并登记到 versions.json， # 使「本期结论」可被回溯核对（权威性来源之一：可复现 + 可回溯）。 snap_list = [] if args.no_snapshot else snap_and_index(out_dir, items, str(data_version), str(data_asof), now) # ---------- 控制台摘要 ---------- print("仓库根目录 :", root) print("输出目录 :", out_dir) print("数据版本 :", data_version, "| 截至", data_asof) print("榜单行数 :", len(rank_cities), "| 全量行数:", len(cities)) print("产出文件 :", ", ".join(items), "+ MANIFEST.txt") print("快照版本数 :", len(snap_list), "（最近一期:", snap_list[-1]["id"] if snap_list else "-", "）") # 复现校验锚点（与站内《方法论》页示例一致） for name, expect in (("成都", 82.2), ("北京", 81.0)): hit = [c for c in rank_cities if c.get("city") == name] if hit: got = hit[0].get("total") flag = "OK" if abs(float(got) - expect) < 0.05 else "!! 与站内示例不符" print(f"校验锚点 : {name} 总分 {got}（期望 {expect}） {flag}") return 0 if __name__ == "__main__": raise SystemExit(main())