#!/usr/bin/env python3 """ Full-Stack Blast Radius Impact Analyzer for FreeExile. Calculates downstream blast radius across: 1. Python Server AST Imports 2. Client WebApp JavaScript ES Modules 3. Dual-Stack Catalogs (Server Python <-> Client JS) 4. Protobuf Network Contracts 5. C++ Native FFI Bridge 6. Game Design Matrix SQLite (17 tables & cross_relationships triples) """ from __future__ import annotations import argparse import ast import json import os import re import sys import time from typing import Dict, List, Optional, Set, Tuple if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8") PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) ACK_STORE_PATH = os.path.join(PROJECT_ROOT, "data", ".blast_radius_ack.json") ACK_TTL_SECONDS = 1800 # 30 minutes def _normalize_target(target: str) -> str: norm = target.replace("\\", "/").strip() if os.path.isabs(norm): norm = os.path.relpath(norm, PROJECT_ROOT).replace("\\", "/") return norm def _get_ack_store() -> Dict[str, float]: if not os.path.exists(ACK_STORE_PATH): return {} try: with open(ACK_STORE_PATH, "r", encoding="utf-8") as f: data = json.load(f) now = time.time() return {k: v for k, v in data.items() if now - v < ACK_TTL_SECONDS} except Exception: return {} def _save_ack(target: str) -> None: store = _get_ack_store() store[target] = time.time() os.makedirs(os.path.dirname(ACK_STORE_PATH), exist_ok=True) try: with open(ACK_STORE_PATH, "w", encoding="utf-8") as f: json.dump(store, f, indent=2) except Exception: pass IGNORED_DIRS = {"node_modules", ".git", ".gemini", "dist", "build", "__pycache__", ".venv"} def _find_py_dependents(target_file: str) -> List[str]: if not target_file.endswith(".py"): return [] stem = os.path.splitext(target_file)[0] mod_full = stem.replace("/", ".").replace("\\", ".") mod_short = os.path.basename(stem) dependents: List[str] = [] for d in ["server", "tools", "tests"]: dir_p = os.path.join(PROJECT_ROOT, d) if not os.path.exists(dir_p): continue for root, dirs, files in os.walk(dir_p): dirs[:] = [x for x in dirs if x not in IGNORED_DIRS] for file in files: if not file.endswith(".py"): continue file_rel = os.path.relpath(os.path.join(root, file), PROJECT_ROOT).replace("\\", "/") if file_rel == target_file: continue try: with open(os.path.join(root, file), "r", encoding="utf-8", errors="ignore") as f: tree = ast.parse(f.read(), filename=file_rel) for node in ast.walk(tree): if isinstance(node, ast.Import): if any(a.name in (mod_full, mod_short) for a in node.names): dependents.append(file_rel) break elif isinstance(node, ast.ImportFrom): if node.module: # Case A: from server.world.server_engine_loop import X if node.module == mod_full or node.module.endswith("." + mod_short) or node.module == mod_short: dependents.append(file_rel) break # Case B: from server.world import server_engine_loop if any(a.name == mod_short for a in node.names): dependents.append(file_rel) break except Exception: continue return sorted(list(set(dependents))) def _find_js_dependents(target_file: str) -> List[str]: if not (target_file.endswith(".js") or target_file.endswith(".ts")): return [] stem = os.path.splitext(os.path.basename(target_file))[0] dependents: List[str] = [] scan_dirs = [os.path.join(PROJECT_ROOT, "client", "webapp", "js"), os.path.join(PROJECT_ROOT, "tests")] for s_dir in scan_dirs: if not os.path.exists(s_dir): continue for root, dirs, files in os.walk(s_dir): dirs[:] = [x for x in dirs if x not in IGNORED_DIRS] for f in files: if not (f.endswith(".js") or f.endswith(".ts")): continue f_rel = os.path.relpath(os.path.join(root, f), PROJECT_ROOT).replace("\\", "/") if f_rel == target_file: continue try: with open(os.path.join(root, f), "r", encoding="utf-8", errors="ignore") as src: content = src.read() if re.search(rf"['\"][^'\"]*{re.escape(stem)}(?:\.js)?['\"]", content): dependents.append(f_rel) except Exception: continue return sorted(list(set(dependents))) def _find_dual_stack_counterparts(target_file: str) -> List[str]: counterparts: List[str] = [] base_name = os.path.splitext(os.path.basename(target_file))[0] # 1. Server Python Catalog -> Client JS Catalog if target_file.startswith("server/") and base_name.endswith("_catalog"): client_candidates = [ f"client/webapp/js/data/{base_name}.js", f"client/webapp/js/ui/{base_name}.js", ] for c in client_candidates: if os.path.exists(os.path.join(PROJECT_ROOT, c)): counterparts.append(c) # 2. Client JS Catalog -> Server Python Catalog elif target_file.startswith("client/") and base_name.endswith("_catalog"): py_candidate = f"server/world/{base_name}.py" if os.path.exists(os.path.join(PROJECT_ROOT, py_candidate)): counterparts.append(py_candidate) return counterparts def _find_proto_dependents(target_file: str) -> List[str]: if not target_file.endswith(".proto"): return [] stem = os.path.splitext(os.path.basename(target_file))[0] deps: List[str] = [] for d in ["server", "client/webapp/js", "tests"]: dir_p = os.path.join(PROJECT_ROOT, d) if not os.path.exists(dir_p): continue for root, dirs, files in os.walk(dir_p): dirs[:] = [x for x in dirs if x not in IGNORED_DIRS] for f in files: if not (f.endswith(".py") or f.endswith(".js") or f.endswith(".ts")): continue f_rel = os.path.relpath(os.path.join(root, f), PROJECT_ROOT).replace("\\", "/") try: with open(os.path.join(root, f), "r", encoding="utf-8", errors="ignore") as src: if stem in src.read(): deps.append(f_rel) except Exception: continue return sorted(list(set(deps))) def _find_cpp_dependents(target_file: str) -> List[str]: if not target_file.startswith("server_cpp/"): return [] bridge = "server/world/native_engine_bridge.py" return [bridge] if os.path.exists(os.path.join(PROJECT_ROOT, bridge)) else [] def _check_game_matrix_impacts(target: str) -> Tuple[List[str], List[str]]: domain_impacts: List[str] = [] affected_files: List[str] = [] db_path = os.path.join(PROJECT_ROOT, "data", "game_design_matrix.db") if not os.path.exists(db_path): return domain_impacts, affected_files import sqlite3 try: conn = sqlite3.connect(db_path) cur = conn.cursor() # 1. Check cross_relationships triples cur.execute( "SELECT relation_id, source_id, relation_type, target_id FROM cross_relationships " "WHERE source_id = ? OR target_id = ?", (target, target), ) for rel in cur.fetchall(): domain_impacts.append(f"[REL] {rel[1]} -({rel[2]})-> {rel[3]}") affected_files.append("data/game_design_matrix.db") # 2. Check entities if target.startswith("quest_"): cur.execute("SELECT title, act_id, zone_id FROM quests WHERE quest_id = ?", (target,)) row = cur.fetchone() if row: domain_impacts.append(f"Quest: '{row[0]}' tại Act {row[1]}, Zone '{row[2]}'.") affected_files.append("wiki/vi/QUESTS_AND_MILESTONES_SPECS.md") elif target.startswith("zone_"): cur.execute("SELECT name, act_id FROM zones WHERE zone_id = ?", (target,)) row = cur.fetchone() if row: cur.execute("SELECT count(*) FROM quests WHERE zone_id = ?", (target,)) q_cnt = cur.fetchone()[0] domain_impacts.append(f"Zone: '{row[0]}' (Act {row[1]}) chứa {q_cnt} nhiệm vụ.") affected_files.append("wiki/vi/POE2_GAMEPLAY_AND_ZONE_ARCHITECTURE.md") elif "matrix" in target.lower(): domain_impacts.append("Core Game Design Matrix: Điều phối Story Acts, Zones, Quests & Affixes.") affected_files.extend(["wiki/vi/QUESTS_AND_MILESTONES_SPECS.md", "data/game_design_matrix.db"]) conn.close() except Exception as e: domain_impacts.append(f"Lưu ý DB: {e}") return domain_impacts, affected_files def analyze_blast_radius(target: str, ack: bool = False) -> Dict[str, object]: norm_target = _normalize_target(target) # 0. Delegate Markdown docs & wiki to Semantic Doc Blast Radius Engine if norm_target.endswith(".md") or norm_target.startswith("docs/") or norm_target.startswith("wiki/"): try: from tools.analysis.doc_blast_radius import analyze_doc_blast_radius doc_res = analyze_doc_blast_radius(norm_target, ack=ack) return { "target": doc_res["target"], "risk_level": doc_res["risk_level"], "blocked": doc_res["blocked"], "acknowledged": doc_res["acknowledged"], "affected_files": doc_res["all_affected_docs"] + doc_res["affected_code_catalogs"], "domain_impacts": doc_res["database_impacts"], "reason": doc_res["reason"], "doc_details": doc_res, } except Exception: pass is_leaf = norm_target.startswith("tests/") or "scratch" in norm_target # Aggregate dependencies across all stacks py_deps = _find_py_dependents(norm_target) js_deps = _find_js_dependents(norm_target) dual_deps = _find_dual_stack_counterparts(norm_target) proto_deps = _find_proto_dependents(norm_target) cpp_deps = _find_cpp_dependents(norm_target) domain_impacts, matrix_files = _check_game_matrix_impacts(norm_target) all_affected = sorted(list(set( py_deps + js_deps + dual_deps + proto_deps + cpp_deps + matrix_files ))) # Determine risk level is_core_contract = ( norm_target.endswith(".proto") or norm_target.startswith("server_cpp/") or "schema" in norm_target or "types" in norm_target or norm_target.endswith("_catalog.py") or norm_target.endswith("_catalog.js") ) if is_leaf: risk_level = "LOW" elif is_core_contract or len(all_affected) >= 5: risk_level = "CRITICAL" elif len(all_affected) >= 2 or len(domain_impacts) > 0: risk_level = "HIGH" elif len(all_affected) == 1: risk_level = "MEDIUM" else: risk_level = "LOW" if ack: _save_ack(norm_target) acknowledged = True else: store = _get_ack_store() acknowledged = norm_target in store blocked = False if risk_level in ("HIGH", "CRITICAL") and not acknowledged: blocked = True reason = ( f"[BLAST RADIUS BLOCKED] File/Entity '{norm_target}' có mức độ rủi ro {risk_level} " f"ảnh hưởng tới {len(all_affected)} file/hệ thống khác. " f"Bắt buộc Agent chạy 'python tools/analysis/blast_radius.py --target {norm_target} --ack' " f"để ghi nhận và mở khóa chỉnh sửa!" ) else: reason = f"Đã thông qua kiểm tra Blast Radius (Rủi ro: {risk_level}, Acknowledged: {acknowledged})." return { "target": norm_target, "risk_level": risk_level, "blocked": blocked, "acknowledged": acknowledged, "affected_files": all_affected, "domain_impacts": domain_impacts, "reason": reason, } def main() -> int: parser = argparse.ArgumentParser(description="Full-Stack Blast Radius Impact Analyzer") parser.add_argument("--target", required=True, help="Target file path or entity ID to analyze") parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format") parser.add_argument("--ack", action="store_true", help="Acknowledge understanding and unlock target for 30 minutes") args = parser.parse_args() result = analyze_blast_radius(args.target, ack=args.ack) if args.format == "json": print(json.dumps(result, indent=2, ensure_ascii=True)) else: print(f"=== BÁO CÁO PHÂN TÍCH VÙNG ẢNH HƯỞNG TOÀN DIỆN (FULL-STACK BLAST RADIUS) ===") print(f"Mục tiêu : {result['target']}") print(f"Mức độ rủi ro : {result['risk_level']}") print(f"Trạng thái khóa : {'[BLOCKED] ĐANG BỊ KHÓA' if result['blocked'] else '[ALLOWED] ĐÃ MỞ KHÓA'}") print(f"Đã duyệt (Ack) : {'Có' if result['acknowledged'] else 'Chưa'}") print(f"\nDanh sách file liên đới ({len(result['affected_files'])}):") for f in result["affected_files"]: print(f" - {f}") if result["domain_impacts"]: print(f"\nẢnh hưởng nghiệp vụ Game Design ({len(result['domain_impacts'])}):") for imp in result["domain_impacts"]: print(f" * {imp}") print(f"\nChi tiết: {result['reason']}") return 0 if __name__ == "__main__": sys.exit(main())