145 lines
4.7 KiB
Python
Executable File
145 lines
4.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import sys
|
|
import json
|
|
import re
|
|
import argparse
|
|
|
|
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": None,
|
|
"test_eseguiti": None,
|
|
"note": None,
|
|
"absorbed_legacy_fragments": None,
|
|
"open_legacy_fragments": None,
|
|
"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)
|
|
for k in fields:
|
|
if k in source_dict and source_dict[k] is not None:
|
|
fields[k] = source_dict[k]
|
|
except Exception:
|
|
pass
|
|
|
|
patterns = {
|
|
"task_id": r"(?:TASK_ID|task_id):\s*(.+)",
|
|
"esito_205": r"(?:ESITO_205|esito_205):\s*(.+)",
|
|
"repository": r"(?:REPOSITORY|repository):\s*(.+)",
|
|
"branch": r"(?:BRANCH|branch):\s*(.+)",
|
|
"commit": r"(?:COMMIT|commit):\s*(.+)",
|
|
"consolidated_unit_id": r"(?:CONSOLIDATED_UNIT_ID|UNITA_ID|consolidated_unit_id):\s*(.+)",
|
|
"blocco_dati": r"(?:BLOCCO_DATI|blocco_dati):\s*(.+)",
|
|
"dates_used": r"(?:DATES_USED|dates_used):\s*(.+)",
|
|
"test_eseguiti": r"(?:TEST_ESEGUITI|test_eseguiti):\s*(.+)",
|
|
"note": r"(?:NOTE|note):\s*(.+)",
|
|
"absorbed_legacy_fragments": r"(?:ABSORBED_LEGACY_FRAGMENTS|absorbed_legacy_fragments):\s*(.+)",
|
|
"open_legacy_fragments": r"(?:OPEN_LEGACY_FRAGMENTS|open_legacy_fragments):\s*(.+)"
|
|
}
|
|
|
|
for key, pat in patterns.items():
|
|
if fields[key] is None:
|
|
m = re.search(pat, text, re.IGNORECASE)
|
|
if m:
|
|
val = m.group(1).strip()
|
|
fields[key] = val
|
|
|
|
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",
|
|
"dates_used",
|
|
"test_eseguiti",
|
|
"note",
|
|
]
|
|
|
|
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)
|
|
|
|
abs_frag = fields.get("absorbed_legacy_fragments")
|
|
open_frag = fields.get("open_legacy_fragments")
|
|
has_abs = abs_frag is not None and (not isinstance(abs_frag, str) or bool(str(abs_frag).strip()))
|
|
has_open = open_frag is not None and (not isinstance(open_frag, str) or bool(str(open_frag).strip()))
|
|
|
|
if not (has_abs or has_open):
|
|
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()
|