tmm621/jcl-cobol-data-create
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

v1.0

Browse Source
This commit is contained in:
tmm621
2026-07-12 14:54:50 +08:00
commit a769e4ae58
99 changed files with 9278 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/superpowers/plans/2026-07-12-testdata-agent-plan.md
+2310
View File
@@ -0,0 +1,2310 @@
# 测试数据生成 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` 规范。