"""式样书解析器 — 读取详细设计书 .md，提取 ProgramMeta 信息。 从外部 agent（jcl-cobol-data-create）移植，适配 V3 接口。 """ import re from dataclasses import dataclass, field from pathlib import Path from typing import Optional @dataclass class FileInfo: no: int = 0 file_db_name: str = "" identifier: str = "" dd_name: str = "" io: str = "" copy_group: str = "" record_format: str = "" record_len: int = 0 medium: str = "" remarks: str = "" @dataclass class KeyInfo: no: int = 0 file_name: str = "" sort_condition: str = "" key_condition: str = "" @dataclass class ModuleInfo: no: int = 0 function: str = "" program_id: str = "" copy_name: str = "" @dataclass class CopyField: level: int = 0 name: str = "" raw_name: str = "" pic_type: str = "" pic_bytes: int = 0 @dataclass class TableColumn: no: int = 0 name_jp: str = "" name_en: str = "" type: str = "" max_len: str = "" decimal_digits: str = "" byte_count: str = "" nullable: bool = True is_pk: bool = False @dataclass class TableInfo: table_name: str = "" db_id: str = "" copy_id: str = "" columns: list = field(default_factory=list) pk_columns: list = field(default_factory=list) @dataclass class ProgramMeta: program_id: str = "" program_name: str = "" system_name: str = "" pgm_type: str = "" pgm_pattern: str = "" summary_lines: list = field(default_factory=list) prerequisites: list = field(default_factory=list) files: list = field(default_factory=list) keys: list = field(default_factory=list) modules: list = field(default_factory=list) process_detail: str = "" output_records: str = "" input_type: str = "file" copy_fields: dict = field(default_factory=dict) db_tables: dict = field(default_factory=dict) def _extract_section(text: str, section_name: str) -> str: """Extract a section by name from markdown text (case-insensitive).""" lines = text.split("\n") result = [] in_section = False section_pattern = re.compile( rf"^#+\s*{re.escape(section_name)}\s*$", re.IGNORECASE ) next_section = re.compile(r"^#+\s", re.IGNORECASE) for line in lines: if section_pattern.match(line): in_section = True continue if in_section and next_section.match(line): break if in_section: result.append(line) return "\n".join(result).strip() def _parse_table_rows(text: str) -> list[dict]: """Parse a markdown table into list of dicts.""" lines = [l.strip() for l in text.split("\n") if l.strip()] if not lines: return [] header_line = None sep_line = None data_start = 0 for i, line in enumerate(lines): if line.startswith("|") and not header_line: header_line = line elif line.startswith("|") and header_line and not sep_line: if set(line.strip("|").replace("-", "").replace(" ", "").replace("|", "")) == set(): sep_line = line data_start = i + 1 break if not header_line: return [] headers = [h.strip() for h in header_line.strip("|").split("|")] rows = [] for line in lines[data_start:]: if not line.startswith("|"): continue cells = [c.strip() for c in line.strip("|").split("|")] row = {} for i, h in enumerate(headers): if i < len(cells): row[h] = cells[i] else: row[h] = "" rows.append(row) return rows class DesignDataInputParser: """解析式样书 .md，返回 ProgramMeta。""" def __init__(self): pass def parse(self, design_md_text: str, source_text: str) -> ProgramMeta: """解析式样书文本和源码文本，返回 ProgramMeta。""" meta = ProgramMeta() self._design_text = design_md_text self._source_text = source_text 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) return meta def _parse_basic_info(self, meta: ProgramMeta): rows = _parse_table_rows(_extract_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) def _parse_use_files(self, meta: ProgramMeta): rows = _parse_table_rows(_extract_section(self._design_text, "使用ファイル一覧")) for row in rows: try: no_str = row.get("NO", "0") no = int(no_str) if no_str and no_str.strip() not in ("", "—") else 0 rec_len_str = row.get("レコード長", "0") rec_len = ( int(rec_len_str) if rec_len_str and rec_len_str.strip() not in ("", "—") 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群", ""), record_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_rows(_extract_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_rows( _extract_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): 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"