mcp: api.py — fix column names to match live ClickHouse schema

docker/mcp/api.py

@@ -0,0 +1,176 @@
+"""
+Manar REST API — FastAPI on port 8092
+Columns: brand_id, platform, account_id, campaign_id, campaign_name,
+         adgroup_id, adgroup_name, hour, date, impressions, clicks,
+         cost_micros, conversions, conv_value, video_views
+"""
+
+import os
+import threading
+import logging
+from datetime import date, timedelta
+
+import clickhouse_connect
+from fastapi import FastAPI, Query
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+
+log = logging.getLogger("manar-api")
+
+CH_HOST  = os.environ.get("CLICKHOUSE_HOST", "clickhouse")
+CH_PORT  = int(os.environ.get("CLICKHOUSE_PORT", 8123))
+CH_DB    = os.environ.get("CLICKHOUSE_DB", "manar")
+CH_USER  = os.environ.get("CLICKHOUSE_USER", "default")
+CH_PASS  = os.environ.get("CLICKHOUSE_PASSWORD", "")
+API_PORT = int(os.environ.get("API_PORT", 8092))
+
+app = FastAPI(title="Manar REST API", version="0.3.0")
+app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "POST"], allow_headers=["*"])
+
+
+def get_ch():
+    return clickhouse_connect.get_client(
+        host=CH_HOST, port=CH_PORT, database=CH_DB,
+        username=CH_USER, password=CH_PASS
+    )
+
+
+@app.get("/api/health")
+def health():
+    return {"status": "ok"}
+
+
+@app.get("/api/{brand}/summary")
+def summary(brand: str, days: int = Query(30, ge=1, le=365)):
+    try:
+        ch = get_ch()
+        result = ch.query(f"""
+            SELECT period,
+                sum(impressions)  AS impressions,
+                sum(clicks)       AS clicks,
+                round(sum(cost_micros)/1e6, 2) AS cost,
+                round(sum(conversions), 1)      AS conversions,
+                round(sum(conv_value), 2)       AS conv_value,
+                round(if(sum(impressions)>0, sum(clicks)/sum(impressions)*100, 0), 2) AS ctr_pct,
+                round(if(sum(clicks)>0, sum(cost_micros)/1e6/sum(clicks), 0), 2)     AS cpc,
+                round(if(sum(cost_micros)>0, sum(conv_value)/(sum(cost_micros)/1e6), 0), 2) AS roas
+            FROM (
+                SELECT if(date >= today()-{days}, 'current', 'prior') AS period,
+                       impressions, clicks, cost_micros, conversions, conv_value
+                FROM manar.ad_metrics_hourly
+                WHERE brand_id='{brand}' AND date >= today()-{days*2}
+            ) GROUP BY period
+        """)
+        data = {}
+        for row in result.result_rows:
+            d = dict(zip(result.column_names, row))
+            data[d.pop("period")] = d
+        current = data.get("current", {})
+        prior   = data.get("prior", {})
+
+        def pct(a, b):
+            try: return round((a - b) / b * 100, 1) if b else None
+            except: return None
+
+        return {
+            "brand": brand, "days": days,
+            "current": current, "prior": prior,
+            "changes": {k: pct(current.get(k, 0), prior.get(k, 0))
+                        for k in ["impressions", "clicks", "cost", "conversions", "roas", "cpc"]}
+        }
+    except Exception as e:
+        log.error(f"summary {brand}: {e}")
+        return {"brand": brand, "days": days, "current": {}, "prior": {}, "changes": {}}
+
+
+@app.get("/api/{brand}/campaigns")
+def campaigns(brand: str, days: int = Query(30, ge=1, le=365)):
+    try:
+        ch = get_ch()
+        today = date.today()
+        date_from = today - timedelta(days=days)
+        result = ch.query(f"""
+            SELECT campaign_id, campaign_name,
+                sum(impressions) AS impressions, sum(clicks) AS clicks,
+                round(sum(cost_micros)/1e6, 2) AS cost,
+                round(sum(conversions), 2) AS conversions,
+                round(sum(conv_value), 2) AS conv_value,
+                round(if(sum(impressions)>0, sum(clicks)/sum(impressions), 0), 4) AS ctr,
+                round(if(sum(clicks)>0, sum(cost_micros)/1e6/sum(clicks), 0), 2) AS cpc,
+                round(if(sum(cost_micros)>0, sum(conv_value)/(sum(cost_micros)/1e6), 0), 2) AS roas
+            FROM manar.ad_metrics_hourly
+            WHERE brand_id='{brand}' AND date>='{date_from}' AND date<='{today}'
+            GROUP BY campaign_id, campaign_name ORDER BY cost DESC LIMIT 50
+        """)
+        return [dict(zip(result.column_names, r)) for r in result.result_rows]
+    except Exception as e:
+        log.error(f"campaigns {brand}: {e}")
+        return []
+
+
+@app.get("/api/{brand}/trend")
+def trend(brand: str, days: int = Query(30, ge=1, le=365)):
+    try:
+        ch = get_ch()
+        today = date.today()
+        date_from = today - timedelta(days=days)
+        result = ch.query(f"""
+            SELECT toString(date) AS date,
+                sum(impressions) AS impressions, sum(clicks) AS clicks,
+                round(sum(cost_micros)/1e6, 2) AS cost,
+                round(sum(conversions), 2) AS conversions
+            FROM manar.ad_metrics_hourly
+            WHERE brand_id='{brand}' AND date>='{date_from}' AND date<='{today}'
+            GROUP BY date ORDER BY date ASC
+        """)
+        return [dict(zip(result.column_names, r)) for r in result.result_rows]
+    except Exception as e:
+        log.error(f"trend {brand}: {e}")
+        return []
+
+
+@app.get("/api/{brand}/search-terms")
+def search_terms(brand: str, days: int = Query(30, ge=1, le=365), limit: int = Query(50, ge=1, le=200)):
+    try:
+        ch = get_ch()
+        today = date.today()
+        date_from = today - timedelta(days=days)
+        result = ch.query(f"""
+            SELECT search_term,
+                sum(impressions) AS impressions, sum(clicks) AS clicks,
+                round(sum(cost_micros)/1e6, 2) AS cost,
+                round(sum(conversions), 2) AS conversions,
+                round(if(sum(clicks)>0, sum(cost_micros)/1e6/sum(clicks), 0), 2) AS cpc
+            FROM manar.search_term_performance
+            WHERE brand_id='{brand}' AND date>='{date_from}' AND date<='{today}'
+            GROUP BY search_term ORDER BY cost DESC LIMIT {limit}
+        """)
+        return [dict(zip(result.column_names, r)) for r in result.result_rows]
+    except Exception as e:
+        log.error(f"search-terms {brand}: {e}")
+        return []
+
+
+@app.get("/api/{brand}/anomalies")
+def anomalies(brand: str):
+    try:
+        ch = get_ch()
+        result = ch.query(f"""
+            SELECT campaign_id, metric, expected_value, actual_value,
+                   deviation_pct, severity, toString(detected_at) AS detected_at
+            FROM manar.anomaly_events
+            WHERE brand_id='{brand}' AND resolved_at IS NULL
+            ORDER BY detected_at DESC LIMIT 50
+        """)
+        return [dict(zip(result.column_names, r)) for r in result.result_rows]
+    except Exception as e:
+        log.error(f"anomalies {brand}: {e}")
+        return []
+
+
+def start_api_thread():
+    def run():
+        uvicorn.run(app, host="0.0.0.0", port=API_PORT, log_level="warning")
+    t = threading.Thread(target=run, daemon=True)
+    t.start()
+    log.info(f"Manar REST API on :{API_PORT}")