tmm621/jcl-cobol-data-create
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
master
jcl-cobol-data-create/docs/superpowers/plans/2026-07-12-testdata-agent-plan.md
T
tmm621 a769e4ae58 v1.0
2026-07-12 14:54:50 +08:00

2311 lines
76 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 测试数据生成 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` 规范。
Reference in New Issue View Git Blame Copy Permalink