#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Dataset / Catalog Doctor V2 Professional analyzer for MVS/zOS dataset, IDCAMS LISTCAT, VSAM, GDG, and JCL dataset references. Design goals: - Work offline on text exports, not on live z/OS catalogs. - Accept mixed content: LISTCAT, IDCAMS messages, JCL snippets, VSAM stats, GDG info. - Produce actionable findings, risk scoring, and structured exports. - Stay dependency-free and portable across Windows, Linux, macOS. """ from __future__ import annotations import argparse import csv import datetime as dt import html import json import os import re import sys from dataclasses import dataclass, field, asdict from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple VERSION = "2.0.0" SEVERITY_WEIGHT = { "INFO": 1, "WARNING": 5, "ERROR": 15, "CRITICAL": 30, } SEVERITY_ORDER = { "INFO": 1, "WARNING": 2, "ERROR": 3, "CRITICAL": 4, } DEFAULT_POLICY: Dict[str, Any] = { "profiles": { "training": { "risk_multiplier": 0.7, "strict_naming": False, "space_primary_min_tracks": 1, "ci_split_warning": 500, "ci_split_critical": 2500, "ca_split_warning": 50, "ca_split_critical": 250, }, "batch": { "risk_multiplier": 1.0, "strict_naming": True, "space_primary_min_tracks": 5, "ci_split_warning": 250, "ci_split_critical": 1500, "ca_split_warning": 25, "ca_split_critical": 150, }, "storage": { "risk_multiplier": 1.1, "strict_naming": True, "space_primary_min_tracks": 10, "ci_split_warning": 100, "ci_split_critical": 1000, "ca_split_warning": 10, "ca_split_critical": 100, }, "security": { "risk_multiplier": 1.2, "strict_naming": True, "space_primary_min_tracks": 5, "ci_split_warning": 250, "ci_split_critical": 1500, "ca_split_warning": 25, "ca_split_critical": 150, }, "production": { "risk_multiplier": 1.15, "strict_naming": True, "space_primary_min_tracks": 10, "ci_split_warning": 200, "ci_split_critical": 1200, "ca_split_warning": 20, "ca_split_critical": 120, }, "strict": { "risk_multiplier": 1.35, "strict_naming": True, "space_primary_min_tracks": 20, "ci_split_warning": 100, "ci_split_critical": 500, "ca_split_warning": 10, "ca_split_critical": 50, }, }, "sensitive_patterns": [ r"\bPROD\b", r"\bPAYROLL\b", r"\bHR\b", r"\bRACF\b", r"\bSECURITY\b", r"\bSYS1\b", r"\bDB2\b", r"\bCICS\b", r"\bIMS\b", ], "allowed_hlq_patterns": [ r"^[A-Z][A-Z0-9]{1,7}$", r"^SYS[0-9A-Z]$", ], "enterprise_dataset_regex": r"^[A-Z0-9#$@]{1,8}(\.[A-Z0-9#$@]{1,8}){1,20}(\([+-]?[0-9]+\))?$", "temporary_dataset_prefix": "&&", "dangerous_dispositions": ["DELETE"], "warning_dispositions": ["OLD", "MOD"], "preferred_output_sms_fields": ["STORCLAS", "MGMTCLAS", "DATACLAS"], "rules": { "missing_catalog_reference": True, "dangerous_delete": True, "new_without_space_or_sms": True, "space_too_small": True, "vsam_split_analysis": True, "gdg_generation_rules": True, "naming_convention": True, "sensitive_dataset_rules": True, "idcams_delete_rules": True, "catalog_error_messages": True, "expiration_rules": True, "sms_rules": True, }, } RULE_CATALOG: List[Dict[str, str]] = [ {"code": "CAT001", "severity": "ERROR", "title": "Referenced dataset missing from catalog", "category": "catalog"}, {"code": "CAT002", "severity": "ERROR", "title": "Catalog or IDCAMS not found message", "category": "catalog"}, {"code": "CAT003", "severity": "WARNING", "title": "Dataset naming convention issue", "category": "governance"}, {"code": "JCL001", "severity": "CRITICAL", "title": "Dangerous DELETE disposition on sensitive dataset", "category": "jcl"}, {"code": "JCL002", "severity": "ERROR", "title": "NEW dataset without SPACE or SMS allocation", "category": "allocation"}, {"code": "JCL003", "severity": "WARNING", "title": "Primary SPACE allocation appears small", "category": "allocation"}, {"code": "JCL004", "severity": "WARNING", "title": "DISP=OLD or MOD on sensitive dataset", "category": "jcl"}, {"code": "JCL005", "severity": "ERROR", "title": "IDCAMS DELETE on sensitive dataset", "category": "idcams"}, {"code": "VSAM001", "severity": "WARNING", "title": "High VSAM CI split count", "category": "vsam"}, {"code": "VSAM002", "severity": "WARNING", "title": "High VSAM CA split count", "category": "vsam"}, {"code": "VSAM003", "severity": "WARNING", "title": "VSAM used percent appears high", "category": "vsam"}, {"code": "GDG001", "severity": "WARNING", "title": "GDG base without generations", "category": "gdg"}, {"code": "GDG002", "severity": "ERROR", "title": "GDG +1 reference without DISP=NEW", "category": "gdg"}, {"code": "GDG003", "severity": "WARNING", "title": "GDG base limit may be too small", "category": "gdg"}, {"code": "SEC001", "severity": "WARNING", "title": "Sensitive dataset referenced", "category": "security"}, {"code": "SMS001", "severity": "WARNING", "title": "Output dataset without SMS class metadata", "category": "sms"}, {"code": "EXP001", "severity": "WARNING", "title": "Expiration or retention policy unclear", "category": "governance"}, ] DATASET_PATTERN = re.compile(r"(?[A-Z0-9#$@]+)\s+DD\s+(?P.*)$", re.IGNORECASE) JCL_EXEC_PATTERN = re.compile(r"^//(?P[A-Z0-9#$@]+)\s+EXEC\s+(?P.*)$", re.IGNORECASE) IDCAMS_DELETE_PATTERN = re.compile(r"\bDELETE\s+([A-Z0-9#$@.]+(?:\([+-]?\d+\))?)", re.IGNORECASE) IDCAMS_DEFINE_PATTERN = re.compile(r"\bDEFINE\s+(?:CLUSTER|GDG|ALTERNATEINDEX|PATH)\b", re.IGNORECASE) GDG_REL_PATTERN = re.compile(r"^(?P[A-Z0-9#$@.]+)\((?P[+-]?\d+)\)$", re.IGNORECASE) SPACE_PATTERN = re.compile(r"SPACE\s*=\s*\((?PCYL|TRK|BLK|K|M|G|AVGREC)\s*,\s*\((?P[0-9]+)", re.IGNORECASE) KV_PATTERN = re.compile(r"\b([A-Z][A-Z0-9_-]{1,30})\s*[-=]\s*([A-Z0-9#$@.()+/-]+)", re.IGNORECASE) @dataclass class DatasetInfo: name: str entry_type: str = "UNKNOWN" cataloged: bool = True is_vsam: bool = False is_gdg_base: bool = False is_generation: bool = False sensitive: bool = False source_lines: List[int] = field(default_factory=list) attributes: Dict[str, Any] = field(default_factory=dict) @dataclass class DatasetReference: name: str base_name: str line_no: int source: str ddname: str = "" step: str = "" disp: str = "" space: str = "" unit: str = "" volser: str = "" sms_classes: Dict[str, str] = field(default_factory=dict) is_gdg_relative: bool = False gdg_relative: Optional[int] = None is_temporary: bool = False raw_line: str = "" @dataclass class Finding: code: str severity: str category: str title: str message: str dataset: str = "" line_no: Optional[int] = None recommendation: str = "" evidence: str = "" @dataclass class AnalysisResult: source_name: str profile: str status: str highest_severity: str risk_score: int datasets: List[DatasetInfo] references: List[DatasetReference] findings: List[Finding] category_scores: Dict[str, int] summary: Dict[str, Any] compare: List[Dict[str, Any]] = field(default_factory=list) def deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: merged = json.loads(json.dumps(base)) for key, value in override.items(): if isinstance(value, dict) and isinstance(merged.get(key), dict): merged[key] = deep_merge(merged[key], value) else: merged[key] = value return merged def read_text(path: str) -> str: with open(path, "r", encoding="utf-8", errors="ignore") as handle: return handle.read() def write_text(path: str, content: str) -> None: ensure_parent(path) with open(path, "w", encoding="utf-8", newline="") as handle: handle.write(content) def ensure_parent(path: str) -> None: parent = os.path.dirname(os.path.abspath(path)) if parent and not os.path.isdir(parent): os.makedirs(parent, exist_ok=True) def normalize_name(name: str) -> str: name = name.strip().upper().strip("'\".,") return re.sub(r"\s+", "", name) def looks_like_dataset_name(name: str) -> bool: """Reject timestamps, dates, and numeric artifacts accidentally matched as DSNs.""" value = normalize_name(name) if not value or "." not in value: return False if value.startswith("&&"): return True parts = value.split(".") if any(not part for part in parts): return False if all(part.isdigit() for part in parts): return False if value.startswith("CATALOG."): return False if len(parts) == 2 and all(part.isdigit() for part in parts): return False if not re.search(r"[A-Z#$@]", value): return False return True def gdg_base_name(name: str) -> str: match = GDG_REL_PATTERN.match(normalize_name(name)) if match: return match.group("base").upper() return normalize_name(name) def parse_params(params: str) -> Dict[str, str]: result: Dict[str, str] = {} text = params.strip() keys = ["DSN", "DSNAME", "DISP", "SPACE", "UNIT", "VOL", "VOLUME", "STORCLAS", "MGMTCLAS", "DATACLAS", "DCB"] for key in keys: pattern = re.compile(r"\b" + re.escape(key) + r"\s*=\s*", re.IGNORECASE) match = pattern.search(text) if not match: continue start = match.end() depth = 0 end = len(text) idx = start while idx < len(text): char = text[idx] if char == "(": depth += 1 elif char == ")" and depth > 0: depth -= 1 elif char == "," and depth == 0: tail = text[idx + 1 :] if re.match(r"\s*(DSN|DSNAME|DISP|SPACE|UNIT|VOL|VOLUME|STORCLAS|MGMTCLAS|DATACLAS|DCB)\s*=", tail, re.IGNORECASE): end = idx break idx += 1 result[key.upper()] = text[start:end].strip().strip("'") if "DSNAME" in result and "DSN" not in result: result["DSN"] = result["DSNAME"] return result def redact_dataset_name(name: str) -> str: if not name: return name base = normalize_name(name) gdg_suffix = "" match = GDG_REL_PATTERN.match(base) if match: base = match.group("base") gdg_suffix = f"({match.group('rel')})" parts = base.split(".") if not parts: return "" if len(parts) == 1: return "" + gdg_suffix return parts[0] + ".." + parts[-1] + gdg_suffix def redact_line(line: str) -> str: return DATASET_PATTERN.sub(lambda m: redact_dataset_name(m.group(0)), line) def is_sensitive(name: str, policy: Dict[str, Any]) -> bool: upper = normalize_name(name) for pattern in policy.get("sensitive_patterns", []): if re.search(pattern, upper, re.IGNORECASE): return True return False def is_valid_dataset_name(name: str, policy: Dict[str, Any]) -> bool: name = normalize_name(name) if name.startswith(policy.get("temporary_dataset_prefix", "&&")): return True regex = policy.get("enterprise_dataset_regex") if regex and not re.match(regex, name, re.IGNORECASE): return False parts = gdg_base_name(name).split(".") if not parts: return False hlq = parts[0] allowed = policy.get("allowed_hlq_patterns") or [] if allowed and not any(re.match(p, hlq, re.IGNORECASE) for p in allowed): return False return True def extract_space_primary(space: str) -> Tuple[str, Optional[int]]: if not space: return "", None match = SPACE_PATTERN.search("SPACE=" + space if not space.upper().startswith("SPACE") else space) if not match: return "", None return match.group("unit").upper(), int(match.group("primary")) def severity_max(values: Iterable[str]) -> str: highest = "INFO" for value in values: if SEVERITY_ORDER.get(value, 0) > SEVERITY_ORDER.get(highest, 0): highest = value return highest def status_from_findings(findings: List[Finding]) -> str: highest = severity_max(f.severity for f in findings) if findings else "INFO" if highest == "CRITICAL": return "BLOCKED" if highest == "ERROR": return "FAILED" if highest == "WARNING": return "WARNING" return "OK" def risk_score(findings: List[Finding], profile_cfg: Dict[str, Any]) -> int: raw = sum(SEVERITY_WEIGHT.get(f.severity, 1) for f in findings) raw = int(raw * float(profile_cfg.get("risk_multiplier", 1.0))) return max(0, min(100, raw)) def category_scores(findings: List[Finding]) -> Dict[str, int]: scores: Dict[str, int] = {} for finding in findings: scores[finding.category] = scores.get(finding.category, 0) + SEVERITY_WEIGHT.get(finding.severity, 1) return {key: min(100, value) for key, value in sorted(scores.items())} def split_continuations(text: str) -> List[str]: """Best-effort normalization for JCL continuations and wrapped LISTCAT lines.""" normalized: List[str] = [] pending = "" for raw in text.splitlines(): line = raw.rstrip("\n") stripped = line.strip() if pending and stripped.startswith("//") and not re.match(r"^//[A-Z0-9#$@]+\s+(DD|EXEC|JOB|PROC|PEND|SET|INCLUDE|JCLLIB)\b", stripped, re.IGNORECASE): pending += " " + stripped[2:].strip() continue if pending: normalized.append(pending) pending = "" if stripped.startswith("//") and stripped.rstrip().endswith(","): pending = stripped else: normalized.append(line) if pending: normalized.append(pending) return normalized class DatasetCatalogAnalyzer: def __init__(self, policy: Dict[str, Any], profile: str = "production", redact: bool = False) -> None: self.policy = policy self.profile = profile self.profile_cfg = policy.get("profiles", {}).get(profile, policy.get("profiles", {}).get("production", {})) self.redact = redact self.datasets: Dict[str, DatasetInfo] = {} self.references: List[DatasetReference] = [] self.findings: List[Finding] = [] self.current_step = "" self.current_dataset: Optional[DatasetInfo] = None self.lines: List[str] = [] self.idcams_deletes: List[Tuple[int, str, str]] = [] def analyze_text(self, text: str, source_name: str = "input") -> AnalysisResult: self.lines = split_continuations(text) for idx, raw in enumerate(self.lines, start=1): line = raw.rstrip() self._parse_line(idx, line) self._post_process() summary = self._build_summary(source_name) highest = severity_max(f.severity for f in self.findings) if self.findings else "INFO" score = risk_score(self.findings, self.profile_cfg) status = status_from_findings(self.findings) return AnalysisResult( source_name=source_name, profile=self.profile, status=status, highest_severity=highest, risk_score=score, datasets=sorted(self.datasets.values(), key=lambda x: x.name), references=self.references, findings=self.findings, category_scores=category_scores(self.findings), summary=summary, ) def _line_display(self, line: str) -> str: return redact_line(line) if self.redact else line def _dataset_display(self, name: str) -> str: return redact_dataset_name(name) if self.redact else name def _get_dataset(self, name: str, entry_type: str = "UNKNOWN", line_no: Optional[int] = None) -> DatasetInfo: name = normalize_name(name) if name not in self.datasets: self.datasets[name] = DatasetInfo(name=name, entry_type=entry_type or "UNKNOWN") ds = self.datasets[name] if entry_type and ds.entry_type == "UNKNOWN": ds.entry_type = entry_type if line_no is not None and line_no not in ds.source_lines: ds.source_lines.append(line_no) ds.sensitive = ds.sensitive or is_sensitive(name, self.policy) return ds def _add_finding(self, code: str, severity: str, category: str, title: str, message: str, dataset: str = "", line_no: Optional[int] = None, recommendation: str = "", evidence: str = "") -> None: finding = Finding( code=code, severity=severity, category=category, title=title, message=message, dataset=self._dataset_display(dataset) if dataset else "", line_no=line_no, recommendation=recommendation, evidence=self._line_display(evidence), ) self.findings.append(finding) def _parse_line(self, line_no: int, line: str) -> None: upper = line.upper() self._parse_listcat_entry(line_no, line) self._parse_listcat_attributes(line_no, line) self._parse_jcl(line_no, line) self._parse_idcams(line_no, line) self._parse_catalog_errors(line_no, line) if "IDCAMS" in upper or "LISTCAT" in upper or "IN-CAT" in upper: return # Catch loose dataset mentions, but avoid turning every system message into catalog entries. for match in DATASET_PATTERN.finditer(line): name = normalize_name(match.group(0)) if looks_like_dataset_name(name) and len(name.split(".")) >= 2: self._get_dataset(gdg_base_name(name), line_no=line_no) def _parse_listcat_entry(self, line_no: int, line: str) -> None: match = LISTCAT_ENTRY_PATTERN.search(line) if not match: return name = normalize_name(match.group(1)) prefix = line.upper().split("----")[0].strip() if "GDG" in prefix or "GENERATION" in prefix: entry_type = "GDG_BASE" elif "CLUSTER" in prefix: entry_type = "CLUSTER" elif "DATA" in prefix: entry_type = "DATA" elif "INDEX" in prefix: entry_type = "INDEX" else: entry_type = prefix.split()[0] if prefix else "UNKNOWN" ds = self._get_dataset(name, entry_type, line_no) ds.cataloged = True ds.is_vsam = entry_type in {"CLUSTER", "DATA", "INDEX", "AIX"} ds.is_gdg_base = entry_type == "GDG_BASE" self.current_dataset = ds def _parse_listcat_attributes(self, line_no: int, line: str) -> None: if self.current_dataset is None: return upper = line.upper() ds = self.current_dataset key_map = { "CI-SPLITS": "ci_splits", "CA-SPLITS": "ca_splits", "RECORDS": "records", "TRACKS": "tracks", "HI-USED-RBA": "hi_used_rba", "HIGH-USED-RBA": "hi_used_rba", "HI-ALLOC-RBA": "hi_alloc_rba", "HIGH-ALLOC-RBA": "hi_alloc_rba", "FREESPACE": "freespace", "SHAREOPTIONS": "shareoptions", "LIMIT": "gdg_limit", "EMPTY": "empty", "NOEMPTY": "noempty", "SCRATCH": "scratch", "NOSCRATCH": "noscratch", "STORAGECLASS": "storclas", "STORCLAS": "storclas", "MANAGEMENTCLASS": "mgmtclas", "MGMTCLAS": "mgmtclas", "DATACLASS": "dataclas", "DATACLAS": "dataclas", "OWNER": "owner", "VOLUMES": "volumes", "VOLSER": "volser", "EXPIRATION": "expiration", "CREATION": "creation", "LAST-REFERENCED": "last_referenced", } for raw_key, attr_key in key_map.items(): # Match both KEY-----value and KEY=value styles. match = re.search(r"\b" + re.escape(raw_key) + r"\b\s*[-=]*\s*([A-Z0-9#$@.,()/+-]+)?", upper) if not match: continue value = match.group(1) or "YES" if value == "YES" and raw_key in {"EMPTY", "NOEMPTY", "SCRATCH", "NOSCRATCH"}: ds.attributes[attr_key] = True else: ds.attributes[attr_key] = value.strip("-") for key, value in KV_PATTERN.findall(upper): if key in {"CI", "CA"}: continue ds.attributes.setdefault(key.lower().replace("-", "_"), value) if "GDG" in upper and "LIMIT" in upper: ds.is_gdg_base = True if "CLUSTER" in ds.entry_type or "CI-SPLITS" in upper or "CA-SPLITS" in upper: ds.is_vsam = True def _parse_jcl(self, line_no: int, line: str) -> None: stripped = line.strip() exec_match = JCL_EXEC_PATTERN.match(stripped) if exec_match: self.current_step = exec_match.group("step").upper() return dd_match = JCL_DD_PATTERN.match(stripped) if not dd_match: return ddname = dd_match.group("ddname").upper() params = dd_match.group("params") parsed = parse_params(params) dsn = parsed.get("DSN", "") if not dsn: return if dsn.strip().upper() in {"DUMMY", "NULLFILE"}: return dsn = normalize_name(dsn) gdg_match = GDG_REL_PATTERN.match(dsn) is_gdg = gdg_match is not None rel = int(gdg_match.group("rel")) if gdg_match else None base = gdg_base_name(dsn) temp = dsn.startswith(self.policy.get("temporary_dataset_prefix", "&&")) ref = DatasetReference( name=dsn, base_name=base, line_no=line_no, source="JCL", ddname=ddname, step=self.current_step, disp=parsed.get("DISP", ""), space=parsed.get("SPACE", ""), unit=parsed.get("UNIT", ""), volser=parsed.get("VOL", parsed.get("VOLUME", "")), sms_classes={k: v for k, v in parsed.items() if k in {"STORCLAS", "MGMTCLAS", "DATACLAS"}}, is_gdg_relative=is_gdg, gdg_relative=rel, is_temporary=temp, raw_line=line, ) self.references.append(ref) if not temp: self._get_dataset(base, line_no=line_no) def _parse_idcams(self, line_no: int, line: str) -> None: delete_match = IDCAMS_DELETE_PATTERN.search(line) if delete_match: name = normalize_name(delete_match.group(1)) if looks_like_dataset_name(name): self.idcams_deletes.append((line_no, name, line)) self._get_dataset(gdg_base_name(name), line_no=line_no) if IDCAMS_DEFINE_PATTERN.search(line): for name in DATASET_PATTERN.findall(line): if looks_like_dataset_name(name): self._get_dataset(gdg_base_name(name), "DEFINED", line_no=line_no) def _parse_catalog_errors(self, line_no: int, line: str) -> None: upper = line.upper() if any(token in upper for token in ["IDC3012I", "IDC3009I", "NOT FOUND", "NOT CATALOGED", "CATALOG ERROR", "ENTRY NOT FOUND"]): datasets = [normalize_name(m.group(0)) for m in DATASET_PATTERN.finditer(line)] dataset = datasets[0] if datasets else "" if dataset and looks_like_dataset_name(dataset): ds = self._get_dataset(gdg_base_name(dataset), line_no=line_no) ds.cataloged = False self._add_finding( "CAT002", "ERROR", "catalog", "Catalog error message detected", "A catalog or IDCAMS error indicates that a referenced dataset or catalog entry could not be resolved.", dataset=dataset, line_no=line_no, recommendation="Verify LISTCAT output, catalog alias, volume, and dataset spelling before rerunning the job.", evidence=line, ) def _post_process(self) -> None: rules = self.policy.get("rules", {}) if rules.get("missing_catalog_reference", True): self._rule_missing_references() if rules.get("dangerous_delete", True): self._rule_dangerous_disp() if rules.get("new_without_space_or_sms", True) or rules.get("space_too_small", True): self._rule_allocation() if rules.get("vsam_split_analysis", True): self._rule_vsam() if rules.get("gdg_generation_rules", True): self._rule_gdg() if rules.get("naming_convention", True): self._rule_naming() if rules.get("sensitive_dataset_rules", True): self._rule_sensitive() if rules.get("idcams_delete_rules", True): self._rule_idcams_delete() if rules.get("sms_rules", True): self._rule_sms() if rules.get("expiration_rules", True): self._rule_expiration() def _rule_missing_references(self) -> None: cataloged = {name for name, ds in self.datasets.items() if ds.cataloged and ds.entry_type != "UNKNOWN"} for ref in self.references: if ref.is_temporary: continue # New datasets are expected not to exist yet. if "NEW" in ref.disp.upper(): continue if ref.base_name not in cataloged and ref.base_name in self.datasets: # UNKNOWN means seen only as reference, not in LISTCAT. if self.datasets[ref.base_name].entry_type == "UNKNOWN": self._add_finding( "CAT001", "ERROR", "catalog", "Referenced dataset not found in LISTCAT", f"JCL references {self._dataset_display(ref.name)} but no LISTCAT entry was detected.", dataset=ref.name, line_no=ref.line_no, recommendation="Run LISTCAT for the dataset, verify GDG base, catalog alias, and spelling. If it is created earlier, validate job dependency order.", evidence=ref.raw_line, ) def _rule_dangerous_disp(self) -> None: for ref in self.references: if ref.is_temporary: continue disp = ref.disp.upper() if "DELETE" in disp: sensitive = is_sensitive(ref.name, self.policy) severity = "CRITICAL" if sensitive else "ERROR" self._add_finding( "JCL001", severity, "jcl", "Dangerous DELETE disposition", f"Dataset {self._dataset_display(ref.name)} uses DISP with DELETE. This can remove data at job end or abnormal termination.", dataset=ref.name, line_no=ref.line_no, recommendation="Review DISP parameters. Avoid DELETE on production or sensitive datasets unless this is an explicit cleanup step.", evidence=ref.raw_line, ) def _rule_allocation(self) -> None: min_primary = int(self.profile_cfg.get("space_primary_min_tracks", 5)) for ref in self.references: disp = ref.disp.upper() if "NEW" not in disp: continue has_sms = any(ref.sms_classes.values()) if not ref.space and not has_sms: self._add_finding( "JCL002", "ERROR", "allocation", "NEW dataset without SPACE or SMS allocation", f"Dataset {self._dataset_display(ref.name)} is created with DISP=NEW but no SPACE or SMS class was detected.", dataset=ref.name, line_no=ref.line_no, recommendation="Add SPACE or SMS allocation metadata such as STORCLAS, MGMTCLAS, or DATACLAS.", evidence=ref.raw_line, ) unit, primary = extract_space_primary(ref.space) if primary is not None and unit in {"TRK", "CYL"} and primary < min_primary: self._add_finding( "JCL003", "WARNING", "allocation", "Primary SPACE allocation appears small", f"Dataset {self._dataset_display(ref.name)} primary SPACE is {primary} {unit}, below profile threshold {min_primary}.", dataset=ref.name, line_no=ref.line_no, recommendation="Review expected volume and allocation unit. A small primary allocation can increase extents or trigger B37/D37 failures.", evidence=ref.raw_line, ) def _int_attr(self, ds: DatasetInfo, key: str) -> Optional[int]: value = ds.attributes.get(key) if value is None: return None text = str(value).replace(",", "") match = re.search(r"\d+", text) return int(match.group(0)) if match else None def _rule_vsam(self) -> None: ci_warn = int(self.profile_cfg.get("ci_split_warning", 250)) ci_crit = int(self.profile_cfg.get("ci_split_critical", 1500)) ca_warn = int(self.profile_cfg.get("ca_split_warning", 25)) ca_crit = int(self.profile_cfg.get("ca_split_critical", 150)) for ds in self.datasets.values(): if not ds.is_vsam: continue ci = self._int_attr(ds, "ci_splits") ca = self._int_attr(ds, "ca_splits") if ci is not None and ci >= ci_warn: severity = "CRITICAL" if ci >= ci_crit else "WARNING" self._add_finding( "VSAM001", severity, "vsam", "High VSAM CI split count", f"VSAM cluster {self._dataset_display(ds.name)} reports {ci} CI splits.", dataset=ds.name, line_no=ds.source_lines[0] if ds.source_lines else None, recommendation="Review FREESPACE, record insert pattern, CA/CI size, and consider REORG if split activity is abnormal.", ) if ca is not None and ca >= ca_warn: severity = "CRITICAL" if ca >= ca_crit else "WARNING" self._add_finding( "VSAM002", severity, "vsam", "High VSAM CA split count", f"VSAM cluster {self._dataset_display(ds.name)} reports {ca} CA splits.", dataset=ds.name, line_no=ds.source_lines[0] if ds.source_lines else None, recommendation="Review dataset growth, space allocation, FREESPACE, and whether a VSAM REORG is needed.", ) used = self._int_attr(ds, "hi_used_rba") alloc = self._int_attr(ds, "hi_alloc_rba") if used and alloc and alloc > 0: pct = round((used / alloc) * 100, 2) ds.attributes["used_percent"] = pct if pct >= 90: self._add_finding( "VSAM003", "WARNING", "vsam", "VSAM cluster appears close to allocated high RBA", f"VSAM cluster {self._dataset_display(ds.name)} used percent is approximately {pct}%.", dataset=ds.name, line_no=ds.source_lines[0] if ds.source_lines else None, recommendation="Review allocation, growth trend, and free space policy before the next heavy update window.", ) def _rule_gdg(self) -> None: gdg_bases = {name for name, ds in self.datasets.items() if ds.is_gdg_base} referenced_generations: Dict[str, List[DatasetReference]] = {} for ref in self.references: if ref.is_gdg_relative: referenced_generations.setdefault(ref.base_name, []).append(ref) if ref.gdg_relative == 1 and "NEW" not in ref.disp.upper(): self._add_finding( "GDG002", "ERROR", "gdg", "GDG +1 reference without DISP=NEW", f"GDG generation {self._dataset_display(ref.name)} is referenced as +1 but DISP does not create a new generation.", dataset=ref.name, line_no=ref.line_no, recommendation="Use DISP=(NEW,CATLG,DELETE) or confirm that the generation is created by a prior step.", evidence=ref.raw_line, ) for base in gdg_bases: refs = referenced_generations.get(base, []) if not refs: # If no concrete GxxxxVyy entry and no relative ref, it may be a stale base in the sample. generation_entries = [name for name in self.datasets if name.startswith(base + ".G")] if not generation_entries: self._add_finding( "GDG001", "WARNING", "gdg", "GDG base without detected generations", f"GDG base {self._dataset_display(base)} was found but no generation entry or relative reference was detected.", dataset=base, recommendation="Verify LISTCAT LEVEL output includes generations and confirm the GDG base is still active.", ) ds = self.datasets[base] limit = self._int_attr(ds, "gdg_limit") if limit is not None and limit < 3: self._add_finding( "GDG003", "WARNING", "gdg", "GDG limit appears small", f"GDG base {self._dataset_display(base)} has LIMIT({limit}), which may be too small for safe rollback or audit retention.", dataset=base, recommendation="Review retention requirements and increase LIMIT if operational rollback depends on multiple generations.", ) def _rule_naming(self) -> None: if not self.profile_cfg.get("strict_naming", False): return for name, ds in self.datasets.items(): if name.startswith(self.policy.get("temporary_dataset_prefix", "&&")): continue if not is_valid_dataset_name(name, self.policy): self._add_finding( "CAT003", "WARNING", "governance", "Dataset naming convention issue", f"Dataset {self._dataset_display(name)} does not match the configured enterprise naming policy.", dataset=name, line_no=ds.source_lines[0] if ds.source_lines else None, recommendation="Validate HLQ and qualifier standards. Update policy JSON if this naming pattern is valid for your site.", ) def _rule_sensitive(self) -> None: for ref in self.references: if not is_sensitive(ref.name, self.policy): continue disp = ref.disp.upper() severity = "WARNING" if "OLD" in disp or "MOD" in disp: severity = "ERROR" if self.profile in {"security", "strict", "production"} else "WARNING" self._add_finding( "SEC001", severity, "security", "Sensitive dataset referenced", f"Sensitive dataset pattern detected for {self._dataset_display(ref.name)}.", dataset=ref.name, line_no=ref.line_no, recommendation="Confirm least privilege, DISP mode, audit requirements, and whether the dataset should be redacted in reports.", evidence=ref.raw_line, ) def _rule_idcams_delete(self) -> None: for line_no, name, line in self.idcams_deletes: if is_sensitive(name, self.policy): self._add_finding( "JCL005", "CRITICAL", "idcams", "IDCAMS DELETE on sensitive dataset", f"IDCAMS DELETE targets sensitive dataset {self._dataset_display(name)}.", dataset=name, line_no=line_no, recommendation="Require manual approval, backup verification, and environment validation before running this control card.", evidence=line, ) def _rule_sms(self) -> None: preferred = self.policy.get("preferred_output_sms_fields", []) for ref in self.references: if "NEW" not in ref.disp.upper() or ref.is_temporary: continue missing = [key for key in preferred if key not in ref.sms_classes] if ref.space and missing: self._add_finding( "SMS001", "WARNING", "sms", "Output dataset without full SMS metadata", f"Dataset {self._dataset_display(ref.name)} is created without full preferred SMS class metadata: {', '.join(missing)}.", dataset=ref.name, line_no=ref.line_no, recommendation="Consider using DATACLAS, STORCLAS, and MGMTCLAS when your site's SMS policy expects explicit classes.", evidence=ref.raw_line, ) def _rule_expiration(self) -> None: for ds in self.datasets.values(): if not ds.sensitive: continue if not any(key in ds.attributes for key in ["expiration", "mgmtclas", "managementclass"]): self._add_finding( "EXP001", "WARNING", "governance", "Retention or expiration policy unclear", f"Sensitive dataset {self._dataset_display(ds.name)} has no visible EXPIRATION or management class metadata in the parsed content.", dataset=ds.name, line_no=ds.source_lines[0] if ds.source_lines else None, recommendation="Verify retention policy, MGMTCLAS, legal hold requirements, and archival rules.", ) def _build_summary(self, source_name: str) -> Dict[str, Any]: sensitive_count = sum(1 for ds in self.datasets.values() if ds.sensitive) vsam_count = sum(1 for ds in self.datasets.values() if ds.is_vsam) gdg_count = sum(1 for ds in self.datasets.values() if ds.is_gdg_base) jcl_ref_count = len(self.references) return { "tool": "Dataset / Catalog Doctor V2", "version": VERSION, "source": source_name, "profile": self.profile, "dataset_count": len(self.datasets), "reference_count": jcl_ref_count, "finding_count": len(self.findings), "sensitive_dataset_count": sensitive_count, "vsam_cluster_count": vsam_count, "gdg_base_count": gdg_count, "generated_at": dt.datetime.now().isoformat(timespec="seconds"), } def compare_results(current: AnalysisResult, baseline: AnalysisResult) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] current_names = {ds.name for ds in current.datasets} baseline_names = {ds.name for ds in baseline.datasets} for name in sorted(current_names - baseline_names): rows.append({"type": "NEW_DATASET", "severity": "INFO", "item": name, "message": "Dataset present in current analysis but not in baseline."}) for name in sorted(baseline_names - current_names): rows.append({"type": "REMOVED_DATASET", "severity": "INFO", "item": name, "message": "Dataset present in baseline but not in current analysis."}) current_codes = [(f.code, f.dataset, f.title) for f in current.findings] baseline_codes = set((f.code, f.dataset, f.title) for f in baseline.findings) for code, dataset, title in current_codes: if (code, dataset, title) not in baseline_codes: rows.append({"type": "NEW_FINDING", "severity": "WARNING", "item": dataset, "message": f"New finding {code}: {title}"}) return rows def result_to_dict(result: AnalysisResult) -> Dict[str, Any]: return { "source_name": result.source_name, "profile": result.profile, "status": result.status, "highest_severity": result.highest_severity, "risk_score": result.risk_score, "summary": result.summary, "category_scores": result.category_scores, "datasets": [asdict(ds) for ds in result.datasets], "references": [asdict(ref) for ref in result.references], "findings": [asdict(f) for f in result.findings], "compare": result.compare, } def write_json(path: str, result: AnalysisResult) -> None: ensure_parent(path) with open(path, "w", encoding="utf-8") as handle: json.dump(result_to_dict(result), handle, indent=2) def write_csv(path: str, rows: Sequence[Dict[str, Any]], fieldnames: Optional[List[str]] = None) -> None: ensure_parent(path) if fieldnames is None: keys: List[str] = [] for row in rows: for key in row.keys(): if key not in keys: keys.append(key) fieldnames = keys with open(path, "w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() for row in rows: serializable = {} for key in fieldnames: value = row.get(key, "") if isinstance(value, (dict, list)): value = json.dumps(value, ensure_ascii=False) serializable[key] = value writer.writerow(serializable) def write_result_csvs(result: AnalysisResult, paths: Dict[str, str]) -> None: if paths.get("findings"): write_csv(paths["findings"], [asdict(f) for f in result.findings]) if paths.get("datasets"): write_csv(paths["datasets"], [asdict(ds) for ds in result.datasets]) if paths.get("references"): write_csv(paths["references"], [asdict(ref) for ref in result.references]) if paths.get("categories"): rows = [{"category": k, "score": v} for k, v in result.category_scores.items()] write_csv(paths["categories"], rows) if paths.get("compare"): write_csv(paths["compare"], result.compare) def html_escape(value: Any) -> str: return html.escape(str(value if value is not None else "")) def severity_badge(severity: str) -> str: cls = { "INFO": "info", "WARNING": "warn", "ERROR": "error", "CRITICAL": "critical", }.get(severity, "info") return f'{html_escape(severity)}' def write_html(path: str, result: AnalysisResult) -> None: ensure_parent(path) rows_findings = "".join( f"{severity_badge(f.severity)}{html_escape(f.code)}{html_escape(f.category)}{html_escape(f.dataset)}{html_escape(f.title)}{html_escape(f.message)}{html_escape(f.recommendation)}" for f in result.findings[:300] ) rows_datasets = "".join( f"{html_escape(ds.name)}{html_escape(ds.entry_type)}{'yes' if ds.cataloged else 'no'}{'yes' if ds.is_vsam else 'no'}{'yes' if ds.is_gdg_base else 'no'}{'yes' if ds.sensitive else 'no'}{html_escape(json.dumps(ds.attributes, ensure_ascii=False))}" for ds in result.datasets[:300] ) rows_refs = "".join( f"{html_escape(ref.line_no)}{html_escape(ref.step)}{html_escape(ref.ddname)}{html_escape(ref.name)}{html_escape(ref.disp)}{html_escape(ref.space)}{html_escape(ref.sms_classes)}" for ref in result.references[:300] ) rows_cat = "".join( f"{html_escape(k)}{html_escape(v)}" for k, v in result.category_scores.items() ) rows_compare = "".join( f"{html_escape(row.get('type'))}{html_escape(row.get('severity'))}{html_escape(row.get('item'))}{html_escape(row.get('message'))}" for row in result.compare[:300] ) content = f""" Dataset / Catalog Doctor V2 Report

Dataset / Catalog Doctor V2 Report

Source: {html_escape(result.source_name)} | Profile: {html_escape(result.profile)} | Generated: {html_escape(result.summary.get('generated_at'))}
{html_escape(result.status)}
Status
{html_escape(result.risk_score)}/100
Risk score
{html_escape(result.highest_severity)}
Highest severity
{html_escape(len(result.findings))}
Findings
{html_escape(len(result.datasets))}
Datasets
{html_escape(len(result.references))}
JCL references
{html_escape(result.summary.get('vsam_cluster_count'))}
VSAM entries
{html_escape(result.summary.get('gdg_base_count'))}
GDG bases
Recommended reading: start with CRITICAL and ERROR findings, then review referenced datasets not found in LISTCAT, dangerous DISP, VSAM split activity, GDG rules, and sensitive dataset usage.

Risk by category

{rows_cat}
CategoryScore

Findings

{rows_findings}
SeverityCodeCategoryDatasetTitleMessageRecommendation

Datasets

{rows_datasets}
NameTypeCatalogedVSAMGDGSensitiveAttributes

JCL references

{rows_refs}
LineStepDDDatasetDISPSPACESMS

Baseline comparison

{rows_compare}
TypeSeverityItemMessage
""" write_text(path, content) def print_console(result: AnalysisResult) -> None: print("Dataset / Catalog Doctor V2") 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"Datasets : {len(result.datasets)}") print(f"JCL references : {len(result.references)}") print(f"Findings : {len(result.findings)}") print(f"Sensitive datasets : {result.summary.get('sensitive_dataset_count')}") print(f"VSAM entries : {result.summary.get('vsam_cluster_count')}") print(f"GDG bases : {result.summary.get('gdg_base_count')}") if result.findings: print("\nTop findings:") for finding in sorted(result.findings, key=lambda f: SEVERITY_ORDER.get(f.severity, 0), reverse=True)[:12]: dataset = f" [{finding.dataset}]" if finding.dataset else "" print(f"- {finding.severity:8} {finding.code:7} {finding.title}{dataset}") if finding.recommendation: print(f" Recommendation: {finding.recommendation}") if result.compare: print("\nBaseline comparison:") for row in result.compare[:10]: print(f"- {row.get('type')}: {row.get('item')} - {row.get('message')}") SAMPLE_MIXED = """//PAYCAT01 JOB (ACCT),'DATASET DOCTOR',CLASS=A,MSGCLASS=X,NOTIFY=&SYSUID //STEP010 EXEC PGM=IDCAMS //SYSPRINT DD SYSOUT=* //SYSIN DD * LISTCAT ENT(PAYROLL.PROD.MASTER) ALL LISTCAT LEVEL(PAYROLL.PROD.GDG) ALL DELETE PAYROLL.PROD.OLD.MASTER PURGE /* //STEP020 EXEC PGM=SORT //SORTIN DD DSN=PAYROLL.PROD.INPUT(+0),DISP=SHR //SORTOUT DD DSN=PAYROLL.PROD.OUTPUT(+1),DISP=(CATLG,CATLG,DELETE),SPACE=(TRK,(1,1)),UNIT=SYSDA //SORTWK01 DD DSN=&&SORTWK,UNIT=SYSDA,SPACE=(TRK,(1,1)),DISP=(NEW,DELETE) //STEP030 EXEC PGM=PAYLOAD //MASTER DD DSN=PAYROLL.PROD.MASTER,DISP=OLD //NEWFILE DD DSN=PAYROLL.PROD.NEWFILE,DISP=(NEW,CATLG,DELETE) //MISSING DD DSN=PAYROLL.PROD.MISSING,DISP=SHR //BADNAME DD DSN=BAD-HLQ.PAYROLL.FILE,DISP=SHR IDCAMS SYSTEM SERVICES TIME: 08:42:10 NONVSAM ------- PAYROLL.PROD.MASTER IN-CAT --- CATALOG.MASTER VOLUMES -- PRD001 CREATION--------2026.110 EXPIRATION------0000.000 MANAGEMENTCLASS-PRODMC STORAGECLASS-PRODFAST CLUSTER ------- PAYROLL.PROD.KSDS.CUSTOMER IN-CAT --- CATALOG.MASTER CI-SPLITS-----1850 CA-SPLITS-----180 HI-USED-RBA---0000000009500000 HI-ALLOC-RBA---0000000010000000 FREESPACE----(10 10) SHAREOPTIONS-(2 3) GDG BASE ------- PAYROLL.PROD.INPUT LIMIT----------2 SCRATCH NOEMPTY IDC3012I ENTRY PAYROLL.PROD.MISSING NOT FOUND """ SAMPLE_BASELINE = """//PAYCAT00 JOB (ACCT),'BASELINE',CLASS=A,MSGCLASS=X //STEP010 EXEC PGM=PAYLOAD //MASTER DD DSN=PAYROLL.PROD.MASTER,DISP=SHR //INPUT DD DSN=PAYROLL.PROD.INPUT(+0),DISP=SHR NONVSAM ------- PAYROLL.PROD.MASTER IN-CAT --- CATALOG.MASTER VOLUMES -- PRD001 CREATION--------2026.100 EXPIRATION------0000.000 MANAGEMENTCLASS-PRODMC STORAGECLASS-PRODFAST GDG BASE ------- PAYROLL.PROD.INPUT LIMIT----------5 SCRATCH NOEMPTY """ SAMPLE_VSAM = """CLUSTER ------- CICS.PROD.KSDS.ORDER IN-CAT --- CATALOG.CICS CI-SPLITS-----3200 CA-SPLITS-----450 HI-USED-RBA---0000000099000000 HI-ALLOC-RBA---0000000100000000 FREESPACE----(0 0) SHAREOPTIONS-(2 3) CLUSTER ------- DB2.PROD.KSDS.INDEX IN-CAT --- CATALOG.DB2 CI-SPLITS-----650 CA-SPLITS-----34 HI-USED-RBA---0000000050000000 HI-ALLOC-RBA---0000000080000000 """ SAMPLE_SECURITY = """//SECJOB01 JOB (SEC),'SECURITY CHECK',CLASS=A,MSGCLASS=X //STEP010 EXEC PGM=IDCAMS //SYSIN DD * DELETE SYS1.RACF.BACKUP PURGE /* //STEP020 EXEC PGM=IEBGENER //SYSUT1 DD DSN=SYS1.RACF.BACKUP,DISP=OLD //SYSUT2 DD DSN=HR.PROD.PAYROLL.EXTRACT,DISP=(NEW,CATLG,DELETE),SPACE=(TRK,(2,1)),UNIT=SYSDA NONVSAM ------- SYS1.RACF.BACKUP IN-CAT --- CATALOG.SECURE NONVSAM ------- HR.PROD.PAYROLL.EXTRACT IN-CAT --- CATALOG.HR """ SAMPLE_GDG = """//GDGJOB01 JOB (ACCT),'GDG CHECK',CLASS=A,MSGCLASS=X //STEP010 EXEC PGM=IEFBR14 //OUT1 DD DSN=FINPROD.BILLING.HISTORY(+1),DISP=SHR,SPACE=(TRK,(1,1)),UNIT=SYSDA //IN1 DD DSN=FINPROD.BILLING.HISTORY(-1),DISP=SHR GDG BASE ------- FINPROD.BILLING.HISTORY LIMIT----------1 SCRATCH EMPTY """ SAMPLE_POLICY = { "sensitive_patterns": [r"\bPROD\b", r"\bPAYROLL\b", r"\bRACF\b", r"^SYS1\.", r"\bHR\b", r"\bFINPROD\b"], "allowed_hlq_patterns": [r"^[A-Z][A-Z0-9]{1,7}$", r"^SYS1$", r"^PAYROLL$", r"^CICS$", r"^DB2$", r"^HR$", r"^FINPROD$"], "profiles": { "production": { "space_primary_min_tracks": 20, "ci_split_warning": 100, "ci_split_critical": 1000, "ca_split_warning": 10, "ca_split_critical": 100, } } } def demo_text(name: str) -> str: demos = { "mixed": SAMPLE_MIXED, "baseline": SAMPLE_BASELINE, "vsam": SAMPLE_VSAM, "security": SAMPLE_SECURITY, "gdg": SAMPLE_GDG, } if name not in demos: raise SystemExit(f"Unknown demo {name}. Choose: {', '.join(sorted(demos))}") return demos[name] def load_policy(path: Optional[str]) -> Dict[str, Any]: policy = json.loads(json.dumps(DEFAULT_POLICY)) if path: with open(path, "r", encoding="utf-8") as handle: custom = json.load(handle) policy = deep_merge(policy, custom) return policy def analyze_input(input_text: str, source_name: str, policy: Dict[str, Any], profile: str, redact: bool) -> AnalysisResult: analyzer = DatasetCatalogAnalyzer(policy=policy, profile=profile, redact=redact) return analyzer.analyze_text(input_text, source_name=source_name) def main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser(description="Dataset / Catalog Doctor V2") parser.add_argument("input_file", nargs="?", help="Text file containing LISTCAT, IDCAMS, JCL dataset references, or mixed content") parser.add_argument("--demo", choices=["mixed", "baseline", "vsam", "security", "gdg"], help="Run with a built-in demo sample") parser.add_argument("--profile", default="production", choices=["production", "batch", "storage", "security", "training", "strict"], help="Analysis profile") parser.add_argument("--custom-policy", help="Custom JSON policy file") parser.add_argument("--compare-with", help="Baseline file to compare with") parser.add_argument("--redact", action="store_true", help="Redact dataset names in findings and report output") parser.add_argument("--json", dest="json_output", help="Write full analysis JSON") parser.add_argument("--html", dest="html_output", help="Write HTML report") parser.add_argument("--csv-findings", help="Write findings CSV") parser.add_argument("--csv-datasets", help="Write datasets CSV") parser.add_argument("--csv-references", help="Write JCL references CSV") parser.add_argument("--csv-categories", help="Write risk categories CSV") parser.add_argument("--csv-compare", help="Write baseline comparison CSV") parser.add_argument("--fail-on", choices=["WARNING", "ERROR", "CRITICAL"], help="Return non-zero if highest severity is at least this level") parser.add_argument("--write-sample-input", help="Write built-in mixed sample to a file") parser.add_argument("--write-baseline-input", help="Write built-in baseline sample to a file") parser.add_argument("--write-policy", help="Write sample policy JSON to a file") parser.add_argument("--export-rules", help="Write built-in rules catalog JSON to a file") args = parser.parse_args(argv) if args.write_sample_input: write_text(args.write_sample_input, SAMPLE_MIXED) print(f"Sample input written to {args.write_sample_input}") return 0 if args.write_baseline_input: write_text(args.write_baseline_input, SAMPLE_BASELINE) print(f"Baseline input written to {args.write_baseline_input}") return 0 if args.write_policy: write_text(args.write_policy, json.dumps(SAMPLE_POLICY, indent=2)) print(f"Sample policy written to {args.write_policy}") return 0 if args.export_rules: write_text(args.export_rules, json.dumps(RULE_CATALOG, indent=2)) print(f"Rules catalog written to {args.export_rules}") return 0 if args.demo: text = demo_text(args.demo) source = f"demo:{args.demo}" elif args.input_file: text = read_text(args.input_file) source = args.input_file else: parser.error("Provide input_file or --demo") return 2 policy = load_policy(args.custom_policy) result = analyze_input(text, source, policy, args.profile, args.redact) if args.compare_with: baseline_text = read_text(args.compare_with) baseline = analyze_input(baseline_text, args.compare_with, policy, args.profile, args.redact) result.compare = compare_results(result, baseline) print_console(result) if args.json_output: write_json(args.json_output, result) print(f"JSON written to {args.json_output}") if args.html_output: write_html(args.html_output, result) print(f"HTML report written to {args.html_output}") write_result_csvs( result, { "findings": args.csv_findings, "datasets": args.csv_datasets, "references": args.csv_references, "categories": args.csv_categories, "compare": args.csv_compare, }, ) for label, path in [ ("Findings CSV", args.csv_findings), ("Datasets CSV", args.csv_datasets), ("References CSV", args.csv_references), ("Categories CSV", args.csv_categories), ("Compare CSV", args.csv_compare), ]: if path: print(f"{label} written to {path}") if args.fail_on and SEVERITY_ORDER.get(result.highest_severity, 0) >= SEVERITY_ORDER[args.fail_on]: return 1 return 0 if __name__ == "__main__": raise SystemExit(main())