import os import time from datetime import datetime, timezone from urllib.parse import urlencode from flask import Flask, request, jsonify, render_template_string, redirect import requests app = Flask(__name__) SUPABASE_URL = os.getenv("SUPABASE_URL", "").rstrip("/") SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY", "") APP_BASE_URL = os.getenv("APP_BASE_URL", "").rstrip("/") NEXT_AGENT = { "trend_hunter": "offer_scout", "offer_scout": "content_strategist", "content_strategist": "seo_writer", "seo_writer": "creative_generator", "creative_generator": "publisher", "publisher": "tracker", "tracker": "analytics", "analytics": "cro_optimizer", "cro_optimizer": "compliance_guard", } # ---------- Infra helpers ---------- def now_iso(): return datetime.now(timezone.utc).isoformat() def provider_config(provider: str): p = provider.lower() if p == "facebook": return { "client_id": os.getenv("FB_CLIENT_ID", ""), "client_secret": os.getenv("FB_CLIENT_SECRET", ""), "auth_url": "https://www.facebook.com/v19.0/dialog/oauth", "token_url": "https://graph.facebook.com/v19.0/oauth/access_token", "scope": os.getenv("FB_SCOPE", "pages_manage_posts,pages_read_engagement"), } if p == "x": return { "client_id": os.getenv("X_CLIENT_ID", ""), "client_secret": os.getenv("X_CLIENT_SECRET", ""), "auth_url": "https://twitter.com/i/oauth2/authorize", "token_url": "https://api.twitter.com/2/oauth2/token", "scope": os.getenv("X_SCOPE", "tweet.read tweet.write users.read offline.access"), } if p == "reddit": return { "client_id": os.getenv("REDDIT_CLIENT_ID", ""), "client_secret": os.getenv("REDDIT_CLIENT_SECRET", ""), "auth_url": "https://www.reddit.com/api/v1/authorize", "token_url": "https://www.reddit.com/api/v1/access_token", "scope": os.getenv("REDDIT_SCOPE", "identity submit read"), } raise RuntimeError(f"Unsupported provider: {provider}") def sb(path, method="GET", params=None, body=None): if not SUPABASE_URL or not SUPABASE_ANON_KEY: raise RuntimeError("Missing SUPABASE_URL / SUPABASE_ANON_KEY") url = f"{SUPABASE_URL}/rest/v1/{path}" headers = { "apikey": SUPABASE_ANON_KEY, "Authorization": f"Bearer {SUPABASE_ANON_KEY}", "Content-Type": "application/json", } if method in ("POST", "PATCH", "DELETE"): headers["Prefer"] = "return=representation" r = requests.request(method, url, headers=headers, params=params, json=body, timeout=25) if r.status_code >= 400: raise RuntimeError(f"Supabase {r.status_code}: {r.text[:250]}") if not r.text: return None return r.json() # ---------- Connection store ---------- def upsert_connection(provider: str, row: dict): existing = sb("social_connections", params={"provider": f"eq.{provider}", "limit": 1}) or [] if existing: sb(f"social_connections?provider=eq.{provider}", method="PATCH", body=row) return {"provider": provider, "status": "updated"} sb("social_connections", method="POST", body=row) return {"provider": provider, "status": "inserted"} def save_connection(provider: str, token_payload: dict): expires_at = None if token_payload.get("expires_in"): try: expires_at = datetime.fromtimestamp( time.time() + int(token_payload.get("expires_in")), tz=timezone.utc ).isoformat() except Exception: expires_at = None row = { "provider": provider, "access_token": token_payload.get("access_token"), "refresh_token": token_payload.get("refresh_token"), "scope": token_payload.get("scope"), "token_type": token_payload.get("token_type"), "expires_at": expires_at, "raw": token_payload, "updated_at": now_iso(), } return upsert_connection(provider, row) def save_make_webhook(webhook_url: str): row = { "provider": "make", "access_token": None, "refresh_token": None, "scope": "webhook", "token_type": "webhook", "expires_at": None, "raw": {"webhook_url": webhook_url}, "updated_at": now_iso(), } return upsert_connection("make", row) # ---------- Affiliate agent logic ---------- def run_agent(agent_code, input_data): prev = (input_data or {}).get("prev_output", {}) if agent_code == "trend_hunter": niche = (input_data or {}).get("niche", "affiliate marketing") base = ["best", "review", "promo", "diskon", "vs"] return { "niche": niche, "keywords": [f"{niche} {b}" for b in base], "search_intent": "commercial+transactional", } if agent_code == "offer_scout": kws = prev.get("keywords", []) return { "offers": [ {"keyword": k, "offer_id": f"offer_{i+1}", "score": 80 - i * 7} for i, k in enumerate(kws[:3]) ] } if agent_code == "content_strategist": offers = prev.get("offers", []) return { "content_plan": [ { "stage": "BOFU" if i == 0 else ("MOFU" if i == 1 else "TOFU"), "title_angle": f"Solusi terbaik untuk {o.get('keyword', 'offer')}", } for i, o in enumerate(offers) ], "affiliate_link": (input_data or {}).get("affiliate_link"), "short_url": (input_data or {}).get("short_url"), } if agent_code == "seo_writer": title = prev.get("content_plan", [{}])[0].get("title_angle", "Panduan affiliate") slug = "-".join([x for x in "".join([c.lower() if c.isalnum() else " " for c in title]).split() if x]) return {"title": title, "slug": slug, "article_html": f"

{title}

Draft siap publish.

"} if agent_code == "creative_generator": title = prev.get("title", "Promo affiliate") return { "creatives": [ {"channel": "facebook", "hook": f"{title} - cek sekarang"}, {"channel": "instagram", "hook": f"🔥 {title}"}, ] } if agent_code == "publisher": return { "publish_refs": [{"channel": "blog", "ref": f"post_{int(time.time())}"}], "short_url": (input_data or {}).get("short_url"), } if agent_code == "tracker": return { "tracking_map": { "utm_source": "hf-space", "utm_medium": "affiliate", "utm_campaign": f"campaign_{int(time.time())}", } } if agent_code == "analytics": return {"kpi_snapshot": {"ctr": 0.07, "cvr": 0.02, "epc": 0.4}} if agent_code == "cro_optimizer": return {"experiments": [{"name": "headline-variant-a", "priority": "high"}]} if agent_code == "compliance_guard": return {"compliance_status": "pass-with-notes", "fixes": ["Tambahkan disclosure affiliate"]} raise RuntimeError(f"No runner for {agent_code}") def run_once(): task_rows = sb("aff_tasks", params={"status": "eq.queued", "order": "scheduled_at.asc", "limit": 1}) or [] if not task_rows: return {"ok": True, "message": "No queued tasks"} task = task_rows[0] task_id = task["id"] attempts = int(task.get("attempts") or 0) + 1 sb( f"aff_tasks?id=eq.{task_id}", method="PATCH", body={"status": "running", "started_at": now_iso(), "attempts": attempts}, ) try: output = run_agent(task["agent_code"], task.get("input") or {}) sb( f"aff_tasks?id=eq.{task_id}", method="PATCH", body={"status": "done", "output": output, "finished_at": now_iso()}, ) next_agent = NEXT_AGENT.get(task["agent_code"]) if next_agent: sb( "aff_tasks", method="POST", body={ "agent_code": next_agent, "status": "queued", "input": {"prev_agent": task["agent_code"], "prev_output": output}, }, ) return {"ok": True, "task_id": task_id, "agent": task["agent_code"], "status": "done"} except Exception as e: sb( f"aff_tasks?id=eq.{task_id}", method="PATCH", body={"status": "error", "error": str(e), "finished_at": now_iso()}, ) return {"ok": False, "task_id": task_id, "agent": task["agent_code"], "error": str(e)} # ---------- Dashboard data ---------- def stats_today(): start = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0).isoformat() rows = ( sb( "aff_tasks", params={"select": "status,agent_code,created_at,output", "created_at": f"gte.{start}", "limit": 2000}, ) or [] ) agg = {"queued": 0, "running": 0, "done": 0, "error": 0} by_agent = {} publish_count = 0 latest_kpi = {"ctr": 0.0, "cvr": 0.0, "epc": 0.0} for row in rows: st = row.get("status", "queued") agg[st] = agg.get(st, 0) + 1 code = row.get("agent_code", "unknown") by_agent[code] = by_agent.get(code, 0) + 1 out = row.get("output") or {} if code == "publisher": refs = out.get("publish_refs") or [] publish_count += len(refs) if code == "analytics" and out.get("kpi_snapshot"): k = out.get("kpi_snapshot") latest_kpi = { "ctr": float(k.get("ctr") or 0.0), "cvr": float(k.get("cvr") or 0.0), "epc": float(k.get("epc") or 0.0), } est_clicks = int(publish_count * 55) est_leads = int(est_clicks * max(latest_kpi["cvr"], 0.01)) return { "total": len(rows), "agg": agg, "by_agent": by_agent, "traffic": { "published_items": publish_count, "estimated_clicks": est_clicks, "estimated_leads": est_leads, "ctr": latest_kpi["ctr"], "cvr": latest_kpi["cvr"], "epc": latest_kpi["epc"], "note": "Estimated from pipeline outputs (publisher + latest analytics snapshot).", }, } def latest_tasks(limit=12): rows = ( sb( "aff_tasks", params={ "select": "id,agent_code,status,created_at,finished_at,error", "order": "created_at.desc", "limit": limit, }, ) or [] ) return rows def connection_overview(): rows = ( sb( "social_connections", params={"select": "provider,scope,token_type,expires_at,updated_at", "order": "updated_at.desc", "limit": 30}, ) or [] ) return rows PAGE = """ OpenClaw Growth Console

âš¡ OpenClaw Growth Console

Status: {{supabase_ok}} · Last refresh: {{refreshed_at}}
Total Tasks Today
{{stats.total}}
Done
{{stats.agg.done}}
Running + Queued
{{stats.agg.running + stats.agg.queued}}
Errors
{{stats.agg.error}}
Published Items
{{stats.traffic.published_items}}
Estimated Clicks
{{stats.traffic.estimated_clicks}}
Estimated Leads
{{stats.traffic.estimated_leads}}
CTR / CVR / EPC
{{'%.2f' % (stats.traffic.ctr*100)}}% · {{'%.2f' % (stats.traffic.cvr*100)}}% · {{'%.2f' % stats.traffic.epc}}
{{stats.traffic.note}}

Connect Make.com

Enqueue Campaign

Connected Platforms

{% for c in connections %} {% endfor %} {% if not connections %} {% endif %}
ProviderScopeTypeExpiresUpdated
{{c.provider}} {{c.scope or '-'}} {{c.token_type or '-'}} {{c.expires_at or '-'}} {{c.updated_at}}
Belum ada koneksi. Connect dulu via tombol di atas.

Latest Tasks

{% for t in latest %} {% endfor %}
CreatedAgentStatusTask IDError
{{t.created_at}} {{t.agent_code}} {% if t.status == 'done' %}done {% elif t.status == 'running' %}running {% elif t.status == 'error' %}error {% else %}queued{% endif %} {{t.id}} {{(t.error or '')[:80]}}
""" # ---------- Routes ---------- @app.get("/") def home(): try: stats = stats_today() conns = connection_overview() latest = latest_tasks(15) return render_template_string( PAGE, supabase_ok="connected", refreshed_at=now_iso(), stats=stats, connections=conns, latest=latest, ) except Exception as e: return render_template_string( PAGE, supabase_ok=f"error: {e}", refreshed_at=now_iso(), stats={"total": 0, "agg": {"done": 0, "running": 0, "queued": 0, "error": 0}, "traffic": {"published_items": 0, "estimated_clicks": 0, "estimated_leads": 0, "ctr": 0, "cvr": 0, "epc": 0, "note": "No data"}}, connections=[], latest=[], ) @app.get("/healthz") def healthz(): return jsonify({"ok": True, "time": now_iso()}) @app.get("/dashboard-data") def dashboard_data(): return jsonify({ "ok": True, "time": now_iso(), "stats": stats_today(), "connections": connection_overview(), "latest": latest_tasks(15), }) @app.get("/oauth/start/") def oauth_start(provider): try: conf = provider_config(provider) if not APP_BASE_URL: return jsonify({"ok": False, "error": "Missing APP_BASE_URL secret"}), 500 if not conf["client_id"] or not conf["client_secret"]: return jsonify({"ok": False, "error": f"Missing client credentials for {provider}"}), 400 redirect_uri = f"{APP_BASE_URL}/oauth/callback/{provider}" state = f"{provider}-{int(time.time())}" params = { "client_id": conf["client_id"], "redirect_uri": redirect_uri, "response_type": "code", "scope": conf["scope"], "state": state, } auth_url = f"{conf['auth_url']}?{urlencode(params)}" return redirect(auth_url) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.get("/oauth/callback/") def oauth_callback(provider): code = request.args.get("code") if not code: return jsonify({"ok": False, "error": "Missing code"}), 400 conf = provider_config(provider) redirect_uri = f"{APP_BASE_URL}/oauth/callback/{provider}" data = { "grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri, "client_id": conf["client_id"], "client_secret": conf["client_secret"], } try: token_res = requests.post(conf["token_url"], data=data, timeout=25) payload = token_res.json() if token_res.status_code >= 400: return jsonify({"ok": False, "error": payload}), 400 saved = save_connection(provider, payload) return jsonify({"ok": True, "provider": provider, "saved": saved}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.get("/connections") def connections(): rows = connection_overview() return jsonify({"ok": True, "connections": rows}) @app.post("/make/connect") def make_connect(): webhook_url = request.form.get("webhook_url") or (request.json or {}).get("webhook_url") if not webhook_url: return jsonify({"ok": False, "error": "webhook_url required"}), 400 if not webhook_url.startswith("https://hook."): return jsonify({"ok": False, "error": "invalid make webhook url"}), 400 saved = save_make_webhook(webhook_url) return jsonify({"ok": True, "saved": saved}) @app.get("/make/test") def make_test(): rows = sb("social_connections", params={"provider": "eq.make", "limit": 1}) or [] if not rows: return jsonify({"ok": False, "error": "make webhook not connected"}), 400 webhook_url = ((rows[0] or {}).get("raw") or {}).get("webhook_url") if not webhook_url: return jsonify({"ok": False, "error": "stored webhook missing"}), 400 payload = { "event": "openclaw_make_test", "timestamp": now_iso(), "source": "hf-space", "message": "Make webhook test OK", } try: r = requests.post(webhook_url, json=payload, timeout=20) return jsonify({"ok": r.status_code < 400, "status": r.status_code, "text": r.text[:180]}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 @app.post("/enqueue") def enqueue(): link = request.form.get("affiliate_link") or (request.json or {}).get("affiliate_link") short = request.form.get("short_url") or (request.json or {}).get("short_url") if not link: return jsonify({"ok": False, "error": "affiliate_link required"}), 400 row = sb( "aff_tasks", method="POST", body={ "agent_code": "content_strategist", "status": "queued", "input": { "niche": "web hosting", "offer": "Hostinger", "affiliate_link": link, "short_url": short, "source": "hf-space", }, }, ) return jsonify({"ok": True, "task": row[0] if isinstance(row, list) and row else row}) @app.get("/run-once") def run_once_route(): return jsonify(run_once()) @app.get("/run-batch") def run_batch(): count = int(request.args.get("count", 5)) out = [] for _ in range(max(1, min(count, 20))): result = run_once() out.append(result) if result.get("message") == "No queued tasks": break return jsonify({"ok": True, "results": out}) @app.get("/stats") def stats_route(): return jsonify(stats_today()) if __name__ == "__main__": port = int(os.getenv("PORT", "7860")) app.run(host="0.0.0.0", port=port)