# 测试数据生成 Agent 实现计划 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 基于 Python 构建一个测试数据生成 Agent，解析 COBOL 详细设计书和 COPYBOOK，通过 DeepSeek API 生成符合规范的测试数据 JSON/SQL 文件。 **Architecture:** 线性管道架构。InputParser 解析输入 → RuleLoader 匹配规则 → PromptBuilder 组装 prompt → APIClient 调用 AI → OutputWriter 保存结果。各模块通过 dataclass 传递结构化数据。 **Tech Stack:** Python 3.9+, requests, pytest --- ### Task 1: 项目初始化 **Files:** - Create: `D:\jcl-cobol-data-create\agent\__init__.py` - Create: `D:\jcl-cobol-data-create\requirements.txt` - Create: `D:\jcl-cobol-data-create\tests\__init__.py` - Create: `D:\jcl-cobol-data-create\tests\conftest.py` - [ ] **Step 1: 创建 agent 包和 __init__.py** ```python # agent/__init__.py __version__ = "0.1.0" ``` - [ ] **Step 2: 创建 requirements.txt** ``` requests>=2.28.0 ``` - [ ] **Step 3: 创建 tests 目录和 conftest.py** ```python # tests/__init__.py # （空文件） ``` ```python # tests/conftest.py import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) @pytest.fixture def data_dir(): return os.path.join(os.path.dirname(__file__), 'test_data') ``` - [ ] **Step 4: 创建 tests/test_data 目录存放测试夹具** Run: `New-Item -ItemType Directory -Path "D:\jcl-cobol-data-create\tests\test_data" -Force` - [ ] **Step 5: 运行测试验证环境** Run: `python -m pytest tests/ -v` Expected: 0 tests collected, no errors. --- ### Task 2: 数据模型定义 **Files:** - Create: `D:\jcl-cobol-data-create\agent\models.py` - Create: `D:\jcl-cobol-data-create\tests\test_models.py` - [ ] **Step 1: 定义数据模型** ```python # agent/models.py from dataclasses import dataclass, field from typing import List, Dict, Optional @dataclass class FileInfo: """使用ファイル一覧 中的一行""" no: int file_db_name: str # 使用ファイル/DB名 identifier: str # 識別子 (R01, W01, etc.) dd_name: str # DD名 io: str # I/O (I, O, I/U/D) copy_group: str # COPY群 format: str # 形式 (FB, VB) record_len: int # レコード長 medium: str # 媒体 (PS, DB) remarks: str # 備考 @dataclass class CopyField: """COPYBOOK 中的单个字段""" level: int # 级别 (03, 05, ...) name: str # 替换后的实际字段名 (如 R01-APPL-ID) raw_name: str # 原始占位符名 (如 (A)APPL-ID) pic_type: str # PIC 定义 (如 "X(008)", "9(008)", "S9(009) COMP-3") pic_bytes: int # 字节数 @dataclass class KeyInfo: """キー項目一覧 中的一行""" no: int file_name: str # ファイル名 sort_condition: str # ソート条件(キー項目) key_condition: str # キー条件(マッチング/キーブレイク) @dataclass class ModuleInfo: """使用モジュール一覧 中的一行""" no: int function: str # 機能 program_id: str # プログラムID copy_name: str # 使用COPY名 @dataclass class TableColumn: """DB 表中的单个字段""" no: int name_jp: str # 项目名(日文) name_en: str # 项目名(英文) type: str # 类型 (CHAR, DECIMAL, etc.) max_len: str # 最大长 decimal_digits: str # 小数桁 byte_count: str # バイト数 nullable: bool # NULL 许可 is_pk: bool # 是否主键 @dataclass class TableInfo: """DB 表定义""" table_name: str # DB名 (如 LEAVE_RECORDS) db_id: str # DB ID (如 EMP_MASTER) copy_id: str # COPY ID columns: List[TableColumn] pk_columns: List[str] # 主键字段名列表 @dataclass class ProgramMeta: """程序完整元数据""" program_id: str program_name: str system_name: str pgm_type: str # PGMタイプ (メイン/サブ) pgm_pattern: str # PGMパターン summary_lines: List[str] # 機能概要 (多行) prerequisites: List[Dict[str, str]] # 前提条件 files: List[FileInfo] keys: List[KeyInfo] modules: List[ModuleInfo] process_detail: str # 処理詳細 全文 (从详细设计书获取) output_records: str # 出力レコード定義 全文 input_type: str # "file", "db", "mixed" # 由 COPYBOOK 解析填充 copy_fields: Dict[str, List[CopyField]] # 識別子 → 字段列表 # DB 相关 db_tables: Dict[str, TableInfo] # DB表名 → 表定义 ``` - [ ] **Step 2: 写测试验证数据模型可实例化** ```python # tests/test_models.py from agent.models import FileInfo, CopyField, ProgramMeta, KeyInfo, TableColumn, TableInfo def test_file_info(): f = FileInfo( no=1, file_db_name="OVT-SORTED", identifier="R01", dd_name="ZAN04R01", io="I", copy_group="ZAN01REC", format="FB", record_len=80, medium="PS", remarks="有効申請" ) assert f.identifier == "R01" assert f.medium == "PS" def test_copy_field(): cf = CopyField(level=3, name="R01-APPL-ID", raw_name="(A)APPL-ID", pic_type="X(008)", pic_bytes=8) assert cf.name == "R01-APPL-ID" assert cf.pic_bytes == 8 def test_program_meta_defaults(): meta = ProgramMeta( program_id="TEST", program_name="", system_name="", pgm_type="", pgm_pattern="", summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail="", output_records="", input_type="file", copy_fields={}, db_tables={} ) assert meta.program_id == "TEST" assert meta.input_type == "file" def test_table_info(): col = TableColumn(no=1, name_jp="社員番号", name_en="EMP_ID", type="CHAR", max_len="8", decimal_digits="", byte_count="8", nullable=False, is_pk=True) table = TableInfo(table_name="EMP_MASTER", db_id="EMP_MASTER", copy_id="", columns=[col], pk_columns=["EMP_ID"]) assert table.pk_columns == ["EMP_ID"] assert len(table.columns) == 1 ``` - [ ] **Step 3: 运行测试** Run: `python -m pytest tests/test_models.py -v` Expected: 4 tests PASS --- ### Task 3: InputParser — 详细设计书解析 **Files:** - Create: `D:\jcl-cobol-data-create\agent\markdown_utils.py` - Create: `D:\jcl-cobol-data-create\agent\input_parser.py` - Create: `D:\jcl-cobol-data-create\tests\test_data\sample_design_ZAN04MAT.md` - Create: `D:\jcl-cobol-data-create\tests\test_markdown_utils.py` - Create: `D:\jcl-cobol-data-create\tests\test_input_parser.py` - [ ] **Step 1: 创建 markdown 工具函数** ```python # agent/markdown_utils.py import re from typing import List, Dict, Optional def extract_section(md_text: str, section_title: str) -> str: """Extract content of a ### section from markdown text. Returns content between this header and the next ## or ### header, or end of file. """ pattern = rf'###\s+{re.escape(section_title)}\s*\n(.*?)(?=\n##|\n###|\Z)' match = re.search(pattern, md_text, re.DOTALL) return match.group(1) if match else '' def parse_table_rows(text: str) -> List[Dict[str, str]]: """Parse a markdown table from text, return list of row dicts. Handles tables with exactly one header row and one separator row. """ lines = [] for line in text.split('\n'): stripped = line.strip() if stripped.startswith('|') and stripped.endswith('|'): # Skip separator rows like |---|------|---| if re.match(r'^\|[\s\-:]+\|', stripped): continue lines.append(stripped) if len(lines) < 1: return [] headers = [cell.strip() for cell in lines[0].split('|')[1:-1]] rows = [] for line in lines[1:]: cells = [cell.strip() for cell in line.split('|')[1:-1]] if len(cells) == len(headers): rows.append(dict(zip(headers, cells))) return rows def parse_table_from_section(md_text: str, section_title: str) -> List[Dict[str, str]]: """Find a ### section and parse its first table.""" section_text = extract_section(md_text, section_title) if not section_text: return [] return parse_table_rows(section_text) def find_row_by_key(rows: List[Dict[str, str]], key_col: str, key_value: str) -> Optional[Dict[str, str]]: """Find a table row where a specific column matches the key value.""" for row in rows: if row.get(key_col, '').strip() == key_value: return row return None ``` - [ ] **Step 2: 写 markdown_utils 的测试** ```python # tests/test_markdown_utils.py from agent.markdown_utils import extract_section, parse_table_rows, parse_table_from_section, find_row_by_key SAMPLE_MD = """ ### 基本情報 | # | 項目 | 内容 | |---|------|------| | 1 | システム名 | 残業統計管理システム | | 4 | PGMパターン | マッチング(1:1) | | 5 | 機能概要 | 取消マッチング処理 | ### 使用ファイル一覧 | NO | 使用ファイル/DB名 | 識別子 | DD名 | I/O | COPY群 | 媒体 | 備考 | |----|------------------|--------|------|-----|--------|------|------| | 1 | OVT-SORTED | R01 | ZAN04R01 | I | ZAN01REC | PS | | | 2 | ERROR-LOG | W01 | ZAN04W01 | O | ZAN05REC | PS | | ### 出力レコード定義 ### 出力ファイル1（W01/OVT-MATCHED） | No | 項目名 | 設定元 | 備考 | |----|--------|--------|------| | 1 | APPL-ID | R01.APPL-ID | | | 2 | EMP-ID | R01.EMP-ID | | """ def test_extract_section(): result = extract_section(SAMPLE_MD, "基本情報") assert "残業統計管理システム" in result assert "使用ファイル一覧" not in result def test_extract_section_not_found(): result = extract_section(SAMPLE_MD, "存在しないセクション") assert result == '' def test_parse_table_rows(): rows = parse_table_from_section(SAMPLE_MD, "基本情報") assert len(rows) >= 3 assert rows[0]['項目'] == 'システム名' def test_find_row_by_key(): rows = parse_table_from_section(SAMPLE_MD, "基本情報") row = find_row_by_key(rows, '項目', 'PGMパターン') assert row is not None assert row['内容'] == 'マッチング(1:1)' def test_parse_use_file_table(): rows = parse_table_from_section(SAMPLE_MD, "使用ファイル一覧") assert len(rows) == 2 assert rows[0]['識別子'] == 'R01' assert rows[0]['DD名'] == 'ZAN04R01' def test_extract_output_records_section(): result = extract_section(SAMPLE_MD, "出力レコード定義") assert "出力ファイル1" in result assert "W01/OVT-MATCHED" in result ``` - [ ] **Step 3: 运行测试确认 markdown_utils 通过** Run: `python -m pytest tests/test_markdown_utils.py -v` Expected: all PASS - [ ] **Step 4: 创建测试用详细设计书 fixture** 使用实际项目中 `詳細設計書_ZAN04MAT.md` 的副本作为测试数据。 Run: `Copy-Item "D:\cobol-tna-system\詳細設計書\詳細設計書_ZAN04MAT.md" "D:\jcl-cobol-data-create\tests\test_data\詳細設計書_ZAN04MAT.md"` - [ ] **Step 5: 实现 InputParser 设计书部分** ```python # agent/input_parser.py import re import os from typing import List, Dict, Optional, Tuple from agent.models import ( FileInfo, KeyInfo, ModuleInfo, ProgramMeta, CopyField, TableColumn, TableInfo ) from agent.markdown_utils import ( extract_section, parse_table_rows, parse_table_from_section, find_row_by_key ) class InputParser: """解析 COBOL 程序的详细设计书和相关文件。""" def __init__(self, design_md_path: str, source_cbl_path: str, file_db_md_path: str, cpy_dir: str, db_md_path: str): self.design_md_path = design_md_path self.source_cbl_path = source_cbl_path self.file_db_md_path = file_db_md_path self.cpy_dir = cpy_dir self.db_md_path = db_md_path self._design_text = '' self._source_text = '' def run(self) -> ProgramMeta: """执行完整解析，返回 ProgramMeta。""" self._design_text = self._read_file(self.design_md_path) self._source_text = self._read_file(self.source_cbl_path) meta = ProgramMeta( program_id='', program_name='', system_name='', pgm_type='', pgm_pattern='', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={} ) self._parse_basic_info(meta) self._parse_use_files(meta) self._parse_keys(meta) self._parse_modules(meta) self._parse_process_detail(meta) self._parse_output_records(meta) self._determine_input_type(meta) self._parse_copybooks(meta) self._parse_db_definition(meta) return meta @staticmethod def _read_file(path: str) -> str: with open(path, 'r', encoding='utf-8') as f: return f.read() def _parse_basic_info(self, meta: ProgramMeta): rows = parse_table_from_section(self._design_text, '基本情報') for row in rows: item = row.get('項目', '') value = row.get('内容', '') if item == 'システム名': meta.system_name = value elif item == 'プログラムID': meta.program_id = value elif item == 'プログラム名': meta.program_name = value elif item == 'PGMタイプ': meta.pgm_type = value elif item == 'PGMパターン': meta.pgm_pattern = value elif item == '機能概要': meta.summary_lines.append(value) elif item == '': # 多行功能概要用空 # 列标识续行 if value: meta.summary_lines.append(value) def _parse_use_files(self, meta: ProgramMeta): rows = parse_table_from_section(self._design_text, '使用ファイル一覧') for row in rows: try: no = int(row.get('NO', '0')) rec_len_str = row.get('レコード長', '0') rec_len = int(rec_len_str) if rec_len_str and rec_len_str != '' else 0 f = FileInfo( no=no, file_db_name=row.get('使用ファイル/DB名', ''), identifier=row.get('識別子', ''), dd_name=row.get('DD名', ''), io=row.get('I/O', ''), copy_group=row.get('COPY群', ''), format=row.get('形式', ''), record_len=rec_len, medium=row.get('媒体', ''), remarks=row.get('備考', '') ) meta.files.append(f) except (ValueError, KeyError): continue def _parse_keys(self, meta: ProgramMeta): rows = parse_table_from_section(self._design_text, 'キー項目一覧') for row in rows: try: k = KeyInfo( no=int(row.get('NO', '0')), file_name=row.get('ファイル名', ''), sort_condition=row.get('ソート条件(キー項目)', ''), key_condition=row.get('キー条件(マッチング/キーブレイク)', '') ) meta.keys.append(k) except (ValueError, KeyError): continue def _parse_modules(self, meta: ProgramMeta): rows = parse_table_from_section(self._design_text, '使用モジュール一覧') for row in rows: try: m = ModuleInfo( no=int(row.get('NO', '0')), function=row.get('機能', ''), program_id=row.get('プログラムID', ''), copy_name=row.get('使用COPY名', '') ) meta.modules.append(m) except (ValueError, KeyError): continue def _parse_process_detail(self, meta: ProgramMeta): meta.process_detail = extract_section(self._design_text, '処理詳細') def _parse_output_records(self, meta: ProgramMeta): meta.output_records = extract_section(self._design_text, '出力レコード定義') def _determine_input_type(self, meta: ProgramMeta): """Determine input type based on input files' medium. Only looks at files with I/O containing 'I' (input). """ input_mediums = set() for f in meta.files: if 'I' in f.io: input_mediums.add(f.medium) if not input_mediums: meta.input_type = 'file' elif input_mediums == {'PS'}: meta.input_type = 'file' elif input_mediums == {'DB'}: meta.input_type = 'db' else: meta.input_type = 'mixed' def _parse_copybooks(self, meta: ProgramMeta): """TODO: 在 Task 4 中实现""" pass def _parse_db_definition(self, meta: ProgramMeta): """TODO: 在 Task 5 中实现""" pass ``` - [ ] **Step 6: 写 InputParser 设计书解析部分的测试** ```python # tests/test_input_parser.py import os from agent.input_parser import InputParser FIXTURE_DIR = os.path.join(os.path.dirname(__file__), 'test_data') def test_parse_design_basic_info(): parser = InputParser( design_md_path=os.path.join(FIXTURE_DIR, '詳細設計書_ZAN04MAT.md'), source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir='dummy_cpy', db_md_path='dummy_db.md' ) parser._design_text = parser._read_file(parser.design_md_path) from agent.models import ProgramMeta meta = ProgramMeta(program_id='', program_name='', system_name='', pgm_type='', pgm_pattern='', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={}) parser._parse_basic_info(meta) assert meta.program_id == 'ZAN04MAT' assert meta.pgm_pattern == 'マッチング(1:1)' assert meta.pgm_type == 'メイン' assert '残業統計管理システム' in meta.system_name def test_parse_use_files(): parser = InputParser( design_md_path=os.path.join(FIXTURE_DIR, '詳細設計書_ZAN04MAT.md'), source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir='dummy_cpy', db_md_path='dummy_db.md' ) parser._design_text = parser._read_file(parser.design_md_path) from agent.models import ProgramMeta meta = ProgramMeta(program_id='', program_name='', system_name='', pgm_type='', pgm_pattern='', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={}) parser._parse_use_files(meta) assert len(meta.files) == 5 r01 = [f for f in meta.files if f.identifier == 'R01'][0] assert r01.file_db_name == 'OVT-SORTED' assert r01.copy_group == 'ZAN01REC' assert r01.medium == 'PS' assert r01.io == 'I' def test_determine_input_type_file(): parser = InputParser( design_md_path=os.path.join(FIXTURE_DIR, '詳細設計書_ZAN04MAT.md'), source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir='dummy_cpy', db_md_path='dummy_db.md' ) from agent.models import ProgramMeta, FileInfo meta = ProgramMeta(program_id='ZAN04MAT', program_name='', system_name='', pgm_type='', pgm_pattern='', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={}) meta.files = [ FileInfo(no=1, file_db_name='F1', identifier='R01', dd_name='DD1', io='I', copy_group='C1', format='FB', record_len=80, medium='PS', remarks=''), FileInfo(no=2, file_db_name='F2', identifier='W01', dd_name='DD2', io='O', copy_group='C2', format='FB', record_len=80, medium='PS', remarks=''), ] parser._determine_input_type(meta) assert meta.input_type == 'file' def test_determine_input_type_mixed(): parser = InputParser( design_md_path=os.path.join(FIXTURE_DIR, '詳細設計書_ZAN04MAT.md'), source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir='dummy_cpy', db_md_path='dummy_db.md' ) from agent.models import ProgramMeta, FileInfo meta = ProgramMeta(program_id='TEST', program_name='', system_name='', pgm_type='', pgm_pattern='', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={}) meta.files = [ FileInfo(no=1, file_db_name='F1', identifier='R01', dd_name='DD1', io='I', copy_group='C1', format='FB', record_len=80, medium='PS', remarks=''), FileInfo(no=2, file_db_name='DB1', identifier='DB', dd_name='', io='I', copy_group='', format='', record_len=0, medium='DB', remarks=''), ] parser._determine_input_type(meta) assert meta.input_type == 'mixed' def test_parse_process_detail(): parser = InputParser( design_md_path=os.path.join(FIXTURE_DIR, '詳細設計書_ZAN04MAT.md'), source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir='dummy_cpy', db_md_path='dummy_db.md' ) parser._design_text = parser._read_file(parser.design_md_path) from agent.models import ProgramMeta meta = ProgramMeta(program_id='', program_name='', system_name='', pgm_type='', pgm_pattern='', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={}) parser._parse_process_detail(meta) assert '1000ITTSOR' in meta.process_detail assert 'マッチの場合' in meta.process_detail def test_parse_output_records(): parser = InputParser( design_md_path=os.path.join(FIXTURE_DIR, '詳細設計書_ZAN04MAT.md'), source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir='dummy_cpy', db_md_path='dummy_db.md' ) parser._design_text = parser._read_file(parser.design_md_path) from agent.models import ProgramMeta meta = ProgramMeta(program_id='', program_name='', system_name='', pgm_type='', pgm_pattern='', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={}) parser._parse_output_records(meta) assert 'OVT-MATCHED' in meta.output_records assert 'OVT-DBCLEAN' in meta.output_records ``` - [ ] **Step 7: 运行测试** Run: `python -m pytest tests/test_input_parser.py -v` Expected: all PASS --- ### Task 4: InputParser — COPYBOOK 解析 **Files:** - Modify: `D:\jcl-cobol-data-create\agent\input_parser.py` (添加 `_parse_copybooks`) - Create: `D:\jcl-cobol-data-create\tests\test_copy_parser.py` - [ ] **Step 1: 添加 COPYBOOK 解析逻辑到 InputParser** 在 `agent/input_parser.py` 的 `_parse_copybooks` 方法中追加实现，替换之前的 `pass`： ```python def _parse_copybooks(self, meta: ProgramMeta): """解析 COPYBOOK 文件，应用 REPLACING 替换。""" # Step 1: 从源码提取 COPY REPLACING 映射 replacing_map = self._extract_copy_replacing() # Step 2: 遍历使用ファイル一覧 中的文件，找到对应的 COPYBOOK for f in meta.files: if not f.copy_group: continue copy_file = os.path.join(self.cpy_dir, f.copy_group + '.cpy') if not os.path.exists(copy_file): continue # 从 REPLACING 映射获取替换前缀 prefix = replacing_map.get(f.copy_group, f.identifier) fields = self._parse_single_copybook(copy_file, prefix) if f.identifier not in meta.copy_fields: meta.copy_fields[f.identifier] = fields def _extract_copy_replacing(self) -> Dict[str, str]: """从 COBOL 源码提取 COPY ... REPLACING ... 映射。 Returns: {copy_name: replacing_value} e.g. {'ZAN01REC': 'R01'} """ result = {} # 匹配模式: COPY ZAN01REC REPLACING ==(A)== BY ==R01==. pattern = r'COPY\s+(\S+)\s+REPLACING\s+==\(A\)==\s+BY\s+==(\S+)==' for match in re.finditer(pattern, self._source_text, re.IGNORECASE): copy_name = match.group(1) replacing_value = match.group(2) result[copy_name] = replacing_value return result def _parse_single_copybook(self, copy_path: str, prefix: str) -> List[CopyField]: """解析单个 COPYBOOK 文件，将 (A) 占位符替换为 'prefix-' 前缀。 e.g. (A)APPL-ID + prefix='R01' → R01-APPL-ID """ copy_text = self._read_file(copy_path) fields = [] for line in copy_text.split('\n'): line = line.strip() if not line or line.startswith('*'): continue # 匹配: 03 (A)FIELD-NAME PIC X(008). # 或: 03 (A)FIELD-NAME PIC S9(009) COMP-3. match = re.match( r'(\d{2})\s+\(A\)(\S+)\s+PIC\s+(.+?)\.', line, re.IGNORECASE ) if not match: continue level = int(match.group(1)) raw_name = '(A)' + match.group(2) pic_full = match.group(3).strip() # 替换占位符 actual_name = prefix + '-' + match.group(2) pic_bytes = self._calculate_pic_bytes(pic_full) fields.append(CopyField( level=level, name=actual_name, raw_name=raw_name, pic_type=pic_full, pic_bytes=pic_bytes )) return fields @staticmethod def _calculate_pic_bytes(pic: str) -> int: """Calculate byte size from PIC definition. Handles: X(n), 9(n), S9(n)V9(m), COMP-3 variants, COMP variants. """ pic_upper = pic.upper().strip() # COMP-3 packed decimal: S9(n)V9(m) COMP-3 or S9(n) COMP-3 or 9(n) COMP-3 if 'COMP-3' in pic_upper: # Remove COMP-3 suffix base = pic_upper.replace('COMP-3', '').strip() # Extract digits: count all 9s total_digits = sum(int(num) for num in re.findall(r'9\((\d+)\)', base)) if base.startswith('S'): return (total_digits + 2) // 2 else: return (total_digits + 1) // 2 # COMP / BINARY: S9(n) COMP, S9(n)V9(m) COMP if 'COMP' in pic_upper or 'BINARY' in pic_upper: base = re.sub(r'\s*(COMP|BINARY)\s*', '', pic_upper).strip() total_digits = sum(int(num) for num in re.findall(r'9\((\d+)\)', base)) if total_digits <= 4: return 2 elif total_digits <= 9: return 4 else: return 8 # Simple PIC: X(n), 9(n), S9(n), S9(n)V9(m) total = 0 # Match X(n) for num in re.findall(r'X\((\d+)\)', pic_upper): total += int(num) # Match G(n) (double-byte) for num in re.findall(r'G\((\d+)\)', pic_upper): total += int(num) * 2 # Match 9(n) for num in re.findall(r'9\((\d+)\)', pic_upper): total += int(num) if total > 0: return total return 0 ``` - [ ] **Step 2: 创建 COPYBOOK 解析测试的 fixture** 将实际 COPYBOOK 文件复制到测试目录： ```powershell Copy-Item "D:\cobol-tna-system\cpy\ZAN01REC.cpy" "D:\jcl-cobol-data-create\tests\test_data\ZAN01REC.cpy" Copy-Item "D:\cobol-tna-system\cpy\ZAN02REC.cpy" "D:\jcl-cobol-data-create\tests\test_data\ZAN02REC.cpy" Copy-Item "D:\cobol-tna-system\cpy\ZAN04REC.cpy" "D:\jcl-cobol-data-create\tests\test_data\ZAN04REC.cpy" ``` - [ ] **Step 3: 写 COPYBOOK 解析测试** ```python # tests/test_copy_parser.py import os import tempfile from agent.input_parser import InputParser FIXTURE_DIR = os.path.join(os.path.dirname(__file__), 'test_data') def test_parse_single_copybook(): parser = InputParser( design_md_path='dummy.md', source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir=FIXTURE_DIR, db_md_path='dummy_db.md' ) fields = parser._parse_single_copybook( os.path.join(FIXTURE_DIR, 'ZAN01REC.cpy'), prefix='R01' ) assert len(fields) == 8 # 第一个字段: (A)APPL-ID → R01-APPL-ID, PIC X(008) f0 = fields[0] assert f0.name == 'R01-APPL-ID' assert f0.raw_name == '(A)APPL-ID' assert 'X' in f0.pic_type assert f0.pic_bytes == 8 # EMP-ID: PIC 9(008) f1 = fields[1] assert f1.name == 'R01-EMP-ID' assert '9' in f1.pic_type assert f1.pic_bytes == 8 # START-TIME: PIC 9(004) f3 = fields[3] assert f3.name == 'R01-START-TIME' assert f3.pic_bytes == 4 # FILLER f7 = fields[7] assert 'FILLER' in f7.name assert f7.pic_bytes == 46 def test_extract_copy_replacing(): source_text = """ FD R01INNFIL. 01 R01INNREC. COPY ZAN01REC REPLACING ==(A)== BY ==R01==. FD R02INNFIL. 01 R02INNREC. COPY ZAN04REC REPLACING ==(A)== BY ==R02==. """ parser = InputParser( design_md_path='dummy.md', source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir=FIXTURE_DIR, db_md_path='dummy_db.md' ) parser._source_text = source_text result = parser._extract_copy_replacing() assert result['ZAN01REC'] == 'R01' assert result['ZAN04REC'] == 'R02' def test_calculate_pic_bytes_x(): assert InputParser._calculate_pic_bytes('X(008)') == 8 assert InputParser._calculate_pic_bytes('X(10)') == 10 def test_calculate_pic_bytes_9(): assert InputParser._calculate_pic_bytes('9(008)') == 8 assert InputParser._calculate_pic_bytes('9(004)') == 4 def test_calculate_pic_bytes_comp3(): # S9(009) COMP-3: 9 digits → (9+2)//2 = 5 assert InputParser._calculate_pic_bytes('S9(009) COMP-3') == 5 # S9(7) COMP-3: 7 digits → (7+2)//2 = 4 assert InputParser._calculate_pic_bytes('S9(7) COMP-3') == 4 # 9(7) COMP-3: 7 digits unsigned → (7+1)//2 = 4 assert InputParser._calculate_pic_bytes('9(7) COMP-3') == 4 def test_calculate_pic_bytes_comp(): # S9(4) COMP: <=4 → 2 bytes assert InputParser._calculate_pic_bytes('S9(4) COMP') == 2 # S9(9) COMP: <=9 → 4 bytes assert InputParser._calculate_pic_bytes('S9(9) COMP') == 4 def test_calculate_pic_bytes_decimal(): # S9(7)V9(2): 7+2=9 bytes assert InputParser._calculate_pic_bytes('S9(7)V9(2)') == 9 ``` - [ ] **Step 4: 运行测试** Run: `python -m pytest tests/test_copy_parser.py -v` Expected: all PASS --- ### Task 5: InputParser — DB 定义解析 **Files:** - Modify: `D:\jcl-cobol-data-create\agent\input_parser.py` (添加 `_parse_db_definition`) - Create: `D:\jcl-cobol-data-create\tests\test_db_parser.py` - [ ] **Step 1: 添加 DB 定义解析逻辑** 在 `agent/input_parser.py` 的 `_parse_db_definition` 方法中追加实现： ```python def _parse_db_definition(self, meta: ProgramMeta): """解析 DB 定义书 .md，提取表结构信息。 只在输入类型涉及 DB 时调用。 """ if meta.input_type not in ('db', 'mixed'): return db_text = self._read_file(self.db_md_path) # 按 ## 切分每个表定义 sections = re.split(r'\n## ', db_text) for section in sections: # 提取表名（## 后的第一部分） table_match = re.match(r'^([\w\-]+)', section) if not table_match: continue table_name = table_match.group(1) # 查找 "DB基本情報" 子节 db_info_text = extract_section('## ' + section, 'DB基本情報') db_info_rows = parse_table_rows(db_info_text) db_id = '' copy_id = '' for row in db_info_rows: db_id = row.get('DB ID', '') or db_id copy_id = row.get('COPY ID', '') or copy_id # 查找 "カラム定義" 子节 column_text = extract_section('## ' + section, 'カラム定義') column_rows = parse_table_rows(column_text) columns = [] pk_columns = [] for row in column_rows: try: no = int(row.get('No', '0')) except ValueError: continue is_pk = row.get('PK', '') == '✓' nullable = row.get('NULL', 'NULL許可') == 'NULL許可' col = TableColumn( no=no, name_jp=row.get('項目名', ''), name_en=row.get('項目名(英字名)', ''), type=row.get('TYPE', ''), max_len=row.get('最大長', ''), decimal_digits=row.get('小数桁', ''), byte_count=row.get('バイト数', ''), nullable=nullable, is_pk=is_pk ) columns.append(col) if is_pk: pk_columns.append(col.name_en) table = TableInfo( table_name=table_name, db_id=db_id, copy_id=copy_id, columns=columns, pk_columns=pk_columns ) meta.db_tables[table_name] = table ``` - [ ] **Step 2: 创建 DB 测试 fixture** ```powershell Copy-Item "D:\cobol-tna-system\詳細設計書\DB定義書.md" "D:\jcl-cobol-data-create\tests\test_data\DB定義書.md" ``` - [ ] **Step 3: 写 DB 解析测试** ```python # tests/test_db_parser.py import os from agent.input_parser import InputParser from agent.models import ProgramMeta, FileInfo FIXTURE_DIR = os.path.join(os.path.dirname(__file__), 'test_data') def test_parse_db_definition(): parser = InputParser( design_md_path='dummy.md', source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir='dummy_cpy', db_md_path=os.path.join(FIXTURE_DIR, 'DB定義書.md') ) meta = ProgramMeta( program_id='TEST', program_name='テスト', system_name='', pgm_type='メイン', pgm_pattern='DB更新', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='mixed', # DB 类型才会触发解析 copy_fields={}, db_tables={} ) parser._parse_db_definition(meta) assert len(meta.db_tables) > 0 assert 'EMP_MASTER' in meta.db_tables emp = meta.db_tables['EMP_MASTER'] assert emp.db_id == 'EMP_MASTER' assert len(emp.columns) >= 3 assert emp.pk_columns == ['EMP_ID'] def test_parse_db_skipped_for_file_type(): parser = InputParser( design_md_path='dummy.md', source_cbl_path='dummy.cbl', file_db_md_path='dummy.md', cpy_dir='dummy_cpy', db_md_path=os.path.join(FIXTURE_DIR, 'DB定義書.md') ) meta = ProgramMeta( program_id='TEST', program_name='テスト', system_name='', pgm_type='メイン', pgm_pattern='マッチング(1:1)', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', # 文件类型，跳过 DB 解析 copy_fields={}, db_tables={} ) parser._parse_db_definition(meta) assert len(meta.db_tables) == 0 ``` - [ ] **Step 4: 运行测试** Run: `python -m pytest tests/test_db_parser.py -v` Expected: all PASS --- ### Task 6: RuleLoader — 规则匹配与特殊功能检测 **Files:** - Create: `D:\jcl-cobol-data-create\agent\rule_loader.py` - Create: `D:\jcl-cobol-data-create\tests\test_rule_loader.py` - [ ] **Step 1: 实现 RuleLoader** ```python # agent/rule_loader.py import os import re from typing import List, Tuple, Optional from agent.models import ProgramMeta # PGMパターン → 规则文件名 映射表 PGM_PATTERN_MAP = { 'マッチング(1:1)': 'マッチング(1-1).md', 'マッチング(1:N)': 'マッチング(1-N).md', 'マッチング(N:1)': 'マッチング(N-1).md', 'マッチング(M:N)': 'マッチング(M-N).md', 'レイアウト編集のみ(GETPUT)': 'レイアウト編集のみ(GETPUT).md', 'レイアウト編集のみ': 'レイアウト編集のみ(GETPUT).md', '項目チェック': '項目チェック.md', '振り分け': '振り分け.md', '振り分け(IF文、EVALUATE文)': '振り分け.md', 'キーブレイク': 'キーブレイク.md', 'キーブレイク(集計、集約)': 'キーブレイク(集計、集約).md', 'キーブレイク(集計、集約の以外)': 'キーブレイク.md', 'DB更新': 'DB更新.md', } # 特殊功能检测：关键词集合 → 规则文件名 SPECIAL_FEATURE_CHECKS = [ (['場合', 'EVALUATE', 'IF'], '条件分支.md'), ] class RuleLoader: """根据程序特征加载对应的数据生成规则。""" def __init__(self, rules_dir: str): """ Args: rules_dir: 规则根目录，包含 pgm_pattern/ 和 special_feature/ 子目录 """ self.pgm_pattern_dir = os.path.join(rules_dir, 'pgm_pattern') self.special_feature_dir = os.path.join(rules_dir, 'special_feature') def load(self, meta: ProgramMeta) -> Tuple[str, List[str], int]: """加载所有相关规则，返回 (合并后的规则文本, 组描述列表, 组数)。 Returns: combined_rules: 合并后的规则文本 group_descriptions: 各组的用途描述列表 group_count: 组数 """ # 1. 加载 PGM 模式规则 pgm_rule = self._load_pgm_pattern_rule(meta.pgm_pattern) if pgm_rule is None: raise FileNotFoundError( f"PGM模式 '{meta.pgm_pattern}' 的规则文件未找到。" f"请在 {self.pgm_pattern_dir} 中添加对应的 .md 文件。" ) # 2. 解析组信息 group_descriptions, group_count = self._parse_group_info(pgm_rule) # 3. 检测特殊功能 parts = [pgm_rule] for keywords, rule_file in SPECIAL_FEATURE_CHECKS: if self._detect_feature(meta.process_detail, keywords): feature_rule = self._read_rule_file( os.path.join(self.special_feature_dir, rule_file) ) if feature_rule: parts.append(feature_rule) combined = '\n\n---\n\n'.join(parts) return combined, group_descriptions, group_count def _load_pgm_pattern_rule(self, pgm_pattern: str) -> Optional[str]: """根据 PGMパターン 加载对应的规则文件。""" # 先精确匹配 filename = PGM_PATTERN_MAP.get(pgm_pattern) if filename: path = os.path.join(self.pgm_pattern_dir, filename) if os.path.exists(path): return self._read_rule_file(path) # 尝试模糊匹配：检查文件名是否包含关键字 for fname in os.listdir(self.pgm_pattern_dir): if fname.endswith('.md'): # 尝试模糊匹配 keyword = pgm_pattern.split('(')[0].strip() if keyword in fname: return self._read_rule_file(os.path.join(self.pgm_pattern_dir, fname)) return None @staticmethod def _read_rule_file(path: str) -> str: with open(path, 'r', encoding='utf-8') as f: return f.read() @staticmethod def _parse_group_info(rule_text: str) -> Tuple[List[str], int]: """从规则文本中解析组信息。 查找表格中 '组' 列对应的行，提取组号和用途。 """ descriptions = [] group_count = 0 # 匹配规则中的组描述表格 # 格式: | 组 | 用途 | ... | for line in rule_text.split('\n'): line = line.strip() if line.startswith('|') and not re.match(r'^\|[\s\-:]+\|', line): cells = [c.strip() for c in line.split('|')[1:-1]] if len(cells) >= 2: try: group_num = int(cells[0]) if group_num > group_count: group_count = group_num descriptions.append(cells[1]) except ValueError: pass return descriptions, group_count @staticmethod def _detect_feature(process_detail: str, keywords: List[str]) -> bool: """检测処理詳細中是否包含特定关键词。""" for kw in keywords: if kw in process_detail: return True return False ``` - [ ] **Step 2: 写 RuleLoader 测试** ```python # tests/test_rule_loader.py import os from agent.rule_loader import RuleLoader, PGM_PATTERN_MAP from agent.models import ProgramMeta RULES_DIR = os.path.join(os.path.dirname(__file__), '..', 'rules') def test_pgm_pattern_map_has_known_patterns(): assert 'マッチング(1:1)' in PGM_PATTERN_MAP assert 'マッチング(1:N)' in PGM_PATTERN_MAP assert 'DB更新' in PGM_PATTERN_MAP def test_load_matching_1_1_rule(): loader = RuleLoader(RULES_DIR) meta = ProgramMeta( program_id='TEST', program_name='', system_name='', pgm_type='メイン', pgm_pattern='マッチング(1:1)', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={} ) combined, descriptions, count = loader.load(meta) assert count > 0 assert len(combined) > 0 assert 'マッチング(1:1)' in combined assert '数据生成步骤' in combined def test_load_matching_1_n_rule(): loader = RuleLoader(RULES_DIR) meta = ProgramMeta( program_id='TEST', program_name='', system_name='', pgm_type='メイン', pgm_pattern='マッチング(1:N)', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={} ) combined, descriptions, count = loader.load(meta) assert count > 0 assert len(combined) > 0 assert 'マッチング(1:N)' in combined def test_load_nonexistent_pattern_raises(): loader = RuleLoader(RULES_DIR) meta = ProgramMeta( program_id='TEST', program_name='', system_name='', pgm_type='メイン', pgm_pattern='不存在的パターン', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields={}, db_tables={} ) try: loader.load(meta) assert False, 'Should have raised FileNotFoundError' except FileNotFoundError: pass def test_detect_conditional_branch(): process = """ 2-1-2.ロジック分岐判定(EVALUATE) 2-1-2-1.STATUS='1'の場合 INSERT処理 """ assert RuleLoader._detect_feature(process, ['場合', 'EVALUATE', 'IF']) is True def test_no_conditional_branch(): process = """ 2-1.マッチの場合 R01をW01に出力 """ # '場合' 也匹配了! 日文中 'の場合' 表示 "在...情况下"，视为条件分支 assert RuleLoader._detect_feature(process, ['場合', 'EVALUATE', 'IF']) is True def test_rule_with_conditional_feature(): loader = RuleLoader(RULES_DIR) meta = ProgramMeta( program_id='TEST', program_name='', system_name='', pgm_type='メイン', pgm_pattern='マッチング(1:1)', summary_lines=[], prerequisites=[], files=[], keys=[], modules=[], process_detail='2-1-2.EVALUATEで分岐する。STATUS=1の場合INSERTする。', output_records='', input_type='file', copy_fields={}, db_tables={} ) combined, descriptions, count = loader.load(meta) assert count > 0 # 条件分支规则应该被包含 assert '分支' in combined or '条件分支' in combined or '条件判断' in combined ``` - [ ] **Step 3: 运行测试** Run: `python -m pytest tests/test_rule_loader.py -v` Expected: all PASS --- ### Task 7: PromptBuilder — Prompt 组装 **Files:** - Create: `D:\jcl-cobol-data-create\agent\prompt_builder.py` - Create: `D:\jcl-cobol-data-create\tests\test_prompt_builder.py` - [ ] **Step 1: 实现 PromptBuilder** ```python # agent/prompt_builder.py from typing import List from agent.models import ProgramMeta, CopyField, TableInfo class PromptBuilder: """将解析后的程序元数据和规则组装成 API prompt。""" def build(self, meta: ProgramMeta, rules_text: str, group_descriptions: List[str], group_count: int) -> str: """构建完整的 API prompt。""" parts = [] # 1. 程序基本情報 parts.append(self._build_basic_info(meta)) # 2. 処理詳細 parts.append(self._build_process_detail(meta)) # 3. 入力構造 (文件 + DB) parts.append(self._build_input_structures(meta)) # 4. 出力レコード定義 parts.append(self._build_output_records(meta)) # 5. データ生成ルール parts.append(rules_text) # 6. 出力格式指示 parts.append(self._build_output_format(meta)) # 7. 生成指示 parts.append(self._build_generation_instruction( group_descriptions, group_count, meta.input_type )) return '\n\n'.join(parts) def _build_basic_info(self, meta: ProgramMeta) -> str: lines = [ f"## 程序基本情報", f"- システム名: {meta.system_name}", f"- プログラムID: {meta.program_id}", f"- プログラム名: {meta.program_name}", f"- PGMパターン: {meta.pgm_pattern}", f"- 入力タイプ: {self._input_type_label(meta.input_type)}", ] if meta.summary_lines: lines.append(f"- 機能概要: {' '.join(meta.summary_lines)}") return '\n'.join(lines) def _build_process_detail(self, meta: ProgramMeta) -> str: return f"## 処理詳細\n\n```\n{meta.process_detail}\n```" def _build_input_structures(self, meta: ProgramMeta) -> str: parts = ["## 入力構造"] # 文件输入结构 input_files = [f for f in meta.files if 'I' in f.io] for f in input_files: parts.append(f"### ファイル {f.identifier} (DD名: {f.dd_name}, COPY: {f.copy_group}, 媒体: {f.medium})") fields = meta.copy_fields.get(f.identifier, []) if fields: parts.append("| 項目名 | PIC | バイト数 |") parts.append("|--------|-----|----------|") for cf in fields: parts.append(f"| {cf.name} | {cf.pic_type} | {cf.pic_bytes} |") else: parts.append("（構造情報なし）") # DB 表结构 db_inputs = [f for f in meta.files if 'I' in f.io and f.medium == 'DB'] if db_inputs or meta.db_tables: parts.append("### DBテーブル構造") for table_name, table in meta.db_tables.items(): parts.append(f"**表名: {table_name}**") parts.append(f"主キー: {', '.join(table.pk_columns)}") parts.append("| 項目名 | 英字名 | タイプ | 最大長 | KEY |") parts.append("|--------|--------|--------|--------|-----|") for col in table.columns: key_mark = '✓' if col.is_pk else '' parts.append(f"| {col.name_jp} | {col.name_en} | {col.type} | {col.max_len} | {key_mark} |") return '\n\n'.join(parts) def _build_output_records(self, meta: ProgramMeta) -> str: return f"## 出力レコード定義\n\n```\n{meta.output_records}\n```" def _build_output_format(self, meta: ProgramMeta) -> str: lines = [ "## 出力形式", "", "### JSON形式（ファイル入力の場合）", f"各グループ1つのJSONファイル: `{{program}}_{{group}}.json`", "", "JSON構造:", "```json", "{", ' "program": "{プログラムID}",', ' "records": [', ' { "input": { "FD名": { "項目名": "値", ... } } },', ' ...', ' ]', "}", "```", "", "### 項目値のルール（PIC → JSON値）", "| PIC | JSON内表示 | 例 |", "|-----|-----------|-----|", "| PIC X(n) | 左詰め + スペース埋め | `\"A0000001\"` |", "| PIC 9(n) | 右詰め + 先行ゼロ | `\"00000101\"` |", "| PIC S9(n) | 符号 + 右詰め + 先行ゼロ | `\"+0000101\"` |", "| PIC S9(n)V9(m) | 符号 + 右詰め + 小数点含む | `\"+001234567\"` |", "| PIC S9(n) COMP | 通常の10進数文字列 | `\"300\"` |", "| PIC S9(n) COMP-3 | 通常の10進数文字列 | `\"1234\"` |", "| PIC 9(n) COMP-3 | 通常の10進数文字列 | `\"1234\"` |", "| FILLER(純粋予約) | グループIDとレコード番号を含む識別可能なパターン | `\"D000000...001\"` |", "| FILLER(業務予約) | 全スペース(PIC X) または 全ゼロ(PIC 9) | |", "", "### SQL形式（DB入力の場合）", "各グループ1つのSQLファイル: `{program}_g{groupId}.sql`", "", "SQL例:", "```sql", "-- Group: 1", "INSERT INTO TABLE_NAME (COL1, COL2) VALUES", "('val1', 'val2');", "```", "", "### データ生成の注意", "- 項目名に意味がある場合（DATE→日付、NAME→氏名、ADDRESS→住所等）、実際の形式に合った値を生成すること", "- 隣接するレコード間で、できるだけ同じ項目に異なる値を設定すること", ] return '\n'.join(lines) def _build_generation_instruction(self, descriptions: List[str], count: int, input_type: str) -> str: lines = [ "## 生成指示", f"生成するグループ数: {count}", "各グループの内容:", ] for i, desc in enumerate(descriptions, 1): lines.append(f" - g{i}: {desc}") if input_type == 'file': lines.append("出力: JSONファイル") elif input_type == 'db': lines.append("出力: SQL INSERTファイル") else: lines.append("出力: JSONファイル + SQL INSERTファイル（混合）") lines.append("") lines.append("すべてのグループのデータを一度にJSONで返してください。") lines.append("出力形式: { \"groups\": { \"g1\": { \"type\": \"json\"|... }, ... } }") return '\n'.join(lines) @staticmethod def _input_type_label(t: str) -> str: labels = {'file': 'ファイル', 'db': 'DB', 'mixed': '混合（ファイル+DB）'} return labels.get(t, t) ``` - [ ] **Step 2: 写 PromptBuilder 测试** ```python # tests/test_prompt_builder.py from agent.prompt_builder import PromptBuilder from agent.models import ProgramMeta, FileInfo, CopyField def test_build_basic_prompt(): meta = ProgramMeta( program_id='ZAN04MAT', program_name='取消マッチング処理', system_name='残業統計管理システム', pgm_type='メイン', pgm_pattern='マッチング(1:1)', summary_lines=['取消申請のマッチング'], prerequisites=[], files=[], keys=[], modules=[], process_detail='1.初期処理...\n2.主処理...', output_records='### 出力ファイル1...', input_type='file', copy_fields={}, db_tables={} ) builder = PromptBuilder() prompt = builder.build( meta=meta, rules_text='# マッチング(1:1) データ生成規則\n...', group_descriptions=['両端不一致', '逆方向両端不一致', '中間不一致'], group_count=3 ) assert 'ZAN04MAT' in prompt assert 'マッチング(1:1)' in prompt assert '1.初期処理' in prompt assert '生成するグループ数: 3' in prompt assert 'g1: 両端不一致' in prompt assert 'g2: 逆方向両端不一致' in prompt assert 'g3: 中間不一致' in prompt assert 'JSON形式' in prompt def test_build_with_copy_fields(): fields = { 'R01': [ CopyField(level=3, name='R01-APPL-ID', raw_name='(A)APPL-ID', pic_type='X(008)', pic_bytes=8), CopyField(level=3, name='R01-EMP-ID', raw_name='(A)EMP-ID', pic_type='9(008)', pic_bytes=8), ] } meta = ProgramMeta( program_id='TEST', program_name='', system_name='', pgm_type='メイン', pgm_pattern='マッチング(1:1)', summary_lines=[], prerequisites=[], files=[ FileInfo(no=1, file_db_name='INPUT-FILE', identifier='R01', dd_name='TESTR01', io='I', copy_group='TESTREC', format='FB', record_len=80, medium='PS', remarks='') ], keys=[], modules=[], process_detail='', output_records='', input_type='file', copy_fields=fields, db_tables={} ) builder = PromptBuilder() prompt = builder.build( meta=meta, rules_text='# 規則', group_descriptions=['テスト'], group_count=1 ) assert 'R01-APPL-ID' in prompt assert 'X(008)' in prompt assert '8' in prompt assert 'R01-EMP-ID' in prompt def test_build_db_prompt(): meta = ProgramMeta( program_id='TESTDB', program_name='DB更新', system_name='', pgm_type='メイン', pgm_pattern='DB更新', summary_lines=[], prerequisites=[], files=[ FileInfo(no=1, file_db_name='INPUT-FILE', identifier='R01', dd_name='TESTR01', io='I', copy_group='TESTREC', format='FB', record_len=80, medium='DB', remarks='') ], keys=[], modules=[], process_detail='', output_records='', input_type='db', copy_fields={}, db_tables={} ) builder = PromptBuilder() prompt = builder.build( meta=meta, rules_text='# DB更新規則', group_descriptions=['テスト'], group_count=1 ) assert 'DB' in prompt or 'SQL' in prompt assert 'INSERT' in prompt ``` - [ ] **Step 3: 运行测试** Run: `python -m pytest tests/test_prompt_builder.py -v` Expected: all PASS --- ### Task 8: APIClient — API 调用与重试 **Files:** - Create: `D:\jcl-cobol-data-create\agent\api_client.py` - Create: `D:\jcl-cobol-data-create\tests\test_api_client.py` - [ ] **Step 1: 实现 APIClient** ```python # agent/api_client.py import json import time from typing import Dict, Any, Optional import requests class APIClient: """DeepSeek API 客户端，含重试逻辑。""" def __init__(self, api_key: str, model: str = 'deepseek-v4-flash', base_url: str = 'https://api.deepseek.com/chat/completions', max_retries: int = 3, timeout: int = 120): self.api_key = api_key self.model = model self.base_url = base_url self.max_retries = max_retries self.timeout = timeout def generate(self, prompt: str) -> Dict[str, Any]: """发送 prompt 并返回 AI 生成的结果。 Returns: AI 返回的 JSON 字典 Raises: RuntimeError: 超过最大重试次数后仍然失败 """ system_prompt = ( "你是COBOL程序的测试数据生成专家。" "请严格按照提供的规则，生成符合格式要求的测试数据。" "输出必须是可被json.loads()直接解析的JSON，不要包裹在```json```代码块中。" "不要在JSON前后添加任何说明文字。" ) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt}, ] last_error = None for attempt in range(1, self.max_retries + 1): try: response = self._call_api(messages) # 提取 AI 返回的文本 content = response['choices'][0]['message']['content'] # 尝试解析 JSON data = self._parse_json(content) if data is not None: return data # JSON 解析失败，反馈给 AI 重试 error_msg = ( f"前回の出力は有効なJSONではありませんでした。" f"必ず有効なJSONのみを出力してください。" f"コードブロック（```）で囲まないでください。" ) messages.append({"role": "assistant", "content": content}) messages.append({"role": "user", "content": error_msg}) except requests.exceptions.RequestException as e: last_error = e if attempt < self.max_retries: time.sleep(2 ** attempt) # 指数退避 continue raise RuntimeError( f"API调用失败，已重试{self.max_retries}次。" f"最后错误: {last_error}" ) def _call_api(self, messages: list) -> dict: """单次 API 调用。""" headers = { 'Authorization': f'Bearer {self.api_key}', 'Content-Type': 'application/json', } payload = { 'model': self.model, 'messages': messages, 'temperature': 0.3, 'max_tokens': 8192, } resp = requests.post( self.base_url, headers=headers, json=payload, timeout=self.timeout ) resp.raise_for_status() return resp.json() @staticmethod def _parse_json(text: str) -> Optional[dict]: """尝试从文本中提取 JSON。""" text = text.strip() # 去除可能的 markdown 代码块包裹 if text.startswith('```json'): text = text[7:] if text.startswith('```'): text = text[3:] if text.endswith('```'): text = text[:-3] text = text.strip() try: return json.loads(text) except json.JSONDecodeError: # 尝试找到第一个 { 和最后一个 } start = text.find('{') end = text.rfind('}') if start >= 0 and end > start: try: return json.loads(text[start:end + 1]) except json.JSONDecodeError: pass return None ``` - [ ] **Step 2: 写 APIClient 测试（mock API）** ```python # tests/test_api_client.py import json from unittest.mock import patch, MagicMock from agent.api_client import APIClient def test_parse_json_valid(): text = '{"groups": {"g1": {"records": []}}}' result = APIClient._parse_json(text) assert result is not None assert 'groups' in result def test_parse_json_with_markdown_wrapper(): text = '```json\n{"key": "value"}\n```' result = APIClient._parse_json(text) assert result is not None assert result['key'] == 'value' def test_parse_json_with_text_before(): text = '少し説明があります。\n{"key": "value"}\n以上です。' result = APIClient._parse_json(text) assert result is not None assert result['key'] == 'value' def test_parse_json_invalid(): text = 'これは有効なJSONではありません。' result = APIClient._parse_json(text) assert result is None @patch('agent.api_client.requests.post') def test_generate_success(mock_post): mock_response = MagicMock() mock_response.json.return_value = { 'choices': [{'message': {'content': '{"groups": {"g1": {"type": "json"}}}'}}] } mock_response.raise_for_status = MagicMock() mock_post.return_value = mock_response client = APIClient(api_key='test-key') result = client.generate("テストプロンプト") assert result['groups']['g1']['type'] == 'json' @patch('agent.api_client.requests.post') def test_generate_retry_on_json_error(mock_post): # 第一次返回无效 JSON，第二次返回有效 JSON bad_response = MagicMock() bad_response.json.return_value = { 'choices': [{'message': {'content': '無効な応答'}}] } bad_response.raise_for_status = MagicMock() good_response = MagicMock() good_response.json.return_value = { 'choices': [{'message': {'content': '{"result": "ok"}'}}] } good_response.raise_for_status = MagicMock() mock_post.side_effect = [bad_response, good_response] client = APIClient(api_key='test-key', max_retries=3) result = client.generate("テスト") assert result['result'] == 'ok' assert mock_post.call_count == 2 # 调用了2次 ``` - [ ] **Step 3: 运行测试** Run: `python -m pytest tests/test_api_client.py -v` Expected: all PASS --- ### Task 9: OutputWriter — 输出写入 **Files:** - Create: `D:\jcl-cobol-data-create\agent\output_writer.py` - Create: `D:\jcl-cobol-data-create\tests\test_output_writer.py` - [ ] **Step 1: 实现 OutputWriter** ```python # agent/output_writer.py import json import os from typing import Dict, Any class OutputWriter: """将 AI 生成的数据写入文件系统。""" def __init__(self, output_dir: str): self.output_dir = output_dir def write(self, program_id: str, ai_result: Dict[str, Any], input_type: str) -> Dict[str, str]: """写入所有组的输出文件。 Args: program_id: 程序ID ai_result: AI 返回的 JSON 数据，格式: {"groups": {"g1": {...}, ...}} input_type: "file", "db", "mixed" Returns: {group_folder: written_file_path} 映射 """ written = {} groups = ai_result.get('groups', ai_result) # 兼容 {"records": [...]} 直接格式 for group_key in sorted(groups.keys()): group_data = groups[group_key] group_dir = os.path.join(self.output_dir, program_id, group_key) os.makedirs(group_dir, exist_ok=True) if input_type in ('file', 'mixed'): json_path = self._write_json(group_dir, program_id, group_key, group_data) written[f"{group_key}/json"] = json_path if input_type in ('db', 'mixed'): sql_path = self._write_sql(group_dir, program_id, group_key, group_data) written[f"{group_key}/sql"] = sql_path return written def _write_json(self, group_dir: str, program_id: str, group_key: str, data: Any) -> str: """写入 JSON 文件。""" filename = f"{program_id}_{group_key}.json" filepath = os.path.join(group_dir, filename) with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) return filepath def _write_sql(self, group_dir: str, program_id: str, group_key: str, data: Any) -> str: """写入 SQL 文件。""" filename = f"{program_id}_{group_key}.sql" filepath = os.path.join(group_dir, filename) # 如果 data 中有 sql 字段，直接使用；否则将整个 data 写入 sql_content = data if isinstance(data, str) else data.get('sql', json.dumps(data, ensure_ascii=False)) with open(filepath, 'w', encoding='utf-8') as f: f.write(sql_content) return filepath ``` - [ ] **Step 2: 写 OutputWriter 测试** ```python # tests/test_output_writer.py import json import os import tempfile from agent.output_writer import OutputWriter def test_write_json_files(): with tempfile.TemporaryDirectory() as tmpdir: writer = OutputWriter(tmpdir) ai_result = { "groups": { "g1": { "program": "TEST", "records": [ {"input": {"R01": {"R01-FIELD": "A001"}}} ] } } } written = writer.write("TESTPGM", ai_result, input_type="file") assert len(written) == 1 json_path = written["g1/json"] assert os.path.exists(json_path) with open(json_path, 'r', encoding='utf-8') as f: content = json.load(f) assert content['program'] == 'TEST' assert len(content['records']) == 1 def test_write_mixed_type(): with tempfile.TemporaryDirectory() as tmpdir: writer = OutputWriter(tmpdir) ai_result = { "groups": { "g1": { "program": "MIXED", "records": [{"input": {"R01": {"F1": "X"}}}], "sql": "INSERT INTO T VALUES ('x');" } } } written = writer.write("MIXEDPGM", ai_result, input_type="mixed") assert len(written) == 2 assert os.path.exists(written["g1/json"]) assert os.path.exists(written["g1/sql"]) def test_write_multiple_groups(): with tempfile.TemporaryDirectory() as tmpdir: writer = OutputWriter(tmpdir) ai_result = { "groups": { "g1": {"records": [{}]}, "g2": {"records": [{}]}, "g3": {"records": [{}]}, } } written = writer.write("MULTI", ai_result, input_type="file") assert len(written) == 3 for g in ['g1', 'g2', 'g3']: assert f'{g}/json' in written assert os.path.exists(written[f'{g}/json']) def test_directory_structure(): with tempfile.TemporaryDirectory() as tmpdir: writer = OutputWriter(tmpdir) ai_result = {"groups": {"g1": {"records": [{}]}}} writer.write("MYPROG", ai_result, input_type="file") expected_dir = os.path.join(tmpdir, "MYPROG", "g1") assert os.path.isdir(expected_dir) assert os.path.isfile(os.path.join(expected_dir, "MYPROG_g1.json")) ``` - [ ] **Step 3: 运行测试** Run: `python -m pytest tests/test_output_writer.py -v` Expected: all PASS --- ### Task 10: main.py — CLI 入口与集成测试 **Files:** - Create: `D:\jcl-cobol-data-create\main.py` - Create: `D:\jcl-cobol-data-create\tests\test_integration.py` - [ ] **Step 1: 确认 agent/__init__.py 导出 generate 函数** 更新 `D:\jcl-cobol-data-create\agent\__init__.py`： ```python # agent/__init__.py __version__ = "0.1.0" from agent.input_parser import InputParser from agent.rule_loader import RuleLoader from agent.prompt_builder import PromptBuilder from agent.api_client import APIClient from agent.output_writer import OutputWriter from agent.models import ProgramMeta, FileInfo, CopyField, KeyInfo, TableColumn, TableInfo def generate(design_md: str, source_cbl: str, file_db_md: str, cpy_dir: str, db_md: str, output_dir: str = "output", api_key: str = "sk-6156cccdc9c14d949cf5bfc5afc67a03", api_model: str = "deepseek-v4-flash", rules_dir: str = "rules") -> dict: """生成测试数据的主入口函数。 Args: design_md: 詳細設計書 .md 路径 source_cbl: COBOL 源码 .cbl 路径 file_db_md: 文件/DB 构造 .md 路径 cpy_dir: COPYBOOK 存放目录路径 db_md: DB 定义书 .md 路径 output_dir: 输出目录 api_key: DeepSeek API key api_model: 模型名称 rules_dir: 规则目录路径 Returns: {"output_files": {"g1/json": "path", ...}, "program_id": "...", "groups": N} """ print(f"== 解析入力: {design_md}") # 1. 解析输入 parser = InputParser(design_md, source_cbl, file_db_md, cpy_dir, db_md) meta = parser.run() if meta.pgm_type == 'サブ': raise ValueError(f"程序 {meta.program_id} はサブプログラムです。主プログラムのみ処理対象です。") print(f" プログラムID: {meta.program_id}, パターン: {meta.pgm_pattern}, 入力タイプ: {meta.input_type}") # 2. 加载规则 loader = RuleLoader(rules_dir) rules_text, group_descriptions, group_count = loader.load(meta) print(f" ルール読み込み完了, グループ数: {group_count}") # 3. 构建 prompt builder = PromptBuilder() prompt = builder.build(meta, rules_text, group_descriptions, group_count) # 4. 调用 API client = APIClient(api_key=api_key, model=api_model) print(f" API呼び出し中...") result = client.generate(prompt) print(f" API応答受信") # 5. 写入输出 writer = OutputWriter(output_dir) output_files = writer.write(meta.program_id, result, meta.input_type) print(f" 出力完了: {len(output_files)} ファイル") return { "output_files": output_files, "program_id": meta.program_id, "groups": group_count, "input_type": meta.input_type, } ``` - [ ] **Step 2: 创建 main.py CLI 入口** ```python # main.py #!/usr/bin/env python """ COBOL テストデータ生成 Agent 使用方法: python main.py \\ --design "D:\\cobol-tna-system\\詳細設計書\\詳細設計書_ZAN04MAT.md" \\ --source "D:\\cobol-tna-system\\src\\ZAN04MAT.cbl" \\ --file-db-md "D:\\cobol-tna-system\\詳細設計書\\COPY句定義書.md" \\ --cpy "D:\\cobol-tna-system\\cpy" \\ --db-md "D:\\cobol-tna-system\\詳細設計書\\DB定義書.md" \\ --output "output" """ import argparse import os import sys from agent import generate DEFAULT_RULES_DIR = os.path.join(os.path.dirname(__file__), 'rules') def main(): parser = argparse.ArgumentParser( description='COBOLテストデータ生成Agent', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 例: python main.py --design 詳細設計書_ZAN04MAT.md --source ZAN04MAT.cbl \\ --cpy cpy/ --db-md DB定義書.md --output output/ """ ) parser.add_argument('--design', required=True, help='詳細設計書 .md のパス') parser.add_argument('--source', required=True, help='COBOL ソース .cbl のパス') parser.add_argument('--file-db-md', required=True, help='ファイル/DB 構造 .md のパス') parser.add_argument('--cpy', required=True, help='COPYBOOK 格納ディレクトリ') parser.add_argument('--db-md', required=True, help='DB 定義書 .md のパス') parser.add_argument('--output', default='output', help='出力ディレクトリ (デフォルト: output)') parser.add_argument('--api-key', default='sk-6156cccdc9c14d949cf5bfc5afc67a03', help='DeepSeek API Key') parser.add_argument('--model', default='deepseek-v4-flash', help='API モデル名') parser.add_argument('--rules', default=DEFAULT_RULES_DIR, help='ルール格納ディレクトリ (デフォルト: rules/)') args = parser.parse_args() # 验证输入文件存在 for name, path in [('--design', args.design), ('--source', args.source)]: if not os.path.exists(path): print(f"エラー: {name} のファイルが見つかりません: {path}", file=sys.stderr) sys.exit(1) result = generate( design_md=args.design, source_cbl=args.source, file_db_md=args.file_db_md, cpy_dir=args.cpy, db_md=args.db_md, output_dir=args.output, api_key=args.api_key, api_model=args.model, rules_dir=args.rules, ) print(f"\n== 完了 ==") print(f"プログラムID: {result['program_id']}") print(f"グループ数: {result['groups']}") print(f"入力タイプ: {result['input_type']}") print(f"出力ファイル:") for key, path in sorted(result['output_files'].items()): print(f" {key}: {path}") if __name__ == '__main__': main() ``` - [ ] **Step 3: 写集成测试（跳过实际 API 调用）** ```python # tests/test_integration.py import os import tempfile from unittest.mock import patch, MagicMock from agent import generate FIXTURE_DIR = os.path.join(os.path.dirname(__file__), 'test_data') @patch('agent.api_client.requests.post') def test_full_pipeline_mock_api(mock_post): """完整流水线测试，API 调用使用 mock。""" # Mock API response mock_response = MagicMock() mock_response.json.return_value = { 'choices': [{'message': {'content': '''{ "groups": { "g1": { "program": "ZAN04MAT", "records": [ { "input": { "R01INNFIL": { "R01-APPL-ID": "A0000001", "R01-EMP-ID": "00000101", "R01-APPL-DATE": "20260101", "R01-START-TIME": "0900", "R01-END-TIME": "1800", "R01-STATUS": "0", "R01-OVT-TYPE": "W", "R01-FILLER": "D000000000000000000000000000000000000000000001" } } } ] } } }'''}}] } mock_response.raise_for_status = MagicMock() mock_post.return_value = mock_response cpy_dir = FIXTURE_DIR output_dir = tempfile.mkdtemp() rules_dir = os.path.join(os.path.dirname(__file__), '..', 'rules') design_md = os.path.join(FIXTURE_DIR, '詳細設計書_ZAN04MAT.md') source_cbl = os.path.join(FIXTURE_DIR, '..', '..', '..', 'cobol-tna-system', 'src', 'ZAN04MAT.cbl') # 如果 source_cbl 不存在，创建一个最小副本 if not os.path.exists(source_cbl): source_cbl = 'dummy.cbl' with open(source_cbl, 'w', encoding='utf-8') as f: f.write("COPY ZAN01REC REPLACING ==(A)== BY ==R01==.\n") try: result = generate( design_md=design_md, source_cbl=source_cbl, file_db_md='dummy.md', cpy_dir=cpy_dir, db_md='dummy_db.md', output_dir=output_dir, api_key='test-key', rules_dir=rules_dir, ) assert result['program_id'] == 'ZAN04MAT' assert result['groups'] > 0 assert len(result['output_files']) > 0 # 验证输出文件存在 for path in result['output_files'].values(): assert os.path.exists(path), f"输出文件不存在: {path}" finally: if source_cbl == 'dummy.cbl': os.remove(source_cbl) ``` - [ ] **Step 4: 运行集成测试** Run: `python -m pytest tests/test_integration.py -v` Expected: PASS - [ ] **Step 5: 运行全部测试** Run: `python -m pytest tests/ -v` Expected: all tests PASS --- ### Task 11: 首次实际运行验证 - [ ] **Step 1: 用 ZAN04MAT 实际运行** ```powershell python main.py ` --design "D:\cobol-tna-system\詳細設計書\詳細設計書_ZAN04MAT.md" ` --source "D:\cobol-tna-system\src\ZAN04MAT.cbl" ` --file-db-md "D:\cobol-tna-system\詳細設計書\COPY句定義書.md" ` --cpy "D:\cobol-tna-system\cpy" ` --db-md "D:\cobol-tna-system\詳細設計書\DB定義書.md" ` --output "D:\jcl-cobol-data-create\output" ``` - [ ] **Step 2: 验证输出** 确认 `output/ZAN04MAT/` 下生成了 g1、g2、g3 三个文件夹，每个文件夹中有对应的 JSON 文件。 - [ ] **Step 3: 验证 JSON 格式** 检查 JSON 文件是否符合 `JSON格式说明v2.0.md` 规范。