feat: 多轮运行 + GCOV 合并 + JSON 出力 + DesignDataGenerator

2026-07-12 21:04:58 +08:00
parent af37e33b98
commit f3be17e5eb
40 changed files with 4397 additions and 198 deletions
@@ -0,0 +1,291 @@
+"""DesignDataGenerator — 式样书驱动测试数据生成器。
+
+从详细设计书 .md + COBOL 源码 + COPYBOOK 中提取业务信息，
+通过 LLM 生成有业务意义的機能テストデータ。
+"""
+
+import json
+import logging
+import os
+import re
+from pathlib import Path
+from typing import Optional
+
+from agents.llm import LLMClient
+from agents.design_data_input_parser import (
+    DesignDataInputParser,
+    ProgramMeta,
+)
+
+logger = logging.getLogger(__name__)
+
+# LLM 提示词
+_SYSTEM_PROMPT = """你是 COBOL 测试数据生成专家。根据详细设计书、COPYBOOK 结构和 DB 定义，
+生成测试数据。数据必须覆盖正常路径和边界条件。
+
+输出格式: {"records": [{"field_name": "value", ...}]} JSON only。"""
+
+
+def _resolve_field_names(
+    records: list[dict],
+    replacing_rules: dict[str, str] | None,
+    v3_field_names: set[str] | None,
+) -> list[dict]:
+    """将外部 Agent 输出的字段名映射为 V3 兼容名称。
+
+    处理顺序：
+    1. REPLACING 展开（(A) → R01）
+    2. 去掉前缀和字段名之间的多余连字符（R01-EMP-ID → R01EMP-ID）
+    3. 尝试直接匹配 V3 字段名
+    4. 尝试以 V3 字段名 prefix 截断匹配
+    5. 无法映射的字段丢弃
+    """
+    if not replacing_rules and not v3_field_names:
+        return records
+
+    # Build prefix map: from REPLACING rules, e.g. (A) → R01, then R01 is the prefix
+    prefixes = set()
+    prefix_from_replacing = {}
+    if replacing_rules:
+        for old, new in replacing_rules.items():
+            if new.strip():
+                prefixes.add(new)
+                prefix_from_replacing[old] = new
+
+    if not v3_field_names:
+        # Just apply replacing, no V3 validation
+        result = []
+        for rec in records:
+            mapped = {}
+            for key, val in rec.items():
+                new_key = key
+                for old, new in prefix_from_replacing.items():
+                    new_key = new_key.replace(old, new)
+                mapped[new_key] = val
+            result.append(mapped)
+        return result
+
+    result = []
+    for rec in records:
+        mapped = {}
+        for key, val in rec.items():
+            new_key = key
+
+            # Step 1: REPLACING 展开
+            for old, new in prefix_from_replacing.items():
+                new_key = new_key.replace(old, new)
+
+            # Step 2: 去掉前缀和字段名间的连字符
+            # Agent 输出: R01-EMP-ID, V3 期望: R01EMP-ID
+            # 去掉 {prefix}- 前缀（如果前缀是 R01，去掉 R01-）
+            for p in prefixes:
+                if new_key.startswith(p + "-"):
+                    new_key = p + new_key[len(p) + 1 :]
+                    break
+
+            # Step 3: 直接匹配
+            if new_key in v3_field_names:
+                mapped[new_key] = val
+                continue
+
+            # Step 4: 去掉所有连字符尝试匹配
+            no_hyphen = new_key.replace("-", "")
+            if no_hyphen in v3_field_names:
+                mapped[no_hyphen] = val
+                continue
+
+            # Step 5: 去掉下划线尝试匹配
+            no_underscore = no_hyphen.replace("_", "")
+            if no_underscore in v3_field_names:
+                mapped[no_underscore] = val
+                continue
+
+            logger.debug(f"  field '{key}' -> '{new_key}' not in V3 fields, dropped")
+
+        result.append(mapped)
+    return result
+
+
+def _extract_replacing_rules(source_text: str) -> dict[str, str]:
+    """从 COBOL 源码的 COPY ... REPLACING 提取替换规则。"""
+    rules = {}
+    for m in re.finditer(
+        r"COPY\s+(\w+)\s+REPLACING\s+==\(A\)==\s+BY\s+==(\w+)==",
+        source_text,
+        re.IGNORECASE,
+    ):
+        rules["(A)"] = m.group(2)
+    return rules
+
+
+def _dedup(
+    main_records: list[dict],
+    additional_records: list[dict],
+    key_fields: list[str] | None = None,
+) -> list[dict]:
+    """合并+去重，additional 优先保留。"""
+    seen = set()
+    result = []
+
+    for rec in additional_records:
+        h = _record_hash(rec, key_fields)
+        if h not in seen:
+            seen.add(h)
+            result.append(rec)
+
+    for rec in main_records:
+        h = _record_hash(rec, key_fields)
+        if h not in seen:
+            seen.add(h)
+            result.append(rec)
+
+    return result
+
+
+def _record_hash(rec: dict, key_fields: list[str] | None) -> tuple:
+    if key_fields:
+        return tuple(rec.get(k, "") for k in key_fields)
+    return tuple(sorted(rec.items()))
+
+
+def _load_rules(rules_dir: str) -> str:
+    """Load all rules from pgm_pattern/ and special_feature/ directories."""
+    texts = []
+    base = Path(rules_dir)
+
+    for subdir in ["pgm_pattern", "special_feature"]:
+        d = base / subdir
+        if d.exists():
+            for f in sorted(d.glob("*.md")):
+                texts.append(f"=== {subdir}/{f.name} ===\n{f.read_text(encoding='utf-8')}")
+
+    return "\n\n".join(texts)
+
+
+class DesignDataGenerator:
+    """式样书驱动测试数据生成器。
+
+    使用例:
+        llm = LLMClient(model="deepseek-v4-flash")
+        gen = DesignDataGenerator(llm, cpy_dirs=["cpy"])
+        records = gen.generate(
+            design_md_text=design_text,
+            source_text=source_text,
+        )
+    """
+
+    def __init__(
+        self,
+        llm_client: LLMClient,
+        cpy_dirs: list[str | Path],
+        rules_dir: str | Path = "rules",
+    ):
+        self.llm = llm_client
+        self.cpy_dirs = cpy_dirs
+        self.rules_dir = Path(rules_dir)
+
+    def generate(
+        self,
+        design_md_text: str,
+        source_text: str,
+        file_db_md_text: str | None = None,
+        db_md_text: str | None = None,
+        replacing_rules: dict[str, str] | None = None,
+        v3_field_names: list[str] | None = None,
+    ) -> list[dict]:
+        """生成机能测试数据。
+
+        Args:
+            design_md_text: 式样书 .md 全文
+            source_text: COBOL 源码全文
+            file_db_md_text: COPY句定义书 .md（可选）
+            db_md_text: DB 定义书 .md（可选）
+            replacing_rules: REPLACING 展开规则
+            v3_field_names: V3 字段名参考列表（用于映射验证）
+
+        Returns:
+            list[dict]: 每条记录为 {field_name: value} 格式
+        """
+        logger.info("  DesignDataGenerator: parsing design document...")
+
+        try:
+            parser = DesignDataInputParser()
+            meta = parser.parse(design_md_text, source_text)
+        except Exception as e:
+            logger.warning(f"  Design doc parsing failed: {e}")
+            return []
+
+        if not meta.pgm_pattern:
+            logger.info("  No PGM pattern found in design doc, skipping")
+            return []
+
+        logger.info(
+            f"  Program: {meta.program_id}, pattern: {meta.pgm_pattern}, "
+            f"type: {meta.input_type}"
+        )
+
+        # 加载规则
+        rules_text = _load_rules(str(self.rules_dir))
+
+        # 构建描述信息
+        files_desc = "\n".join(
+            f"  {f.identifier}: {f.file_db_name} (I/O={f.io}, 媒体={f.medium})"
+            for f in meta.files
+        )
+        keys_desc = "\n".join(
+            f"  {k.file_name}: sort={k.sort_condition}, key={k.key_condition}"
+            for k in meta.keys
+        )
+
+        user_prompt = f"""## プログラム情報
+- プログラムID: {meta.program_id}
+- PGMパターン: {meta.pgm_pattern}
+- 入力タイプ: {meta.input_type}
+
+## 使用ファイル一覧
+{files_desc or '（なし）'}
+
+## キー項目一覧
+{keys_desc or '（なし）'}
+
+## 処理詳細
+{meta.process_detail[:2000] if meta.process_detail else '（なし）'}
+
+## 出力レコード定義
+{meta.output_records[:1000] if meta.output_records else '（なし）'}
+
+## データ生成ルール
+{rules_text[:2000] if rules_text else '（なし）'}
+
+以下の JSON 形式でテストデータを生成してください：
+{{"records": [{{"field1": "value1", "field2": "value2", ...}}]}}"""
+
+        try:
+            response = self.llm.call(
+                [
+                    {"role": "system", "content": _SYSTEM_PROMPT},
+                    {"role": "user", "content": user_prompt},
+                ]
+            )
+            logger.info("  LLM response received")
+        except Exception as e:
+            logger.warning(f"  LLM call failed: {e}")
+            return []
+
+        try:
+            parsed = json.loads(response)
+            raw_records = parsed.get("records", [])
+        except (json.JSONDecodeError, KeyError) as e:
+            logger.warning(f"  LLM response parse failed: {e}")
+            return []
+
+        if not raw_records:
+            logger.info("  No records generated")
+            return []
+
+        # 字段名映射
+        v3_names_set = set(v3_field_names) if v3_field_names else None
+        mapped = _resolve_field_names(raw_records, replacing_rules, v3_names_set)
+
+        logger.info(f"  Generated {len(mapped)} records")
+        return mapped