#!/usr/bin/env python3 """Set client-portal env via Dokploy authenticated fetch (in-page), then Deploy.""" from __future__ import annotations import json import re import sys import time import urllib.request 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") SHOT = ROOT / "scripts/dokploy-shots" 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 dokploy_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 load_env_text() -> str: vals: dict[str, str] = {} for line in (ROOT / ".env.local").read_text(encoding="utf-8").splitlines(): if not line.strip() or line.strip().startswith("#") or "=" not in line: continue k, v = line.split("=", 1) vals[k.strip()] = v.strip().strip('"').strip("'") parts = [] for k in ( "NEXT_PUBLIC_SUPABASE_URL", "NEXT_PUBLIC_SUPABASE_ANON_KEY", "SUPABASE_SERVICE_ROLE_KEY", ): if not vals.get(k): raise SystemExit(f"missing {k}") parts.append(f"{k}={vals[k]}") return "\n".join(parts) + "\n" def main() -> int: env_text = load_env_text() email = dokploy_cred("DOKPLOY_EMAIL") or "marketing@4ventures.es" password = dokploy_cred("DOKPLOY_PASSWORD") with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page(viewport={"width": 1500, "height": 1200}) page.goto(BASE + "/", wait_until="domcontentloaded", timeout=90000) page.wait_for_timeout(800) 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) # Discover API by probing common Dokploy endpoints from authenticated page page.goto(APP, wait_until="networkidle", timeout=120000) page.wait_for_timeout(1000) probe = page.evaluate( """async ({ appId, env }) => { const tries = []; const payloads = [ { path: '/api/application.update', body: { applicationId: appId, env } }, { path: '/api/application.update', body: { applicationId: appId, applicationStatus: 'done', env } }, { path: '/api/trpc/application.update', body: { json: { applicationId: appId, env } } }, { path: '/api/trpc/application.saveEnvironment?batch=1', body: { '0': { json: { applicationId: appId, env } } } }, { path: '/api/trpc/application.update?batch=1', body: { '0': { json: { applicationId: appId, env } } } }, ]; for (const t of payloads) { try { 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, 240) }); if (r.ok) return { ok: true, tries }; } catch (e) { tries.push({ path: t.path, error: String(e) }); } } // Also GET one application to learn shape try { const r = await fetch('/api/trpc/application.one?batch=1&input=' + encodeURIComponent(JSON.stringify({ '0': { json: { applicationId: appId } } })), { credentials: 'include' }); tries.push({ path: 'application.one', status: r.status, body: (await r.text()).slice(0, 500) }); } catch (e) { tries.push({ path: 'application.one', error: String(e) }); } return { ok: false, tries }; }""", {"appId": APP_ID, "env": env_text}, ) print(json.dumps(probe, indent=2)[:4000], flush=True) if not probe.get("ok"): # UI-only: careful eye click inside Environment Settings card page.goto(APP + "?tab=environment", wait_until="networkidle", timeout=120000) page.wait_for_timeout(1200) page.screenshot(path=str(SHOT / "env_api_fail.png"), full_page=True) # Click ONLY the eye in the Environment Settings header row clicked = page.evaluate( """() => { const labels = [...document.querySelectorAll('*')].filter(el => el.childNodes.length && [...el.childNodes].some(n => n.nodeType===3 && n.textContent.trim()==='Environment Settings')); for (const label of labels) { let root = label; for (let i=0;i<6;i++) { root = root.parentElement; if (!root) break; const btns = [...root.querySelectorAll('button')].filter(b => b.querySelector('svg') && !b.innerText.trim()); if (btns.length) { btns[0].click(); return true; } } } return false; }""" ) print("careful eye", clicked, flush=True) page.wait_for_timeout(800) page.screenshot(path=str(SHOT / "env_careful.png"), full_page=True) # fill first textarea or contenteditable filled = page.evaluate( """(val) => { const tas = [...document.querySelectorAll('textarea')]; if (tas.length) { const ta = tas[0]; const setter = Object.getOwnPropertyDescriptor(window.HTMLTextAreaElement.prototype, 'value').set; setter.call(ta, val); ta.dispatchEvent(new Event('input', { bubbles: true })); ta.dispatchEvent(new Event('change', { bubbles: true })); return 'textarea'; } const cm = document.querySelector('.cm-content'); if (cm) { cm.setAttribute('contenteditable','true'); return 'cm-present'; } return 'none'; }""", env_text, ) print("filled", filled, flush=True) if filled == "cm-present": page.locator(".cm-content").first.click(force=True) page.keyboard.press("Meta+A") page.keyboard.insert_text(env_text) save = page.get_by_role("button", name=re.compile(r"^Save$", re.I)) if save.count(): save.first.click(force=True) page.wait_for_timeout(2000) print("saved ui", flush=True) # Deploy regardless if API ok page.goto(APP, wait_until="networkidle", timeout=120000) page.wait_for_timeout(800) 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(700) conf = page.get_by_role("button", name=re.compile(r"^Confirm$", re.I)) if conf.count(): conf.first.click() print("DEPLOY_CLICKED", flush=True) page.get_by_role("tab", name=re.compile(r"^Deployments$", re.I)).first.click() deadline = time.time() + 780 while time.time() < deadline: page.reload(wait_until="networkidle") body = page.inner_text("body") m = re.search(r"1\.\s*(Error|Done|Running)", body) status = m.group(1) if m else "?" print("status", status, flush=True) if status == "Done": break if status == "Error": browser.close() return 6 time.sleep(15) browser.close() for _ in range(30): try: with urllib.request.urlopen(HEALTH, timeout=12) as r: body = r.read().decode() print("health", body[:220], flush=True) if '"serviceRoleConfigured":true' in body: print("SUCCESS", flush=True) return 0 except Exception as e: print("wait", type(e).__name__, flush=True) time.sleep(6) print("PARTIAL", flush=True) return 5 if __name__ == "__main__": raise SystemExit(main())