| """netlify_deployer.py — Deploy, block, and unblock sites via Netlify REST API.""" |
|
|
| import logging |
| import time |
| import requests |
| from config import NETLIFY_TOKEN, NETLIFY_API |
|
|
| log = logging.getLogger(__name__) |
| HEADERS = {"Authorization": f"Bearer {NETLIFY_TOKEN}", "Content-Type": "application/json"} |
|
|
|
|
| def _api(method, path, **kwargs): |
| r = requests.request(method, f"{NETLIFY_API}{path}", headers=HEADERS, timeout=60, **kwargs) |
| r.raise_for_status() |
| return r.json() |
|
|
|
|
| def create_site(site_name: str) -> dict: |
| safe = "".join(c if c.isalnum() or c=="-" else "-" for c in site_name.lower())[:55] |
| safe = safe.strip("-") or "webreach-site" |
| data = _api("POST", "/sites", json={"name": safe}) |
| log.info("🌐 Site created: %s → %s", data["name"], data.get("ssl_url","")) |
| return data |
|
|
|
|
| def deploy_zip(site_id: str, zip_path: str) -> str: |
| """Upload ZIP, poll until live. Returns live URL.""" |
| url = f"{NETLIFY_API}/sites/{site_id}/deploys" |
| hdrs = {"Authorization": f"Bearer {NETLIFY_TOKEN}", "Content-Type": "application/zip"} |
| with open(zip_path, "rb") as f: |
| resp = requests.post(url, headers=hdrs, data=f.read(), timeout=120) |
| resp.raise_for_status() |
| deploy_id = resp.json()["id"] |
| log.info("🚀 Deploy %s started", deploy_id) |
|
|
| for _ in range(40): |
| time.sleep(4) |
| status = _api("GET", f"/deploys/{deploy_id}") |
| state = status.get("state") |
| if state == "ready": |
| live = status.get("ssl_url") or status.get("url","") |
| log.info("✅ Live: %s", live) |
| return live |
| if state in ("error","failed"): |
| raise RuntimeError(f"Netlify deploy failed: {status.get('error_message')}") |
|
|
| raise TimeoutError("Netlify deploy timed out") |
|
|
|
|
| def block_site(site_id: str) -> str: |
| """Replace site content with a payment-pending page.""" |
| BLOCK_HTML = b"""<!DOCTYPE html><html><head><meta charset="UTF-8"> |
| <title>Payment Pending</title> |
| <style>body{font-family:sans-serif;display:flex;align-items:center; |
| justify-content:center;height:100vh;margin:0;background:#f5f5f5} |
| .box{text-align:center;padding:3rem;background:#fff;border-radius:12px; |
| box-shadow:0 4px 24px rgba(0,0,0,.1)}h1{color:#e53e3e}p{color:#555} |
| </style></head><body><div class="box"> |
| <h1>⏳ Payment Verification Pending</h1> |
| <p>This website is temporarily unavailable while payment is being verified.</p> |
| <p>Please contact us if you have any questions.</p> |
| </div></body></html>""" |
| import io, zipfile, tempfile, os |
| tmp = tempfile.mktemp(suffix=".zip") |
| with zipfile.ZipFile(tmp, "w") as zf: |
| zf.writestr("index.html", BLOCK_HTML) |
| url = f"{NETLIFY_API}/sites/{site_id}/deploys" |
| hdrs = {"Authorization": f"Bearer {NETLIFY_TOKEN}", "Content-Type": "application/zip"} |
| with open(tmp, "rb") as f: |
| resp = requests.post(url, headers=hdrs, data=f.read(), timeout=60) |
| os.unlink(tmp) |
| resp.raise_for_status() |
| log.info("🔒 Site %s blocked (payment page)", site_id) |
| return resp.json().get("ssl_url","") |
|
|
|
|
| def unblock_site(site_id: str, zip_path: str) -> str: |
| """Redeploy the real site. Returns live URL.""" |
| log.info("🔓 Unblocking site %s", site_id) |
| return deploy_zip(site_id, zip_path) |
|
|
|
|
| def set_custom_domain(site_id: str, domain: str) -> str: |
| data = _api("PUT", f"/sites/{site_id}", json={"custom_domain": domain}) |
| url = data.get("ssl_url") or f"https://{domain}" |
| log.info("🔗 Domain set: %s → %s", domain, url) |
| return url |
|
|
|
|
| def deploy_for_lead(lead_id: int, business_name: str, zip_path: str) -> tuple: |
| """Create site + deploy. Returns (site_id, live_url).""" |
| site = create_site(f"{business_name}-wr{lead_id}") |
| return site["id"], deploy_zip(site["id"], zip_path) |
|
|
