131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
"""データ統合 — 白盒 + 机能 + 策略データの統合と合併。
|
|
|
|
generate_all_data() エントリポイント:
|
|
① generate_data() → 白盒（MC/DC パスカバレッジ）
|
|
② DesignDataGenerator → 机能（式样书から LLM 生成）
|
|
③ strategy_supplement() → 策略（HINA 分類に基づく境界条件）
|
|
④ 重複除去 + フィールド名正規化
|
|
⑤ 統合リスト返却
|
|
"""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from cobol_testgen import extract_structure, generate_data
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _dedup(
|
|
main_records: list[dict],
|
|
additional_records: list[dict] | None = None,
|
|
key_fields: list[str] | None = None,
|
|
) -> list[dict]:
|
|
"""合并+去重，additional 优先保留。"""
|
|
if not additional_records:
|
|
return list(main_records)
|
|
|
|
seen = set()
|
|
result = []
|
|
|
|
def _hash(rec, keys):
|
|
if keys:
|
|
return tuple(rec.get(k, "") for k in keys)
|
|
return tuple(sorted(rec.items()))
|
|
|
|
for rec in additional_records:
|
|
h = _hash(rec, key_fields)
|
|
if h not in seen:
|
|
seen.add(h)
|
|
result.append(rec)
|
|
|
|
for rec in main_records:
|
|
h = _hash(rec, key_fields)
|
|
if h not in seen:
|
|
seen.add(h)
|
|
result.append(rec)
|
|
|
|
return result
|
|
|
|
|
|
def generate_all_data(
|
|
program_id: str,
|
|
src_text: str,
|
|
st: dict | None = None,
|
|
copybook_dirs: list[str | Path] | None = None,
|
|
design_doc_dir: str | Path | None = None,
|
|
llm_client=None,
|
|
config=None,
|
|
merge_strategy: str = "merge_to_normal",
|
|
) -> list[dict]:
|
|
"""白盒 + 机能 + 策略 全量生成と統合。
|
|
|
|
Args:
|
|
program_id: プログラム ID
|
|
src_text: COBOL ソーステキスト
|
|
st: extract_structure() 結果（省略時は内部で再解析）
|
|
copybook_dirs: COPYBOOK 探索パス
|
|
design_doc_dir: 式样书配置ディレクトリ
|
|
llm_client: LLMClient インスタンス（None で LLM 系スキップ）
|
|
config: Config インスタンス
|
|
merge_strategy: merge_to_normal / as_separate_scenes / auto
|
|
|
|
Returns:
|
|
list[dict]: 統合済みレコードリスト
|
|
"""
|
|
cbd = [str(d) for d in (copybook_dirs or [])]
|
|
|
|
# ① 白盒データ
|
|
if st is None:
|
|
st = extract_structure(src_text, copybook_dirs=cbd)
|
|
whitebox = generate_data(src_text, st, copybook_dirs=cbd)
|
|
logger.info(f" White-box records: {len(whitebox)}")
|
|
|
|
# ② 机能データ（式样书 + LLM）
|
|
func_data: list[dict] = []
|
|
if design_doc_dir and llm_client:
|
|
design_path = Path(design_doc_dir) / f"詳細設計書_{program_id}.md"
|
|
if design_path.exists():
|
|
from agents.design_data import DesignDataGenerator, _extract_replacing_rules
|
|
|
|
gen = DesignDataGenerator(llm_client, cbd)
|
|
v3_names = list(st.get("field_names", [])) if st else None
|
|
replacing = _extract_replacing_rules(src_text)
|
|
|
|
try:
|
|
func_data = gen.generate(
|
|
design_md_text=design_path.read_text(encoding="utf-8"),
|
|
source_text=src_text,
|
|
replacing_rules=replacing,
|
|
v3_field_names=v3_names,
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f" DesignDataGenerator failed: {e}")
|
|
else:
|
|
logger.info(f" Design doc not found: {design_path}")
|
|
|
|
logger.info(f" Functional records: {len(func_data)}")
|
|
|
|
# ③ 策略データ
|
|
strategy_data: list[dict] = []
|
|
try:
|
|
from hina.strategy import supplement
|
|
|
|
strat_raw = supplement([], {})
|
|
for s in strat_raw:
|
|
if isinstance(s, dict) and "fields" in s:
|
|
strategy_data.append(s["fields"])
|
|
except Exception as e:
|
|
logger.debug(f" Strategy supplement skipped: {e}")
|
|
|
|
logger.info(f" Strategy records: {len(strategy_data)}")
|
|
|
|
# ④ 統合
|
|
all_records = _dedup(whitebox, func_data)
|
|
if strategy_data:
|
|
all_records = _dedup(all_records, strategy_data)
|
|
|
|
logger.info(f" Total merged records: {len(all_records)}")
|
|
return all_records
|