import os import re from typing import List, Tuple, Optional from agent.models import ProgramMeta PGM_PATTERN_MAP = { 'マッチング(1:1)': 'matching-1-1.md', 'マッチング(1:1)': 'matching-1-1.md', 'マッチング(1:N)': 'matching-1-n.md', 'マッチング(1:N)': 'matching-1-n.md', 'マッチング(1:N) + EVALUATE 4パターン分岐': 'matching-1-n.md', 'マッチング(N:1)': 'matching-n-1.md', 'マッチング(N:1)': 'matching-n-1.md', 'マッチング(M:N)': 'matching-m-n.md', 'マッチングM:N⇒出力M件': 'matching-m-n.md', 'マッチングM:N⇒出力N件': 'matching-m-n.md', 'マッチングM:N⇒出力M×N件': 'matching-m-n.md', '18(M:N→M件マッチング)': 'matching-m-n.md', '19(M:N→N件マッチング+集計)': 'matching-m-n.md', '20(M:N→M×N件直積出力)': 'matching-m-n.md', 'レイアウト編集のみ(GETPUT)': 'layout-edit-getput.md', 'レイアウト編集のみ(GETPUT)': 'layout-edit-getput.md', 'レイアウト編集のみ': 'layout-edit-getput.md', 'GETPUT(編集出力)': 'getput-edit-output.md', '項目チェック': 'item-check.md', '項目チェック(重複なし)': 'item-check.md', '項目チェック(重複なし)': 'item-check.md', '項目チェック(重複含まず)': 'item-check.md', '項目チェック(重複含む)': 'item-check.md', '項目チェック(半角20桁/4桁)': 'item-check-halfwidth.md', '27(半角20桁/4桁チェック)': 'item-check-halfwidth.md', '半角20桁/4桁': 'item-check-halfwidth.md', '振り分け': 'dispatch.md', '振り分け(IF文)': 'dispatch.md', '振り分け(IF文、EVALUATE文)': 'dispatch.md', '振り分け(EVALUATE文)': 'dispatch-evaluate.md', '振り分け(EVALUATE文)': 'dispatch-evaluate.md', 'キーブレイク': 'keybreak-sum.md', 'キーブレイク(集計)': 'keybreak-sum.md', 'キーブレイク(集計)': 'keybreak-sum.md', 'キーブレイク(集約)': 'keybreak-collect.md', 'キーブレイク(集約)': 'keybreak-collect.md', 'キーブレイク(集計、集約)': 'keybreak-sum.md', 'キーブレイク(集計、集約の以外)': 'keybreak-aggregation.md', 'キーブレイク(集計集約以外)': 'keybreak-aggregation.md', 'キーブレイク(集計集約以外)': 'keybreak-aggregation.md', 'キーブレイク(非集計)': 'keybreak-aggregation.md', '1:Nキーブレイク(同キー集約)': 'keybreak-sum.md', '1:N+キーブレイク(同キー)': 'keybreak-sum.md', '1:N+同キーキーブレイク': 'keybreak-sum.md', '32(1:N+同キーキーブレイク)': 'keybreak-sum.md', '1:N+同キー': 'keybreak-sum.md', '1:N+キーブレイク(異キー)': 'keybreak-diff-key.md', '1:N+キーブレイク(異キー)': 'keybreak-diff-key.md', '1:N+異キーキーブレイク': 'keybreak-diff-key.md', '33(1:N+異キーキーブレイク)': 'keybreak-diff-key.md', '1:N+異キー': 'keybreak-diff-key.md', 'DB更新': 'db-update.md', 'DB更新 + SYSIN読込(P28)': 'db-update.md', 'SELECT処理': 'select-process.md', 'SELECT条件': 'select-process.md', 'SELECT条件': 'select-process.md', 'DB検索': 'select-process.md', 'DB検索': 'select-process.md', '50分割': 'split-50.md', '50分割': 'split-50.md', '25分割': 'split-25.md', '25分割': 'split-25.md', '11(25分割)': 'split-25.md', '100分割': 'split-100.md', '100分割': 'split-100.md', '12(100分割)': 'split-100.md', 'MERGE(複数ファイル結合)': 'merge.md', 'MERGE(複数ファイル結合)': 'merge.md', 'CSV→FB変換(改行あり)': 'csv-to-fb.md', 'CSV→FB変換(改行あり)': 'csv-to-fb.md', 'CSV→FB変換(改行なし)': 'csv-to-fb.md', 'CSV→FB変換(改行なし)': 'csv-to-fb.md', '2段階マッチング(1:1⇒1:1)': 'two-stage-matching-1-1.md', '2段階マッチング(1:1⇒1:1)': 'two-stage-matching-1-1.md', '2段階マッチング(N:1⇒N:1)': 'two-stage-matching-n-1.md', '2段階マッチング(N:1⇒N:1)': 'two-stage-matching-n-1.md', '2段階マッチング(M:N⇒M:N)': 'two-stage-matching-m-n.md', '2段階マッチング(M:N⇒M:N)': 'two-stage-matching-m-n.md', 'マッチング(1:1→1:1 2段階)': 'two-stage-matching-1-1.md', 'マッチング(N:1→N:1 2段階)': 'two-stage-matching-n-1.md', 'マッチング(M:N→M:N 2段階)': 'two-stage-matching-m-n.md', '内部テーブル検索': 'internal-table-search.md', '24(内部テーブル検索)': 'internal-table-search.md', 'サブプログラム使用': 'subprogram.md', 'サブプログラム': 'subprogram.md', 'サブPGM': 'subprogram.md', '25(サブPGM)': 'subprogram.md', 'SYSIN読込': 'sysin-read.md', 'SYSIN読込': 'sysin-read.md', '28(SYSIN読込)': 'sysin-read.md', 'ASCII→EBCDIC変換': 'ascii-to-ebcdic.md', 'ASCII→EBCDIC変換': 'ascii-to-ebcdic.md', '29(ASCII→EBCDIC変換)': 'ascii-to-ebcdic.md', 'SORT(INPUT/OUTPUT PROCEDURE)': 'sort.md', '34(SORT INPUT/OUTPUT PROCEDURE)': 'sort.md', 'SORT': 'sort.md', 'オンラインPGM': 'online-pgm.md', 'オンラインPGM': 'online-pgm.md', 'オンライン': 'online-pgm.md', '編集出力(ランキング生成)': 'ranking.md', '36(ランキング生成)': 'ranking.md', 'ランキング生成': 'ranking.md', '編集出力(印刷制御)': 'getput-edit-output.md', '編集出力': 'getput-edit-output.md', } SPECIAL_FEATURE_CHECKS = [ (['場合', 'EVALUATE', 'IF'], '条件分支.md'), ] class RuleLoader: """根据程序特征加载对应的数据生成规则。""" def __init__(self, rules_dir: str): self.pgm_pattern_dir = os.path.join(rules_dir, 'pgm_pattern') self.special_feature_dir = os.path.join(rules_dir, 'special_feature') self.last_rule_file = None def load(self, meta: ProgramMeta) -> Tuple[str, List[str], int]: """加载所有相关规则,返回 (合并后的规则文本, 组描述列表, 组数)。""" 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 ファイルを追加してください。" ) group_descriptions, group_count = self._parse_group_info(pgm_rule) 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): self.last_rule_file = filename return self._read_rule_file(path) if os.path.isdir(self.pgm_pattern_dir): available = sorted(os.listdir(self.pgm_pattern_dir)) # 对每个规则文件名(去掉.md),检查它是否出现在 PGMパターン 中 # 半角/全角括弧に加え、'1:N' vs '1-N' のコロン/ハイフン差異も吸収する pgm_lower = pgm_pattern.lower().replace(':', '-') for fname in available: if fname.endswith('.md'): rule_name = fname[:-3].lower() if rule_name in pgm_lower: self.last_rule_file = 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]: """从规则文本中解析组信息。 只统计表头首列为"组"的表格(生成的数据组), 避免把记录明细表中的数字(如 record序号 / 第1条)误算成组号。 """ descriptions = [] group_count = 0 in_group_table = False for line in rule_text.split('\n'): stripped = line.strip() if not stripped.startswith('|'): in_group_table = False continue # 分隔行(|---|)跳过 if re.match(r'^\|[\s\-:]+\|', stripped): continue cells = [c.strip() for c in stripped.split('|')[1:-1]] if not cells: continue # 表头行:首列为"组"时进入组表格 if cells[0] == '组': in_group_table = True continue if not in_group_table: continue 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