tei/CM95
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

CM93 MySQL importer validated baseline

Browse Source
This commit is contained in:
kaotei
2026-09-02 09:50:12 +08:00
commit 9d56592cad
62 changed files with 24778 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

613
cm93_importer_production_source_snapshot_v1.py Normal file
View File

@@ -0,0 +1,613 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CM93 Importer 生产源码快照与接入点提取
Version: production-snapshot-v1
目的
====
完全独立，不连接数据库，不读取 V10/V11/V12，不扫描历史审计脚本。
只针对当前项目真正的生产源码：
- src/cm93db/importer.py
- src/cm93db/geometry.py
- src/cm93db/cm93_binary.py
- src/cm93db/cell_parser.py
- src/cm93db/audit.py
- sql/001_schema_mysql57.sql
- README.md
- AGENTS.md（若存在）
生成：
1. 上述文件的完整只读快照副本；
2. SHA-256；
3. Python 函数/类清单；
4. 与 normalization / geometry issue / geotype / bbox /
counters / DB write / raw source 相关的完整函数摘录；
5. SQL schema 中关键表完整 CREATE TABLE 片段；
6. 一个小 ZIP，供 ChatGPT 精确分析后生成真正修改版。
不做：
- 不修改源码
- 不连接 MySQL
- 不执行 SQL
- 不生成 patch
- 不生成 PBF
"""
from __future__ import annotations
import argparse
import ast
import csv
import hashlib
import json
import re
import shutil
import sys
import zipfile
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
PRODUCTION_FILES = [
"src/cm93db/importer.py",
"src/cm93db/geometry.py",
"src/cm93db/cm93_binary.py",
"src/cm93db/cell_parser.py",
"src/cm93db/audit.py",
"sql/001_schema_mysql57.sql",
"README.md",
"AGENTS.md",
]
KEY_PATTERNS = {
"normalization": [
r"\bis_valid\b",
r"\bmake_valid\b",
r"\bpolygonize\b",
r"\bpolygonize_full\b",
r"\bunary_union\b",
r"\bnormalize\b",
r"\bexplain_validity\b",
r"\bbuffer\s*\(\s*0",
],
"geometry_issue": [
r"geometry_issue",
r"geometry missing",
r"geometry_missing",
r"geometry_repaired",
r"invalid geometry",
],
"geotype": [
r"raw_geotype",
r"\bgeotype\b",
r"geometry_type",
r"\bMETA\b",
r"\bAREA\b",
],
"bbox": [
r"bbox_min_lon",
r"bbox_min_lat",
r"bbox_max_lon",
r"bbox_max_lat",
r"\bbbox\b",
r"\bbounds\b",
],
"counters": [
r"geometry_invalid_count",
r"geometry_repaired_count",
r"summary_json",
r"chart_import_job",
],
"db_write": [
r"chart_area_geometry",
r"chart_coverage",
r"chart_feature",
r"INSERT\s+INTO",
r"UPDATE\s+",
r"executemany\s*\(",
r"execute\s*\(",
],
"raw_source": [
r"raw_feature_blob",
r"chart_raw_record",
r"source_record_index",
r"raw_blob",
r"raw_metadata_json",
],
"geometry_build": [
r"\bPolygon\s*\(",
r"\bMultiPolygon\s*\(",
r"\bLineString\s*\(",
r"\bPoint\s*\(",
r"\bwkb\b",
r"\bshapely\b",
],
"srid": [
r"\bSRID\b",
r"\bsrid\b",
r"4326",
r"normalized_srid",
],
}
SCHEMA_TABLES = [
"chart_feature",
"chart_area_geometry",
"chart_coverage",
"chart_import_job",
"chart_raw_record",
]
def sha256_file(path: Path) -> str:
h = hashlib.sha256()
with path.open("rb") as f:
while True:
chunk = f.read(1024 * 1024)
if not chunk:
break
h.update(chunk)
return h.hexdigest()
def safe_text(path: Path) -> Optional[str]:
for enc in ("utf-8", "utf-8-sig", "cp932", "latin-1"):
try:
return path.read_text(encoding=enc)
except Exception:
pass
return None
def write_csv(path: Path, rows: Iterable[Dict[str, Any]]) -> None:
rows = list(rows)
keys: List[str] = []
seen = set()
for row in rows:
for key in row:
if key not in seen:
seen.add(key)
keys.append(key)
with path.open("w", encoding="utf-8-sig", newline="") as f:
if not keys:
return
w = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
w.writeheader()
for row in rows:
out = {}
for key in keys:
v = row.get(key)
if isinstance(v, (list, tuple, dict)):
v = json.dumps(v, ensure_ascii=False, default=str)
out[key] = "" if v is None else v
w.writerow(out)
def ast_functions(path: Path, text: str, rel: str):
functions: List[Dict[str, Any]] = []
classes: List[Dict[str, Any]] = []
if path.suffix.lower() != ".py":
return functions, classes
try:
tree = ast.parse(text)
except Exception as e:
functions.append({
"file": rel,
"function_name": "__AST_PARSE_ERROR__",
"line": None,
"end_line": None,
"error": f"{type(e).__name__}: {e}",
})
return functions, classes
class_stack: List[str] = []
class Visitor(ast.NodeVisitor):
def visit_ClassDef(self, node: ast.ClassDef):
classes.append({
"file": rel,
"class_name": node.name,
"line": node.lineno,
"end_line": getattr(node, "end_lineno", None),
})
class_stack.append(node.name)
self.generic_visit(node)
class_stack.pop()
def visit_FunctionDef(self, node: ast.FunctionDef):
functions.append({
"file": rel,
"class_name": ".".join(class_stack),
"function_name": node.name,
"line": node.lineno,
"end_line": getattr(node, "end_lineno", None),
"async": 0,
"args": [a.arg for a in node.args.args],
})
self.generic_visit(node)
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef):
functions.append({
"file": rel,
"class_name": ".".join(class_stack),
"function_name": node.name,
"line": node.lineno,
"end_line": getattr(node, "end_lineno", None),
"async": 1,
"args": [a.arg for a in node.args.args],
})
self.generic_visit(node)
Visitor().visit(tree)
return functions, classes
def line_hits(rel: str, text: str) -> List[Dict[str, Any]]:
rows = []
lines = text.splitlines()
compiled = {
group: [re.compile(p, re.I) for p in pats]
for group, pats in KEY_PATTERNS.items()
}
for no, line in enumerate(lines, 1):
for group, pats in compiled.items():
matched = [p.pattern for p in pats if p.search(line)]
if not matched:
continue
start = max(1, no - 3)
end = min(len(lines), no + 3)
context = "\n".join(
f"{i:05d}: {lines[i-1]}"
for i in range(start, end + 1)
)
rows.append({
"file": rel,
"line": no,
"group": group,
"code": line.strip(),
"matched_patterns": matched,
"context": context,
})
return rows
def function_key_groups(
fn: Dict[str, Any],
text: str,
) -> List[str]:
if fn.get("line") is None:
return []
lines = text.splitlines()
start = max(1, int(fn["line"]))
end = fn.get("end_line")
if end is None:
end = min(len(lines), start + 80)
else:
end = min(len(lines), int(end))
body = "\n".join(lines[start - 1:end])
groups = []
for group, pats in KEY_PATTERNS.items():
if any(re.search(p, body, re.I) for p in pats):
groups.append(group)
return groups
def make_function_excerpts(
root: Path,
function_rows: List[Dict[str, Any]],
texts: Dict[str, str],
) -> str:
chunks: List[str] = []
for fn in sorted(
function_rows,
key=lambda x: (
x.get("file") or "",
int(x.get("line") or 0),
)
):
rel = fn.get("file")
text = texts.get(rel)
if not text or fn.get("line") is None:
continue
groups = function_key_groups(fn, text)
if not groups:
continue
lines = text.splitlines()
start = max(1, int(fn["line"]))
end_val = fn.get("end_line")
end = (
min(len(lines), int(end_val))
if end_val is not None
else min(len(lines), start + 100)
)
# 防止单个超大函数让结果失控，但保留足够上下文。
if end - start > 260:
end = start + 260
cls = fn.get("class_name") or ""
name = fn.get("function_name") or ""
display = f"{cls}.{name}" if cls else name
chunks.append("=" * 110)
chunks.append(
f"FILE={rel} FUNCTION={display} "
f"LINES={start}-{fn.get('end_line')} "
f"GROUPS={','.join(groups)}"
)
chunks.append("=" * 110)
for no in range(start, end + 1):
chunks.append(f"{no:05d}: {lines[no-1]}")
chunks.append("")
return "\n".join(chunks)
def extract_create_table(sql_text: str, table: str) -> Optional[str]:
# 兼容 CREATE TABLE `name` (...) ENGINE=...
pat = re.compile(
rf"CREATE\s+TABLE\s+`?{re.escape(table)}`?\s*\(",
re.I,
)
m = pat.search(sql_text)
if not m:
return None
start = m.start()
pos = m.end() - 1
depth = 0
in_single = False
in_double = False
escape = False
for i in range(pos, len(sql_text)):
ch = sql_text[i]
if escape:
escape = False
continue
if ch == "\\":
escape = True
continue
if ch == "'" and not in_double:
in_single = not in_single
continue
if ch == '"' and not in_single:
in_double = not in_double
continue
if in_single or in_double:
continue
if ch == "(":
depth += 1
elif ch == ")":
depth -= 1
if depth == 0:
# 到分号为止
semi = sql_text.find(";", i)
if semi == -1:
semi = i
return sql_text[start:semi + 1]
return None
def parse_args():
p = argparse.ArgumentParser(
description="CM93 importer production source snapshot"
)
p.add_argument("--project-root", default=".")
p.add_argument("--output", default=None)
return p.parse_args()
def main():
args = parse_args()
root = Path(args.project_root).expanduser().resolve()
if not root.is_dir():
print(f"项目目录不存在：{root}")
sys.exit(2)
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
outdir = Path(
args.output
or root / f"cm93_importer_production_snapshot_v1_{stamp}"
).expanduser().resolve()
outdir.mkdir(parents=True, exist_ok=True)
snapshot_dir = outdir / "source_snapshot"
snapshot_dir.mkdir(parents=True, exist_ok=True)
inventory = []
all_functions = []
all_classes = []
all_hits = []
texts: Dict[str, str] = {}
print("[1/5] 收集生产文件 ...")
for rel in PRODUCTION_FILES:
src = root / rel
if not src.exists():
inventory.append({
"file": rel,
"exists": 0,
"size_bytes": None,
"sha256": None,
})
continue
dest = snapshot_dir / rel
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dest)
inventory.append({
"file": rel,
"exists": 1,
"size_bytes": src.stat().st_size,
"sha256": sha256_file(src),
})
text = safe_text(src)
if text is None:
continue
texts[rel] = text
funcs, classes = ast_functions(src, text, rel)
all_functions.extend(funcs)
all_classes.extend(classes)
all_hits.extend(line_hits(rel, text))
print("[2/5] 提取相关完整函数 ...")
excerpts = make_function_excerpts(
root,
all_functions,
texts,
)
(outdir / "生产源码关键函数完整摘录_v1.txt").write_text(
excerpts,
encoding="utf-8",
)
print("[3/5] 提取关键 schema ...")
schema_text = texts.get("sql/001_schema_mysql57.sql", "")
schema_chunks = []
for table in SCHEMA_TABLES:
ddl = extract_create_table(schema_text, table)
schema_chunks.append("=" * 100)
schema_chunks.append(f"TABLE: {table}")
schema_chunks.append("=" * 100)
schema_chunks.append(
ddl if ddl is not None else "NOT FOUND"
)
schema_chunks.append("")
(outdir / "关键表DDL_v1.txt").write_text(
"\n".join(schema_chunks),
encoding="utf-8",
)
print("[4/5] 写清单与报告 ...")
write_csv(outdir / "生产文件SHA256_v1.csv", inventory)
write_csv(outdir / "生产Python函数清单_v1.csv", all_functions)
write_csv(outdir / "生产Python类清单_v1.csv", all_classes)
write_csv(outdir / "生产源码关键命中_v1.csv", all_hits)
group_counts = {}
for row in all_hits:
group_counts[row["group"]] = (
group_counts.get(row["group"], 0) + 1
)
report = {
"version": "production-snapshot-v1",
"generated_at": datetime.now().isoformat(timespec="seconds"),
"project_root": str(root),
"database_dependency": False,
"history_dependency": False,
"read_only": True,
"pbf": "OUT_OF_SCOPE",
"production_files": inventory,
"function_count": len(all_functions),
"class_count": len(all_classes),
"key_hit_count": len(all_hits),
"key_hit_group_counts": group_counts,
"snapshot_contains_only_explicit_production_files": True,
}
(outdir / "生产源码快照报告_v1.json").write_text(
json.dumps(
report,
ensure_ascii=False,
indent=2,
default=str,
),
encoding="utf-8",
)
summary = [
"CM93 Importer 生产源码快照与接入点提取 v1",
"=" * 90,
f"Project: {root}",
"",
"仅包含明确指定的生产文件，不扫描任何 V* 历史审计脚本。",
"",
"FILES",
]
for row in inventory:
summary.append(
f" {'OK' if row['exists'] else 'MISSING':7s} "
f"{row['file']}"
)
summary += [
"",
f"Python functions: {len(all_functions)}",
f"Python classes: {len(all_classes)}",
f"Key hits: {len(all_hits)}",
"",
"NO DATABASE / NO SOURCE MODIFY / NO PBF",
]
(outdir / "生产源码快照摘要_v1.txt").write_text(
"\n".join(summary),
encoding="utf-8",
)
print("[5/5] 压缩 ...")
zip_path = outdir.with_suffix(".zip")
with zipfile.ZipFile(
zip_path,
"w",
compression=zipfile.ZIP_DEFLATED,
) as z:
for p in sorted(outdir.rglob("*")):
if p.is_file():
z.write(
p,
p.relative_to(outdir.parent),
)
print("")
print("PRODUCTION SNAPSHOT COMPLETE: YES")
print(f"Output: {outdir}")
print(f"ZIP: {zip_path}")
print("把这个 ZIP 上传给 ChatGPT。")
if __name__ == "__main__":
main()