#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ COBOL / Copybook Impact Analyzer V2 A standalone modernization analyzer for COBOL ecosystems: - COBOL programs - Copybooks - Data Division fields - File Section, Working-Storage, Linkage Section - COPY dependencies - DB2 embedded SQL - CICS commands and COMMAREA - Procedure Division call graph - JSON Schema and OpenAPI drafts - Impact, risk and modernization findings No external dependencies required. """ from __future__ import annotations import argparse import csv import html import json import os import re import sys from collections import Counter, defaultdict from dataclasses import dataclass, asdict, field from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple SEVERITY_ORDER = {"INFO": 0, "WARNING": 1, "ERROR": 2, "CRITICAL": 3} COBOL_EXTENSIONS = {".cbl", ".cob", ".cobol", ".pgm"} COPYBOOK_EXTENSIONS = {".cpy", ".copy", ".cobcpy", ".inc"} ALL_EXTENSIONS = COBOL_EXTENSIONS | COPYBOOK_EXTENSIONS | {".txt"} DEFAULT_POLICY = { "sensitive_name_patterns": [ "SSN", "SOCIAL", "PASSWORD", "PASSWD", "PIN", "CARD", "PAN", "CREDIT", "IBAN", "ACCOUNT", "SALARY", "PAYROLL", "DOB" ], "high_risk_program_calls": ["IKJEFT01", "IDCAMS", "DSNUTILB"], "banned_statements": ["ALTER"], "warning_statements": ["GO TO"], "api_candidate_keywords": ["REQUEST", "RESPONSE", "COMMAREA", "INPUT", "OUTPUT"], "copybook_root_candidates": ["REQUEST", "RESPONSE", "COMMAREA", "CUSTOMER", "ACCOUNT"], "max_commarea_bytes": 32767, "max_record_bytes_warning": 4096, "max_occurs_warning": 1000, "max_copy_depth_warning": 5, "naming_regex": r"^[A-Z0-9][A-Z0-9-]{1,29}$", } @dataclass class SourceFile: path: str name: str kind: str logical_name: str line_count: int raw_text: str @dataclass class FieldDef: source: str section: str level: int name: str parent: str path: str pic: str = "" usage: str = "" occurs: Optional[int] = None occurs_depending_on: str = "" redefines: str = "" value: str = "" line_no: int = 0 byte_length: Optional[int] = None decimal_digits: Optional[int] = None signed: bool = False json_type: str = "string" is_group: bool = False raw_line: str = "" @dataclass class ProgramInfo: source: str program_id: str = "" divisions: List[str] = field(default_factory=list) files: List[str] = field(default_factory=list) fd_records: List[str] = field(default_factory=list) copybooks: List[str] = field(default_factory=list) copybooks_missing: List[str] = field(default_factory=list) called_programs: List[str] = field(default_factory=list) performed_paragraphs: List[str] = field(default_factory=list) paragraphs: List[str] = field(default_factory=list) goto_targets: List[str] = field(default_factory=list) statements: Counter = field(default_factory=Counter) @dataclass class CopybookInfo: source: str name: str fields: int = 0 root_fields: List[str] = field(default_factory=list) used_by_programs: List[str] = field(default_factory=list) estimated_bytes: int = 0 has_redefines: bool = False has_occurs: bool = False has_occurs_depending: bool = False @dataclass class Dependency: source: str target: str kind: str detail: str = "" line_no: int = 0 @dataclass class DB2Reference: source: str statement: str table: str cursor: str = "" line_no: int = 0 raw_text: str = "" @dataclass class CICSReference: source: str command: str program: str = "" transid: str = "" commarea: str = "" length: str = "" line_no: int = 0 raw_text: str = "" @dataclass class Finding: severity: str category: str code: str source: str line_no: int title: str detail: str recommendation: str @dataclass class AnalysisResult: status: str profile: str risk_score: int highest_severity: str files: List[SourceFile] programs: List[ProgramInfo] copybooks: List[CopybookInfo] fields: List[FieldDef] dependencies: List[Dependency] db2: List[DB2Reference] cics: List[CICSReference] findings: List[Finding] stats: Dict[str, Any] json_schema: Dict[str, Any] openapi: Dict[str, Any] mermaid: str compare_rows: List[Dict[str, str]] = field(default_factory=list) def read_text(path: Path) -> str: for encoding in ("utf-8", "cp037", "latin-1"): try: return path.read_text(encoding=encoding, errors="ignore") except Exception: continue return path.read_bytes().decode("utf-8", errors="ignore") def normalize_logical_name(path: Path) -> str: name = path.stem.upper() name = re.sub(r"[^A-Z0-9_-]+", "-", name) return name def classify_file(path: Path, text: str) -> str: ext = path.suffix.lower() upper = text.upper() if ext in COPYBOOK_EXTENSIONS: return "copybook" if ext in COBOL_EXTENSIONS: return "program" if "IDENTIFICATION DIVISION" in upper or "PROGRAM-ID" in upper or "PROCEDURE DIVISION" in upper: return "program" if re.search(r"^\s*\d{2}\s+[A-Z0-9-]+", upper, re.MULTILINE): return "copybook" return "text" def collect_input_files(paths: Sequence[str], input_dirs: Sequence[str]) -> List[Path]: result: List[Path] = [] seen = set() def add_file(p: Path) -> None: if not p.exists() or not p.is_file(): return key = str(p.resolve()) if key in seen: return if p.suffix.lower() in ALL_EXTENSIONS: seen.add(key) result.append(p) for raw in paths: p = Path(raw) if p.is_file(): add_file(p) elif p.is_dir(): for child in sorted(p.rglob("*")): add_file(child) for raw in input_dirs: p = Path(raw) if p.is_dir(): for child in sorted(p.rglob("*")): add_file(child) return result def strip_sequence_area(line: str) -> str: raw = line.rstrip("\n") if len(raw) >= 7: marker = raw[6:7] if marker in ("*", "/", "D"): return "" if raw[:6].strip().isdigit(): return raw[6:].rstrip() if raw.lstrip().startswith("*"): return "" return raw.rstrip() def logical_lines(text: str) -> List[Tuple[int, str]]: out: List[Tuple[int, str]] = [] buffer = "" start_line = 0 for idx, line in enumerate(text.splitlines(), 1): stripped = strip_sequence_area(line) if not stripped.strip(): continue # Drop inline comment markers common in samples. if stripped.strip().startswith("*>"): continue cont = False if len(line) >= 7 and line[6:7] == "-": cont = True if cont and buffer: buffer += " " + stripped.strip() else: if buffer: out.append((start_line, buffer.strip())) buffer = stripped.strip() start_line = idx if buffer.endswith("."): out.append((start_line, buffer.strip())) buffer = "" start_line = 0 if buffer: out.append((start_line, buffer.strip())) return out def sentence_blocks(lines: List[Tuple[int, str]]) -> List[Tuple[int, str]]: blocks: List[Tuple[int, str]] = [] buf = "" start = 0 for line_no, text in lines: if not buf: start = line_no buf += " " + text.strip() if text.strip().endswith(".") or text.upper().strip().startswith("EXEC SQL") or text.upper().strip().startswith("EXEC CICS"): blocks.append((start, buf.strip())) buf = "" start = 0 if buf: blocks.append((start, buf.strip())) return blocks def extract_exec_blocks(lines: List[Tuple[int, str]], family: str) -> List[Tuple[int, str]]: """Extract multi-line EXEC SQL / EXEC CICS blocks.""" marker = f"EXEC {family.upper()}" blocks: List[Tuple[int, str]] = [] active = False start = 0 buf: List[str] = [] for line_no, text in lines: upper = text.upper() if not active and marker in upper: active = True start = line_no buf = [text.strip()] if "END-EXEC" in upper: blocks.append((start, " ".join(buf))) active = False buf = [] continue if active: buf.append(text.strip()) if "END-EXEC" in upper: blocks.append((start, " ".join(buf))) active = False buf = [] if active and buf: blocks.append((start, " ".join(buf))) return blocks def normalize_ws(text: str) -> str: return re.sub(r"\s+", " ", text.strip()) def load_policy(path: Optional[str]) -> Dict[str, Any]: policy = dict(DEFAULT_POLICY) if not path: return policy with open(path, "r", encoding="utf-8") as fh: custom = json.load(fh) for key, value in custom.items(): if isinstance(value, list) and isinstance(policy.get(key), list): policy[key] = list(policy[key]) + value else: policy[key] = value return policy def parse_pic(pic: str, usage: str) -> Tuple[Optional[int], Optional[int], bool, str]: if not pic: return None, None, False, "object" p = pic.upper().replace(" ", "") signed = p.startswith("S") or "SIGN" in usage.upper() decimal_digits = 0 def count_symbol(symbol: str, value: str) -> int: total = 0 idx = 0 while idx < len(value): ch = value[idx] if ch == symbol: m = re.match(rf"{symbol}\((\d+)\)", value[idx:]) if m: total += int(m.group(1)) idx += len(m.group(0)) continue total += 1 idx += 1 return total x_count = count_symbol("X", p) + count_symbol("A", p) n_count = count_symbol("9", p) if "V" in p: after_v = p.split("V", 1)[1] decimal_digits = count_symbol("9", after_v) if x_count > 0 and n_count == 0: return x_count, None, signed, "string" if n_count > 0: usage_u = usage.upper() if "COMP-3" in usage_u or "PACKED-DECIMAL" in usage_u: byte_length = int((n_count + 2) / 2) elif "COMP" in usage_u or "BINARY" in usage_u: if n_count <= 4: byte_length = 2 elif n_count <= 9: byte_length = 4 else: byte_length = 8 else: byte_length = n_count return byte_length, decimal_digits, signed, "number" if decimal_digits else "integer" return None, None, signed, "string" def extract_quoted_literals(text: str) -> List[str]: return re.findall(r"['\"]([^'\"]+)['\"]", text) class CobolAnalyzer: def __init__(self, files: List[Path], profile: str, policy: Dict[str, Any], redact: bool = False): self.paths = files self.profile = profile self.policy = policy self.redact = redact self.sources: List[SourceFile] = [] self.programs: Dict[str, ProgramInfo] = {} self.copybooks: Dict[str, CopybookInfo] = {} self.fields: List[FieldDef] = [] self.dependencies: List[Dependency] = [] self.db2: List[DB2Reference] = [] self.cics: List[CICSReference] = [] self.findings: List[Finding] = [] self.source_by_logical: Dict[str, SourceFile] = {} self.copybook_names_available: set[str] = set() def analyze(self, compare_with: Optional[str] = None) -> AnalysisResult: self._load_sources() self._parse_all_sources() self._resolve_copybook_usage() self._run_rules() compare_rows = self._compare_copybook_or_program(compare_with) if compare_with else [] if compare_rows: for row in compare_rows: if row["change"] in ("REMOVED_FIELD", "ADDED_FIELD"): self.findings.append(Finding( severity="WARNING", category="COMPARE", code=row["change"], source=row["source"], line_no=0, title=row["change"].replace("_", " ").title(), detail=f"{row['field']} changed in comparison source.", recommendation="Review copybook compatibility and downstream consumers." )) stats = self._build_stats(compare_rows) risk_score, highest, status = self._score() json_schema = self._build_json_schema() openapi = self._build_openapi(json_schema) mermaid = self._build_mermaid() return AnalysisResult( status=status, profile=self.profile, risk_score=risk_score, highest_severity=highest, files=self.sources, programs=list(self.programs.values()), copybooks=list(self.copybooks.values()), fields=self.fields, dependencies=self.dependencies, db2=self.db2, cics=self.cics, findings=self.findings, stats=stats, json_schema=json_schema, openapi=openapi, mermaid=mermaid, compare_rows=compare_rows, ) def _load_sources(self) -> None: for path in self.paths: text = read_text(path) kind = classify_file(path, text) source = SourceFile( path=str(path), name=path.name, kind=kind, logical_name=normalize_logical_name(path), line_count=len(text.splitlines()), raw_text=text, ) self.sources.append(source) self.source_by_logical[source.logical_name] = source if kind == "copybook": self.copybook_names_available.add(source.logical_name) def _parse_all_sources(self) -> None: for source in self.sources: self._parse_source(source) def _parse_source(self, source: SourceFile) -> None: lines = logical_lines(source.raw_text) blocks = sentence_blocks(lines) blocks.extend(extract_exec_blocks(lines, "SQL")) blocks.extend(extract_exec_blocks(lines, "CICS")) if source.kind == "program": program = self._parse_program_skeleton(source, lines, blocks) self.programs[source.logical_name] = program elif source.kind == "copybook": self.copybooks[source.logical_name] = CopybookInfo(source=source.name, name=source.logical_name) self._parse_data_fields(source, lines) self._parse_sql_blocks(source, blocks) self._parse_cics_blocks(source, blocks) self._parse_dependencies(source, lines, blocks) def _parse_program_skeleton(self, source: SourceFile, lines: List[Tuple[int, str]], blocks: List[Tuple[int, str]]) -> ProgramInfo: program = ProgramInfo(source=source.name) text_u = source.raw_text.upper() m = re.search(r"PROGRAM-ID\.\s*([A-Z0-9_-]+)", text_u) if m: program.program_id = m.group(1) else: program.program_id = source.logical_name for _, line in lines: u = line.upper() for div in ("IDENTIFICATION DIVISION", "ENVIRONMENT DIVISION", "DATA DIVISION", "PROCEDURE DIVISION"): if div in u: program.divisions.append(div) m = re.match(r"^([A-Z0-9-]+)\.\s*$", u) if m and " DIVISION" not in u and " SECTION" not in u: program.paragraphs.append(m.group(1)) for copy in re.findall(r"\bCOPY\s+([A-Z0-9_-]+)", u): program.copybooks.append(copy.replace("_", "-")) sel = re.search(r"\bSELECT\s+([A-Z0-9-]+)\s+ASSIGN\s+TO\s+([A-Z0-9'\"._-]+)", u) if sel: program.files.append(sel.group(1)) self.dependencies.append(Dependency(source=source.name, target=sel.group(2).strip("'\""), kind="FILE_ASSIGN", detail=sel.group(1), line_no=_)) fd = re.match(r"^FD\s+([A-Z0-9-]+)", u) if fd: program.fd_records.append(fd.group(1)) for line_no, block in blocks: u = block.upper() for call in re.findall(r"\bCALL\s+['\"]?([A-Z0-9_-]+)['\"]?", u): if call not in ("USING", "RETURNING"): program.called_programs.append(call) self.dependencies.append(Dependency(source=source.name, target=call, kind="CALL", line_no=line_no)) for perf in re.findall(r"\bPERFORM\s+([A-Z0-9-]+)", u): if perf not in ("UNTIL", "VARYING"): program.performed_paragraphs.append(perf) self.dependencies.append(Dependency(source=source.name, target=perf, kind="PERFORM", line_no=line_no)) for goto in re.findall(r"\bGO\s+TO\s+([A-Z0-9-]+)", u): program.goto_targets.append(goto) self.dependencies.append(Dependency(source=source.name, target=goto, kind="GOTO", line_no=line_no)) for statement in ("MOVE", "COMPUTE", "IF", "EVALUATE", "PERFORM", "CALL", "READ", "WRITE", "REWRITE", "DELETE", "START", "OPEN", "CLOSE", "GO TO", "EXEC SQL", "EXEC CICS"): if statement in u: program.statements[statement] += 1 program.copybooks = sorted(set(program.copybooks)) program.called_programs = sorted(set(program.called_programs)) program.performed_paragraphs = sorted(set(program.performed_paragraphs)) program.goto_targets = sorted(set(program.goto_targets)) program.files = sorted(set(program.files)) program.fd_records = sorted(set(program.fd_records)) return program def _current_section(self, line_u: str, current: str) -> str: if "FILE SECTION" in line_u: return "FILE SECTION" if "WORKING-STORAGE SECTION" in line_u: return "WORKING-STORAGE" if "LOCAL-STORAGE SECTION" in line_u: return "LOCAL-STORAGE" if "LINKAGE SECTION" in line_u: return "LINKAGE" if "PROCEDURE DIVISION" in line_u: return "PROCEDURE" return current def _parse_data_fields(self, source: SourceFile, lines: List[Tuple[int, str]]) -> None: section = "" stack: List[FieldDef] = [] local_fields: List[FieldDef] = [] for line_no, line in lines: u = line.upper() section = self._current_section(u, section) if section == "PROCEDURE": continue m = re.match(r"^(0[1-9]|[1-4][0-9]|66|77|88)\s+([A-Z0-9-]+)(.*?)(?:\.)?$", u) if not m: continue level = int(m.group(1)) name = m.group(2) rest = m.group(3) or "" while stack and stack[-1].level >= level: stack.pop() parent = stack[-1].name if stack else "" parent_path = stack[-1].path if stack else "" path = f"{parent_path}.{name}" if parent_path else name pic_match = re.search(r"\bPIC(?:TURE)?\s+([A-Z0-9()VSP+\-.,]+)", rest) pic = pic_match.group(1).rstrip(".") if pic_match else "" usage = "" usage_match = re.search(r"\b(?:USAGE\s+)?(COMP-3|COMP-5|COMP-4|COMP-2|COMP-1|COMP|BINARY|PACKED-DECIMAL|DISPLAY|INDEX)\b", rest) if usage_match: usage = usage_match.group(1) occurs = None occurs_depending_on = "" occurs_match = re.search(r"\bOCCURS\s+(\d+)", rest) if occurs_match: occurs = int(occurs_match.group(1)) dep_match = re.search(r"\bDEPENDING\s+ON\s+([A-Z0-9-]+)", rest) if dep_match: occurs_depending_on = dep_match.group(1) redefines = "" red_match = re.search(r"\bREDEFINES\s+([A-Z0-9-]+)", rest) if red_match: redefines = red_match.group(1) value = "" val_match = re.search(r"\bVALUE\s+(.+?)(?:\.|$)", rest) if val_match: value = val_match.group(1).strip() byte_length, decimal_digits, signed, json_type = parse_pic(pic, usage) is_group = not pic and level not in (66, 77, 88) field_def = FieldDef( source=source.name, section=section or "DATA", level=level, name=name, parent=parent, path=path, pic=pic, usage=usage, occurs=occurs, occurs_depending_on=occurs_depending_on, redefines=redefines, value=value, line_no=line_no, byte_length=byte_length, decimal_digits=decimal_digits, signed=signed, json_type=json_type, is_group=is_group, raw_line=line.strip(), ) self.fields.append(field_def) local_fields.append(field_def) if level not in (66, 77, 88): stack.append(field_def) if source.kind == "copybook": info = self.copybooks.setdefault(source.logical_name, CopybookInfo(source=source.name, name=source.logical_name)) info.fields = len(local_fields) info.root_fields = [f.name for f in local_fields if f.level == 1] info.estimated_bytes = sum((f.byte_length or 0) * (f.occurs or 1) for f in local_fields if not f.is_group) info.has_redefines = any(f.redefines for f in local_fields) info.has_occurs = any(f.occurs for f in local_fields) info.has_occurs_depending = any(f.occurs_depending_on for f in local_fields) def _parse_sql_blocks(self, source: SourceFile, blocks: List[Tuple[int, str]]) -> None: for line_no, block in blocks: u = normalize_ws(block.upper()) if "EXEC SQL" not in u: continue statement = "SQL" if " SELECT " in f" {u} ": statement = "SELECT" elif " INSERT " in f" {u} ": statement = "INSERT" elif " UPDATE " in f" {u} ": statement = "UPDATE" elif " DELETE " in f" {u} ": statement = "DELETE" elif " DECLARE " in f" {u} " and " CURSOR " in f" {u} ": statement = "DECLARE_CURSOR" cursor = "" cur = re.search(r"\bDECLARE\s+([A-Z0-9_-]+)\s+CURSOR", u) if cur: cursor = cur.group(1) tables = set() patterns = [ r"\bFROM\s+([A-Z0-9_.-]+)", r"\bJOIN\s+([A-Z0-9_.-]+)", r"\bUPDATE\s+([A-Z0-9_.-]+)", r"\bINTO\s+([A-Z0-9_.-]+)", r"\bDELETE\s+FROM\s+([A-Z0-9_.-]+)", ] for pat in patterns: for table in re.findall(pat, u): if table not in {"HOST", "CORRESPONDING", "DESCRIPTOR"}: tables.add(table.rstrip(",.")) for table in sorted(tables): self.db2.append(DB2Reference(source=source.name, statement=statement, table=table, cursor=cursor, line_no=line_no, raw_text=block)) self.dependencies.append(Dependency(source=source.name, target=table, kind="DB2_TABLE", detail=statement, line_no=line_no)) def _parse_cics_blocks(self, source: SourceFile, blocks: List[Tuple[int, str]]) -> None: for line_no, block in blocks: u = normalize_ws(block.upper()) if "EXEC CICS" not in u: continue if "END-EXEC" not in u and not any(token in u for token in ("PROGRAM", "COMMAREA", "TRANSID", "LENGTH")): continue cmd_match = re.search(r"EXEC\s+CICS\s+([A-Z]+)", u) command = cmd_match.group(1) if cmd_match else "CICS" program = "" transid = "" commarea = "" length = "" prog_match = re.search(r"\bPROGRAM\s*\(\s*([A-Z0-9_-]+)\s*\)", u) if prog_match: program = prog_match.group(1) else: literals = extract_quoted_literals(block) if command in ("LINK", "XCTL") and literals: program = literals[0].upper() tran_match = re.search(r"\bTRANSID\s*\(\s*([A-Z0-9_-]+)\s*\)", u) if tran_match: transid = tran_match.group(1) comm_match = re.search(r"\bCOMMAREA\s*\(\s*([A-Z0-9_-]+)\s*\)", u) if comm_match: commarea = comm_match.group(1) len_match = re.search(r"\bLENGTH\s*\(\s*([A-Z0-9_-]+|\d+)\s*\)", u) if len_match: length = len_match.group(1) self.cics.append(CICSReference(source=source.name, command=command, program=program, transid=transid, commarea=commarea, length=length, line_no=line_no, raw_text=block)) target = program or transid or command self.dependencies.append(Dependency(source=source.name, target=target, kind="CICS", detail=command, line_no=line_no)) def _parse_dependencies(self, source: SourceFile, lines: List[Tuple[int, str]], blocks: List[Tuple[int, str]]) -> None: for line_no, line in lines: u = line.upper() for copy in re.findall(r"\bCOPY\s+([A-Z0-9_-]+)", u): target = copy.replace("_", "-") self.dependencies.append(Dependency(source=source.name, target=target, kind="COPY", line_no=line_no)) for include in re.findall(r"\bEXEC\s+SQL\s+INCLUDE\s+([A-Z0-9_-]+)", u): self.dependencies.append(Dependency(source=source.name, target=include, kind="SQL_INCLUDE", line_no=line_no)) def _find_copybook_info(self, copy_name: str) -> Optional[CopybookInfo]: candidates = [copy_name, copy_name.replace("_", "-"), copy_name.replace("-", "_")] for name in candidates: if name in self.copybooks: return self.copybooks[name] for info in self.copybooks.values(): if copy_name in info.root_fields or copy_name.replace("_", "-") in info.root_fields: return info stem = Path(info.source).stem.upper().replace("_", "-") if copy_name in stem or stem.endswith(copy_name): return info return None def _resolve_copybook_usage(self) -> None: for program in self.programs.values(): missing = [] for copy in program.copybooks: info = self._find_copybook_info(copy) if not info: missing.append(copy) else: info.used_by_programs.append(program.program_id or program.source) program.copybooks_missing = sorted(set(missing)) for info in self.copybooks.values(): info.used_by_programs = sorted(set(info.used_by_programs)) def _run_rules(self) -> None: self._rule_missing_copybooks() self._rule_risky_cobol_constructs() self._rule_data_layout_complexity() self._rule_sensitive_fields() self._rule_api_candidates() self._rule_db2_cics() self._rule_naming() def _add(self, severity: str, category: str, code: str, source: str, line_no: int, title: str, detail: str, recommendation: str) -> None: self.findings.append(Finding(severity=severity, category=category, code=code, source=source, line_no=line_no, title=title, detail=detail, recommendation=recommendation)) def _rule_missing_copybooks(self) -> None: for p in self.programs.values(): for missing in p.copybooks_missing: self._add("ERROR", "DEPENDENCY", "MISSING_COPYBOOK", p.source, 0, "Missing copybook dependency", f"Program references COPY {missing}, but no matching copybook was provided.", "Add the copybook directory with --input-dir or verify COPY library resolution.") def _rule_risky_cobol_constructs(self) -> None: for p in self.programs.values(): if p.goto_targets: self._add("WARNING", "MAINTAINABILITY", "GOTO_USAGE", p.source, 0, "GO TO statements detected", f"Targets: {', '.join(p.goto_targets[:10])}", "Review control flow before modernization or automated refactoring.") if not p.program_id: self._add("ERROR", "STRUCTURE", "MISSING_PROGRAM_ID", p.source, 0, "Missing PROGRAM-ID", "No PROGRAM-ID could be detected.", "Verify source format and Identification Division.") for call in p.called_programs: if call in self.policy.get("high_risk_program_calls", []): self._add("WARNING", "DEPENDENCY", "HIGH_RISK_CALL", p.source, 0, "High impact called program", f"Program calls {call}.", "Review runtime environment, security, and job control assumptions.") for stmt in self.policy.get("banned_statements", []): if p.statements.get(stmt, 0): self._add("CRITICAL", "MAINTAINABILITY", "BANNED_STATEMENT", p.source, 0, "Banned COBOL statement detected", f"Statement {stmt} appears in procedure division.", "Remove or isolate this construct before modernization.") def _rule_data_layout_complexity(self) -> None: max_record = int(self.policy.get("max_record_bytes_warning", 4096)) max_occurs = int(self.policy.get("max_occurs_warning", 1000)) for info in self.copybooks.values(): if info.estimated_bytes > max_record: self._add("WARNING", "DATA_LAYOUT", "LARGE_RECORD", info.source, 0, "Large copybook record", f"Estimated leaf field bytes: {info.estimated_bytes}.", "Review API payload size, OCCURS arrays, and binary field mapping.") if info.has_redefines: self._add("WARNING", "DATA_LAYOUT", "REDEFINES_PRESENT", info.source, 0, "REDEFINES detected", "The copybook contains overlay fields.", "Manual mapping review is recommended before JSON/OpenAPI exposure.") if info.has_occurs_depending: self._add("WARNING", "DATA_LAYOUT", "OCCURS_DEPENDING_ON", info.source, 0, "OCCURS DEPENDING ON detected", "Variable-length arrays were detected.", "Represent this carefully in JSON Schema and validate bounds.") for f in self.fields: if f.occurs and f.occurs > max_occurs: self._add("WARNING", "DATA_LAYOUT", "LARGE_OCCURS", f.source, f.line_no, "Large OCCURS clause", f"{f.path} OCCURS {f.occurs}.", "Check memory, payload size, and downstream API limits.") if f.usage.upper() in ("COMP-3", "PACKED-DECIMAL"): self._add("INFO", "DATA_MAPPING", "PACKED_DECIMAL", f.source, f.line_no, "Packed decimal field", f"{f.path} uses {f.usage}.", "Map precision and scale explicitly in JSON Schema.") if f.redefines: self._add("WARNING", "DATA_LAYOUT", "FIELD_REDEFINES", f.source, f.line_no, "Field redefines another field", f"{f.path} redefines {f.redefines}.", "Review mutually exclusive layouts and API representation.") def _rule_sensitive_fields(self) -> None: patterns = [p.upper() for p in self.policy.get("sensitive_name_patterns", [])] for f in self.fields: name = f.name.upper() if any(p in name for p in patterns): self._add("WARNING", "SECURITY", "SENSITIVE_FIELD", f.source, f.line_no, "Sensitive field candidate", f"Field {f.path} matches sensitive naming policy.", "Classify the data, mask in lower environments, and review API exposure.") for dep in self.dependencies: target = dep.target.upper() if any(p in target for p in patterns): self._add("WARNING", "SECURITY", "SENSITIVE_DEPENDENCY", dep.source, dep.line_no, "Sensitive dependency candidate", f"Dependency target {dep.target} matches sensitive naming policy.", "Review security, access control, and data exposure rules.") def _rule_api_candidates(self) -> None: keywords = [p.upper() for p in self.policy.get("api_candidate_keywords", [])] for info in self.copybooks.values(): key = " ".join([info.name] + info.root_fields).upper() if any(k in key for k in keywords): self._add("INFO", "API", "API_PAYLOAD_CANDIDATE", info.source, 0, "API payload candidate", f"Copybook {info.name} looks like a request/response or COMMAREA payload.", "Generate JSON Schema and validate it with application SMEs.") def _rule_db2_cics(self) -> None: if self.db2: tables = sorted(set(x.table for x in self.db2)) sev = "WARNING" if len(tables) > 3 else "INFO" self._add(sev, "DB2", "DB2_DEPENDENCIES", "GLOBAL", 0, "DB2 dependencies detected", f"Tables: {', '.join(tables[:20])}", "Review transaction boundary, SQL access path, and data ownership.") if self.cics: commands = sorted(set(x.command for x in self.cics)) self._add("INFO", "CICS", "CICS_DEPENDENCIES", "GLOBAL", 0, "CICS dependencies detected", f"Commands: {', '.join(commands)}", "Review COMMAREA layout, transaction scope, and target programs.") for c in self.cics: if c.commarea: self._add("INFO", "CICS", "COMMAREA_DETECTED", c.source, c.line_no, "CICS COMMAREA detected", f"Command {c.command} uses COMMAREA {c.commarea}.", "Map COMMAREA copybook as request/response contract.") def _rule_naming(self) -> None: naming_regex = self.policy.get("naming_regex") if not naming_regex: return rx = re.compile(naming_regex) for f in self.fields: if not rx.match(f.name): self._add("INFO", "NAMING", "FIELD_NAMING", f.source, f.line_no, "Field naming policy mismatch", f"Field {f.name} does not match configured naming regex.", "Review naming conventions if this impacts generated APIs.") def _compare_copybook_or_program(self, path: Optional[str]) -> List[Dict[str, str]]: if not path: return [] compare_files = collect_input_files([path], []) if not compare_files: return [] other = CobolAnalyzer(compare_files, profile=self.profile, policy=self.policy, redact=self.redact) other._load_sources() other._parse_all_sources() current_fields = {f.path: f for f in self.fields} other_fields = {f.path: f for f in other.fields} rows: List[Dict[str, str]] = [] for name in sorted(set(current_fields) | set(other_fields)): if name not in other_fields: rows.append({"source": "current", "change": "ADDED_FIELD", "field": name, "detail": ""}) elif name not in current_fields: rows.append({"source": "current", "change": "REMOVED_FIELD", "field": name, "detail": ""}) else: a = current_fields[name] b = other_fields[name] if (a.pic, a.usage, a.occurs) != (b.pic, b.usage, b.occurs): rows.append({"source": "current", "change": "CHANGED_FIELD", "field": name, "detail": f"{b.pic}/{b.usage}/{b.occurs} -> {a.pic}/{a.usage}/{a.occurs}"}) self.compare_rows = rows return rows def _build_stats(self, compare_rows: List[Dict[str, str]]) -> Dict[str, Any]: return { "file_count": len(self.sources), "program_count": len(self.programs), "copybook_count": len(self.copybooks), "field_count": len(self.fields), "dependency_count": len(self.dependencies), "db2_reference_count": len(self.db2), "cics_reference_count": len(self.cics), "finding_count": len(self.findings), "db2_tables": sorted(set(x.table for x in self.db2)), "cics_commands": sorted(set(x.command for x in self.cics)), "compare_change_count": len(compare_rows), "categories": dict(Counter(f.category for f in self.findings)), } def _score(self) -> Tuple[int, str, str]: score = 0 highest = "INFO" for f in self.findings: highest = max(highest, f.severity, key=lambda x: SEVERITY_ORDER.get(x, 0)) if f.severity == "CRITICAL": score += 35 elif f.severity == "ERROR": score += 20 elif f.severity == "WARNING": score += 8 else: score += 2 score += min(len(self.db2) * 2, 10) score += min(len(self.cics) * 2, 10) score = min(score, 100) if score >= 75 or highest == "CRITICAL": status = "HIGH_RISK" elif score >= 40 or highest == "ERROR": status = "REVIEW_REQUIRED" elif score >= 15: status = "ATTENTION" else: status = "OK" return score, highest, status def _field_to_schema(self, f: FieldDef) -> Dict[str, Any]: schema: Dict[str, Any] = {"type": f.json_type} if f.json_type in ("integer", "number"): if f.decimal_digits: schema["multipleOf"] = 10 ** (-f.decimal_digits) if f.byte_length: schema["description"] = f"COBOL PIC {f.pic} {f.usage}".strip() elif f.byte_length: schema["maxLength"] = f.byte_length if f.occurs: return {"type": "array", "maxItems": f.occurs, "items": schema} return schema def _build_json_schema(self) -> Dict[str, Any]: root_name = "CobolPayload" candidates = [c for c in self.copybooks.values() if c.fields] if candidates: root_name = candidates[0].name.title().replace("-", "") properties: Dict[str, Any] = {} required: List[str] = [] # Keep leaf fields only for a pragmatic draft. for f in self.fields: if f.is_group or f.level == 88: continue json_name = f.name.lower().replace("-", "_") properties[json_name] = self._field_to_schema(f) if not f.value: required.append(json_name) return { "$schema": "https://json-schema.org/draft/2020-12/schema", "title": root_name, "type": "object", "properties": properties, "required": sorted(set(required))[:200], "x-cobol-note": "Draft generated from COBOL fields. Manual review required for REDEFINES, OCCURS DEPENDING ON, signed decimals, and binary fields.", } def _build_openapi(self, json_schema: Dict[str, Any]) -> Dict[str, Any]: program = next(iter(self.programs.values()), None) name = (program.program_id if program else "cobol-service").lower().replace("_", "-") path = f"/{name}" return { "openapi": "3.0.3", "info": { "title": f"{name} API draft", "version": "0.1.0", "description": "Draft OpenAPI contract generated from COBOL and copybook analysis. Manual validation required." }, "paths": { path: { "post": { "summary": f"Invoke {name}", "operationId": f"invoke_{name.replace('-', '_')}", "requestBody": { "required": True, "content": { "application/json": { "schema": {"$ref": "#/components/schemas/CobolPayload"} } } }, "responses": { "200": { "description": "Successful response", "content": { "application/json": { "schema": {"$ref": "#/components/schemas/CobolPayload"} } } } } } } }, "components": {"schemas": {"CobolPayload": json_schema}}, } def _build_mermaid(self) -> str: lines = ["graph TD"] for dep in self.dependencies: src = re.sub(r"[^A-Za-z0-9_]", "_", Path(dep.source).stem) tgt = re.sub(r"[^A-Za-z0-9_]", "_", dep.target) label = dep.kind lines.append(f" {src}[{Path(dep.source).stem}] -->|{label}| {tgt}[{dep.target}]") if len(lines) == 1: lines.append(" INPUT[COBOL Source] --> OUTPUT[Analysis]") return "\n".join(lines) def json_safe(value: Any) -> Any: if hasattr(value, "__dataclass_fields__"): return json_safe(asdict(value)) if isinstance(value, Counter): return {str(k): json_safe(v) for k, v in value.items()} if isinstance(value, dict): return {str(k): json_safe(v) for k, v in value.items()} if isinstance(value, (list, tuple, set)): return [json_safe(v) for v in value] return value def result_to_dict(result: AnalysisResult, include_raw: bool = False) -> Dict[str, Any]: data = json_safe(result) if not include_raw: for f in data.get("files", []): f.pop("raw_text", None) return data def write_json(path: str, data: Any) -> None: Path(path).write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") def write_csv(path: str, rows: Iterable[Any], fieldnames: Optional[List[str]] = None) -> None: rows_list = list(rows) if not fieldnames: if rows_list: first = rows_list[0] if hasattr(first, "__dataclass_fields__"): fieldnames = list(asdict(first).keys()) elif isinstance(first, dict): fieldnames = list(first.keys()) else: fieldnames = ["empty"] with open(path, "w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=fieldnames) writer.writeheader() for row in rows_list: if hasattr(row, "__dataclass_fields__"): writer.writerow(asdict(row)) elif isinstance(row, dict): writer.writerow(row) else: writer.writerow({"value": row}) def redact_text(text: str) -> str: text = re.sub(r"\b[A-Z0-9]{1,8}(?:\.[A-Z0-9#$@_-]{1,8}){1,8}\b", "", text) text = re.sub(r"\b[A-Z][A-Z0-9_-]{2,7}\b", lambda m: m.group(0) if m.group(0) in {"COBOL", "CICS", "DB2", "SQL", "COPY", "EXEC"} else m.group(0), text) return text def write_html_report(path: str, result: AnalysisResult) -> None: sev_class = { "INFO": "info", "WARNING": "warning", "ERROR": "error", "CRITICAL": "critical", } findings_rows = [] for f in result.findings[:200]: findings_rows.append( f"{html.escape(f.severity)}" f"{html.escape(f.category)}{html.escape(f.code)}" f"{html.escape(f.source)}:{f.line_no}{html.escape(f.title)}" f"{html.escape(f.detail)}{html.escape(f.recommendation)}" ) field_rows = [] for f in result.fields[:200]: field_rows.append( f"{html.escape(f.source)}{f.line_no}{html.escape(f.section)}" f"{f.level}{html.escape(f.path)}{html.escape(f.pic)}" f"{html.escape(f.usage)}{'' if f.byte_length is None else f.byte_length}" f"{'' if f.occurs is None else f.occurs}{html.escape(f.redefines)}" ) dep_rows = [] for d in result.dependencies[:200]: dep_rows.append( f"{html.escape(d.kind)}{html.escape(d.source)}" f"{html.escape(d.target)}{html.escape(d.detail)}{d.line_no}" ) db2_rows = [ f"{html.escape(x.source)}{x.line_no}{html.escape(x.statement)}{html.escape(x.table)}{html.escape(x.cursor)}" for x in result.db2[:200] ] cics_rows = [ f"{html.escape(x.source)}{x.line_no}{html.escape(x.command)}{html.escape(x.program)}{html.escape(x.commarea)}{html.escape(x.length)}" for x in result.cics[:200] ] css = """ body { font-family: Segoe UI, Arial, sans-serif; margin: 0; color: #111827; background: #f5f7fb; } header { background: #102a3a; color: white; padding: 26px 34px; } main { max-width: 1180px; margin: 24px auto; background: white; padding: 28px; border-radius: 12px; box-shadow: 0 8px 24px rgba(0,0,0,.08); } h1 { margin: 0 0 8px; } h2 { border-left: 5px solid #00b4d8; padding-left: 12px; color: #22313f; margin-top: 34px; } .grid { display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 14px; } .card { border: 1px solid #dce3ea; border-radius: 10px; padding: 14px; background: #fbfdff; } .value { font-size: 1.5rem; font-weight: 800; color: #0077b6; } .label { color: #667085; font-size: .9rem; } table { width: 100%; border-collapse: collapse; font-size: .9rem; margin: 16px 0; } th { background: #2c3e50; color: white; text-align: left; padding: 9px; border: 1px solid #243442; } td { padding: 8px; border: 1px solid #dce3ea; vertical-align: top; } tr:nth-child(even) td { background: #f8fafc; } .badge { display: inline-block; border-radius: 999px; padding: 2px 9px; font-weight: 700; font-size: .78rem; } .info { background: #e3f2fd; color: #0d47a1; } .warning { background: #fff3e0; color: #e65100; } .error { background: #ffebee; color: #8b0000; } .critical { background: #b71c1c; color: #fff; } pre { background: #282c34; color: #abb2bf; padding: 16px; border-radius: 8px; overflow-x: auto; } """ html_doc = f""" COBOL Copybook Impact Analyzer V2

COBOL / Copybook Impact Analyzer V2

Status: {html.escape(result.status)} | Risk score: {result.risk_score}/100 | Highest severity: {html.escape(result.highest_severity)}
{result.stats.get('file_count')}
Files
{result.stats.get('program_count')}
Programs
{result.stats.get('copybook_count')}
Copybooks
{result.stats.get('field_count')}
Fields
{result.stats.get('dependency_count')}
Dependencies
{result.stats.get('db2_reference_count')}
DB2 refs
{result.stats.get('cics_reference_count')}
CICS refs
{result.stats.get('finding_count')}
Findings

Findings

{''.join(findings_rows) or ''}
SeverityCategoryCodeLocationTitleDetailRecommendation
No findings

Fields

{''.join(field_rows)}
SourceLineSectionLevelPathPICUsageBytesOccursRedefines

Dependencies

{''.join(dep_rows)}
KindSourceTargetDetailLine

DB2 References

{''.join(db2_rows) or ''}
SourceLineStatementTableCursor
No DB2 references

CICS References

{''.join(cics_rows) or ''}
SourceLineCommandProgramCOMMAREALength
No CICS references

Mermaid Dependency Graph

{html.escape(result.mermaid)}
""" Path(path).write_text(html_doc, encoding="utf-8") DEMO_PROGRAM = """ IDENTIFICATION DIVISION. PROGRAM-ID. CUSTAPI. ENVIRONMENT DIVISION. INPUT-OUTPUT SECTION. FILE-CONTROL. SELECT CUSTOMER-FILE ASSIGN TO 'PROD.CUSTOMER.MASTER' ORGANIZATION IS INDEXED ACCESS MODE IS DYNAMIC RECORD KEY IS CUSTOMER-ID. DATA DIVISION. FILE SECTION. FD CUSTOMER-FILE. 01 CUSTOMER-RECORD. COPY CUSTOMER-ACCOUNT-RECORD. WORKING-STORAGE SECTION. 01 WS-RETURN-CODE PIC S9(4) COMP. 01 WS-SQLCODE PIC S9(9) COMP. 01 WS-DB2-CUSTOMER-ID PIC X(12). 01 WS-API-TRACE-ID PIC X(36). 01 WS-COUNTER PIC 9(4) COMP. 01 WS-ERROR-FLAG PIC X VALUE 'N'. 01 WS-CUSTOMER-REQUEST. COPY CUSTOMER-REQUEST. 01 WS-CUSTOMER-RESPONSE. COPY CUSTOMER-RESPONSE. LINKAGE SECTION. 01 DFHCOMMAREA. COPY CUSTOMER-REQUEST. PROCEDURE DIVISION. MAIN-LOGIC. EXEC CICS LINK PROGRAM('CUSTVALD') COMMAREA(DFHCOMMAREA) LENGTH(32500) END-EXEC. EXEC SQL DECLARE CUSTCUR CURSOR FOR SELECT CUSTOMER_ID, CUSTOMER_NAME, ACCOUNT_BALANCE FROM PRODDB.CUSTOMER WHERE CUSTOMER_ID = :WS-DB2-CUSTOMER-ID END-EXEC. EXEC SQL OPEN CUSTCUR END-EXEC. PERFORM READ-CUSTOMER. PERFORM BUILD-RESPONSE. GOBACK. READ-CUSTOMER. EXEC SQL FETCH CUSTCUR INTO :WS-DB2-CUSTOMER-ID, :CR-CUSTOMER-NAME, :CR-ACCOUNT-BALANCE END-EXEC. IF WS-SQLCODE NOT = 0 MOVE 'Y' TO WS-ERROR-FLAG END-IF. BUILD-RESPONSE. MOVE CR-CUSTOMER-ID TO RESP-CUSTOMER-ID. MOVE CR-CUSTOMER-NAME TO RESP-CUSTOMER-NAME. """ DEMO_COPYBOOK_REQUEST = """ 01 CUSTOMER-REQUEST. 05 REQ-HEADER. 10 REQ-TRACE-ID PIC X(36). 10 REQ-CHANNEL PIC X(10). 10 REQ-TIMESTAMP PIC X(26). 05 REQ-CUSTOMER. 10 REQ-CUSTOMER-ID PIC X(12). 10 REQ-ACCOUNT-ID PIC X(18). 10 REQ-SSN PIC X(11). 10 REQ-CARD-NUMBER PIC X(19). 05 REQ-OPTIONS. 10 REQ-INCLUDE-HISTORY PIC X. 10 REQ-MAX-ROWS PIC 9(4) COMP. 05 REQ-FILTERS OCCURS 25 TIMES. 10 REQ-FILTER-NAME PIC X(20). 10 REQ-FILTER-VALUE PIC X(80). """ DEMO_COPYBOOK_RESPONSE = """ 01 CUSTOMER-RESPONSE. 05 RESP-HEADER. 10 RESP-TRACE-ID PIC X(36). 10 RESP-STATUS-CODE PIC X(8). 10 RESP-MESSAGE PIC X(120). 05 RESP-CUSTOMER. 10 RESP-CUSTOMER-ID PIC X(12). 10 RESP-CUSTOMER-NAME PIC X(60). 10 RESP-ACCOUNT-BALANCE PIC S9(11)V99 COMP-3. 05 RESP-HISTORY-COUNT PIC 9(4) COMP. 05 RESP-HISTORY OCCURS 500 TIMES DEPENDING ON RESP-HISTORY-COUNT. 10 RESP-HIST-DATE PIC X(10). 10 RESP-HIST-AMOUNT PIC S9(9)V99 COMP-3. 10 RESP-HIST-CODE PIC X(8). """ DEMO_COPYBOOK_ACCOUNT = """ 01 CUSTOMER-ACCOUNT-RECORD. 05 CR-CUSTOMER-ID PIC X(12). 05 CR-CUSTOMER-NAME PIC X(60). 05 CR-ACCOUNT-ID PIC X(18). 05 CR-ACCOUNT-BALANCE PIC S9(11)V99 COMP-3. 05 CR-ACCOUNT-STATUS PIC X(1). 05 CR-SECURITY-GROUP PIC X(8). 05 CR-RAW-DATA PIC X(120). 05 CR-REDEFINED-DATA REDEFINES CR-RAW-DATA. 10 CR-SEGMENT-CODE PIC X(4). 10 CR-SEGMENT-PAYLOAD PIC X(116). """ def write_demo_files(directory: str) -> List[str]: base = Path(directory) base.mkdir(parents=True, exist_ok=True) files = { "sample_cobol_v2_customer_api.cbl": DEMO_PROGRAM, "sample_cobol_v2_customer_request.cpy": DEMO_COPYBOOK_REQUEST, "sample_cobol_v2_customer_response.cpy": DEMO_COPYBOOK_RESPONSE, "sample_cobol_v2_customer_account_record.cpy": DEMO_COPYBOOK_ACCOUNT, } paths = [] for name, content in files.items(): p = base / name p.write_text(content, encoding="utf-8") paths.append(str(p)) policy = { "sensitive_name_patterns": ["SSN", "CARD", "SECURITY", "BALANCE"], "max_record_bytes_warning": 2048, "max_occurs_warning": 300, "naming_regex": "^[A-Z0-9][A-Z0-9-]{1,30}$" } policy_path = base / "sample_cobol_v2_policy.json" policy_path.write_text(json.dumps(policy, indent=2), encoding="utf-8") return paths def print_console_summary(result: AnalysisResult) -> None: print("COBOL / Copybook Impact Analyzer V2") print("=" * 48) print(f"Status : {result.status}") print(f"Highest severity : {result.highest_severity}") print(f"Risk score : {result.risk_score}/100") print(f"Profile : {result.profile}") print(f"Files : {result.stats.get('file_count')}") print(f"Programs : {result.stats.get('program_count')}") print(f"Copybooks : {result.stats.get('copybook_count')}") print(f"Fields : {result.stats.get('field_count')}") print(f"Dependencies : {result.stats.get('dependency_count')}") print(f"DB2 references : {result.stats.get('db2_reference_count')}") print(f"CICS references : {result.stats.get('cics_reference_count')}") print(f"Findings : {result.stats.get('finding_count')}") if result.stats.get("db2_tables"): print(f"DB2 tables : {', '.join(result.stats.get('db2_tables'))}") if result.stats.get("cics_commands"): print(f"CICS commands : {', '.join(result.stats.get('cics_commands'))}") print() print("Top findings") print("-" * 48) for f in sorted(result.findings, key=lambda x: SEVERITY_ORDER.get(x.severity, 0), reverse=True)[:10]: print(f"[{f.severity}] {f.code} - {f.title} ({f.source}:{f.line_no})") print(f" {f.detail}") print(f" Recommendation: {f.recommendation}") def export_all(args: argparse.Namespace, result: AnalysisResult) -> None: if args.json: write_json(args.json, result_to_dict(result)) if args.html: write_html_report(args.html, result) if args.csv_fields: write_csv(args.csv_fields, result.fields) if args.csv_dependencies: write_csv(args.csv_dependencies, result.dependencies) if args.csv_findings: write_csv(args.csv_findings, result.findings) if args.csv_programs: write_csv(args.csv_programs, result.programs) if args.csv_copybooks: write_csv(args.csv_copybooks, result.copybooks) if args.csv_db2: write_csv(args.csv_db2, result.db2) if args.csv_cics: write_csv(args.csv_cics, result.cics) if args.csv_files: rows = [] for f in result.files: d = asdict(f) d.pop("raw_text", None) rows.append(d) write_csv(args.csv_files, rows) if getattr(args, "csv_compare", None): write_csv(args.csv_compare, result.compare_rows, fieldnames=["source", "change", "field", "detail"]) if args.json_schema: write_json(args.json_schema, result.json_schema) if args.openapi: write_json(args.openapi, result.openapi) if args.mermaid: Path(args.mermaid).write_text(result.mermaid, encoding="utf-8") def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="COBOL / Copybook Impact Analyzer V2") parser.add_argument("paths", nargs="*", help="COBOL or copybook files/directories") parser.add_argument("--input-dir", action="append", default=[], help="Additional directory to scan") parser.add_argument("--profile", default="modernization", choices=["modernization", "api", "db2", "cics", "batch", "security", "training"], help="Analysis profile") parser.add_argument("--custom-policy", help="JSON policy file") parser.add_argument("--compare-with", help="Baseline source/copybook/directory to compare with") parser.add_argument("--demo", action="store_true", help="Run built-in demo") parser.add_argument("--write-demo-files", help="Write demo files to directory and exit") parser.add_argument("--redact", action="store_true", help="Redact sensitive names in selected textual outputs") parser.add_argument("--fail-on", choices=["INFO", "WARNING", "ERROR", "CRITICAL"], help="Return exit code 2 when highest severity is at least this value") parser.add_argument("--json", help="Write full JSON analysis") parser.add_argument("--html", help="Write HTML report") parser.add_argument("--csv-fields", help="Write field inventory CSV") parser.add_argument("--csv-dependencies", help="Write dependencies CSV") parser.add_argument("--csv-findings", help="Write findings CSV") parser.add_argument("--csv-programs", help="Write programs CSV") parser.add_argument("--csv-copybooks", help="Write copybooks CSV") parser.add_argument("--csv-db2", help="Write DB2 references CSV") parser.add_argument("--csv-cics", help="Write CICS references CSV") parser.add_argument("--csv-files", help="Write source files CSV") parser.add_argument("--csv-compare", help="Write baseline comparison CSV") parser.add_argument("--json-schema", help="Write JSON Schema draft") parser.add_argument("--openapi", help="Write OpenAPI draft") parser.add_argument("--mermaid", help="Write Mermaid dependency graph") return parser def main(argv: Optional[Sequence[str]] = None) -> int: parser = build_arg_parser() args = parser.parse_args(argv) if args.write_demo_files: paths = write_demo_files(args.write_demo_files) print("Demo files written:") for p in paths: print(p) return 0 if args.demo: demo_dir = Path("cobol_copybook_v2_demo") paths = write_demo_files(str(demo_dir)) args.paths = paths files = collect_input_files(args.paths, args.input_dir) if not files: parser.error("No input files found. Provide COBOL/copybook files, --input-dir, or --demo.") policy = load_policy(args.custom_policy) analyzer = CobolAnalyzer(files, profile=args.profile, policy=policy, redact=args.redact) result = analyzer.analyze(compare_with=args.compare_with) print_console_summary(result) export_all(args, result) if args.fail_on: if SEVERITY_ORDER.get(result.highest_severity, 0) >= SEVERITY_ORDER[args.fail_on]: return 2 return 0 if __name__ == "__main__": raise SystemExit(main())