ligbox-ops-platform/deploy/vm112-wizard/fix-provision-zone.py

#!/usr/bin/env python3
"""Replace provision_cloudflare_zone body on VM112."""
from pathlib import Path

ROUTER = Path("/opt/ligbox-wizard/backend/app/routers/onboarding.py")
CF = Path("/opt/ligbox-wizard/backend/app/services/cloudflare.py")
text = ROUTER.read_text(encoding="utf-8")

# imports
if "provision_ligbox_zone" not in text:
    text = text.replace(
        "from app.services.dns_account_registry import resolve_ligbox_zone, resolve_payload",
        "from app.services.dns_account_registry import (\n"
        "    pick_provision_account,\n"
        "    provision_ligbox_zone,\n"
        "    resolve_ligbox_zone,\n"
        "    resolve_payload,\n"
        ")",
    )

OLD_START = '@router.post("/dns/cloudflare/provision-zone")\ndef provision_cloudflare_zone(body: DomainRequest, request: Request):'
OLD_END = '@router.get("/dns/portal-onboarding/{domain}")'

start = text.find(OLD_START)
end = text.find(OLD_END)
if start < 0 or end < 0:
    raise SystemExit("anchors not found")

NEW_FUNC = '''@router.post("/dns/cloudflare/provision-zone")
def provision_cloudflare_zone(body: DomainRequest, request: Request):
    get_session_from_request(request)
    domain = normalize_domain(body.domain)
    try:
        validate_primary_domain(domain)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=invalid_domain_http_detail()) from e

    match = resolve_ligbox_zone(domain)
    if match:
        activity_log.info(
            f"Zona {domain} já na conta {match.account.id} — sem create",
            source="cloudflare",
        )
        payload = _portal_onboarding_payload(domain, zone_created=False)
        payload["zone_id"] = match.zone_id
        payload["ligbox_account_id"] = match.account.id
        payload["dns_mode"] = match.dns_mode
        payload["message"] = (
            f"Domínio {domain} já está na Cloudflare Ligbox ({match.account.label})."
        )
        if nameservers := wizard_nameservers(match.zone):
            payload["nameservers"] = nameservers
            payload["status"]["nameservers"] = nameservers
        return payload

    provision_acct = pick_provision_account()
    if not provision_acct:
        raise HTTPException(
            400,
            "Nenhuma conta Cloudflare Ligbox configurada. Ver dns-accounts.yaml e secrets/.",
        )

    try:
        activity_log.info(
            f"Criar zona cliente {domain} na CF Ligbox ({provision_acct.id})",
            source="cloudflare",
        )
        provision_acct.client().verify_token()
        prov = provision_ligbox_zone(domain, account=provision_acct)
        ligbox_acct = prov["account"]
        cf = ligbox_acct.client()
        zone = prov["zone"]
        created = prov["created"]
        sandbox = is_cloudflare_sandbox(cf.token)
        if sandbox:
            activity_log.info(
                "Cloudflare sandbox: zona simulada (token de teste — não cria zona real)",
                source="cloudflare",
            )
        activity_log.ok(
            f"Zona {domain} {'criada' if created else 'já existia'} (id {zone.get('id', '?')})",
            source="cloudflare",
        )
        payload = _portal_onboarding_payload(domain, zone_created=created)
        payload["zone_id"] = zone.get("id")
        payload["ligbox_account_id"] = ligbox_acct.id
        payload["dns_mode"] = f"ligbox_cf_{ligbox_acct.id}"
        payload["sandbox"] = sandbox
        payload["message"] = (
            f"Domínio {domain} {'criado' if created else 'já existia'} na Cloudflare Ligbox "
            f"({ligbox_acct.label}). "
            + (
                "Ambiente de teste: apontamentos simulados automaticamente."
                if sandbox
                else "Altere os nameservers no registrador."
            )
        )
        if nameservers := wizard_nameservers(zone):
            payload["nameservers"] = nameservers
            payload["status"]["nameservers"] = nameservers
        if sandbox:
            zone_id = zone.get("id")
            applied = []
            for rec in mail_dns_records(domain, settings.mail_public_ip, []):
                upsert = cf.upsert_record(
                    zone_id,
                    rec["type"],
                    rec["name"],
                    rec["content"],
                    priority=rec.get("priority"),
                    proxied=rec.get("proxied", False),
                    zone_name=domain,
                )
                applied.append({"type": rec["type"], "name": rec["name"], "id": upsert.get("id")})
            verification = dns_verify.verify_mail_dns(domain, settings.mail_public_ip)
            payload["applied"] = applied
            payload["verification"] = verification
            for step in payload.get("steps") or []:
                if step.get("order") == 3:
                    step["done"] = True
                    step["detail"] = "Apontamentos simulados (sandbox)."
            activity_log.ok("Cloudflare sandbox: apontamentos de email simulados", source="cloudflare")
        return payload
    except CloudflareError as e:
        activity_log.error(f"Provision zona falhou: {e}", source="cloudflare")
        raise HTTPException(
            status_code=502,
            detail=cloudflare_error_http_detail(e, domain),
        ) from e


'''

text = text[:start] + NEW_FUNC + text[end:]
ROUTER.write_text(text, encoding="utf-8")

cf_text = CF.read_text(encoding="utf-8")
if "account_id: str | None = None" not in cf_text:
    cf_text = cf_text.replace(
        '    def ensure_zone(self, domain: str) -> dict:\n'
        '        """Garante que a zona existe na conta Ibytera; cria se necessário."""\n'
        "        zone = self.get_zone_by_name(domain)\n"
        "        if zone:\n"
        '            return {"zone": zone, "created": False}\n'
        "        zone = self.create_zone(domain)\n"
        '        return {"zone": zone, "created": True}',
        '    def ensure_zone(self, domain: str, account_id: str | None = None) -> dict:\n'
        '        """Garante que a zona existe na conta CF; cria se necessário."""\n'
        "        zone = self.get_zone_by_name(domain)\n"
        "        if zone:\n"
        '            return {"zone": zone, "created": False}\n'
        "        zone = self.create_zone(domain, account_id=account_id)\n"
        '        return {"zone": zone, "created": True}',
    )
    CF.write_text(cf_text, encoding="utf-8")

import ast

ast.parse(ROUTER.read_text())
print("fix-provision-zone OK")