#!/usr/bin/env python3
"""Read-only HubSpot/LinkedIn handoff check. No account access or mutation."""
import base64
import json
import sys
from datetime import datetime
from pathlib import Path

SCHEMA = "maestro.hubspot-linkedin-integration.v1"
STEPS = ("hubspot_ads_permission", "linkedin_account_role", "account_status", "app_authorization", "connection", "reporting", "auto_tracking", "final_url", "redirect", "tracking_parameters", "page_role", "lead_sync", "submission", "contact", "ad_event", "form_event", "routing")
COUNTS = ("submitted", "synced", "contacts", "ad_events", "form_events", "routed_contacts", "duplicate_submissions", "resolved_submissions", "rejected_submissions", "unresolved_submissions")
OWNERS = {"hubspot_ads_permission":"HubSpot portal admin", "linkedin_account_role":"LinkedIn ad-account owner", "account_status":"LinkedIn ad-account owner", "app_authorization":"LinkedIn app owner", "connection":"HubSpot Ads owner", "reporting":"HubSpot Ads owner", "auto_tracking":"HubSpot Ads owner", "final_url":"LinkedIn campaign owner", "redirect":"Web destination owner", "tracking_parameters":"HubSpot Ads owner", "page_role":"LinkedIn Page admin", "lead_sync":"HubSpot lead-sync owner", "submission":"LinkedIn form owner", "contact":"HubSpot CRM owner", "ad_event":"HubSpot CRM owner", "form_event":"HubSpot CRM owner", "routing":"HubSpot workflow owner", "test_authorization":"LinkedIn form owner", "population_reconciliation":"Marketing operations owner", "population_evidence":"Marketing operations owner", "planning":"Marketing operations owner", "import_contract":"Record owner", "report_filter":"HubSpot Ads owner"}
ACTION = {"hubspot_ads_permission":"Read back Publish access to HubSpot Ads for the connecting user in the named portal.", "linkedin_account_role":"Read back the signed-in LinkedIn user and Account Manager role for the exact ad account.", "account_status":"Read back the ad-account hold status; resolve a hold before reconnecting.", "app_authorization":"Read back current app permissions with the original LinkedIn user; review all connected portals before any reset.", "connection":"Read back the connected ad-account ID and LinkedIn user in HubSpot Ads settings.", "reporting":"Read back the exact campaign/ad and selected date range in HubSpot Ads.", "auto_tracking":"Read back auto-tracking for this exact ad account.", "final_url":"Read back the full final destination URL for the actual ad.", "redirect":"Trace the final destination and remove third-party or HTTP-to-HTTPS redirects before checking parameters.", "tracking_parameters":"Read back applied UTM and hsa parameters on the eligible ad, not a planned template.", "page_role":"Read back the connected user's qualifying role on the exact LinkedIn Page.", "lead_sync":"Read back selected lead-sync account, Page/form scope and Leads-tab Error types.", "submission":"Provide the authorized LinkedIn submission ID and timestamp for the exact form and cutoff.", "contact":"Read back the matched HubSpot contact ID and whether it existed before the submission.", "ad_event":"Read back the contact's LinkedIn ad event for this submission.", "form_event":"Read back the contact's lead-ad form submission event for this form.", "routing":"Read back actual workflow enrollment or owner assignment for the contact and submission window.", "test_authorization":"Provide the form owner's dated authorization for the exact test cohort, Page and form before submitting.", "population_reconciliation":"Name each unresolved submission ID, its last proven receipt and the next native receipt needed to resolve or reject it.", "population_evidence":"Provide a dated, same-scope native count for the named population. Keep blank, Unknown and evidenced zero distinct.", "planning":"Complete the named planned field without inventing an observed result.", "import_contract":"Repair the named field and reevaluate the preserved source.", "report_filter":"Read back the contact events, whether the contact predated the form, the report view and date range before diagnosing sync."}
PLAN = ("organization", "portal_id", "linkedin_user_ref", "ad_account_id", "page_id", "form_id", "campaign_id", "ad_id", "ad_format", "connection_owner", "lead_sync_owner", "routing_owner", "test_authorized_by")
ALLOWED_FORMATS = {"Sponsored Content with Lead Gen Form", "Single Image", "InMail with Lead Gen Form", "Text", "Carousel Image", "Video", "Job Spotlight", "Boosted Social"}
UNSUPPORTED_FORMATS = {"InMail without Lead Gen Form", "Event Ad", "Document Ad", "Thought Leader Ad"}
NATIVE_TYPES = {"native_readback", "native_receipt", "source_report", "fictional_example_receipt"}
INPUT_RAW = ""
INPUT_BYTES = b""

def kind(value):
    if value is None: return "null"
    if isinstance(value, bool): return "boolean"
    if isinstance(value, dict): return "object"
    if isinstance(value, list): return "array"
    if isinstance(value, str): return "string"
    if isinstance(value, (int, float)): return "number"
    return type(value).__name__

def timestamp(value):
    if not isinstance(value, str) or not value: return False
    try: return datetime.fromisoformat(value.replace("Z", "+00:00")).tzinfo is not None
    except ValueError: return False

def instant(value):
    return datetime.fromisoformat(value.replace("Z", "+00:00"))

def diagnostic(path, expected, value, message, missing=False):
    return {"path":path, "expected_kind":expected, "received_kind":"missing" if missing else kind(value), "value":"<missing>" if missing else value, "message":message}

def result(source, disposition, state, path, reason, diagnostics=()):
    owner = OWNERS.get(state, OWNERS["import_contract"])
    last_proven = None
    cutoff = source.get("cutoff") if isinstance(source, dict) and isinstance(source.get("cutoff"), dict) else None
    if isinstance(source, dict):
        observations = source.get("observations") if isinstance(source.get("observations"), dict) else {}
        if state in observations and isinstance(observations[state], dict):
            supplied_owner = observations[state].get("owner")
            if isinstance(supplied_owner, str) and supplied_owner.strip(): owner = supplied_owner.strip()
        elif state == "test_authorization":
            supplied_owner = source.get("plan", {}).get("test_authorized_by") if isinstance(source.get("plan"), dict) else None
            if isinstance(supplied_owner, str) and supplied_owner.strip(): owner = supplied_owner.strip()
        refs = []
        if state in observations and isinstance(observations[state], dict): refs.append(observations[state].get("evidence_ref"))
        if state in STEPS:
            for step in reversed(STEPS[:STEPS.index(state)]):
                row = observations.get(step)
                if isinstance(row, dict) and row.get("evidence_ref"): refs.append(row["evidence_ref"])
        if state in ("population_evidence", "population_reconciliation") and isinstance(source.get("populations"), dict):
            name = path.split(".")[1] if isinstance(path, str) and path.startswith("populations.") and len(path.split(".")) > 1 else "unresolved_submissions"
            row = source["populations"].get(name)
            if isinstance(row, dict): refs.insert(0, row.get("evidence_ref"))
        if state == "report_filter" and isinstance(source.get("report_diagnosis"), dict): refs.insert(0, source["report_diagnosis"].get("evidence_ref"))
        if state == "test_authorization" and isinstance(source.get("plan"), dict): refs.insert(0, source["plan"].get("test_authorization_ref"))
        evidence = {e.get("id"):e for e in source.get("evidence", []) if isinstance(e, dict) and e.get("id")} if isinstance(source.get("evidence"), list) else {}
        for ref in refs:
            item = evidence.get(ref)
            if item:
                last_proven = {"evidence_ref":ref, "observed_at":item.get("observed_at"), "detail":item.get("detail")}
                break
    request = {"id":"HL-"+state.upper(), "owner":owner, "path":path, "receipt":ACTION.get(state, ACTION["import_contract"]), "last_proven_receipt":last_proven, "cutoff":cutoff, "privacy_note":"Keep contact personal data, tokens and credentials in approved systems; share only safe IDs or redacted receipt references.", "reason":reason} if state else None
    return {"schema_version":"maestro.hubspot-linkedin-check.v1", "disposition":disposition, "first_broken_state":state, "path":path, "reason":reason, "next_request":request, "diagnostics":list(diagnostics), "source":source, "input_preserved_json":INPUT_RAW, "input_bytes_base64":base64.b64encode(INPUT_BYTES).decode("ascii")}

def check(source):
    if not isinstance(source, dict):
        d=diagnostic("$", "object", source, "Record must be an object.")
        return result(source,"REPAIR_REQUIRED","import_contract","$",d["message"],[d])
    errors=[]
    def require(obj, path, key, expected):
        if key not in obj: errors.append(diagnostic(path+key,expected,None,"Required member is missing.",True)); return
        v=obj[key]
        if kind(v)!=expected: errors.append(diagnostic(path+key,expected,v,"Required member has the wrong kind."))
    for key,typ in (("schema_version","string"),("record_id","string"),("example_only","boolean"),("evaluation_phase","string"),("scope","object"),("plan","object"),("cutoff","object"),("observations","object"),("populations","object"),("evidence","array"),("report_diagnosis","object")):
        require(source,"",key,typ)
    if errors: return result(source,"REPAIR_REQUIRED","import_contract",errors[0]["path"],errors[0]["message"],errors)
    if source["schema_version"]!=SCHEMA or source["evaluation_phase"] not in ("draft","launch","post_test"):
        path="schema_version" if source["schema_version"]!=SCHEMA else "evaluation_phase"
        d=diagnostic(path,"supported literal",source[path],"Unsupported schema or phase.")
        return result(source,"REPAIR_REQUIRED","import_contract",path,d["message"],[d])
    scope,plan,cutoff,obs,pops= (source[x] for x in ("scope","plan","cutoff","observations","populations"))
    for key in PLAN: require(scope,"scope.",key,"string") if key in PLAN[:9] else require(plan,"plan.",key,"string")
    for key in ("destination_url","planned_route","planned_report_view","test_authorized_at","test_authorization_ref"): require(plan,"plan.",key,"string")
    for key in ("start","end","timezone","cohort","ad_account_id","page_id","form_id"): require(cutoff,"cutoff.",key,"string")
    for key in ("contact_existed_before","view","contact_absent_from_report","evidence_ref"): require(source["report_diagnosis"],"report_diagnosis.",key,"string")
    for step in STEPS:
        require(obs,"observations.",step,"object")
        row=obs.get(step)
        if isinstance(row,dict):
            for key in ("state","value","evidence_ref","observed_at","owner"): require(row,"observations."+step+".",key,"string")
            if isinstance(row.get("state"),str) and row["state"] not in ("BLANK","UNKNOWN","PASS","FAIL","NOT_APPLICABLE"):
                errors.append(diagnostic("observations."+step+".state","BLANK|UNKNOWN|PASS|FAIL|NOT_APPLICABLE",row["state"],"Invalid observation state."))
    for name in COUNTS:
        require(pops,"populations.",name,"object")
        row=pops.get(name)
        if isinstance(row,dict):
            for key in ("state","evidence_ref","captured_at","definition"): require(row,"populations."+name+".",key,"string")
            require(row,"populations."+name+".","scope","object")
            if "value" not in row: errors.append(diagnostic("populations."+name+".value","null or nonnegative integer",None,"Required member is missing.",True))
            elif row.get("state") == "KNOWN" and (kind(row["value"])!="number" or not isinstance(row["value"],int) or row["value"]<0 or row["value"]>9007199254740991): errors.append(diagnostic("populations."+name+".value","safe nonnegative integer",row["value"],"Known count needs a safe nonnegative integer."))
            elif row.get("state") in ("BLANK","UNKNOWN") and row["value"] is not None: errors.append(diagnostic("populations."+name+".value","null",row["value"],"Blank and Unknown cannot have a count."))
            if row.get("state") not in ("BLANK","UNKNOWN","KNOWN"): errors.append(diagnostic("populations."+name+".state","BLANK|UNKNOWN|KNOWN",row.get("state"),"Invalid population state."))
    for i,e in enumerate(source["evidence"]):
        if not isinstance(e,dict): errors.append(diagnostic(f"evidence[{i}]","object",e,"Evidence member must be an object.")); continue
        for key in ("id","type","observed_at","scope","detail"): require(e,f"evidence[{i}].",key,"object" if key=="scope" else "string")
    if errors: return result(source,"REPAIR_REQUIRED","import_contract",errors[0]["path"],errors[0]["message"],errors)
    if source["example_only"] is False and any(e["type"]=="fictional_example_receipt" for e in source["evidence"]):
        return result(source,"REPAIR_REQUIRED","import_contract","evidence","Fictional evidence cannot verify a live record.")
    evidence={e["id"]:e for e in source["evidence"] if e["id"]}
    expected={"portal_id":scope["portal_id"],"linkedin_user_ref":scope["linkedin_user_ref"],**{k:cutoff[k] for k in ("ad_account_id","page_id","form_id")}}
    def valid_receipt(ref,at):
        e=evidence.get(ref)
        return bool(e and e["type"] in NATIVE_TYPES and timestamp(e["observed_at"]) and e["observed_at"]==at and all(e["scope"].get(k)==v for k,v in expected.items()) and e["detail"].strip())
    for field in ("ad_account_id","page_id","form_id"):
        if cutoff[field] and scope[field] and cutoff[field]!=scope[field]: return result(source,"REPAIR_REQUIRED","import_contract","cutoff."+field,"Cutoff scope differs from the planned identity.")
    if scope["ad_format"] and scope["ad_format"] not in ALLOWED_FORMATS|UNSUPPORTED_FORMATS:
        return result(source,"REPAIR_REQUIRED","import_contract","scope.ad_format","Ad format is outside the documented support matrix.")
    missing=[("scope."+k if k in scope else "plan."+k) for k in PLAN if k!="test_authorized_by" and not (scope.get(k) if k in scope else plan.get(k))]
    missing += ["cutoff."+k for k in ("start","end","timezone","cohort","ad_account_id","page_id","form_id") if not cutoff[k]]
    if scope["ad_format"] in ALLOWED_FORMATS and not plan["destination_url"]: missing.append("plan.destination_url")
    if not plan["planned_route"]: missing.append("plan.planned_route")
    if missing: return result(source,"LAUNCH_HOLD","planning",missing[0],"Planned scope or authorization is incomplete: "+", ".join(missing))
    if not timestamp(cutoff["start"]) or not timestamp(cutoff["end"]) or cutoff["start"]>=cutoff["end"]:
        return result(source,"REPAIR_REQUIRED","import_contract","cutoff","Cutoff needs ordered ISO-8601 timestamps with time zones.")
    if source["evaluation_phase"]!="draft":
        auth=evidence.get(plan["test_authorization_ref"])
        if not plan["test_authorized_by"] or not timestamp(plan["test_authorized_at"]) or instant(plan["test_authorized_at"]) >= instant(cutoff["start"]) or not auth or auth["type"] not in ("operator_authorization","fictional_example_receipt") or (auth["type"]=="fictional_example_receipt" and not source["example_only"]) or auth["observed_at"]!=plan["test_authorized_at"] or any(auth["scope"].get(k)!=v for k,v in expected.items()) or not auth["detail"].strip():
            return result(source,"LAUNCH_HOLD","test_authorization","plan.test_authorization_ref","The planned test lacks a dated, same-scope owner authorization receipt.")
    role_values={"hubspot_ads_permission":{"Publish"},"linkedin_account_role":{"Account Manager"},"page_role":{"Super admin","Content admin","Lead Gen Forms Manager"}}
    exact={"app_authorization":("current","Current app authorization must be read back."),"connection":(scope["ad_account_id"],"Connected account ID differs from planned scope."),"auto_tracking":("on","Auto-tracking is not on for the scoped account."),"final_url":(plan["destination_url"],"Observed final URL differs from the planned destination.")}
    needle_values={"reporting":(scope["campaign_id"],scope["ad_id"]),"lead_sync":(scope["ad_account_id"],scope["page_id"],scope["form_id"]),"tracking_parameters":("utm_source=linkedin","utm_medium=paid","hsa_")}
    for step in STEPS:
        if scope["ad_format"] in UNSUPPORTED_FORMATS and step in ("auto_tracking","final_url","redirect","tracking_parameters"): continue
        row=obs[step]
        if row["state"] in ("PASS","FAIL") and (not row["value"].strip() or not valid_receipt(row["evidence_ref"],row["observed_at"])):
            return result(source,"LAUNCH_HOLD",step,"observations."+step,"An observed state needs a specific dated native receipt for this portal, LinkedIn user, account, Page and form.")
        if row["state"]=="FAIL": return result(source,"REPAIR_REQUIRED",step,"observations."+step,"Observed failure: "+row["value"])
        if row["state"]=="PASS" and step in role_values and row["value"] not in role_values[step]:
            return result(source,"REPAIR_REQUIRED",step,"observations."+step+".value","The observed role does not satisfy this permission.")
        if row["state"]=="PASS" and step=="account_status" and row["value"]!="active":
            return result(source,"REPAIR_REQUIRED",step,"observations.account_status.value","An account on hold cannot be treated as active.")
        if row["state"]=="PASS" and step=="redirect" and row["value"]!="none":
            return result(source,"REPAIR_REQUIRED",step,"observations.redirect.value","A redirect cannot be marked clear.")
        if row["state"]=="PASS" and step in exact and row["value"]!=exact[step][0]:
            return result(source,"REPAIR_REQUIRED",step,"observations."+step+".value",exact[step][1])
        if row["state"]=="PASS" and step in needle_values and any(n not in row["value"] for n in needle_values[step]):
            return result(source,"REPAIR_REQUIRED",step,"observations."+step+".value","Observed value omits a required scoped ID or applied parameter.")
    if source["report_diagnosis"]["contact_absent_from_report"]=="YES":
        d=source["report_diagnosis"]
        if d["contact_existed_before"]=="YES" and d["view"]=="First form submission" and obs["contact"]["state"]=="PASS" and obs["ad_event"]["state"]=="PASS" and obs["form_event"]["state"]=="PASS":
            return result(source,"REPAIR_REQUIRED","report_filter","report_diagnosis.view","A pre-existing contact with both events may be hidden by this report view; check the view and date range.")
    for name in COUNTS:
        row=pops[name]
        if row["state"]=="KNOWN":
            if row["scope"]!=cutoff or not timestamp(row["captured_at"]) or instant(row["captured_at"]) < instant(cutoff["end"]) or not valid_receipt(row["evidence_ref"],row["captured_at"]):
                return result(source,"LAUNCH_HOLD","population_evidence","populations."+name,"Count lacks a same-scope dated native receipt.")
    unresolved=pops["unresolved_submissions"]
    if unresolved["state"]=="KNOWN" and unresolved["value"]>0:
        return result(source,"REPAIR_REQUIRED","population_reconciliation","populations.unresolved_submissions.value",f"{unresolved['value']} submission(s) remain unresolved at the cutoff.")
    known={k:pops[k]["value"] for k in COUNTS if pops[k]["state"]=="KNOWN"}
    if len(known)==len(COUNTS):
        minimum_resolved_contacts=max(1 if known["resolved_submissions"] else 0,known["resolved_submissions"]-known["duplicate_submissions"])
        checks=((known["submitted"]==known["resolved_submissions"]+known["rejected_submissions"]+known["unresolved_submissions"],"Submitted IDs must partition into resolved, rejected and unresolved."),(known["synced"]<=known["submitted"],"Synced submissions exceed submitted IDs."),(known["contacts"]<=known["synced"],"Distinct contacts exceed synced submissions."),(known["routed_contacts"]<=known["contacts"],"Routed contacts exceed distinct contacts."),(known["duplicate_submissions"]<=known["submitted"],"Duplicate submissions exceed submitted IDs."),(known["resolved_submissions"]<=known["synced"],"Resolved submissions exceed synced submission receipts."),(known["resolved_submissions"]<=known["ad_events"],"Resolved submissions exceed evidenced in-cohort ad events."),(known["resolved_submissions"]<=known["form_events"],"Resolved submissions exceed evidenced in-cohort form events."),(known["contacts"]>=minimum_resolved_contacts,"Resolved submissions lack the minimum distinct contact receipts after duplicate mapping."),(known["routed_contacts"]>=minimum_resolved_contacts,"Resolved submissions lack the minimum routed-contact receipts after duplicate mapping."))
        for ok,why in checks:
            if not ok: return result(source,"REPAIR_REQUIRED","population_reconciliation","populations",why)
        for step,name in (("contact","contacts"),("ad_event","ad_events"),("form_event","form_events"),("routing","routed_contacts")):
            if obs[step]["state"]=="PASS" and known[name]==0:
                return result(source,"REPAIR_REQUIRED","population_reconciliation","populations."+name+".value",f"{step} is marked PASS while the evidenced {name} population is zero.")
    if source["evaluation_phase"]=="draft": return result(source,"DRAFT_READY",None,None,"Planned record is reviewable; native verification is still pending.")
    for step in STEPS:
        if scope["ad_format"] in UNSUPPORTED_FORMATS and step in ("auto_tracking","final_url","redirect","tracking_parameters"):
            if obs[step]["state"]!="NOT_APPLICABLE" or not obs[step]["value"].strip(): return result(source,"LAUNCH_HOLD",step,"observations."+step,"Document the unsupported format and its source; do not claim tracking passed.")
            continue
        if obs[step]["state"]!="PASS": return result(source,"LAUNCH_HOLD",step,"observations."+step,"Native state is blank or Unknown; obtain the next receipt.")
    for name in COUNTS:
        if pops[name]["state"]!="KNOWN": return result(source,"LAUNCH_HOLD","population_evidence","populations."+name,"Population is blank or Unknown at the cutoff.")
    if known["submitted"]<1: return result(source,"LAUNCH_HOLD","submission","populations.submitted.value","Verification requires at least one authorized test submission.")
    if known["unresolved_submissions"]!=0: return result(source,"REPAIR_REQUIRED","population_reconciliation","populations.unresolved_submissions.value","Unresolved count must be evidenced zero.")
    return result(source,"VERIFIED",None,None,"Dated native receipts and a disjoint zero-unresolved cutoff reconcile.")

def markdown(out):
    lines=["# HubSpot–LinkedIn integration check", "", "**Decision:** "+out["disposition"], "", out["reason"], ""]
    if out["next_request"]:
        q=out["next_request"]
        last = "None recorded" if q["last_proven_receipt"] is None else json.dumps(q["last_proven_receipt"], ensure_ascii=False, separators=(",", ":"))
        cutoff = "Not available" if q["cutoff"] is None else json.dumps(q["cutoff"], ensure_ascii=False, separators=(",", ":"))
        lines += ["## Next evidence request", "", "- ID: `"+q["id"]+"`", "- Owner: "+q["owner"], "- Field: `"+str(q["path"])+"`", "- Last proven receipt: "+last, "- Receipt/action: "+q["receipt"], "- Cutoff: "+cutoff, "- Privacy: "+q["privacy_note"], ""]
    if out["diagnostics"]:
        lines += ["## Source diagnostics", ""]
        for d in out["diagnostics"]: lines.append("- `"+d["path"]+"`: expected "+d["expected_kind"]+", received "+d["received_kind"]+"; value `"+json.dumps(d["value"],ensure_ascii=False,separators=(",",":"))+"`.")
        lines.append("")
    lines += ["## Preserved input bytes", "", "The exact UTF-8 source follows. The JSON result also includes `input_bytes_base64` for byte-exact recovery.", "", "```json", out["input_preserved_json"], "```", ""]
    return "\n".join(lines)

def main():
    if len(sys.argv)<2 or len(sys.argv)>3 or (len(sys.argv)==3 and sys.argv[2]!="--markdown"):
        print("usage: hubspot-linkedin-checker.py INPUT.json [--markdown]",file=sys.stderr); return 2
    global INPUT_RAW, INPUT_BYTES
    INPUT_BYTES=Path(sys.argv[1]).read_bytes()
    try: raw=INPUT_BYTES.decode("utf-8")
    except UnicodeDecodeError:
        raw=INPUT_BYTES.decode("utf-8",errors="replace")
        INPUT_RAW=raw
        d={"path":"$","expected_kind":"UTF-8 JSON text","received_kind":"invalid bytes","value":base64.b64encode(INPUT_BYTES).decode("ascii"),"message":"Input is not valid UTF-8; exact bytes are preserved in base64."}
        out=result(raw,"REPAIR_REQUIRED","import_contract","$","Input is not valid UTF-8; exact bytes are preserved in base64.",[d])
    else:
        INPUT_RAW=raw
        def reject_constant(_): raise ValueError("Nonstandard JSON constant")
        try: source=json.loads(raw,parse_constant=reject_constant); out=check(source)
        except (json.JSONDecodeError,ValueError):
            d={"path":"$","expected_kind":"valid JSON","received_kind":"malformed text","value":raw,"message":"Invalid JSON text."}
            out=result(raw,"REPAIR_REQUIRED","import_contract","$","JSON could not be parsed; original text is preserved.",[d])
    print(markdown(out) if len(sys.argv)==3 else json.dumps(out,ensure_ascii=False,indent=2))
    return 0

if __name__=="__main__": raise SystemExit(main())