"""ImpactAgent：变更点定位（Impact Agent MVP，确定性规则，无 LLM）。 从要件定義各表（機能/画面/DB/IF/バッチ）取 変更区分 + 既存対応 列， 与 ExistingSystemInfo 连接比对，输出 ImpactReport（影响调查书）。 """ from __future__ import annotations import logging from datetime import date from pathlib import Path from genesis.data_models import ( ChangeAnalysis, ChangeElement, ChangeType, DesignReference, ExcelTable, ImpactReport, ImpactWarning, SheetType, StructuredSource, ) _LOGGER = logging.getLogger(__name__) _SHEET_ELEMENT_TYPE = { SheetType.FUNCTION: "機能", SheetType.SCREEN: "画面", SheetType.REPORT: "帳票", SheetType.DATABASE: "DB", SheetType.INTERFACE: "IF", SheetType.BATCH: "バッチ", } _CHANGE_TYPE_MAP = { "新規": ChangeType.NEW, "変更": ChangeType.MODIFIED, "削除": ChangeType.DELETED, "不变": ChangeType.UNCHANGED, } def _element_to_dict(el: ChangeElement) -> dict: """ChangeElement → JSON 可序列化 dict（ChangeType 枚举转 value）。""" return { "element_id": el.element_id, "element_type": el.element_type, "name": el.name, "change_type": el.change_type.value, "existing_mapping": list(el.existing_mapping), "impacted_existing": list(el.impacted_existing), "evidence": el.evidence, "status": el.status, } def impact_report_to_dict(report: ImpactReport) -> dict: """影响调查书 → JSON 可序列化 dict（供下载/日志，api-design §2.4 impact-result）。""" ca = report.change_analysis return { "metadata": dict(report.metadata), "change_analysis": { "project_type": ca.project_type, "new_elements": [_element_to_dict(e) for e in ca.new_elements], "modified_elements": [_element_to_dict(e) for e in ca.modified_elements], "deleted_elements": [_element_to_dict(e) for e in ca.deleted_elements], "unchanged_elements": [_element_to_dict(e) for e in ca.unchanged_elements], "warnings": [{"element_id": w.element_id, "issue": w.issue} for w in ca.warnings], }, "summary": dict(report.summary), "design_references": [ {"doc_name": r.doc_name, "identifier": r.identifier, "snippet": r.snippet} for r in report.design_references ], } def _header_index(headers: list[str], *keywords: str) -> int | None: """按关键词定位列索引（如 変更区分 / 既存対応）。""" for i, h in enumerate(headers): hl = str(h).strip() if any(k in hl for k in keywords): return i return None class ImpactAgent: """变更点定位 → 影响调查书（MVP）。""" def run( self, structured_source: StructuredSource, session_id: str = "impact", scope: dict | None = None, ) -> ImpactReport: existing = structured_source.existing_system if existing is None: raise ValueError("未提供既有系统（existing_system），无法执行影响调查") lookup = self._build_lookup(existing) new_elements: list[ChangeElement] = [] modified_elements: list[ChangeElement] = [] deleted_elements: list[ChangeElement] = [] warnings: list[ImpactWarning] = [] matched: set[str] = set() for table in structured_source.tables: self._classify_table( table, lookup, new_elements, modified_elements, deleted_elements, warnings, matched ) if scope: # scope 参数预留：MVP 默认全量调查；模块/深度收窄由调用方确认后传入，当前忽略 _LOGGER.warning("scope 参数预留（MVP 默认全量调查），当前忽略: %s", scope) unchanged = self._unchanged_count(existing, matched) change_analysis = ChangeAnalysis( project_type="enhancement", new_elements=new_elements, modified_elements=modified_elements, deleted_elements=deleted_elements, unchanged_elements=[], warnings=warnings, ) summary = { "total": len(new_elements) + len(modified_elements) + len(deleted_elements), "new": len(new_elements), "modified": len(modified_elements), "deleted": len(deleted_elements), "unchanged": unchanged, "warnings": len(warnings), } design_references = self._cross_ref_design_docs(structured_source, lookup) return ImpactReport( metadata={ "version": "v1", "session_id": session_id, "created_at": date.today().isoformat(), "llm_model": "none", # MVP 确定性规则，无 LLM 参与 "source": existing.source_path, }, change_analysis=change_analysis, summary=summary, design_references=design_references, ) def _cross_ref_design_docs( self, structured_source: StructuredSource, lookup: dict[str, list[dict]] ) -> list[DesignReference]: """既有设计文档确定性交叉引用（Type A 辅助证据）。 以既有系统（existing_system）解析出的标识符（类/方法/模块名）为锚， 在 design_docs 的 markdown 文本中检索命中，输出 DesignReference 列表。 无 LLM 参与，纯字符串匹配。 """ design_docs = getattr(structured_source, "design_docs", []) or [] if not design_docs: return [] identifiers = [k for k in lookup.keys() if len(k) >= 3] refs: list[DesignReference] = [] for doc in design_docs: text = getattr(doc, "markdown_content", "") or "" if not text: continue doc_name = Path(getattr(doc, "source_path", "design.docx")).name text_lower = text.lower() for ident in identifiers: il = ident.lower() idx = text_lower.find(il) if idx >= 0: token = text[idx:idx + len(il)] # 设计文档中的原始大小写 start = max(0, idx - 30) end = min(len(text), idx + len(il) + 30) snippet = text[start:end].replace("\n", " ").strip() refs.append(DesignReference(doc_name=doc_name, identifier=token, snippet=snippet)) return refs # ---------- 内部 ---------- def _classify_table( self, table: ExcelTable, lookup: dict[str, list[dict]], new_elements: list[ChangeElement], modified_elements: list[ChangeElement], deleted_elements: list[ChangeElement], warnings: list[ImpactWarning], matched: set[str], ) -> None: headers = [str(h) for h in table.headers] change_idx = _header_index(headers, "変更区分", "区分") if change_idx is None: return # 无变更区分列的表（如新規開発的帳票一覧）不参与变更点定位 mapping_idx = _header_index(headers, "既存対応") element_type = _SHEET_ELEMENT_TYPE.get(table.detected_type, table.name) name_idx = _header_index(headers, "名") or 1 for row in table.rows: change_val = self._cell(row, headers, change_idx) if not change_val: continue change_type = _CHANGE_TYPE_MAP.get(str(change_val).strip()) if change_type is None: continue element_id = str(self._cell(row, headers, 0) or "") name = str(self._cell(row, headers, name_idx) or "") mapping = ( str(self._cell(row, headers, mapping_idx) or "") if mapping_idx is not None else "" ) tokens = [ t.strip() for t in mapping.replace("，", ",").replace(" ", "").split(",") if t.strip() ] hits = self._match_tokens(tokens, lookup) impacted = [h["class_name"] for h in hits] evidence = hits[0]["source_uri"] if hits else "" for h in hits: matched.add(h["class_name"]) status = "ok" issue: str | None = None if change_type is ChangeType.NEW and tokens: status = "conflict" issue = f"新規要素却声明了既存対応: {mapping}" elif change_type in (ChangeType.MODIFIED, ChangeType.DELETED): if not tokens: status = "warning" issue = "缺少既存対応，无法定位修改/删除对象" elif not hits: status = "warning" issue = f"既存対応无法匹配既有类: {mapping}" element = ChangeElement( element_id=element_id, element_type=element_type, name=name, change_type=change_type, existing_mapping=tokens, impacted_existing=impacted, evidence=evidence, status=status, ) if change_type is ChangeType.NEW: new_elements.append(element) elif change_type is ChangeType.MODIFIED: modified_elements.append(element) elif change_type is ChangeType.DELETED: deleted_elements.append(element) if issue: warnings.append(ImpactWarning(element_id=element_id, issue=issue)) @staticmethod def _build_lookup(existing) -> dict[str, list[dict]]: """类名/表名（小写）→ 既有要素索引，供 token 匹配。""" lookup: dict[str, list[dict]] = {} def add(key: str, item: dict) -> None: lookup.setdefault(key.lower(), []).append(item) for c in existing.controller_layer: add(c.class_name, {"kind": "controller", "class_name": c.class_name, "source_uri": c.source_uri}) for s in existing.service_layer: add(s.class_name, {"kind": "service", "class_name": s.class_name, "source_uri": s.source_uri}) for e in existing.entity_layer: add(e.class_name, {"kind": "entity", "class_name": e.class_name, "source_uri": e.source_uri}) if e.table_name: add(e.table_name, {"kind": "entity", "class_name": e.class_name, "source_uri": e.source_uri}) return lookup def _match_tokens(self, tokens: list[str], lookup: dict[str, list[dict]]) -> list[dict]: """token → 既有类命中列表（去重）。匹配规则：类名/表名完全相等 或 类名前缀匹配。""" hits: list[dict] = [] seen: set[tuple[str, str]] = set() for token in tokens: key = token.lower() for candidate in lookup.get(key, []): if (candidate["kind"], candidate["class_name"]) not in seen: seen.add((candidate["kind"], candidate["class_name"])) hits.append(candidate) # 前缀匹配：token 是类名前缀（如 OrderService → OrderServiceImpl） for k, items in lookup.items(): if k.startswith(key) and k != key: for candidate in items: if (candidate["kind"], candidate["class_name"]) not in seen: seen.add((candidate["kind"], candidate["class_name"])) hits.append(candidate) return hits @staticmethod def _cell(row: dict, headers: list[str], idx: int): if idx is None or idx >= len(headers): return "" cell = row.get(headers[idx]) return cell.value if cell is not None else "" @staticmethod def _unchanged_count(existing, matched: set[str]) -> int: all_classes = ( {c.class_name for c in existing.controller_layer} | {s.class_name for s in existing.service_layer} | {e.class_name for e in existing.entity_layer} ) return len(all_classes - matched)