78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
import requests
|
|
|
|
|
|
def _raise_with_detail(r: requests.Response):
|
|
"""Raise an HTTPError but include upstream error body for easier debugging."""
|
|
try:
|
|
detail = r.json()
|
|
except Exception:
|
|
detail = r.text
|
|
http_error_msg = f"{r.status_code} {r.url}: {detail}"
|
|
raise requests.HTTPError(http_error_msg, response=r)
|
|
from django.conf import settings
|
|
from django.db.models import Sum
|
|
from .models import Test, ResultEntry, TextVariant
|
|
|
|
def agents_generate_texts(payload: dict) -> dict:
|
|
url = f"{settings.AGENTS_SERVICE_URL}/api/v1/texts/generate"
|
|
r = requests.post(url, json=payload, timeout=180)
|
|
|
|
if r.status_code >= 400:
|
|
_raise_with_detail(r)
|
|
|
|
return r.json()
|
|
|
|
def agents_analyze(rows: list[dict], objective: str, policy: dict | None = None) -> dict:
|
|
url = f"{settings.AGENTS_SERVICE_URL}/api/v1/tests/analyze"
|
|
payload = {"rows": rows, "objective": objective}
|
|
if policy:
|
|
payload["policy"] = policy
|
|
r = requests.post(url, json=payload, timeout=90)
|
|
|
|
if r.status_code >= 400:
|
|
_raise_with_detail(r)
|
|
|
|
return r.json()
|
|
|
|
def aggregate_test_rows(test: Test) -> list[dict]:
|
|
qs = (ResultEntry.objects.filter(test=test).values("variant_id").annotate(
|
|
impressions=Sum("impressions"), clicks=Sum("clicks"), conversions=Sum("conversions"), leads=Sum("leads"), spend=Sum("spend")
|
|
))
|
|
rows=[]
|
|
for row in qs:
|
|
v = TextVariant.objects.get(id=row["variant_id"])
|
|
rows.append({
|
|
"variant_id": v.id, "format": v.format,
|
|
"impressions": int(row["impressions"] or 0),
|
|
"clicks": int(row["clicks"] or 0),
|
|
"conversions": int(row["conversions"] or 0),
|
|
"leads": int(row["leads"] or 0),
|
|
"spend": float(row["spend"] or 0.0),
|
|
})
|
|
return rows
|
|
|
|
|
|
def aggregate_test_rows_by_segment(test: Test) -> list[dict]:
|
|
"""Aggregate stats per (segment, variant)."""
|
|
qs = (ResultEntry.objects.filter(test=test)
|
|
.values("segment_id", "variant_id")
|
|
.annotate(impressions=Sum("impressions"), clicks=Sum("clicks"),
|
|
conversions=Sum("conversions"), leads=Sum("leads"), spend=Sum("spend")))
|
|
rows = []
|
|
# preload formats
|
|
fmt_map = {tv.id: tv.format for tv in TextVariant.objects.filter(brief=test.brief)}
|
|
seg_map = {s.id: s.name for s in test.segments.all()}
|
|
for r in qs:
|
|
rows.append({
|
|
"variant_id": r["variant_id"],
|
|
"format": fmt_map.get(r["variant_id"]),
|
|
"segment_id": r["segment_id"],
|
|
"segment_name": seg_map.get(r["segment_id"]),
|
|
"impressions": int(r["impressions"] or 0),
|
|
"clicks": int(r["clicks"] or 0),
|
|
"conversions": int(r["conversions"] or 0),
|
|
"leads": int(r["leads"] or 0),
|
|
"spend": float(r["spend"] or 0.0),
|
|
})
|
|
return rows
|