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

#!/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())