292 lines
8.8 KiB
Python
292 lines
8.8 KiB
Python
"""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
|