"""orchestrator_jcl.py — jcl-cobol-data-create + cobol-java-v3 联合管道 从 jcl-cobol-data-create 生成 JSON 测试数据，转换为 COBOL 二进制平文件， 编译运行 COBOL/Java，逐 group 比对并生成报告。 用法: python main.py --mode agent \ --design "D:/cobol-tna-system/詳細設計書/詳細設計書_ZAN04MAT.md" \ --cobol-src "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" \ --java-src "D:/path/to/java" \ --mapping "D:/path/to/mapping.yaml" \ --output "output" """ import json import logging import os import re import shutil import subprocess import sys import time from pathlib import Path from cobol_testgen.flatfile import analyze_fd_layout, write_all_files from cobol_testgen.file_io import read_output_file from runners.cobol_runner import CobolRunner from runners.native_java_runner import NativeJavaRunner from runners.spark_java_runner import SparkJavaRunner from comparator import align_records, compare_field from data.diff_result import VerificationRun, FieldResult from config import Config JCL_ROOT = r"D:\jcl-cobol-data-create" if JCL_ROOT not in sys.path: sys.path.insert(0, JCL_ROOT) logger = logging.getLogger(__name__) DEFAULT_API_KEY = "sk-6156cccdc9c14d949cf5bfc5afc67a03" DEFAULT_API_MODEL = "deepseek-v4-flash" def run_jcl_pipeline( cfg: Config, design_md: str, cobol_src: str, file_db_md: str, cpy_dir: str, db_md: str, java_src: str = "", mapping: str = "", output_dir: str = "output", api_key: str = DEFAULT_API_KEY, api_model: str = DEFAULT_API_MODEL, ) -> dict: """执行 jcl agent 模式全管道。 Returns: {group_name: VerificationRun} 字典 """ t0 = time.time() results = {} # ── Phase 1: 调用 jcl-cobol-data-create 生成 JSON ── print("=" * 60) print("Phase 1: Generating test data via jcl-cobol-data-create...") print("=" * 60) from agent import generate as jcl_generate jcl_result = jcl_generate( design_md=design_md, source_cbl=cobol_src, file_db_md=file_db_md, cpy_dir=cpy_dir, db_md=db_md, output_dir=output_dir, api_key=api_key, api_model=api_model, ) program_id = jcl_result["program_id"] groups = jcl_result["groups"] input_type = jcl_result["input_type"] print(f" program_id={program_id} groups={groups} input_type={input_type}") if input_type in ("db", "mixed"): raise NotImplementedError( f"DB input type ({input_type}) not yet supported in agent mode" ) # ── Phase 2: 解析元数据 ── print("=" * 60) print("Phase 2: Parsing COBOL metadata...") print("=" * 60) src_text = _read_file(cobol_src) cpy_dirs = [cpy_dir, str(Path(cobol_src).parent)] layouts = analyze_fd_layout(src_text, copybook_dirs=cpy_dirs) input_fds = {n: i for n, i in layouts.items() if i["direction"] != "OUTPUT"} output_fds = {n: i for n, i in layouts.items() if i["direction"] == "OUTPUT"} print(f" {len(layouts)} FDs total: {list(layouts.keys())}") print(f" Input FDs: {list(input_fds.keys())}") print(f" Output FDs: {list(output_fds.keys())}") # ── Phase 3: 编译 COBOL + Java（一次）── print("=" * 60) print("Phase 3: Compiling...") print("=" * 60) cob_runner = CobolRunner() work_dir = (Path(output_dir) / program_id / "build").resolve() if work_dir.exists(): shutil.rmtree(str(work_dir)) work_dir.mkdir(parents=True, exist_ok=True) cob_build = cob_runner.compile_with_links( cobol_src, str(work_dir), copybook_dirs=cpy_dirs, ) if not cob_build.success: print(f" COBOL compile FAILED:\n{cob_build.log[-400:]}") return { program_id: VerificationRun( program=program_id, runner=cfg.runner_mode, status="BLOCKED", exit_code=2, ) } print(f" COBOL OK → {cob_build.artifact_path}") # 编译子程序（扫描 CALL 语句 + 默认 sub 目录） sub_dirs = [ Path(cobol_src).parent.parent / "sub", Path(cobol_src).parent / "sub", ] sub_bin_dirs = [ Path(cobol_src).parent.parent / "bin", Path(cobol_src).parent / "bin", ] _compile_subprograms(src_text, cpy_dirs, work_dir, sub_dirs) _sub_lib_paths = [str(work_dir)] java_runner = None # type: NativeJavaRunner | SparkJavaRunner | None java_build = None if java_src and Path(java_src).exists(): java_runner = ( SparkJavaRunner(cfg.spark_master) if cfg.runner_mode == "spark" else NativeJavaRunner() ) java_build = java_runner.compile(java_src) if not java_build.success: print(f" Java compile FAILED:\n{java_build.log[-400:]}") return { program_id: VerificationRun( program=program_id, runner=cfg.runner_mode, status="BLOCKED", exit_code=2, ) } print(f" Java OK") # ── Phase 4: 逐组循环 ── print("=" * 60) print("Phase 4: Running per-group pipeline...") print("=" * 60) prog_dir = Path(output_dir) / program_id group_dirs = sorted( [d for d in prog_dir.iterdir() if d.is_dir() and d.name.startswith("g")], key=lambda d: int(d.name[1:]) if d.name[1:].isdigit() else 0, ) for group_dir in group_dirs: group_name = group_dir.name print(f"\n── Group {group_name} ──") vr = VerificationRun(program=f"{program_id}/{group_name}", runner=cfg.runner_mode) try: # 4a. 读取 JSON，展平为 flat records json_files = list(group_dir.glob(f"{program_id}_*.json")) if not json_files: json_files = list(group_dir.glob("*.json")) if not json_files: print(f" SKIP: no JSON file found") continue json_path = json_files[0] with open(json_path, encoding="utf-8") as f: data = json.load(f) flat_records = [] for record in data.get("records", []): flat = {} inp = record.get("input", {}) for dd_name, fields in inp.items(): for fname, fval in fields.items(): # 归一化: jcl "R01-APPL-ID" → cobol "R01APPL-ID" flat[_normalize_field_name(fname)] = fval if flat: flat_records.append(flat) print(f" {len(flat_records)} records from {json_path.name}") if not flat_records: print(f" SKIP: empty records") vr.status = "BLOCKED" vr.exit_code = 2 results[group_name] = vr continue # 4b. 写入二进制平文件（通过 write_all_files 自动按 FD 拆分） run_dir = Path(output_dir) / program_id / "runtime" / group_name if run_dir.exists(): shutil.rmtree(str(run_dir)) run_dir.mkdir(parents=True, exist_ok=True) written = write_all_files( flat_records, src_text, run_dir, copybook_dirs=cpy_dirs, ) print(f" Wrote {len(written)} flat files: {[w[0] for w in written]}") # 4c. 运行 COBOL exe_path = cob_build.artifact_path print(f" Running COBOL...") try: env = os.environ.copy() env["COB_LIBRARY_PATH"] = os.pathsep.join(_sub_lib_paths) p = subprocess.run( [exe_path], capture_output=True, text=True, timeout=120, cwd=str(run_dir), env=env, ) cobol_log = (p.stdout or "") + "\n" + (p.stderr or "") cobol_ok = (p.returncode == 0) log_dir = run_dir / "logs" log_dir.mkdir(exist_ok=True) (log_dir / "cobol_run.log").write_text(cobol_log, encoding="utf-8") if not cobol_ok: print(f" COBOL FAILED (rc={p.returncode})") # Check if output files were produced despite non-zero rc any_output = any( (run_dir / fd_name).exists() for fd_name in output_fds ) if not any_output: vr.status = "ERROR" vr.exit_code = 3 results[group_name] = vr continue print(f" Output files exist despite non-zero rc, continuing...") else: print(f" COBOL OK (rc=0)") except subprocess.TimeoutExpired: print(f" COBOL TIMEOUT") vr.status = "ERROR" vr.exit_code = 3 results[group_name] = vr continue # 4d. 读取 COBOL 输出文件 cobol_output_records = [] for fd_name, layout in output_fds.items(): out_file = run_dir / fd_name if not out_file.exists(): continue if not layout.get("records"): continue # 取最长 record 的字段定义作为 fdict best_rec = max( layout["records"], key=lambda r: (len(r.get("fields", [])), r.get("record_length", 0)), ) fd_field_dicts = best_rec.get("fields", []) if not fd_field_dicts: continue try: recs = read_output_file(str(out_file), fd_field_dicts) cobol_output_records.extend(recs) print(f" {fd_name}: {len(recs)} records") except Exception as e: print(f" {fd_name}: read error - {e}") logger.warning(f"read_output_file {fd_name}: {e}", exc_info=True) print(f" Total COBOL output records: {len(cobol_output_records)}") # 4e. 运行 Java（如果可用） java_output_records = [] if java_runner is not None and java_build is not None and java_build.success: java_input = run_dir / "java_input.json" with open(java_input, "w", encoding="utf-8") as f: for rec in flat_records: f.write(json.dumps(rec, ensure_ascii=False) + "\n") java_out_dir = run_dir / "java_output" java_out_dir.mkdir(exist_ok=True) try: jr = java_runner.run( java_build.artifact_path, str(java_input), str(java_out_dir), ) if jr.success and jr.records: java_output_records = jr.records print(f" Java OK: {len(java_output_records)} records") else: print(f" Java: no output records (success={jr.success})") except Exception as e: print(f" Java error: {e}") logger.warning("Java run failed", exc_info=True) # 4f. 比对 if cobol_output_records and java_output_records: print(f" Comparing: COBOL={len(cobol_output_records)} vs Java={len(java_output_records)}") aligned = align_records( cobol_output_records, java_output_records, key_field="CUST-ID" ) frs = [] for c, j, st in aligned: if st == "MISSING_IN_SPARK": frs.append(FieldResult(field_name="unknown", status="MISSING_IN_SPARK")) continue if st == "EXTRA_IN_SPARK": frs.append(FieldResult(field_name="unknown", status="EXTRA")) continue for key in c: if key == "CUST-ID": continue cv = str(c.get(key, "")) jv = str(j.get(key, "") if j else "") frs.append(compare_field(key, cv, jv, "string", cfg.tolerance)) matched = sum(1 for f in frs if f.status in ("PASS", "TOLERATED")) mismatched = sum(1 for f in frs if f.status not in ("PASS", "TOLERATED")) vr.fields_matched = matched vr.fields_mismatched = mismatched vr.field_results = frs vr.status = "PASS" if mismatched == 0 else "MISMATCH" vr.exit_code = 0 if mismatched == 0 else 1 print(f" Result: {matched} matched, {mismatched} mismatched") elif cobol_output_records: vr.fields_matched = len(cobol_output_records) vr.status = "PASS" vr.exit_code = 0 print(f" COBOL-only: {len(cobol_output_records)} output records") else: vr.status = "PASS" vr.exit_code = 0 print(f" No output to compare") # 4g. 生成报告 rd = Path("reports") / program_id / group_name rd.mkdir(parents=True, exist_ok=True) from report import ReportGenerator rg = ReportGenerator() rg.generate_json(vr, rd / "result.json") rg.generate_html(vr, rd / "report.html") rg.generate_machine_json(vr, rd / "machine.json") vr.report_path = str(rd) print(f" Report: {rd}") except Exception as e: logger.exception(f"Group {group_name} failed") vr.status = "ERROR" vr.exit_code = 3 vr.report_path = str(e)[:200] vr.duration_s = time.time() - t0 results[group_name] = vr # ── Summary ── print("\n" + "=" * 60) print("Pipeline complete") print("=" * 60) for gn, vr in results.items(): t = vr.fields_matched + vr.fields_mismatched if t: print(f" {gn}: {vr.status} ({vr.fields_matched}/{t}, {vr.duration_s:.0f}s)") else: print(f" {gn}: {vr.status} ({vr.duration_s:.0f}s)") return results def _compile_subprograms(src_text: str, copybook_dirs: list, work_dir: Path, sub_dirs: list): """扫描 COBOL 源码中的 CALL 语句，编译对应的子程序为 DLL。 Args: src_text: COBOL 源码文本 copybook_dirs: COPYBOOK 搜索路径 work_dir: 编译产物输出目录 sub_dirs: 子程序源码目录列表（按优先级搜索） """ called = set() for m in re.finditer(r"CALL\s+['\"]([^'\"]+)['\"]", src_text, re.IGNORECASE): called.add(m.group(1)) if not called: return print(f" Subprograms to compile: {sorted(called)}") for sub_name in sorted(called): # 查找子程序源码 sub_path = None for sd in sub_dirs: p = sd / f"{sub_name}.cbl" if p.exists(): sub_path = p break if sub_path is None: print(f" {sub_name}: source not found in {[str(d) for d in sub_dirs]}") continue dll_path = work_dir / f"{sub_name}.dll" cmd = ["cobc", "-m", "-o", f"{sub_name}.dll", str(sub_path)] for cpy in copybook_dirs: cmd.extend(["-I", str(cpy)]) try: p = subprocess.run( cmd, capture_output=True, text=True, timeout=60, cwd=str(work_dir), ) if p.returncode == 0: print(f" {sub_name}: OK") else: err = (p.stdout + p.stderr)[:200] print(f" {sub_name}: FAIL - {err}") except Exception as e: print(f" {sub_name}: ERROR - {e}") def _normalize_field_name(name: str) -> str: """将 jcl 格式的字段名归一化为 cobol FD layout 格式。 jcl: R01-APPL-ID (prefix + '-' + field body) cobol: R01APPL-ID (prefix + field body, 无额外 hyphen) 规则: 如果字段名以 2-4 位大写字母/数字开头后紧跟 '-'， 则移除该 '-'。如 R01-APPL-ID → R01APPL-ID。 """ m = re.match(r"^([A-Z0-9]{2,4})-", name) if m: prefix = m.group(1) return prefix + name[len(prefix) + 1:] return name def _read_file(path: str) -> str: p = Path(path) for enc in ("utf-8-sig", "utf-8", "shift_jis", "cp932"): try: return p.read_text(encoding=enc) except (UnicodeDecodeError, UnicodeError): continue return p.read_text(encoding="utf-8", errors="replace")