1+ # -*- coding: utf-8 -*-
2+ """
3+ ArchAlignmentVerifier — 架构对齐验证器（5.0 Phase 2）
4+
5+ 重构后确定性验证现状与目标架构的对齐度，输出四维评分 + 差距复检。
6+
7+ 对齐度评分（0–100）：
8+ 职责对齐 40% 已归属模块数 / 总模块数（游离模块惩罚）
9+ 依赖健康 30% 1 - 违例依赖边 / 总模块依赖边
10+ 业务覆盖 20% 有实现角色数 / 总角色数
11+ 代码健康 10% 复用 arch_audit 健康度归一化（0–10 → 0–1）
12+
13+ 增量验证：传入本次改动文件集合时，只对受影响子图（含改动文件的调用闭包）
14+ 复检，快速反馈单张任务卡是否达标。
15+ """
16+
17+ import os
18+ from typing import Any , Dict , List , Optional
19+ from core .arch_gap_analyzer import analyze_gap , _is_test_module , _match_module_ids
20+ from core .arch_audit import locate_kg_db , module_of
21+ from core .graph_closure import load_graph , file_base
22+
23+ WEIGHTS = {
24+ "responsibility" : 0.40 ,
25+ "dependency" : 0.30 ,
26+ "business" : 0.20 ,
27+ "health" : 0.10 ,
28+ }
29+
30+
31+ class ArchAlignmentVerifier :
32+ """架构对齐验证器（5.0 Phase 2）"""
33+
34+ def verify (
35+ self ,
36+ project_path : str ,
37+ target_arch : Optional [dict ] = None ,
38+ changed_files : Optional [List [str ]] = None ,
39+ db_path : Optional [str ] = None ,
40+ ) -> dict :
41+ """架构对齐验证主入口。
42+
43+ Args:
44+ project_path: 目标项目路径。
45+ target_arch: 目标架构（缺省读取已存储）。
46+ changed_files: 本次改动文件集合（增量验证时传入；None = 全量）。
47+ db_path: 知识图谱 db（缺省自动定位）。
48+
49+ Returns:
50+ score 各维度 + 总对齐度 + 差距复检。
51+ """
52+ db = db_path or locate_kg_db (project_path )
53+ if not db or not os .path .exists (db ):
54+ return {"ok" : False , "project_path" : project_path ,
55+ "message" : "知识图谱不存在，需先构建" , "score" : 0.0 ,
56+ "dimensions" : {}, "gaps" : []}
57+
58+ if target_arch is None :
59+ target_arch = self ._load_stored_arch (project_path )
60+ if target_arch is None :
61+ return {"ok" : False , "project_path" : project_path ,
62+ "message" : "缺少目标架构，请先设置或传入" , "score" : 0.0 ,
63+ "dimensions" : {}, "gaps" : []}
64+
65+ # 1. 全量差距（供评分）
66+ gap_result = analyze_gap (project_path , target_arch , db_path = db )
67+
68+ nodes , adj = load_graph (db )
69+
70+ # 2. 子图裁剪（增量验证）：只对改动文件影响的差距复检
71+ if changed_files :
72+ focus = self ._focus_set (changed_files , nodes )
73+ gaps = self ._filter_gaps_in_focus (gap_result .get ("gaps" , []), focus )
74+ else :
75+ gaps = gap_result .get ("gaps" , [])
76+
77+ # 3. 四维评分
78+ dimensions = self ._score_dimensions (project_path , target_arch ,
79+ nodes , adj , gap_result , db )
80+ total = round (
81+ dimensions ["responsibility" ]["score" ] * WEIGHTS ["responsibility" ]
82+ + dimensions ["dependency" ]["score" ] * WEIGHTS ["dependency" ]
83+ + dimensions ["business" ]["score" ] * WEIGHTS ["business" ]
84+ + dimensions ["health" ]["score" ] * WEIGHTS ["health" ],
85+ 2 ,
86+ )
87+
88+ return {
89+ "ok" : True ,
90+ "project_path" : project_path ,
91+ "tool" : "coderef_arch_verify" ,
92+ "mode" : "incremental" if changed_files else "full" ,
93+ "changed_files" : changed_files or [],
94+ "score" : total ,
95+ "dimensions" : dimensions ,
96+ "gaps" : gaps ,
97+ "remaining" : {
98+ "total" : len (gaps ),
99+ "high" : sum (1 for g in gaps if g ["severity" ] == "high" ),
100+ "medium" : sum (1 for g in gaps if g ["severity" ] == "medium" ),
101+ "low" : sum (1 for g in gaps if g ["severity" ] == "low" ),
102+ },
103+ }
104+
105+ # ────────────────────────────────────────────────
106+ # 内部方法
107+ # ────────────────────────────────────────────────
108+
109+ def _load_stored_arch (self , project_path : str ) -> Optional [dict ]:
110+ import json
111+ path = os .path .join (project_path , ".coderef" , "target_arch.json" )
112+ if os .path .exists (path ):
113+ with open (path , "r" , encoding = "utf-8" ) as f :
114+ return json .load (f )
115+ return None
116+
117+ def _score_dimensions (self , project_path , target_arch , nodes , adj ,
118+ gap_result , db ) -> dict :
119+ roles = target_arch .get ("tech_roles" ) or []
120+ total_mods = sum (1 for n in nodes .values ()
121+ if n .get ("type" ) == "module"
122+ and not _is_test_module (module_of (n , project_path ) or n .get ("name" , "" )))
123+ assigned_ids = set ()
124+ for role in roles :
125+ assigned_ids |= _match_module_ids (nodes , project_path ,
126+ role .get ("target_modules" , []))
127+ responsibility = len (assigned_ids ) / total_mods if total_mods else 1.0
128+
129+ # 依赖健康：违例边 / 文件级模块依赖边
130+ module_edges = self ._count_module_edges (nodes , adj , project_path )
131+ violation_edges = 0
132+ for g in gap_result .get ("gaps" , []):
133+ if g .get ("type" ) == "dependency_violation" :
134+ violation_edges += 1
135+ elif g .get ("type" ) == "cycle" :
136+ # 环中边近似 = 环长-1（保守）
137+ violation_edges += max (1 , len (g .get ("modules" , [])) - 1 )
138+ dependency = max (0.0 , 1.0 - violation_edges / max (1 , module_edges ))
139+
140+ business = self ._score_business (target_arch , nodes , project_path )
141+
142+ health = self ._score_health (project_path , db )
143+
144+ return {
145+ "responsibility" : {
146+ "score" : round (responsibility , 2 ),
147+ "weight" : WEIGHTS ["responsibility" ],
148+ "assigned" : len (assigned_ids ),
149+ "total_modules" : total_mods ,
150+ },
151+ "dependency" : {
152+ "score" : round (dependency , 2 ),
153+ "weight" : WEIGHTS ["dependency" ],
154+ "violations" : violation_edges ,
155+ "module_edges" : module_edges ,
156+ },
157+ "business" : {
158+ "score" : round (business ["ratio" ], 2 ),
159+ "weight" : WEIGHTS ["business" ],
160+ "implemented_roles" : business ["implemented" ],
161+ "total_roles" : business ["total" ],
162+ },
163+ "health" : {
164+ "score" : round (health , 2 ),
165+ "weight" : WEIGHTS ["health" ],
166+ "arch_health_raw" : None ,
167+ },
168+ }
169+
170+ def _count_module_edges (self , nodes : dict , adj : Dict [str , list ],
171+ project_path : str ) -> int :
172+ """统计文件级模块依赖边（去重 pair，排除测试模块）。"""
173+ pairs = set ()
174+ for nid , targets in adj .items ():
175+ src_mod = module_of (nodes .get (nid , {}), project_path ) or ""
176+ if not src_mod or _is_test_module (src_mod ):
177+ continue
178+ for t in targets :
179+ dst_mod = module_of (nodes .get (t , {}), project_path ) or ""
180+ if not dst_mod or _is_test_module (dst_mod ) or dst_mod == src_mod :
181+ continue
182+ pairs .add ((src_mod , dst_mod ))
183+ return len (pairs )
184+
185+ def _score_business (self , target_arch , nodes , project_path ) -> dict :
186+ roles = target_arch .get ("tech_roles" ) or []
187+ implemented = 0
188+ for role in roles :
189+ specs = role .get ("target_modules" , [])
190+ if any (self ._module_exists (project_path , s , nodes ) for s in specs ):
191+ implemented += 1
192+ total = len (roles )
193+ return {"implemented" : implemented , "total" : total ,
194+ "ratio" : implemented / total if total else 1.0 }
195+
196+ def _module_exists (self , project_path : str , spec : str , nodes : dict ) -> bool :
197+ from core .arch_gap_analyzer import _module_exists
198+ return _module_exists (project_path , spec , nodes )
199+
200+ def _score_health (self , project_path : str , db : str ) -> float :
201+ try :
202+ from core .arch_audit import audit as arch_audit
203+ r = arch_audit (project_path , db_path = db )
204+ health_score = r .get ("health_score" )
205+ if isinstance (health_score , (int , float )):
206+ return min (1.0 , health_score / 10.0 )
207+ except Exception :
208+ pass
209+ return 0.5 # 保守默认（未计入健康维度）
210+
211+ def _focus_set (self , changed_files : List [str ], nodes : dict ) -> set :
212+ """扩展本次改动文件为"受影响文件闭包"（文件级）。"""
213+ bases = {os .path .basename (f ).replace (".py" , "" ) for f in changed_files if f }
214+ focus = set (bases )
215+ for nid , n in nodes .items ():
216+ base = file_base (n )
217+ b = (os .path .splitext (base )[0 ] if base else "" )
218+ if b in bases :
219+ focus .add (base )
220+ return focus
221+
222+ def _filter_gaps_in_focus (self , gaps : List [dict ], focus : set ) -> List [dict ]:
223+ """增量：只保留差距涉及的目标模块在受影响闭包内的差距。"""
224+ if not focus :
225+ return gaps
226+ out = []
227+ for g in gaps :
228+ mod = g .get ("module" ) or ""
229+ base = mod .split ("/" )[- 1 ] if mod else ""
230+ match = (base in focus ) or (
231+ g .get ("type" ) == "cycle"
232+ and any ((m .split ("/" )[- 1 ]) in focus for m in g .get ("modules" , []))
233+ )
234+ if match :
235+ out .append (g )
236+ return out
0 commit comments