audit: phase 0 tooling for file-by-file 40250 porting
- refroot.py: one reference-root resolver (MT_40250_SOURCE, then manifest) used by every audit script; missing tree or fingerprint input is an error - port_map.py: per-function inventory of active 40250 units plus the 40250 Python root, status/queue/init/show/check, 40250: tag scan - manifest: fix 18 wrong reference paths (Python UI now points at the 40250 Client/Eternexus/root, not the m2dev assets/root), drop 2 deleted implementation files, move 64 prose test entries to evidence.commands - first port-map entry: PythonPlayerEventHandler.cpp - roadmap replaced by a batch/unit queue Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
shenlei
@@ -0,0 +1,295 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Per-function 40250 port-map: inventory, skeleton creation, coverage and queue.
|
||||
|
||||
Units are the active 40250 C++ sources (from the .vcxproj files) plus the 40250 Python
|
||||
root (`../../Client/Eternexus/root/*.py` relative to the reference root). Port-map entries
|
||||
live in `audit/port-map/<unit path>.json`; functions without an entry count as TODO.
|
||||
|
||||
port_map.py status [--lib UserInterface] coverage per library (and per unit with --units)
|
||||
port_map.py queue [--limit 20] next units by priority, then by TODO count
|
||||
port_map.py init <unit> [<unit>...] create or extend entries (new functions as TODO)
|
||||
port_map.py show <unit> per-function status of one unit
|
||||
port_map.py check stale hashes, unknown functions, tag mismatches
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import fnmatch
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
|
||||
import refroot # script directory is on sys.path when run directly
|
||||
from audit_source_coverage import active_project_sources
|
||||
|
||||
STATUSES = ("TODO", "PORTED", "ADAPTED", "N_A", "DIVERGENT", "NEEDS_LIVE")
|
||||
DONE = {"PORTED", "ADAPTED", "N_A"}
|
||||
PY_ROOT = "../../Client/Eternexus/root"
|
||||
|
||||
# Order follows SKILL.md "Priority"; the first matching pattern wins.
|
||||
PRIORITY = [
|
||||
("P0", ["UserInterface/InstanceBase*.cpp", "GameLib/ActorInstance.cpp", "GameLib/ActorInstanceMotion*.cpp",
|
||||
"GameLib/ActorInstanceRotation.cpp", "GameLib/ActorInstancePosition.cpp", "GameLib/ActorInstanceEvent.cpp",
|
||||
"GameLib/ActorInstanceCollisionDetection.cpp", "GameLib/ActorInstanceSync.cpp",
|
||||
"UserInterface/PythonPlayer.cpp", "UserInterface/PythonPlayerInput*.cpp",
|
||||
"UserInterface/PythonPlayerEventHandler.cpp", "UserInterface/PythonCharacterManager.cpp"]),
|
||||
("P1", ["GameLib/ActorInstanceBattle.cpp", "GameLib/ActorInstanceAttach*.cpp", "GameLib/ActorInstanceFly.cpp",
|
||||
"GameLib/FlyingObject*.cpp", "GameLib/FlyTarget.cpp", "GameLib/FlyingInstance.cpp", "GameLib/FlyTrace.cpp",
|
||||
"UserInterface/PythonPlayerSkill.cpp", "UserInterface/PythonSkill.cpp", "GameLib/RaceData*.cpp",
|
||||
"GameLib/RaceMotionData*.cpp", "GameLib/ActorInstanceData.cpp", "GameLib/WeaponTrace.cpp"]),
|
||||
("P2", ["UserInterface/PythonNetworkStream*.cpp", "UserInterface/AccountConnector.cpp",
|
||||
"UserInterface/ServerStateChecker.cpp", "EterLib/NetStream.cpp", "EterLib/NetPacketHeaderMap.cpp"]),
|
||||
("P3", ["UserInterface/PythonItem*.cpp", "UserInterface/PythonExchange*.cpp", "UserInterface/PythonSafeBox.cpp",
|
||||
"UserInterface/PythonShop.cpp", "UserInterface/PythonQuest.cpp", "UserInterface/PythonGuild.cpp",
|
||||
"UserInterface/PythonMessenger.cpp", "UserInterface/PythonChat*.cpp", "UserInterface/PythonTextTail.cpp",
|
||||
"UserInterface/PythonMiniMap.cpp", "GameLib/ItemData.cpp", "GameLib/ItemManager.cpp", f"{PY_ROOT}/*.py"]),
|
||||
]
|
||||
DEFAULT_PRIORITY = "P4"
|
||||
|
||||
# Column-0 definition: `<ret> [Class::]name(` where the body `{` follows on this or the next line.
|
||||
_DEF = re.compile(r"^(?!return\b|else\b|if\b|while\b|for\b|switch\b|case\b|#|//|/\*|\s)"
|
||||
r"[^;(){}=]*?\b((?:[A-Za-z_]\w*::)*~?[A-Za-z_]\w*)\s*\(")
|
||||
_PY_CLASS = re.compile(r"^class\s+(\w+)")
|
||||
_PY_DEF = re.compile(r"^(\s*)def\s+(\w+)\s*\(")
|
||||
_TAG = re.compile(r"(?://|#)\s*40250:\s*([A-Za-z_][\w:.~]*)")
|
||||
|
||||
|
||||
def sha256(path: Path) -> str:
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
def cpp_functions(path: Path) -> list[str]:
|
||||
lines = path.read_bytes().decode("latin-1").splitlines()
|
||||
names: list[str] = []
|
||||
for i, line in enumerate(lines):
|
||||
m = _DEF.match(line)
|
||||
if not m or line.rstrip().endswith(";"):
|
||||
continue
|
||||
name = m.group(1)
|
||||
if name.isupper() or name.split("::")[-1].isupper(): # macros
|
||||
continue
|
||||
rest = "\n".join(lines[i:i + 8])
|
||||
body = rest.find("{")
|
||||
semi = rest.find(";")
|
||||
if body < 0 or (0 <= semi < body):
|
||||
continue
|
||||
if name not in names:
|
||||
names.append(name)
|
||||
return names
|
||||
|
||||
|
||||
def py_functions(path: Path) -> list[str]:
|
||||
names: list[str] = []
|
||||
cls = ""
|
||||
for line in path.read_bytes().decode("latin-1").splitlines():
|
||||
if m := _PY_CLASS.match(line):
|
||||
cls = m.group(1)
|
||||
continue
|
||||
if m := _PY_DEF.match(line):
|
||||
indent, fn = m.groups()
|
||||
if not indent:
|
||||
cls = ""
|
||||
name = f"{cls}.{fn}" if cls and indent else fn
|
||||
if name not in names:
|
||||
names.append(name)
|
||||
return names
|
||||
|
||||
|
||||
def inventory(ref: Path) -> dict[str, Path]:
|
||||
units = {p: ref / p for p in sorted(active_project_sources(ref)) if p.endswith(".cpp")}
|
||||
py_dir = (ref / PY_ROOT).resolve()
|
||||
if py_dir.is_dir():
|
||||
for p in sorted(py_dir.glob("*.py")):
|
||||
units[f"{PY_ROOT}/{p.name}"] = p
|
||||
return units
|
||||
|
||||
|
||||
def functions(path: Path) -> list[str]:
|
||||
return py_functions(path) if path.suffix == ".py" else cpp_functions(path)
|
||||
|
||||
|
||||
def priority(unit: str) -> str:
|
||||
for level, patterns in PRIORITY:
|
||||
if any(fnmatch.fnmatch(unit, pat) for pat in patterns):
|
||||
return level
|
||||
return DEFAULT_PRIORITY
|
||||
|
||||
|
||||
def entry_path(repo: Path, unit: str) -> Path:
|
||||
rel = unit.replace("../../Client/Eternexus/", "Client/")
|
||||
return repo / "audit" / "port-map" / (rel + ".json")
|
||||
|
||||
|
||||
def load_entry(repo: Path, unit: str) -> dict | None:
|
||||
path = entry_path(repo, unit)
|
||||
return json.loads(path.read_text(encoding="utf-8")) if path.is_file() else None
|
||||
|
||||
|
||||
def lib_of(unit: str) -> str:
|
||||
return "Client/root" if unit.startswith(PY_ROOT) else unit.split("/")[0]
|
||||
|
||||
|
||||
def unit_counts(repo: Path, unit: str, path: Path) -> Counter:
|
||||
entry = load_entry(repo, unit) or {"functions": {}}
|
||||
recorded = entry["functions"]
|
||||
counts = Counter()
|
||||
for fn in functions(path):
|
||||
counts[recorded.get(fn, {}).get("status", "TODO")] += 1
|
||||
return counts
|
||||
|
||||
|
||||
def resolve_unit(units: dict[str, Path], name: str) -> str:
|
||||
if name in units:
|
||||
return name
|
||||
hits = [u for u in units if u.endswith("/" + name) or Path(u).name == name]
|
||||
if len(hits) != 1:
|
||||
raise SystemExit(f"unit {name!r}: {'ambiguous: ' + ', '.join(hits) if hits else 'not an active 40250 unit'}")
|
||||
return hits[0]
|
||||
|
||||
|
||||
def cmd_status(repo: Path, units: dict[str, Path], args) -> int:
|
||||
per_lib: dict[str, Counter] = {}
|
||||
rows = []
|
||||
for unit, path in units.items():
|
||||
lib = lib_of(unit)
|
||||
if args.lib and lib != args.lib:
|
||||
continue
|
||||
counts = unit_counts(repo, unit, path)
|
||||
per_lib.setdefault(lib, Counter()).update(counts)
|
||||
rows.append((unit, counts))
|
||||
total = Counter()
|
||||
print(f"{'library':<16}{'funcs':>7}{'done':>7} " + " ".join(f"{s:>10}" for s in STATUSES))
|
||||
for lib, c in sorted(per_lib.items()):
|
||||
total.update(c)
|
||||
n = sum(c.values())
|
||||
done = sum(c[s] for s in DONE)
|
||||
print(f"{lib:<16}{n:>7}{done:>7} " + " ".join(f"{c[s]:>10}" for s in STATUSES))
|
||||
n = sum(total.values())
|
||||
done = sum(total[s] for s in DONE)
|
||||
print(f"{'TOTAL':<16}{n:>7}{done:>7} " + " ".join(f"{total[s]:>10}" for s in STATUSES)
|
||||
+ f" ({100.0 * done / max(n, 1):.1f}% done)")
|
||||
if args.units:
|
||||
for unit, c in rows:
|
||||
if sum(c.values()) != c["TODO"]:
|
||||
print(f" {unit}: " + ", ".join(f"{s}={c[s]}" for s in STATUSES if c[s]))
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_queue(repo: Path, units: dict[str, Path], args) -> int:
|
||||
rows = []
|
||||
for unit, path in units.items():
|
||||
c = unit_counts(repo, unit, path)
|
||||
open_ = c["TODO"] + c["DIVERGENT"]
|
||||
if open_:
|
||||
rows.append((priority(unit), -open_, unit, c))
|
||||
for level, neg, unit, c in sorted(rows)[: args.limit]:
|
||||
print(f"{level}\t{-neg:>4} open\t{unit}\t(TODO={c['TODO']} DIVERGENT={c['DIVERGENT']})")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_init(repo: Path, units: dict[str, Path], args) -> int:
|
||||
for name in args.units:
|
||||
unit = resolve_unit(units, name)
|
||||
path = units[unit]
|
||||
entry = load_entry(repo, unit) or {
|
||||
"reference": unit, "reference_sha256": sha256(path), "priority": priority(unit),
|
||||
"contracts": [], "functions": {},
|
||||
}
|
||||
added = 0
|
||||
for fn in functions(path):
|
||||
if fn not in entry["functions"]:
|
||||
entry["functions"][fn] = {"status": "TODO"}
|
||||
added += 1
|
||||
out = entry_path(repo, unit)
|
||||
out.parent.mkdir(parents=True, exist_ok=True)
|
||||
out.write_text(json.dumps(entry, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
print(f"{out.relative_to(repo)}: {added} function(s) added, {len(entry['functions'])} total")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_show(repo: Path, units: dict[str, Path], args) -> int:
|
||||
unit = resolve_unit(units, args.unit)
|
||||
recorded = (load_entry(repo, unit) or {"functions": {}})["functions"]
|
||||
for fn in functions(units[unit]):
|
||||
info = recorded.get(fn, {"status": "TODO"})
|
||||
impl = ", ".join(info.get("impl", []))
|
||||
print(f"{info['status']:<11}{fn}" + (f" -> {impl}" if impl else "") + (f" # {info['note']}" if info.get("note") else ""))
|
||||
return 0
|
||||
|
||||
|
||||
def scan_tags(repo: Path) -> dict[str, list[str]]:
|
||||
tags: dict[str, list[str]] = {}
|
||||
for base, pattern in (("project", "*.gd"), ("extension/src", "*.[ch]*")):
|
||||
for path in (repo / base).rglob(pattern):
|
||||
if "/addons/" in path.as_posix():
|
||||
continue
|
||||
for i, line in enumerate(path.read_text(encoding="utf-8", errors="replace").splitlines(), 1):
|
||||
for m in _TAG.finditer(line):
|
||||
tags.setdefault(m.group(1), []).append(f"{path.relative_to(repo)}:{i}")
|
||||
return tags
|
||||
|
||||
|
||||
def cmd_check(repo: Path, units: dict[str, Path], args) -> int:
|
||||
problems = 0
|
||||
recorded_ported: set[str] = set()
|
||||
for entry_file in sorted((repo / "audit" / "port-map").rglob("*.json")):
|
||||
entry = json.loads(entry_file.read_text(encoding="utf-8"))
|
||||
unit = entry.get("reference", "")
|
||||
label = entry_file.relative_to(repo)
|
||||
if unit not in units:
|
||||
print(f"ERROR {label}: reference {unit!r} is not an active 40250 unit")
|
||||
problems += 1
|
||||
continue
|
||||
if entry.get("reference_sha256") != sha256(units[unit]):
|
||||
print(f"STALE {label}: reference changed; recheck non-TODO functions, then update reference_sha256")
|
||||
problems += 1
|
||||
known = set(functions(units[unit]))
|
||||
for fn, info in entry.get("functions", {}).items():
|
||||
status = info.get("status")
|
||||
if fn not in known:
|
||||
print(f"ERROR {label}: {fn} is not defined in {unit}")
|
||||
problems += 1
|
||||
if status not in STATUSES:
|
||||
print(f"ERROR {label}: {fn} has invalid status {status!r}")
|
||||
problems += 1
|
||||
if status in {"PORTED", "ADAPTED", "NEEDS_LIVE"} and not info.get("impl"):
|
||||
print(f"ERROR {label}: {fn} is {status} without impl")
|
||||
problems += 1
|
||||
if status in {"ADAPTED", "N_A", "DIVERGENT", "NEEDS_LIVE"} and not info.get("note"):
|
||||
print(f"ERROR {label}: {fn} is {status} without note")
|
||||
problems += 1
|
||||
if status == "ADAPTED" and not info.get("test"):
|
||||
print(f"ERROR {label}: {fn} is ADAPTED without test")
|
||||
problems += 1
|
||||
if status in {"PORTED", "ADAPTED"}:
|
||||
recorded_ported.add(fn.split("::")[-1] if "::" not in fn else fn)
|
||||
recorded_ported.add(fn)
|
||||
tags = scan_tags(repo)
|
||||
untracked = sorted(t for t in tags if t not in recorded_ported)
|
||||
for tag in untracked:
|
||||
print(f"WARN tag `40250: {tag}` at {tags[tag][0]} has no PORTED/ADAPTED port-map entry")
|
||||
print(f"check: {problems} error(s), {len(untracked)} untracked tag(s), {len(tags)} tag(s) scanned")
|
||||
return 1 if problems else 0
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
sub = parser.add_subparsers(dest="command", required=True)
|
||||
p = sub.add_parser("status"); p.add_argument("--lib"); p.add_argument("--units", action="store_true")
|
||||
p = sub.add_parser("queue"); p.add_argument("--limit", type=int, default=20)
|
||||
p = sub.add_parser("init"); p.add_argument("units", nargs="+")
|
||||
p = sub.add_parser("show"); p.add_argument("unit")
|
||||
sub.add_parser("check")
|
||||
args = parser.parse_args()
|
||||
repo = refroot.repo_root()
|
||||
units = inventory(refroot.reference_root(repo))
|
||||
return {"status": cmd_status, "queue": cmd_queue, "init": cmd_init, "show": cmd_show,
|
||||
"check": cmd_check}[args.command](repo, units, args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user