netgescon-day0/scripts/ops/antigravity-cli/parse_205_runner_output.py

#!/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()