"""NiFi2.5 REST configuration; keeps TLS validation and creates isolated flow."""
import argparse
import json
import os
from pathlib import Path
import requests

def main():
    ap=argparse.ArgumentParser()
    ap.add_argument("--url",default="https://localhost:18443/nifi-api")
    ap.add_argument("--ca",type=Path,required=True,help="Trusted exported CA/server certificate with valid hostname")
    ap.add_argument("--username",default="handbook")
    ap.add_argument("--password-env",default="HANDBOOK_NIFI_PASSWORD")
    ap.add_argument("--start",action="store_true",help="Start newly created demo downstream first")
    a=ap.parse_args()
    password=os.environ[a.password_env]
    session=requests.Session();session.verify=str(a.ca)
    def api(method,path,**kwargs):
        response=session.request(method,a.url+path,timeout=30,**kwargs)
        response.raise_for_status()
        return response
    token=api("POST","/access/token",data={"username":a.username,"password":password}).text
    session.headers["Authorization"]="Bearer "+token
    about=api("GET","/flow/about").json()["about"]["version"]
    if not about.startswith("2.5."):raise RuntimeError("Expected NiFi2.5 API; inspect schema before changing baseline")
    blueprint=json.loads(Path(__file__).with_name("flow.json").read_text(encoding="utf-8"))
    root=api("GET","/flow/process-groups/root").json()["processGroupFlow"]
    if any(g.get("component",{}).get("name")==blueprint["name"] for g in root["flow"].get("processGroups",[])):
        raise RuntimeError("Demo group already exists; inspect it in UI, no automatic duplicate/overwrite")
    group=api("POST",f"/process-groups/{root['id']}/process-groups",
              json={"revision":{"version":0},"component":{"name":blueprint["name"]}}).json()["id"]
    types=api("GET","/flow/processor-types").json()["processorTypes"]
    created={}
    for i,spec in enumerate(blueprint["processors"]):
        found=[t for t in types if t["type"]==spec["type"] and t["bundle"]["version"].startswith("2.5.")]
        if len(found)!=1:raise RuntimeError("Missing/ambiguous processor bundle")
        entity=api("POST",f"/process-groups/{group}/processors",json={
          "revision":{"version":0},"component":{"name":spec["name"],"type":spec["type"],
           "bundle":found[0]["bundle"],"position":{"x":0,"y":i*200}}}).json()
        component=entity["component"]
        descriptors=component["config"].get("descriptors",{})
        properties=dict(spec.get("dynamicProperties",{}))
        for display,value in spec.get("displayProperties",{}).items():
            names=[k for k,d in descriptors.items() if d.get("displayName")==display or k==display]
            if len(names)!=1:raise RuntimeError("Property descriptor changed: "+display)
            properties[names[0]]=value
        config={"properties":properties,"autoTerminatedRelationships":spec.get("terminate",[])}
        if "period" in spec:config["schedulingPeriod"]=spec["period"]
        updated=api("PUT",f"/processors/{entity['id']}",json={
          "revision":entity["revision"],"component":{"id":entity["id"],"config":config}}).json()
        created[spec["name"]]=updated["id"]
    for edge in blueprint["connections"]:
        api("POST",f"/process-groups/{group}/connections",json={
          "revision":{"version":0},"component":{
            "source":{"id":created[edge["from"]],"groupId":group,"type":"PROCESSOR"},
            "destination":{"id":created[edge["to"]],"groupId":group,"type":"PROCESSOR"},
            "selectedRelationships":[edge["relationship"]],
            "backPressureObjectThreshold":100,"backPressureDataSizeThreshold":"1 MB"}})
    if a.start:
        for spec in reversed(blueprint["processors"]):
            pid=created[spec["name"]]
            entity=api("GET",f"/processors/{pid}").json()
            if entity["component"].get("validationStatus")!="VALID":
                raise RuntimeError("Wait for validation/inspect stopped group in UI")
            api("PUT",f"/processors/{pid}/run-status",
                json={"revision":entity["revision"],"state":"RUNNING"})
    print(json.dumps({"group_id":group,"created_stopped":not a.start,"nifi_version":about}))
    # API/schema/property validation happens on the real server; static JSON
    # validation of flow.json does not establish runtime compatibility.

if __name__=="__main__":main()