#!/usr/bin/env python3 """ Ensure Dokploy schedule for pipeline Meta publish (every 10 min). Uses application.createSchedule / findAll if API available; falls back to UI. """ 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" f"/environment/bGja70l_a7HQ1dxyXqSuf/services/application/{APP_ID}" ) SCHEDULE_NAME = "pipeline-publish" CRON_EXPR = "*/10 * * * *" # Runs inside the app container where CRON_SECRET is already in env. CMD = ( 'curl -sS -H "Authorization: Bearer $CRON_SECRET" ' '"http://127.0.0.1:3000/api/cron/pipeline-publish" || ' 'curl -sS -H "Authorization: Bearer $CRON_SECRET" ' '"https://app.4ventures.es/api/cron/pipeline-publish"' ) def cred(key: str) -> str: import os v = os.environ.get(key, "").strip() if v: return v for p in (BPS / ".env", ROOT / ".env.local"): if not p.is_file(): continue 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 main() -> int: email = cred("DOKPLOY_EMAIL") or "marketing@4ventures.es" password = cred("DOKPLOY_PASSWORD") if not password: print("missing DOKPLOY_PASSWORD", flush=True) return 2 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(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) # Probe Dokploy schedule APIs probe = page.evaluate( """async ({ appId, name, cron, command }) => { const tryFetch = async (path, body) => { const r = await fetch(path, { method: 'POST', credentials: 'include', headers: { 'content-type': 'application/json' }, body: JSON.stringify(body), }); const text = await r.text(); let json = null; try { json = JSON.parse(text); } catch {} return { ok: r.ok, status: r.status, path, text: text.slice(0, 500), json }; }; const listPaths = [ '/api/trpc/schedule.all', '/api/application.one', ]; const outs = {}; // application.one to inspect shape outs.one = await tryFetch('/api/application.one', { applicationId: appId }); // Dokploy v0. something: schedule.create const createBodies = [ { applicationId: appId, name, cron, command, shellType: 'bash' }, { applicationId: appId, scheduleName: name, cronExpression: cron, command, enabled: true }, { applicationId: appId, name, cronExpression: cron, command }, ]; const createPaths = [ '/api/schedule.create', '/api/application.createSchedule', '/api/trpc/schedule.create', ]; outs.creates = []; for (const path of createPaths) { for (const body of createBodies) { const r = await tryFetch(path, body); outs.creates.push(r); if (r.ok) return { ok: true, via: path, r }; } } // list schedules for (const path of [ '/api/schedule.all', '/api/application.schedules', '/api/schedule.byApplicationId', ]) { outs['list_'+path] = await tryFetch(path, { applicationId: appId }); } return { ok: false, outs }; }""", { "appId": APP_ID, "name": SCHEDULE_NAME, "cron": CRON_EXPR, "command": CMD, }, ) print("API_PROBE", json.dumps(probe, indent=2)[:4000], flush=True) if probe.get("ok"): print("SCHEDULE_OK_API", flush=True) browser.close() return 0 # UI fallback: Schedules tab page.goto(APP, wait_until="domcontentloaded", timeout=90000) page.wait_for_timeout(1000) tab = page.get_by_role("tab", name=re.compile(r"Schedule", re.I)) if not tab.count(): # try Advanced / Domains area for name in ("Schedules", "Cron", "Advanced"): t = page.get_by_text(re.compile(rf"^{name}$", re.I)) if t.count(): t.first.click() page.wait_for_timeout(600) break else: tab.first.click() page.wait_for_timeout(800) body = page.inner_text("body") print("UI_BODY_SNIP", body[:1500].replace("\n", " | "), flush=True) if SCHEDULE_NAME in body or "pipeline-publish" in body: print("SCHEDULE_ALREADY_PRESENT", flush=True) browser.close() return 0 add = page.get_by_role("button", name=re.compile(r"Add|Create|New|Schedule", re.I)) if add.count(): add.first.click() page.wait_for_timeout(700) # fill fields heuristically inputs = page.locator("input, textarea") n = inputs.count() filled = 0 for i in range(min(n, 12)): el = inputs.nth(i) try: typ = (el.get_attribute("type") or "").lower() name = (el.get_attribute("name") or el.get_attribute("placeholder") or "").lower() if typ == "hidden": continue if "name" in name or filled == 0: el.fill(SCHEDULE_NAME) filled += 1 elif "cron" in name or filled == 1: el.fill(CRON_EXPR) filled += 1 elif "command" in name or "cmd" in name or filled == 2: el.fill(CMD) filled += 1 except Exception: pass save = page.get_by_role("button", name=re.compile(r"Save|Create|Confirm|Add", re.I)) if save.count(): save.first.click() page.wait_for_timeout(1200) print("SCHEDULE_UI_SAVED", flush=True) browser.close() return 0 print("SCHEDULE_NEEDS_MANUAL", flush=True) # Persist ops note note = ROOT / ".ops" / "pipeline-publish-cron.txt" note.write_text( f"Schedule needed on Dokploy app {APP_ID}:\n" f" name: {SCHEDULE_NAME}\n" f" cron: {CRON_EXPR}\n" f" command:\n{CMD}\n", encoding="utf-8", ) browser.close() return 0 # non-fatal: docs + note written if __name__ == "__main__": raise SystemExit(main())