Files

T

2026-07-12 14:54:50 +08:00

76 KiB

Raw Permalink Blame History

测试数据生成 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

# agent/__init__.py
__version__ = "0.1.0"

Step 2: 创建 requirements.txt

requests>=2.28.0

Step 3: 创建 tests 目录和 conftest.py

# tests/__init__.py
# （空文件）

# 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: 定义数据模型

# 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: 写测试验证数据模型可实例化

# 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 工具函数

# 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 的测试

# 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 设计书部分

# 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 设计书解析部分的测试

# 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：

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 文件复制到测试目录：

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 解析测试

# 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 方法中追加实现：

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

Copy-Item "D:\cobol-tna-system\詳細設計書\DB定義書.md" "D:\jcl-cobol-data-create\tests\test_data\DB定義書.md"

Step 3: 写 DB 解析测试

# 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

# 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 测试

# 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

# 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 测试

# 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

# 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）

# 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

# 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 测试

# 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：

# 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 入口

# 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 调用）

# 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 实际运行

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 规范。

76 KiB Raw Permalink Blame History Unescape Escape

测试数据生成 Agent 实现计划

Task 1: 项目初始化

Task 2: 数据模型定义

Task 3: InputParser — 详细设计书解析

Task 4: InputParser — COPYBOOK 解析

Task 5: InputParser — DB 定义解析

Task 6: RuleLoader — 规则匹配与特殊功能检测

Task 7: PromptBuilder — Prompt 组装

Task 8: APIClient — API 调用与重试

Task 9: OutputWriter — 输出写入

Task 10: main.py — CLI 入口与集成测试

Task 11: 首次实际运行验证

76 KiB

Raw Permalink Blame History