#!/usr/bin/env python3 """Deploy the DAX Feishu approval definition from the repo artifact. This script is intentionally deploy-side, not Helix runtime. It reads the reviewed definition artifact, builds the Feishu v4 approval-definition payload, and creates or updates the approval definition through Feishu OpenAPI. Required environment: FEISHU_APP_ID FEISHU_APP_SECRET """ from __future__ import annotations import argparse import json import os import sys import urllib.error import urllib.request from pathlib import Path from typing import Any FEISHU_BASE_URL = "https://open.feishu.cn" TOKEN_PATH = "/open-apis/auth/v3/tenant_access_token/internal" class DeployError(RuntimeError): pass def _read_json(path: Path) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise DeployError(f"Failed to read JSON artifact {path}: {exc}") from exc if not isinstance(payload, dict): raise DeployError(f"Artifact {path} must be a JSON object") return payload def _request_json( method: str, url: str, *, body: dict[str, Any], token: str | None = None, ) -> dict[str, Any]: data = json.dumps(body, ensure_ascii=False).encode("utf-8") headers = {"Content-Type": "application/json; charset=utf-8"} if token: headers["Authorization"] = f"Bearer {token}" request = urllib.request.Request( url, data=data, headers=headers, method=method, ) try: with urllib.request.urlopen(request, timeout=30) as response: raw = response.read().decode("utf-8") except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") raise DeployError( f"Feishu HTTP {exc.code} for {url}: {raw[:500]}", ) from exc except urllib.error.URLError as exc: raise DeployError(f"Feishu request failed for {url}: {exc}") from exc try: payload = json.loads(raw) except json.JSONDecodeError as exc: raise DeployError( f"Feishu returned non-JSON response for {url}: {raw[:500]}", ) from exc if not isinstance(payload, dict): raise DeployError(f"Feishu returned non-object response for {url}") return payload def _tenant_access_token(*, app_id: str, app_secret: str) -> str: payload = _request_json( "POST", FEISHU_BASE_URL + TOKEN_PATH, body={"app_id": app_id, "app_secret": app_secret}, ) if payload.get("code") != 0: raise DeployError( "Feishu tenant token failed: " f"code={payload.get('code')} msg={payload.get('msg')}", ) token = payload.get("tenant_access_token") if not isinstance(token, str) or not token: raise DeployError("Feishu token response did not include tenant_access_token") return token def build_approval_definition_payload( definition: dict[str, Any], *, approval_code: str | None = None, ) -> dict[str, Any]: contract = definition.get("feishu_v4_api_contract") if not isinstance(contract, dict): raise DeployError("definition is missing feishu_v4_api_contract") i18n_keys = contract.get("i18n_keys") if not isinstance(i18n_keys, dict): raise DeployError("definition contract is missing i18n_keys") widget_i18n = i18n_keys.get("form_widgets") if not isinstance(widget_i18n, dict): raise DeployError("definition contract is missing widget i18n keys") widgets = definition.get("form_widgets") if not isinstance(widgets, list) or not widgets: raise DeployError("definition is missing form_widgets") form_content: list[dict[str, Any]] = [] for widget in widgets: if not isinstance(widget, dict): raise DeployError("form_widgets entries must be objects") canonical_field = str(widget.get("canonical_field") or "") custom_id = str(widget.get("custom_id") or "") widget_type = str(widget.get("type") or "") name_key = str(widget_i18n.get(canonical_field) or "") if not canonical_field or not custom_id or not widget_type or not name_key: raise DeployError( "form widget is incomplete: " f"canonical_field={canonical_field!r} custom_id={custom_id!r}", ) form_content.append({ "id": custom_id, "name": name_key, "type": widget_type, "required": bool(widget.get("required", True)), }) body: dict[str, Any] = { "approval_name": contract["approval_name_key"], "description": contract["description_key"], "form": { "form_content": json.dumps(form_content, ensure_ascii=False), }, "node_list": contract["node_list"], "viewers": contract["viewers"], "i18n_resources": contract["i18n_resources"], } if approval_code: body["approval_code"] = approval_code return body def _approval_code_from_response(payload: dict[str, Any]) -> str: data = payload.get("data") candidates: list[Any] = [payload.get("approval_code")] if isinstance(data, dict): candidates.extend([ data.get("approval_code"), data.get("code"), ]) for candidate in candidates: if isinstance(candidate, str) and candidate.strip(): return candidate.strip() raise DeployError( "Feishu approval deploy response did not include approval_code: " + json.dumps(payload, ensure_ascii=False, sort_keys=True)[:500], ) def deploy( *, definition_path: Path, approval_code: str | None, dry_run: bool, payload_out: Path | None, ) -> str: definition = _read_json(definition_path) contract = definition["feishu_v4_api_contract"] endpoint = str(contract.get("endpoint") or "") if not endpoint.startswith("/open-apis/approval/v4/approvals"): raise DeployError(f"Unexpected Feishu endpoint in definition: {endpoint!r}") body = build_approval_definition_payload( definition, approval_code=approval_code, ) if payload_out: payload_out.write_text( json.dumps(body, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) if dry_run: print("approval_code=" + (approval_code or "DRY_RUN_NEW_DEFINITION")) return approval_code or "DRY_RUN_NEW_DEFINITION" app_id = os.environ.get("FEISHU_APP_ID", "").strip() app_secret = os.environ.get("FEISHU_APP_SECRET", "").strip() if not app_id or not app_secret: raise DeployError("FEISHU_APP_ID and FEISHU_APP_SECRET are required") token = _tenant_access_token(app_id=app_id, app_secret=app_secret) payload = _request_json( "POST", FEISHU_BASE_URL + endpoint, body=body, token=token, ) if payload.get("code") != 0: raise DeployError( "Feishu approval deploy failed: " f"code={payload.get('code')} msg={payload.get('msg')} " f"payload={json.dumps(payload, ensure_ascii=False)[:500]}", ) published_code = _approval_code_from_response(payload) print("approval_code=" + published_code) return published_code def main() -> int: parser = argparse.ArgumentParser( description="Deploy the DAX Feishu approval definition.", ) parser.add_argument( "--definition", type=Path, default=Path(__file__).parent / "approval-definitions" / "data_access_group_grant" / "definition.json", ) parser.add_argument( "--approval-code", help="Existing Feishu approval_code to update. Omit to create.", ) parser.add_argument( "--dry-run", action="store_true", help="Build/validate payload without calling Feishu.", ) parser.add_argument( "--payload-out", type=Path, help="Write the Feishu request body JSON for inspection.", ) args = parser.parse_args() try: deploy( definition_path=args.definition, approval_code=(args.approval_code or "").strip() or None, dry_run=args.dry_run, payload_out=args.payload_out, ) except DeployError as exc: print(f"ERROR: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())