271 lines
8.1 KiB
Python
271 lines
8.1 KiB
Python
"""式样书解析器 — 读取详细设计书 .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"
|