614 lines
15 KiB
Python
614 lines
15 KiB
Python
#!/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()
|