lwc/bin
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(projekt-matching): deployment scripts and programmatic flow builder

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
tlg
2026-07-09 13:00:06 +02:00
parent a248dec9ea
commit 92b1c23c3a
2 changed files with 384 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

317
projekt-matching/deploy/build_flows.py Normal file
View File

@@ -0,0 +1,317 @@
#!/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):
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())

67
projekt-matching/deploy/deploy_files.sh Executable file
View File

@@ -0,0 +1,67 @@
#!/bin/bash
# Copy the projektmatch package and vorgaben files into the Langflow data dir
# (mounted at /app/langflow in the container). Idempotent.
#
# NOTE: /home/tlg/mkt/bewerb/vorgaben/Lebenslauf_Dr-Ing_Thomas_Langer.md and
# /home/tlg/mkt/bewerb/.secrets/espocrm-api.md are chmod 600, owned by tlg.
# This script normally runs as the lwc automation user, which cannot read
# those two files (verified: `cp` fails with EACCES). Rather than aborting
# the whole deploy (set -e) on a permission error, each vorgaben file is
# copied individually and a missing/unreadable file only produces a WARN,
# so the parts of the pipeline that don't need it still deploy correctly.
set -e
SRC="$(cd "$(dirname "$0")/.." && pwd)"
DEST="$HOME/.local/share/langflow_pod/langflow-data"
rsync -a --delete --exclude __pycache__ "$SRC/projektmatch/" "$DEST/projektmatch/"
mkdir -p "$DEST/vorgaben"
for f in /home/tlg/mkt/bewerb/vorgaben/Lebenslauf_Dr-Ing_Thomas_Langer.md \
/home/tlg/mkt/bewerb/vorgaben/rahmenbedingungen.md; do
if [ -r "$f" ]; then
cp "$f" "$DEST/vorgaben/"
else
echo "WARN: cannot read $f as $(whoami) (permission denied) - skipping" >&2
fi
done
# Langflow runs as uid 1000 gid 0 -> needs group read
chmod -R g+rX "$DEST/projektmatch" "$DEST/vorgaben"
# Make /app/langflow importable as a Python path root inside the container.
# The component files (Task 9) do `sys.path.insert(0, "/app/langflow")` as
# a *runtime* statement, but Langflow 1.10's custom-component loader
# (lfx/custom/validate.py: prepare_global_scope/create_class) statically
# scans the module's top-level `import`/`from ... import` AST nodes and
# resolves them via importlib BEFORE executing any other code -- an `if:`
# block containing the sys.path hack is a plain ast.If node, which that
# loader silently drops without ever executing it. So "import projektmatch"
# 404s both at design time (POST /custom_component, used by build_flows.py)
# and at real flow-run time, regardless of the in-file sys.path trick,
# unless /app/langflow is already on sys.path when the interpreter starts.
# Fix: drop a .pth file into the venv's site-packages (only takes effect
# for *new* Python processes, so the running langflow server needs a
# restart to pick it up -- the venv itself lives in the container's
# ephemeral overlay, not the bind-mounted data dir, so this must be
# redone after any container recreation).
CTR="langflow_ctr"
if podman inspect "$CTR" >/dev/null 2>&1; then
SITE_PKGS="$(podman exec "$CTR" python -c 'import site; print(site.getsitepackages()[0])')"
PTH_FILE="$SITE_PKGS/projektmatch.pth"
CURRENT="$(podman exec "$CTR" cat "$PTH_FILE" 2>/dev/null || true)"
if [ "$CURRENT" != "/app/langflow" ]; then
echo "Registering /app/langflow on container sys.path (restart required)..."
podman exec "$CTR" sh -c "echo /app/langflow > '$PTH_FILE'"
podman restart "$CTR" >/dev/null
for i in $(seq 1 40); do
code=$(curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8090/api/v1/auto_login || true)
[ "$code" = "200" ] && break
sleep 3
done
if [ "$code" != "200" ]; then
echo "WARN: $CTR did not come back healthy after restart (last health code: $code)" >&2
fi
fi
else
echo "WARN: container $CTR not found - skipping sys.path fix" >&2
fi
echo "Deployed to $DEST"