332 lines
14 KiB
Python
332 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Create Langflow global variables, API key, and both PM flows via API.
|
|
|
|
Idempotent: variables are upserted; existing flows with the same name are
|
|
deleted and recreated. Run AFTER deploy_files.sh. Needs secrets.local.env
|
|
sourced into the environment (PM_IMAP_PASSWORD, PM_ESPO_API_KEY,
|
|
PM_ESPO_BASE).
|
|
|
|
Node/edge shapes and the two builtin_template category keys below were
|
|
verified live against Langflow 1.10.0's GET /api/v1/all and the existing
|
|
"vLLM Smoke Test" flow's edge dump (see task-10-report.md for the raw
|
|
evidence). Notable deviations from the naive assumption:
|
|
- Webhook lives in the "input_output" category, not "data".
|
|
- Every custom Component's Data-typed `out` output reports its Langflow
|
|
wire type as "JSON" (not "Data") in outputs[].types.
|
|
- DataInput/JSONInput fields (`ctx_in`, `trigger`) report input_types as
|
|
the full ["Data", "JSON"], not just ["Data"].
|
|
Edges are built from the live template's outputs[].types /
|
|
template[field].input_types instead of hardcoded guesses, so any future
|
|
Langflow version drift here is easy to re-derive by rerunning the Step-3
|
|
probe in the task brief.
|
|
|
|
IMPORTANT (found while debugging Step 6, see task-10-report.md): Langflow's
|
|
`Component` base class defines `ctx` as a reserved @property (returns
|
|
`self.graph.context`, the flow-level shared context store). The Task-9
|
|
components originally named their inter-stage DataInput field "ctx", which
|
|
silently shadows the framework property at attribute-read time -- the edge
|
|
correctly delivers data into `self._attributes["ctx"]`, but `self.ctx`
|
|
inside build_out() resolves to the *class property* instead (an empty
|
|
dotdict), not the input. There is no error; the payload just silently comes
|
|
back empty one hop downstream. Task 10 renamed the field to `ctx_in` in all
|
|
five affected components (pm_extract.py, pm_match.py, pm_rules.py,
|
|
pm_crm.py, pm_notify.py); the edges below target "ctx_in".
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
BASE = "http://127.0.0.1:8090/api/v1"
|
|
COMPONENTS = Path(__file__).resolve().parent.parent / "components"
|
|
ENV_FILE = Path.home() / ".config" / "projekt-matching" / "env"
|
|
|
|
GENERIC_VARS = {
|
|
"PM_ESPO_BASE": os.environ.get("PM_ESPO_BASE", ""),
|
|
"PM_NOTIFY_TO": "chancen@destengs.com",
|
|
"PM_ALERT_TO": "chancen@destengs.com",
|
|
"PM_THRESHOLD": "85",
|
|
"PM_VLLM_BASE": "http://host.containers.internal:8081/v1",
|
|
"PM_VLLM_MODEL": "AxionML/Qwen3.5-9B-NVFP4",
|
|
"PM_FLOW2_ID": "placeholder",
|
|
}
|
|
SECRET_VARS = ("PM_IMAP_PASSWORD", "PM_ESPO_API_KEY")
|
|
|
|
|
|
def login():
|
|
r = requests.get(f"{BASE}/auto_login", timeout=15)
|
|
r.raise_for_status()
|
|
token = r.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def upsert_variable(hdr, name, value, vtype):
|
|
existing = {v["name"]: v["id"] for v in
|
|
requests.get(f"{BASE}/variables/", headers=hdr,
|
|
timeout=15).json()}
|
|
body = {"name": name, "value": value, "type": vtype,
|
|
"default_fields": []}
|
|
if name in existing:
|
|
requests.patch(f"{BASE}/variables/{existing[name]}", headers=hdr,
|
|
json={"id": existing[name], "name": name,
|
|
"value": value}, timeout=15).raise_for_status()
|
|
else:
|
|
requests.post(f"{BASE}/variables/", headers=hdr, json=body,
|
|
timeout=15).raise_for_status()
|
|
|
|
|
|
def get_api_key(hdr):
|
|
# Idempotent delete+recreate: the key VALUE is only returned at creation
|
|
# time, so reuse is impossible -- instead remove any previous
|
|
# "projekt-matching" keys (each earlier run left one behind otherwise)
|
|
# and mint a fresh one. Callers then refresh both the env file and the
|
|
# PM_LANGFLOW_API_KEY Credential variable, so nothing still references
|
|
# the deleted keys. Live API shape (Langflow 1.10.0): GET /api_key/
|
|
# returns {"total_count", "user_id", "api_keys": [{"id", "name", ...}]};
|
|
# DELETE /api_key/{id} removes one key.
|
|
r = requests.get(f"{BASE}/api_key/", headers=hdr, timeout=15)
|
|
r.raise_for_status()
|
|
for key in r.json().get("api_keys", []):
|
|
if key["name"] == "projekt-matching":
|
|
requests.delete(f"{BASE}/api_key/{key['id']}", headers=hdr,
|
|
timeout=15).raise_for_status()
|
|
r = requests.post(f"{BASE}/api_key/", headers=hdr,
|
|
json={"name": "projekt-matching"}, timeout=15)
|
|
r.raise_for_status()
|
|
return r.json()["api_key"]
|
|
|
|
|
|
def component_template(hdr, code):
|
|
r = requests.post(f"{BASE}/custom_component", headers=hdr,
|
|
json={"code": code}, timeout=60)
|
|
r.raise_for_status()
|
|
out = r.json()
|
|
return out["data"], out["type"]
|
|
|
|
|
|
def builtin_template(hdr, category, name):
|
|
r = requests.get(f"{BASE}/all", headers=hdr, timeout=60)
|
|
r.raise_for_status()
|
|
return r.json()[category][name]
|
|
|
|
|
|
def handle_str(d):
|
|
return json.dumps(d, separators=(",", ":")).replace('"', "œ")
|
|
|
|
|
|
def make_node(node_id, comp_type, template, x):
|
|
return {"id": node_id, "type": "genericNode",
|
|
"position": {"x": x, "y": 0},
|
|
"data": {"type": comp_type, "id": node_id, "node": template,
|
|
"showNode": True}}
|
|
|
|
|
|
def out_spec(template, out_name):
|
|
"""Look up an output's live wire types from a fetched node template."""
|
|
for o in template["outputs"]:
|
|
if o["name"] == out_name:
|
|
return o["types"]
|
|
raise KeyError(f"output {out_name!r} not found")
|
|
|
|
|
|
def in_spec(template, field_name):
|
|
"""Look up a field's live (input_types, type) from a node template."""
|
|
spec = template["template"][field_name]
|
|
return spec["input_types"], spec["type"]
|
|
|
|
|
|
def make_edge(src_id, src_type, out_name, out_types, tgt_id, field,
|
|
input_types, field_type):
|
|
sh = {"dataType": src_type, "id": src_id, "name": out_name,
|
|
"output_types": out_types}
|
|
th = {"fieldName": field, "id": tgt_id, "inputTypes": input_types,
|
|
"type": field_type}
|
|
return {"id": f"reactflow__edge-{src_id}-{tgt_id}",
|
|
"source": src_id, "target": tgt_id,
|
|
"sourceHandle": handle_str(sh), "targetHandle": handle_str(th),
|
|
"data": {"sourceHandle": sh, "targetHandle": th},
|
|
"className": ""}
|
|
|
|
|
|
def edge_between(src_id, src_type, src_tpl, out_name,
|
|
tgt_id, tgt_tpl, field):
|
|
out_types = out_spec(src_tpl, out_name)
|
|
input_types, field_type = in_spec(tgt_tpl, field)
|
|
return make_edge(src_id, src_type, out_name, out_types,
|
|
tgt_id, field, input_types, field_type)
|
|
|
|
|
|
def _matches_flow(f, name, endpoint):
|
|
fname, fep = f["name"], f.get("endpoint_name") or ""
|
|
return (fname == name or fname.startswith(f"{name} (")
|
|
or fep == endpoint
|
|
or (fep.startswith(f"{endpoint}-")
|
|
and fep[len(endpoint) + 1:].isdigit()))
|
|
|
|
|
|
def create_flow(hdr, name, endpoint, nodes, edges):
|
|
# Delete every previous incarnation. Langflow auto-renames on collision
|
|
# ("Name (1)", endpoint "endpoint-1") instead of rejecting, so a plain
|
|
# exact-name match leaves suffixed zombies behind that then keep the
|
|
# endpoint slot occupied on the next run -- match the auto-suffixed
|
|
# variants (by name AND endpoint) too, and fail loudly if a delete
|
|
# doesn't stick. Then wait until the listing no longer shows any of
|
|
# them: the create-time uniqueness check otherwise still sees the
|
|
# just-deleted rows and suffixes the new flow anyway (observed live
|
|
# on 1.10.0).
|
|
flows = requests.get(f"{BASE}/flows/", headers=hdr, timeout=30).json()
|
|
for f in flows:
|
|
if _matches_flow(f, name, endpoint):
|
|
requests.delete(f"{BASE}/flows/{f['id']}", headers=hdr,
|
|
timeout=15).raise_for_status()
|
|
for _ in range(20):
|
|
flows = requests.get(f"{BASE}/flows/", headers=hdr,
|
|
timeout=30).json()
|
|
if not any(_matches_flow(f, name, endpoint) for f in flows):
|
|
break
|
|
time.sleep(0.5)
|
|
|
|
body = {"name": name, "endpoint_name": endpoint, "is_component": False,
|
|
"data": {"nodes": nodes, "edges": edges,
|
|
"viewport": {"x": 0, "y": 0, "zoom": 0.5}}}
|
|
r = requests.post(f"{BASE}/flows/", headers=hdr, json=body, timeout=30)
|
|
r.raise_for_status()
|
|
created = r.json()
|
|
|
|
# Belt-and-braces: if the name/endpoint still got suffixed, PATCH them
|
|
# back. The PATCH can 404 transiently right after the POST (read-your-
|
|
# writes lag observed live), so retry briefly.
|
|
fixes = {}
|
|
if created["name"] != name:
|
|
fixes["name"] = name
|
|
if (created.get("endpoint_name") or "") != endpoint:
|
|
fixes["endpoint_name"] = endpoint
|
|
if fixes:
|
|
for attempt in range(10):
|
|
pr = requests.patch(f"{BASE}/flows/{created['id']}",
|
|
headers=hdr, json=fixes, timeout=15)
|
|
if pr.status_code == 404:
|
|
time.sleep(1)
|
|
continue
|
|
pr.raise_for_status()
|
|
break
|
|
else:
|
|
raise RuntimeError(
|
|
f"could not normalize flow {created['id']} to "
|
|
f"name={name!r} endpoint={endpoint!r}")
|
|
return created["id"]
|
|
|
|
|
|
def main():
|
|
hdr = login()
|
|
for name in SECRET_VARS:
|
|
value = os.environ.get(name, "")
|
|
if not value:
|
|
# Langflow rejects empty Credential values outright ("Variable
|
|
# value cannot be empty"). PM_ESPO_API_KEY legitimately comes
|
|
# back empty here when secrets.local.env's grep against
|
|
# /home/tlg/mkt/bewerb/.secrets/espocrm-api.md can't read that
|
|
# chmod-600 file (see task-10-report.md). Use an obviously-fake
|
|
# placeholder so the deploy can proceed -- stage_crm() only
|
|
# calls espocrm with this value when ctx["decision"] ==
|
|
# "consider", so the reject-path smoke test (Step 6) is
|
|
# unaffected; the accept-path / real CRM writes remain
|
|
# unverified until a real key is supplied.
|
|
value = f"UNSET-{name}-blocked-by-file-permissions"
|
|
print(f"WARN: {name} is empty in the environment, using "
|
|
f"placeholder credential value", file=sys.stderr)
|
|
upsert_variable(hdr, name, value, "Credential")
|
|
for name, value in GENERIC_VARS.items():
|
|
if name == "PM_FLOW2_ID":
|
|
continue
|
|
if not value:
|
|
# Same empty-value rejection as above, for PM_ESPO_BASE.
|
|
value = f"UNSET-{name}-blocked-by-file-permissions"
|
|
print(f"WARN: {name} is empty in the environment, using "
|
|
f"placeholder value", file=sys.stderr)
|
|
upsert_variable(hdr, name, value, "Generic")
|
|
api_key = get_api_key(hdr)
|
|
# PMIngest reads the run-API key via load_from_db -> must exist as a
|
|
# Credential global variable, not only in the host env file.
|
|
upsert_variable(hdr, "PM_LANGFLOW_API_KEY", api_key, "Credential")
|
|
|
|
def custom(fname):
|
|
code = (COMPONENTS / fname).read_text()
|
|
return component_template(hdr, code)
|
|
|
|
# ---- Flow 2: TextInput -> PMFetch -> ... -> PMNotify -> TextOutput
|
|
text_in = builtin_template(hdr, "input_output", "TextInput")
|
|
text_out = builtin_template(hdr, "input_output", "TextOutput")
|
|
chain = [("TextInput-pm2i", "TextInput", text_in)]
|
|
for fname, nid in [("pm_fetch.py", "PMFetch-pm2a"),
|
|
("pm_extract.py", "PMExtract-pm2b"),
|
|
("pm_match.py", "PMMatch-pm2c"),
|
|
("pm_rules.py", "PMRules-pm2d"),
|
|
("pm_crm.py", "PMCrm-pm2e"),
|
|
("pm_notify.py", "PMNotify-pm2f")]:
|
|
tpl, ctype = custom(fname)
|
|
chain.append((nid, ctype, tpl))
|
|
chain.append(("TextOutput-pm2o", "TextOutput", text_out))
|
|
nodes2 = [make_node(nid, ctype, tpl, x=i * 380)
|
|
for i, (nid, ctype, tpl) in enumerate(chain)]
|
|
|
|
by_id = {nid: tpl for nid, _, tpl in chain}
|
|
by_type = {nid: ctype for nid, ctype, _ in chain}
|
|
|
|
edges2 = [
|
|
edge_between("TextInput-pm2i", "TextInput", by_id["TextInput-pm2i"],
|
|
"text", "PMFetch-pm2a", by_id["PMFetch-pm2a"],
|
|
"payload"),
|
|
edge_between("PMFetch-pm2a", by_type["PMFetch-pm2a"],
|
|
by_id["PMFetch-pm2a"], "out",
|
|
"PMExtract-pm2b", by_id["PMExtract-pm2b"], "ctx_in"),
|
|
edge_between("PMExtract-pm2b", by_type["PMExtract-pm2b"],
|
|
by_id["PMExtract-pm2b"], "out",
|
|
"PMMatch-pm2c", by_id["PMMatch-pm2c"], "ctx_in"),
|
|
edge_between("PMMatch-pm2c", by_type["PMMatch-pm2c"],
|
|
by_id["PMMatch-pm2c"], "out",
|
|
"PMRules-pm2d", by_id["PMRules-pm2d"], "ctx_in"),
|
|
edge_between("PMRules-pm2d", by_type["PMRules-pm2d"],
|
|
by_id["PMRules-pm2d"], "out",
|
|
"PMCrm-pm2e", by_id["PMCrm-pm2e"], "ctx_in"),
|
|
edge_between("PMCrm-pm2e", by_type["PMCrm-pm2e"],
|
|
by_id["PMCrm-pm2e"], "out",
|
|
"PMNotify-pm2f", by_id["PMNotify-pm2f"], "ctx_in"),
|
|
edge_between("PMNotify-pm2f", by_type["PMNotify-pm2f"],
|
|
by_id["PMNotify-pm2f"], "out",
|
|
"TextOutput-pm2o", by_id["TextOutput-pm2o"],
|
|
"input_value"),
|
|
]
|
|
flow2_id = create_flow(hdr, "PM Projekt bewerten", "pm-flow2",
|
|
nodes2, edges2)
|
|
upsert_variable(hdr, "PM_FLOW2_ID", flow2_id, "Generic")
|
|
|
|
# ---- Flow 1: Webhook -> PMIngest -> TextOutput
|
|
webhook = builtin_template(hdr, "input_output", "Webhook")
|
|
tpl_ing, ctype_ing = custom("pm_ingest.py")
|
|
nodes1 = [make_node("Webhook-pm1w", "Webhook", webhook, 0),
|
|
make_node("PMIngest-pm1a", ctype_ing, tpl_ing, 380),
|
|
make_node("TextOutput-pm1o", "TextOutput", text_out, 760)]
|
|
edges1 = [
|
|
edge_between("Webhook-pm1w", "Webhook", webhook, "output_data",
|
|
"PMIngest-pm1a", tpl_ing, "trigger"),
|
|
edge_between("PMIngest-pm1a", ctype_ing, tpl_ing, "out",
|
|
"TextOutput-pm1o", text_out, "input_value"),
|
|
]
|
|
flow1_id = create_flow(hdr, "PM Ingest", "pm-ingest", nodes1, edges1)
|
|
|
|
ENV_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
ENV_FILE.write_text(
|
|
f"LANGFLOW_API_KEY={api_key}\n"
|
|
f"FLOW1_ID={flow1_id}\nFLOW2_ID={flow2_id}\n"
|
|
f"FLOW1_WEBHOOK=http://127.0.0.1:8090/api/v1/webhook/pm-ingest\n")
|
|
ENV_FILE.chmod(0o600)
|
|
print(f"flow1={flow1_id} flow2={flow2_id}\nenv -> {ENV_FILE}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|