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"
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 = """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)