import os import time from datetime import datetime, timezone from flask import Flask, request, jsonify, render_template_string import requests app = Flask(__name__) SUPABASE_URL = os.getenv("SUPABASE_URL", "").rstrip("/") SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY", "") 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", } def now_iso(): return datetime.now(timezone.utc).isoformat() 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() 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": t = prev.get("title", "Promo affiliate") return {"creatives": [{"channel": "facebook", "hook": f"{t} - cek sekarang"}, {"channel": "instagram", "hook": f"🔥 {t}"}]} 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"} t = task_rows[0] task_id = t["id"] attempts = int(t.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(t["agent_code"], t.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(t["agent_code"]) if next_agent: sb("aff_tasks", method="POST", body={"agent_code": next_agent, "status": "queued", "input": {"prev_agent": t["agent_code"], "prev_output": output}}) return {"ok": True, "task_id": task_id, "agent": t["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": t["agent_code"], "error": str(e)} 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", "created_at": f"gte.{start}", "limit": 1000}) or [] agg = {"queued": 0, "running": 0, "done": 0, "error": 0} by_agent = {} for r in rows: st = r.get("status", "queued") agg[st] = agg.get(st, 0) + 1 a = r.get("agent_code", "unknown") by_agent[a] = by_agent.get(a, 0) + 1 return {"total": len(rows), "agg": agg, "by_agent": by_agent} PAGE = """ OpenClaw Affiliate Ops

âš¡ OpenClaw Affiliate Ops (HF Space)

Supabase: {{supabase_ok}}

{{stats}}
""" @app.get("/") def home(): try: s = stats_today() return render_template_string(PAGE, supabase_ok="connected", stats=s) except Exception as e: return render_template_string(PAGE, supabase_ok=f"error: {e}", stats={}) @app.get("/healthz") def healthz(): return jsonify({"ok": True, "time": now_iso()}) @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))): r = run_once() out.append(r) if r.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)