4ventures-client-portal/scripts/dokploy_set_superhuman_env.py

#!/usr/bin/env python3
"""Merge SUPERHUMAN_DOCS_API_KEY into client-portal Dokploy env, then Deploy."""
from __future__ import annotations

import json
import re
import sys
import time
from pathlib import Path

from playwright.sync_api import sync_playwright

ROOT = Path("/Users/4ventures/4ventures-client-portal")
BPS = Path("/Users/4ventures/Projects/bps-facturacion")
BASE = "https://dokploy.4ventures.es"
APP_ID = "NZS797t9Rnq9Pj4mHiii3"
APP = f"{BASE}/dashboard/project/Z_4M-uDOH1_--itGSTh4l/environment/bGja70l_a7HQ1dxyXqSuf/services/application/{APP_ID}"
HEALTH = "https://app.4ventures.es/api/health"


def cred(key: str) -> str:
    import os

    v = os.environ.get(key, "").strip()
    if v:
        return v
    p = BPS / ".env"
    if not p.is_file():
        return ""
    for line in p.read_text(encoding="utf-8").splitlines():
        if line.startswith(f"{key}="):
            return line.split("=", 1)[1].strip().strip('"').strip("'")
    return ""


def local_key() -> str:
    for line in (ROOT / ".env.local").read_text(encoding="utf-8").splitlines():
        if line.startswith("SUPERHUMAN_DOCS_API_KEY="):
            return line.split("=", 1)[1].strip().strip('"').strip("'")
    p = ROOT / ".ops" / "superhuman-docs-api-key.txt"
    if p.is_file():
        return p.read_text(encoding="utf-8").strip()
    return ""


def parse_env(text: str) -> dict[str, str]:
    out: dict[str, str] = {}
    for line in (text or "").splitlines():
        if not line.strip() or line.strip().startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        out[k.strip()] = v.strip()
    return out


def dump_env(vals: dict[str, str]) -> str:
    return "\n".join(f"{k}={v}" for k, v in vals.items()) + "\n"


def main() -> int:
    sh_key = local_key()
    if not sh_key:
        print("missing SUPERHUMAN_DOCS_API_KEY locally", flush=True)
        return 2
    email = cred("DOKPLOY_EMAIL") or "marketing@4ventures.es"
    password = cred("DOKPLOY_PASSWORD")

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(viewport={"width": 1400, "height": 1000})
        page.goto(BASE + "/", wait_until="domcontentloaded", timeout=90000)
        page.wait_for_timeout(600)
        if "/dashboard" not in page.url:
            page.locator('input[type="email"], input[name="email"]').first.fill(email)
            page.locator('input[type="password"]').first.fill(password)
            page.locator('button[type="submit"]').first.click()
            page.wait_for_url(re.compile(r"/dashboard"), timeout=90000)
        print("login ok", flush=True)

        page.goto(APP, wait_until="networkidle", timeout=120000)
        page.wait_for_timeout(800)

        result = page.evaluate(
            """async ({ appId, shKey }) => {
              const oneUrl = '/api/trpc/application.one?batch=1&input=' + encodeURIComponent(JSON.stringify({ '0': { json: { applicationId: appId } } }));
              const one = await fetch(oneUrl, { credentials: 'include' });
              const oneText = await one.text();
              if (!one.ok) return { ok: false, step: 'one', status: one.status, body: oneText.slice(0, 400) };
              let envText = '';
              try {
                const j = JSON.parse(oneText);
                const data = j[0]?.result?.data?.json || j[0]?.result?.data || j;
                envText = data?.env || data?.environment || '';
                if (typeof envText !== 'string') envText = '';
              } catch (e) {
                return { ok: false, step: 'parse', error: String(e), body: oneText.slice(0, 400) };
              }
              const lines = envText.split('\\n').filter(Boolean);
              const map = {};
              for (const line of lines) {
                const i = line.indexOf('=');
                if (i < 0) continue;
                map[line.slice(0, i)] = line.slice(i + 1);
              }
              map['SUPERHUMAN_DOCS_API_KEY'] = shKey;
              const merged = Object.keys(map).map(k => k + '=' + map[k]).join('\\n') + '\\n';
              const keys = Object.keys(map);

              const tries = [];
              const payloads = [
                { path: '/api/application.update', body: { applicationId: appId, env: merged } },
                { path: '/api/trpc/application.update?batch=1', body: { '0': { json: { applicationId: appId, env: merged } } } },
                { path: '/api/trpc/application.saveEnvironment?batch=1', body: { '0': { json: { applicationId: appId, env: merged } } } },
              ];
              for (const t of payloads) {
                const r = await fetch(t.path, {
                  method: 'POST',
                  credentials: 'include',
                  headers: { 'content-type': 'application/json' },
                  body: JSON.stringify(t.body),
                });
                const text = await r.text();
                tries.push({ path: t.path, status: r.status, body: text.slice(0, 180) });
                if (r.ok) return { ok: true, keys, tries, hadKey: keys.includes('SUPERHUMAN_DOCS_API_KEY') };
              }
              return { ok: false, keys, tries, envLen: envText.length };
            }""",
            {"appId": APP_ID, "shKey": sh_key},
        )
        print(
            json.dumps(
                {
                    "ok": result.get("ok"),
                    "hadKey": result.get("hadKey"),
                    "keyCount": len(result.get("keys") or []),
                    "tries": result.get("tries"),
                    "step": result.get("step"),
                    "status": result.get("status"),
                },
                indent=2,
            )[:2000],
            flush=True,
        )
        if not result.get("ok"):
            browser.close()
            return 6

        # Deploy
        page.goto(APP, wait_until="domcontentloaded", timeout=90000)
        page.wait_for_timeout(700)
        page.get_by_role("tab", name=re.compile(r"^General$", re.I)).first.click()
        page.wait_for_timeout(500)
        page.get_by_role("button", name=re.compile(r"^Deploy$", re.I)).first.click()
        page.wait_for_timeout(600)
        conf = page.get_by_role("button", name=re.compile(r"^Confirm$", re.I))
        if conf.count():
            conf.first.click()
        print("DEPLOY_CLICKED", flush=True)

        for i in range(24):
            time.sleep(10)
            try:
                import urllib.request

                code = urllib.request.urlopen(HEALTH, timeout=12).getcode()
            except Exception as e:
                code = str(e)
            print(f"health_try{i}", code, flush=True)
            if code == 200:
                break

        browser.close()
        return 0


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