331 lines
12 KiB
Python
331 lines
12 KiB
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
|
|
|
|
def _parse_copybooks(self, meta: ProgramMeta):
|
|
replacing_list = self._extract_copy_replacing()
|
|
|
|
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
|
|
|
|
prefix = f.identifier
|
|
for i, (cpy_name, rep_val) in enumerate(replacing_list):
|
|
if cpy_name == f.copy_group and rep_val == f.identifier:
|
|
prefix = rep_val
|
|
replacing_list.pop(i)
|
|
break
|
|
else:
|
|
for i, (cpy_name, rep_val) in enumerate(replacing_list):
|
|
if cpy_name == f.copy_group:
|
|
prefix = rep_val
|
|
replacing_list.pop(i)
|
|
break
|
|
|
|
fields = self._parse_single_copybook(copy_file, prefix)
|
|
if f.identifier not in meta.copy_fields:
|
|
meta.copy_fields[f.identifier] = fields
|
|
|
|
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)
|
|
|
|
section_with_header = '## ' + section
|
|
|
|
db_info_text = extract_section(section_with_header, '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_with_header, 'カラム定義')
|
|
column_rows = parse_table_rows(column_text)
|
|
|
|
if not column_rows:
|
|
continue
|
|
|
|
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
|
|
|
|
def _extract_copy_replacing(self) -> List[Tuple[str, str]]:
|
|
"""从 COBOL 源码提取 COPY ... REPLACING ... 映射。
|
|
Returns: [(copy_name, replacing_value), ...] 按出现顺序保存。
|
|
"""
|
|
result = []
|
|
pattern = r'COPY\s+(\S+)\s+REPLACING\s+==\(A\)==\s+BY\s+==(\S+)=='
|
|
for match in re.finditer(pattern, self._source_text, re.IGNORECASE):
|
|
result.append((match.group(1), match.group(2)))
|
|
return result
|
|
|
|
def _parse_single_copybook(self, copy_path: str, prefix: str) -> List[CopyField]:
|
|
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
|
|
|
|
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:
|
|
pic_upper = pic.upper().strip()
|
|
|
|
if 'COMP-3' in pic_upper:
|
|
base = pic_upper.replace('COMP-3', '').strip()
|
|
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
|
|
|
|
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
|
|
|
|
total = 0
|
|
for num in re.findall(r'X\((\d+)\)', pic_upper):
|
|
total += int(num)
|
|
for num in re.findall(r'G\((\d+)\)', pic_upper):
|
|
total += int(num) * 2
|
|
for num in re.findall(r'9\((\d+)\)', pic_upper):
|
|
total += int(num)
|
|
if total > 0:
|
|
return total
|
|
|
|
return 0
|
|
|
|
@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_str = row.get('NO', '0')
|
|
try:
|
|
no = int(no_str) if no_str and no_str.strip() not in ('', '—') else 0
|
|
except ValueError:
|
|
no = 0
|
|
rec_len_str = row.get('レコード長', '0')
|
|
try:
|
|
rec_len = int(rec_len_str) if rec_len_str and rec_len_str.strip() not in ('', '—') else 0
|
|
except ValueError:
|
|
rec_len = 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):
|
|
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'
|