#!/usr/bin/env python3 import sys import json import re import argparse PLACEHOLDERS = {"none", "nessuno", "nessuna", "n/a", "non applicabile", "null"} def clean_list_item(item: str) -> str: item = item.strip() if item.startswith("- "): item = item[2:].strip() elif item.startswith("-"): item = item[1:].strip() return item def is_placeholder(item: str) -> bool: clean = clean_list_item(item).lower() return clean in PLACEHOLDERS or not clean def parse_list_field(val) -> list: if val is None: return [] items = [] if isinstance(val, list): for el in val: s = str(el).strip() if not is_placeholder(s): items.append(clean_list_item(s)) elif isinstance(val, str): lines = val.strip().splitlines() for line in lines: s = line.strip() if not s: continue if not is_placeholder(s): items.append(clean_list_item(s)) return items def parse_output(text: str) -> dict: fields = { "task_id": None, "esito_205": None, "repository": None, "branch": None, "commit": None, "consolidated_unit_id": None, "blocco_dati": None, "dates_used": [], "test_eseguiti": [], "note": [], "absorbed_legacy_fragments": [], "open_legacy_fragments": [], "raw_text": text } try: json_match = re.search(r'\{.*\}', text, re.DOTALL) if json_match: data = json.loads(json_match.group(0)) if isinstance(data, dict): source_dict = data.get("parsed", data) if isinstance(source_dict, dict): for k in fields: if k in ["raw_text"]: continue if k in ["dates_used", "test_eseguiti", "note", "absorbed_legacy_fragments", "open_legacy_fragments"]: if k in source_dict and source_dict[k] is not None: fields[k] = parse_list_field(source_dict[k]) else: if k in source_dict and source_dict[k] is not None: fields[k] = str(source_dict[k]).strip() except Exception: pass HEADER_MAP = { "TASK_ID": "task_id", "ESITO_205": "esito_205", "REPOSITORY": "repository", "BRANCH": "branch", "COMMIT": "commit", "CONSOLIDATED_UNIT_ID": "consolidated_unit_id", "UNITA_ID": "consolidated_unit_id", "BLOCCO_DATI": "blocco_dati", "ABSORBED_LEGACY_FRAGMENTS": "absorbed_legacy_fragments", "OPEN_LEGACY_FRAGMENTS": "open_legacy_fragments", "DATES_USED": "dates_used", "TEST_ESEGUITI": "test_eseguiti", "NOTE": "note", } lines = text.splitlines() current_key = None for line in lines: stripped = line.strip() if not stripped: continue header_match = None for h_name, field_key in HEADER_MAP.items(): pattern = rf"^(?:{h_name}|{field_key}):\s*(.*)$" m = re.match(pattern, stripped, re.IGNORECASE) if m: header_match = (field_key, m.group(1).strip()) break if header_match: field_key, inline_val = header_match current_key = field_key if field_key in ["dates_used", "test_eseguiti", "note", "absorbed_legacy_fragments", "open_legacy_fragments"]: if inline_val and not is_placeholder(inline_val): fields[field_key].append(clean_list_item(inline_val)) else: if not fields[field_key] and inline_val: fields[field_key] = inline_val elif current_key and current_key in ["dates_used", "test_eseguiti", "note", "absorbed_legacy_fragments", "open_legacy_fragments"]: if not is_placeholder(stripped): fields[current_key].append(clean_list_item(stripped)) for k in ["dates_used", "test_eseguiti", "note", "absorbed_legacy_fragments", "open_legacy_fragments"]: cleaned = [] for el in fields[k]: c = clean_list_item(str(el)) if c and not is_placeholder(c) and c not in cleaned: cleaned.append(c) fields[k] = cleaned return fields def validate_fields(fields: dict, expected_repo: str = None, expected_branch: str = None, expected_commit: str = None) -> tuple[bool, list[str]]: missing = [] if expected_repo and fields.get("repository") != expected_repo: missing.append("repository") elif not fields.get("repository"): missing.append("repository") if expected_branch and fields.get("branch") != expected_branch: missing.append("branch") elif not fields.get("branch"): missing.append("branch") if expected_commit and fields.get("commit") != expected_commit: missing.append("commit") elif not fields.get("commit"): missing.append("commit") mandatory_single_fields = [ "task_id", "esito_205", "consolidated_unit_id", "blocco_dati", ] for key in mandatory_single_fields: val = fields.get(key) if val is None or (isinstance(val, str) and not val.strip()): missing.append(key) mandatory_list_fields = [ "dates_used", "test_eseguiti", "note", ] for key in mandatory_list_fields: val = fields.get(key) if not isinstance(val, list) or len(val) == 0: missing.append(key) abs_frag = fields.get("absorbed_legacy_fragments") or [] open_frag = fields.get("open_legacy_fragments") or [] if len(abs_frag) == 0 and len(open_frag) == 0: missing.append("absorbed_legacy_fragments|open_legacy_fragments") dedup_missing = [] for item in missing: if item not in dedup_missing: dedup_missing.append(item) is_valid = len(dedup_missing) == 0 return is_valid, dedup_missing def main(): parser = argparse.ArgumentParser(description="Parse and validate 205 runner output.") parser.add_argument("--expected-repo", help="Expected git repository URL") parser.add_argument("--expected-branch", help="Expected git branch name") parser.add_argument("--expected-commit", help="Expected git commit hash") args = parser.parse_args() input_text = sys.stdin.read() fields = parse_output(input_text) is_valid, missing = validate_fields( fields, expected_repo=args.expected_repo, expected_branch=args.expected_branch, expected_commit=args.expected_commit ) if is_valid: output_data = { "ok": True, "parsed": fields } print(json.dumps(output_data, ensure_ascii=False, indent=2)) sys.exit(0) else: output_data = { "ok": False, "missing": missing, "parsed": fields } print(json.dumps(output_data, ensure_ascii=False, indent=2)) sys.exit(1) if __name__ == "__main__": main()