feat(OQ-31): GET /opax/build-status + GET /notify/channels i main.py

This commit is contained in:
chrischristiansen-glitch 2026-06-13 05:53:13 +02:00
parent 7bd1d3fb50
commit 4dc5e278e8

525
main.py
View File

@ -11,6 +11,8 @@ CG5b: POST /notify/email — SendGrid digest til kunde-e-post.
CG5c: POST /notify/sms Twilio SMS, Guard+-tier, spike-varsler.
CG5-onboard: GET /onboard + POST /onboard/complete klient-onboarding wizard.
DOCS: GET /docs/{path} proxy til privat GitHub-repo via ADC/PAT.
OQ-29: GET /notify/channels list delivery-kanaler per gcp_project.
OQ-31: GET /opax/build-status ekte Cloud Build-status.
"""
import os
@ -51,6 +53,7 @@ import sendgrid
from sendgrid.helpers.mail import Mail
AGENT_ID = "osvauco-opax"
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5")
# Guard+-tiers som får tilgang til SMS-varsler
SMS_ALLOWED_TIERS = {"guard", "shield", "enterprise"}
@ -61,8 +64,7 @@ app = FastAPI(
version="0.1.0",
)
# Paths exempt from IAP enforcement (health/readiness probes reach Cloud Run
# directly without the IAP-injected x-goog-authenticated-user-email header).
# Paths exempt from IAP enforcement
IAP_EXEMPT_PATHS = {"/health", "/healthz", "/readiness", "/liveness", "/onboard"}
@app.middleware("http")
@ -148,16 +150,11 @@ if _static_dir.is_dir():
# ── DOCS PROXY (MD-filer fra privat GitHub-repo) ──────────────────────────────
_GH_REPO = "vauco-saas/OSVauco"
_GH_REPO = "vauco-saas/OSVauco"
_GH_BRANCH = os.environ.get("DOCS_BRANCH", "main")
@app.get("/docs/{path:path}")
async def proxy_doc(path: str):
"""
Henter vilkårlig fil fra privat GitHub-repo.
Bruker GITHUB_PAT env-var (Secret Manager Cloud Run).
Fallback: forsøker uten auth (fungerer ikke for private repos, men feiler pent).
"""
token = os.environ.get("GITHUB_PAT")
url = f"https://raw.githubusercontent.com/{_GH_REPO}/{_GH_BRANCH}/{path}"
headers = {"Authorization": f"token {token}"} if token else {}
@ -175,6 +172,83 @@ async def proxy_doc(path: str):
return Response(content=resp.text, media_type="text/plain; charset=utf-8")
# ── OQ-31: Cloud Build status ─────────────────────────────────────────────────
@app.get("/opax/build-status")
async def build_status():
"""
OQ-31 Ekte Cloud Build-status fra GCP.
Erstatter mock/manglende endepunkt. Henter siste bygg via Cloud Build API.
"""
try:
from google.cloud.devtools import cloudbuild_v1
client = cloudbuild_v1.CloudBuildClient()
request_obj = cloudbuild_v1.ListBuildsRequest(
project_id=PROJECT_ID,
filter='trigger_id!=""',
page_size=5,
)
builds = list(client.list_builds(request=request_obj))
if not builds:
return JSONResponse({"status": "unknown", "message": "Ingen builds funnet"})
b = builds[0]
status_map = {
1: "queued", 2: "working", 3: "success",
4: "failure", 5: "internal_error", 6: "timeout", 7: "cancelled"
}
status_str = status_map.get(int(b.status), "unknown")
duration_s = None
if b.start_time and b.finish_time:
duration_s = int(b.finish_time.seconds - b.start_time.seconds)
return JSONResponse({
"status": status_str,
"build_id": b.id,
"trigger_id": b.build_trigger_id or "",
"branch": (b.substitutions or {}).get("BRANCH_NAME", "main"),
"commit": (b.substitutions or {}).get("SHORT_SHA", ""),
"duration_s": duration_s,
"start_time": str(b.start_time) if b.start_time else None,
"finish_time": str(b.finish_time) if b.finish_time else None,
"log_url": b.log_url or "",
})
except Exception as e:
import logging
logging.getLogger(__name__).error(f"[/opax/build-status] Failed: {e}", exc_info=True)
return JSONResponse({"status": "error", "message": str(e)}, status_code=200)
# ── OQ-29: Notify channels per kunde ─────────────────────────────────────────
@app.get("/notify/channels")
async def notify_channels(gcp_project: str):
"""
OQ-29 List delivery-kanaler konfigurert for en kunde.
Leser fra Firestore: onboarded_customers/{gcp_project}.channels
Returformat: { "gcp_project": "xxx", "channels": ["webhook", "email", "sms"] }
"""
if not db:
return JSONResponse(
status_code=503,
content={"error": "Firestore ikke tilgjengelig"}
)
try:
doc = db.collection("onboarded_customers").document(gcp_project).get()
if not doc.exists:
return JSONResponse(
status_code=404,
content={"gcp_project": gcp_project, "channels": [], "detail": "Kunde ikke funnet"}
)
data = doc.to_dict()
return JSONResponse({
"gcp_project": gcp_project,
"company": data.get("company", ""),
"tier": data.get("tier", ""),
"channels": data.get("channels", []),
"webhook_url": data.get("webhook_url") or None,
"email_to": data.get("email_to") or None,
})
except Exception as e:
return JSONResponse(status_code=500, content={"error": str(e)})
# ── MODELLER ──────────────────────────────────────────────────────────────────
class RunRequest(BaseModel):
message: str
@ -218,29 +292,10 @@ class EmailNotifyRequest(BaseModel):
payload: Optional[Dict[str, Any]] = Field(None, description="Valgfri rådata å inkludere i e-post")
class SmsNotifyRequest(BaseModel):
"""
CG5c: SMS-varsel via Twilio REST API.
Kun tilgjengelig for Guard+-tier (guard | shield | enterprise).
Brukes primært til kritiske kostnadsspikes hold meldinger korte (<160 tegn).
"""
to: str = Field(
...,
description="Mottakers mobilnummer i E.164-format, f.eks. +4791234567",
pattern=r"^\+[1-9]\d{7,14}$",
)
body: str = Field(
...,
description="Meldingstekst. Maks 160 tegn anbefalt (ett SMS-segment).",
max_length=320,
)
event: str = Field(
"spike",
description="Event-type: spike | budget | anomaly | custom",
)
tier: str = Field(
...,
description="Kundens tier. Må være guard, shield eller enterprise.",
)
to: str = Field(..., description="Mottakers mobilnummer i E.164-format", pattern=r"^\+[1-9]\d{7,14}$")
body: str = Field(..., description="Meldingstekst. Maks 160 tegn anbefalt.", max_length=320)
event: str = Field("spike", description="Event-type: spike | budget | anomaly | custom")
tier: str = Field(..., description="Kundens tier. Må være guard, shield eller enterprise.")
# ── ONBOARDING MODEL (CG5-onboard) ────────────────────────────────────────────
@ -257,14 +312,6 @@ class OnboardCompleteRequest(BaseModel):
# ── NOTIFICATION SERVICE (CG5a + CG5b + CG5c) ─────────────────────────────────
class NotificationService:
"""
Felles delivery layer for CostGuard-varsler.
CG5a: send_webhook() poster til hvilken som helst webhook-URL
CG5b: send_email() sender via SendGrid
CG5c: send_sms() sender via Twilio REST (Guard+-tier)
"""
# Event-type → emoji + standardtekst
EVENT_META = {
"spike": {"emoji": "🚨", "label": "Kostnadsspike oppdaget"},
"digest": {"emoji": "📊", "label": "Daglig kostnadsoppsummering"},
@ -277,7 +324,6 @@ class NotificationService:
return self.EVENT_META.get(event, self.EVENT_META["custom"])
def _build_slack_payload(self, req: WebhookNotifyRequest) -> dict:
"""Slack/Teams/Google Chat-kompatibelt JSON-format."""
meta = self._event_meta(req.event)
return {
"text": f"{meta['emoji']} *{req.title}*\n{req.body}",
@ -293,19 +339,10 @@ class NotificationService:
}
async def send_webhook(self, req: WebhookNotifyRequest) -> dict:
"""
CG5a: POST JSON til oppgitt webhook-URL.
Støtter Slack, Teams, Google Chat, Discord og alle custom webhooks.
"""
slack_payload = self._build_slack_payload(req)
try:
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(
req.url,
json=slack_payload,
headers={"Content-Type": "application/json"},
)
print(f"[NotificationService] webhook → {req.url} status={resp.status_code}")
resp = await client.post(req.url, json=slack_payload, headers={"Content-Type": "application/json"})
if resp.status_code >= 400:
return {"status": "error", "http_status": resp.status_code, "detail": resp.text[:200]}
return {"status": "ok", "http_status": resp.status_code, "event": req.event}
@ -315,130 +352,61 @@ class NotificationService:
return {"status": "error", "detail": str(e)}
async def send_email(self, req: EmailNotifyRequest) -> dict:
"""
CG5b: Send e-post via SendGrid.
Hvis body_html er None, bygges innhold fra billing-data automatisk.
"""
api_key = os.environ.get("SENDGRID_API_KEY")
to_email = req.to or ALERT_EMAIL
if not api_key:
return {"status": "not_configured", "detail": "SENDGRID_API_KEY mangler"}
if not to_email:
return {"status": "not_configured", "detail": "Ingen mottaker-e-post (to eller ALERT_EMAIL)"}
return {"status": "not_configured", "detail": "Ingen mottaker-e-post"}
meta = self._event_meta(req.event)
subject = req.subject or f"{meta['emoji']} CostGuard: {meta['label']}{datetime.date.today()}"
if req.body_html:
html_content = req.body_html
else:
html_content = await self._build_digest_html(req.payload)
html_content = req.body_html if req.body_html else await self._build_digest_html(req.payload)
try:
message = Mail(
from_email="costguard@osvauco.no",
to_emails=to_email,
subject=subject,
html_content=html_content,
)
message = Mail(from_email="costguard@osvauco.no", to_emails=to_email, subject=subject, html_content=html_content)
sg = sendgrid.SendGridAPIClient(api_key)
response = sg.send(message)
msg_id = response.headers.get("X-Message-Id", "")
print(f"[NotificationService] email → {to_email} msg_id={msg_id}")
return {"status": "ok", "message_id": msg_id, "to": to_email, "event": req.event}
except Exception as e:
print(f"[NotificationService] email error: {e}", file=sys.stderr)
return {"status": "error", "detail": str(e)}
async def send_sms(self, req: SmsNotifyRequest) -> dict:
"""
CG5c: Send SMS via Twilio REST API (ingen Twilio SDK kun httpx).
Guard+-tier påkrevd. Konfigureres via env-vars:
TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_FROM_NUMBER
"""
# Guard+-tier-sjekk
if req.tier.lower() not in SMS_ALLOWED_TIERS:
return {
"status": "not_allowed",
"detail": f"SMS krever Guard+-tier. Aktuæll tier: '{req.tier}'. "
f"Oppgrader på opax.vauco.no/onboard.",
}
account_sid = os.environ.get("TWILIO_ACCOUNT_SID")
auth_token = os.environ.get("TWILIO_AUTH_TOKEN")
from_number = os.environ.get("TWILIO_FROM_NUMBER")
return {"status": "not_allowed", "detail": f"SMS krever Guard+-tier. Aktuell tier: '{req.tier}'."}
account_sid = os.environ.get("TWILIO_ACCOUNT_SID")
auth_token = os.environ.get("TWILIO_AUTH_TOKEN")
from_number = os.environ.get("TWILIO_FROM_NUMBER")
if not all([account_sid, auth_token, from_number]):
return {
"status": "not_configured",
"detail": "En eller flere Twilio-env-vars mangler: "
"TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_FROM_NUMBER",
}
return {"status": "not_configured", "detail": "Twilio env-vars mangler"}
meta = self._event_meta(req.event)
# Bygg meldingstekst med event-prefix hvis den ikke allerede starter med emoji
sms_body = req.body
if not sms_body.startswith(meta["emoji"]):
sms_body = f"{meta['emoji']} CostGuard: {sms_body}"
# Twilio Messages API — form-encoded POST
sms_body = req.body if req.body.startswith(meta["emoji"]) else f"{meta['emoji']} CostGuard: {req.body}"
twilio_url = f"https://api.twilio.com/2010-04-01/Accounts/{account_sid}/Messages.json"
try:
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.post(
twilio_url,
data={"From": from_number, "To": req.to, "Body": sms_body},
auth=(account_sid, auth_token),
)
resp = await client.post(twilio_url, data={"From": from_number, "To": req.to, "Body": sms_body}, auth=(account_sid, auth_token))
data = resp.json()
print(
f"[NotificationService] sms → {req.to} "
f"sid={data.get('sid','?')} status={data.get('status','?')}"
)
if resp.status_code >= 400:
return {
"status": "error",
"http_status": resp.status_code,
"twilio_code": data.get("code"),
"detail": data.get("message", resp.text[:200]),
}
return {
"status": "ok",
"sid": data.get("sid"),
"to": req.to,
"sms_status": data.get("status"),
"event": req.event,
}
return {"status": "error", "http_status": resp.status_code, "detail": data.get("message", "")}
return {"status": "ok", "sid": data.get("sid"), "to": req.to, "sms_status": data.get("status"), "event": req.event}
except httpx.TimeoutException:
return {"status": "error", "detail": "Twilio timeout (>15s)"}
except Exception as e:
print(f"[NotificationService] sms error: {e}", file=sys.stderr)
return {"status": "error", "detail": str(e)}
async def _build_digest_html(self, extra_payload: Optional[dict] = None) -> str:
"""Henter billing-data og bygger HTML-digest."""
try:
billing = BillingAgent()
summary = billing.get_summary().get("summary", [])
forecast = billing.get_forecast()
mtd = forecast.get("month_to_date_cost", 0)
daily_avg = forecast.get("daily_average_last_7_days", 0)
top3 = "".join(
f"<li>{s['service']}: kr {s.get('total_cost', 0):.2f}</li>"
for s in summary[:3]
)
top3 = "".join(f"<li>{s['service']}: kr {s.get('total_cost', 0):.2f}</li>" for s in summary[:3])
except Exception:
mtd, daily_avg, top3 = 0, 0, "<li>Data ikke tilgjengelig</li>"
extra_rows = ""
if extra_payload:
extra_rows = "".join(
f"<tr><td style='padding:4px 8px;color:#666'>{k}</td>"
f"<td style='padding:4px 8px'><strong>{v}</strong></td></tr>"
for k, v in extra_payload.items()
)
extra_rows = "".join(
f"<tr><td style='padding:4px 8px;color:#666'>{k}</td><td style='padding:4px 8px'><strong>{v}</strong></td></tr>"
for k, v in (extra_payload or {}).items()
)
return f"""
<div style='font-family:sans-serif;max-width:600px;margin:0 auto'>
<div style='background:#1a1a2e;padding:20px;border-radius:8px 8px 0 0'>
@ -447,82 +415,53 @@ class NotificationService:
</div>
<div style='background:#f9f9f9;padding:20px;border-radius:0 0 8px 8px'>
<table style='width:100%;border-collapse:collapse'>
<tr><td style='padding:4px 8px;color:#666'>Kostnad MTD</td>
<td style='padding:4px 8px'><strong>kr {mtd:.2f}</strong></td></tr>
<tr><td style='padding:4px 8px;color:#666'>Daglig snitt (7d)</td>
<td style='padding:4px 8px'><strong>kr {daily_avg:.2f}</strong></td></tr>
<tr><td style='padding:4px 8px;color:#666'>Kostnad MTD</td><td style='padding:4px 8px'><strong>kr {mtd:.2f}</strong></td></tr>
<tr><td style='padding:4px 8px;color:#666'>Daglig snitt (7d)</td><td style='padding:4px 8px'><strong>kr {daily_avg:.2f}</strong></td></tr>
{extra_rows}
</table>
<h4 style='margin-top:16px'>Topp 3 GCP-tjenester</h4>
<ul>{top3}</ul>
<hr style='border:none;border-top:1px solid #ddd;margin:16px 0'>
<p style='color:#999;font-size:12px'>CostGuard by Vauco AS &middot; opax.vauco.no</p>
<p style='color:#999;font-size:12px'>CostGuard by Vauco AS · opax.vauco.no</p>
</div>
</div>
"""
# Singleton
_notifier = NotificationService()
# ── NOTIFY ENDPOINTS (CG5a + CG5b + CG5c) ─────────────────────────────────────
async def _notify_webhook(request: Request, req: WebhookNotifyRequest):
result = await _notifier.send_webhook(req)
status_code = 200 if result["status"] == "ok" else 502
return JSONResponse(status_code=status_code, content=result)
return JSONResponse(status_code=200 if result["status"] == "ok" else 502, content=result)
app.add_api_route("/notify/webhook", endpoint=require_auth(_notify_webhook), methods=["POST"])
async def _notify_email(request: Request, req: EmailNotifyRequest):
result = await _notifier.send_email(req)
if result["status"] == "not_configured":
return JSONResponse(status_code=503, content=result)
status_code = 200 if result["status"] == "ok" else 500
return JSONResponse(status_code=status_code, content=result)
return JSONResponse(status_code=200 if result["status"] == "ok" else 500, content=result)
app.add_api_route("/notify/email", endpoint=require_auth(_notify_email), methods=["POST"])
async def _notify_sms(request: Request, req: SmsNotifyRequest):
"""
CG5c: POST /notify/sms Guard+-tier SMS via Twilio.
HTTP-svar:
200 sendt OK
402 tier ikke Guard+ (payment required / upgrade)
503 Twilio ikke konfigurert (env-vars mangler)
500 Twilio-feil
"""
result = await _notifier.send_sms(req)
if result["status"] == "not_allowed":
return JSONResponse(status_code=402, content=result)
if result["status"] == "not_configured":
return JSONResponse(status_code=503, content=result)
status_code = 200 if result["status"] == "ok" else 500
return JSONResponse(status_code=status_code, content=result)
return JSONResponse(status_code=200 if result["status"] == "ok" else 500, content=result)
app.add_api_route("/notify/sms", endpoint=require_auth(_notify_sms), methods=["POST"])
# ── ONBOARDING ENDPOINTS (CG5-onboard) ───────────────────────────────────────
@app.get("/onboard")
def onboard_wizard():
"""
CG5-onboard: Server onboarding-wizard HTML.
IAP-exempt kunder når denne uten eksisterende sesjon.
"""
return FileResponse("static/onboard.html")
@app.post("/onboard/complete")
async def onboard_complete(req: OnboardCompleteRequest):
"""
CG5-onboard: Lagrer kundeprofil i Firestore og sender velkomstvarsler
via eksisterende _notifier (send_webhook + send_email).
"""
if db:
try:
db.collection("onboarded_customers").document(req.gcp_project).set({
@ -536,87 +475,29 @@ async def onboard_complete(req: OnboardCompleteRequest):
"email_to": req.email_to,
"onboarded_at": firestore.SERVER_TIMESTAMP,
})
print(f"[Onboard] {req.gcp_project} lagret i Firestore")
except Exception as e:
print(f"[Onboard] Firestore-feil (ikke kritisk): {e}", file=sys.stderr)
print(f"[Onboard] Firestore-feil: {e}", file=sys.stderr)
notify_results: Dict[str, Any] = {}
if req.webhook_url and "webhook" in req.channels:
notify_results["webhook"] = await _notifier.send_webhook(WebhookNotifyRequest(
url=req.webhook_url,
event="custom",
url=req.webhook_url, event="custom",
title=f"🎉 {req.company} er koblet til CostGuard!",
body=(
f"Hei {req.contact}! CostGuard overvåker nå *{req.gcp_project}*. "
f"Du mottar første kostnadsrapport innen 24 timer."
),
payload={
"bedrift": req.company,
"tier": req.tier,
"prosjekt": req.gcp_project,
"dato": datetime.date.today().isoformat(),
},
body=f"Hei {req.contact}! CostGuard overvåker nå *{req.gcp_project}*.",
payload={"bedrift": req.company, "tier": req.tier, "prosjekt": req.gcp_project, "dato": datetime.date.today().isoformat()},
))
if req.email_to and "email" in req.channels:
TIER_LABELS = {
"starter": "Starter ($499/mnd)",
"guard": "Guard ($999/mnd)",
"shield": "Shield ($1.999/mnd)",
"enterprise": "Enterprise ($3.500+/mnd)",
}
tier_label = TIER_LABELS.get(req.tier, req.tier)
TIER_LABELS = {"starter": "Starter ($499/mnd)", "guard": "Guard ($999/mnd)", "shield": "Shield ($1.999/mnd)", "enterprise": "Enterprise ($3.500+/mnd)"}
notify_results["email"] = await _notifier.send_email(EmailNotifyRequest(
to=req.email_to,
event="custom",
to=req.email_to, event="custom",
subject=f"✅ Velkommen til CostGuard — {req.company}",
body_html=f"""
<div style='font-family:sans-serif;max-width:600px;margin:0 auto;padding:32px'>
<div style='background:linear-gradient(135deg,#7c5cff,#00d4a8);padding:28px;
border-radius:12px 12px 0 0;text-align:center'>
<h1 style='color:#fff;margin:0;font-size:26px'>🎉 Velkommen til CostGuard!</h1>
</div>
<div style='background:#f9f9f9;padding:28px;border-radius:0 0 12px 12px'>
<p style='font-size:16px'>Hei <strong>{req.contact}</strong>,</p>
<p>CostGuard overvåker ditt GCP-prosjekt og vil varsle deg ved kostnadsspikes,
anomalier og budsjettoverskridelser.</p>
<table style='width:100%;border-collapse:collapse;margin:20px 0'>
<tr style='background:#efefef'>
<td style='padding:8px 12px;font-weight:700'>Bedrift</td>
<td style='padding:8px 12px'>{req.company}</td></tr>
<tr>
<td style='padding:8px 12px;font-weight:700'>GCP-prosjekt</td>
<td style='padding:8px 12px'><code>{req.gcp_project}</code></td></tr>
<tr style='background:#efefef'>
<td style='padding:8px 12px;font-weight:700'>Tier</td>
<td style='padding:8px 12px'>{tier_label}</td></tr>
<tr>
<td style='padding:8px 12px;font-weight:700'>Kanaler</td>
<td style='padding:8px 12px'>{', '.join(req.channels)}</td></tr>
</table>
<p>Du mottar din første kostnadsrapport innen 24 timer.</p>
<div style='text-align:center;margin:24px 0'>
<a href='https://opax.vauco.no/static/costguard-dashboard.html'
style='background:linear-gradient(135deg,#7c5cff,#00d4a8);
color:#050509;padding:12px 28px;border-radius:99px;
text-decoration:none;font-weight:700;font-size:14px'>
Åpne dashboard
</a>
</div>
<hr style='border:none;border-top:1px solid #ddd'>
<p style='font-size:12px;color:#999'>CostGuard by Vauco AS · opax.vauco.no</p>
</div>
</div>
""",
body_html=f"<p>Hei {req.contact}, CostGuard overvåker nå {req.gcp_project} (tier: {TIER_LABELS.get(req.tier, req.tier)}).</p>",
))
internal_wh = os.environ.get("VAUCO_INTERNAL_WEBHOOK")
if internal_wh:
try:
await _notifier.send_webhook(WebhookNotifyRequest(
url=internal_wh,
event="custom",
url=internal_wh, event="custom",
title="🆕 Ny CostGuard-kunde onboardet",
body=f"{req.company} ({req.gcp_project}) — tier: {req.tier}",
payload={"kontakt": req.contact, "e-post": req.email, "kanaler": ", ".join(req.channels)},
@ -624,13 +505,7 @@ async def onboard_complete(req: OnboardCompleteRequest):
except Exception as e:
print(f"[Onboard] Intern Slack-feil: {e}", file=sys.stderr)
return {
"status": "ok",
"customer": req.gcp_project,
"company": req.company,
"channels": req.channels,
"notify": notify_results,
}
return {"status": "ok", "customer": req.gcp_project, "company": req.company, "channels": req.channels, "notify": notify_results}
# ── ROOT LANDING PAGE ───────────────────────────────────────────────────
@ -655,7 +530,7 @@ def service_worker():
return FileResponse("static/sw.js")
# ── AUTH — User Authentication ──────────────────────────────────────────
# ── ADMIN ─────────────────────────────────────────────────────────────────────
@app.get('/admin')
@require_auth
async def admin_panel(request: Request):
@ -664,76 +539,48 @@ async def admin_panel(request: Request):
@app.post('/admin/create-customer')
@require_auth
async def create_customer(request: Request):
import subprocess, shlex
import subprocess
data = await request.json()
customer_name = data.get("customer_name", "").strip()
project_id = data.get("project_id", "").strip()
customer_name = data.get("customer_name", "").strip()
project_id = data.get("project_id", "").strip()
billing_account_id = data.get("billing_account_id", "").strip()
alert_email = data.get("alert_email", "").strip()
container_image = data.get("container_image", "").strip()
region = data.get("region", "europe-north1").strip()
alert_email = data.get("alert_email", "").strip()
container_image = data.get("container_image", "").strip()
region = data.get("region", "europe-north1").strip()
if not all([customer_name, project_id, billing_account_id, alert_email, container_image]):
raise HTTPException(status_code=400, detail="Alle felt er påkrevd")
import re
if not re.match(r'^[a-z0-9_-]+$', customer_name):
raise HTTPException(status_code=400, detail="customer_name kan kun inneholde a-z, 0-9, - og _")
customer_dir = f"infrastructure/terraform/customers/{customer_name}"
template_dir = "infrastructure/terraform/customers/_template"
import os, shutil
import shutil
if os.path.exists(customer_dir):
raise HTTPException(status_code=409, detail=f"Kunde {customer_name} eksisterer allerede")
shutil.copytree(template_dir, customer_dir)
tfvars_content = f'''customer_id = "{customer_name}"
project_id = "{project_id}"
region = "{region}"
billing_account_id = "{billing_account_id}"
alert_email = "{alert_email}"
billing_viewer_emails = ["chris.christiansen@vauco.no", "jason.vauger@vauco.no"]
container_image = "{container_image}"
'''
tfvars_content = f'''customer_id = "{customer_name}"\nproject_id = "{project_id}"\nregion = "{region}"\nbilling_account_id = "{billing_account_id}"\nalert_email = "{alert_email}"\nbilling_viewer_emails = ["chris.christiansen@vauco.no", "jason.vauger@vauco.no"]\ncontainer_image = "{container_image}"\n'''
with open(f"{customer_dir}/terraform.tfvars", "w") as f:
f.write(tfvars_content)
try:
init = subprocess.run(
["terraform", f"-chdir={customer_dir}", "init", "-no-color"],
capture_output=True, text=True, timeout=120
)
init = subprocess.run(["terraform", f"-chdir={customer_dir}", "init", "-no-color"], capture_output=True, text=True, timeout=120)
if init.returncode != 0:
shutil.rmtree(customer_dir)
raise HTTPException(status_code=500, detail=f"terraform init feilet: {init.stderr[-500:]}")
apply = subprocess.run(
["terraform", f"-chdir={customer_dir}", "apply", "-auto-approve", "-no-color"],
capture_output=True, text=True, timeout=600
)
apply = subprocess.run(["terraform", f"-chdir={customer_dir}", "apply", "-auto-approve", "-no-color"], capture_output=True, text=True, timeout=600)
if apply.returncode != 0:
raise HTTPException(status_code=500, detail=f"terraform apply feilet: {apply.stderr[-500:]}")
return {"status": "ok", "customer": customer_name, "project_id": project_id}
except subprocess.TimeoutExpired:
raise HTTPException(status_code=504, detail="Terraform tok for lang tid (>10 min)")
# ── BILLING ENDPOINTS (CG1 + CG2) ────────────────────────────────────────────
async def get_budget(request: Request):
if not db:
return JSONResponse(status_code=500, content={"error": "Firestore is not configured"})
try:
doc_ref = db.collection("settings").document("budget")
doc = doc_ref.get()
if doc.exists:
return JSONResponse(content={"budget": doc.to_dict().get("limit", 500)})
else:
return JSONResponse(content={"budget": 500})
doc = db.collection("settings").document("budget").get()
return JSONResponse(content={"budget": doc.to_dict().get("limit", 500) if doc.exists else 500})
except Exception as e:
return JSONResponse(status_code=500, content={"error": str(e)})
@ -741,8 +588,7 @@ async def set_budget(request: Request, payload: BudgetUpdateRequest):
if not db:
return JSONResponse(status_code=500, content={"error": "Firestore is not configured"})
try:
doc_ref = db.collection("settings").document("budget")
doc_ref.set({"limit": payload.budget})
db.collection("settings").document("budget").set({"limit": payload.budget})
return JSONResponse(content={"status": "ok", "budget": payload.budget})
except Exception as e:
return JSONResponse(status_code=500, content={"error": str(e)})
@ -756,16 +602,12 @@ async def trigger_email_report():
result = await _notifier.send_email(EmailNotifyRequest(event="digest"))
if result["status"] == "not_configured":
return JSONResponse(status_code=200, content={"status": "not_configured", "error": result["detail"]})
if result["status"] == "error":
return JSONResponse(status_code=500, content=result)
return JSONResponse(content=result)
return JSONResponse(status_code=200 if result["status"] == "ok" else 500, content=result)
@app.post("/billing/snapshot")
async def create_daily_snapshot(request: Request):
email_response = await trigger_email_report()
if email_response.status_code != 200:
print(f"Daily snapshot: email report failed. Status: {email_response.status_code}", file=sys.stderr)
await trigger_email_report()
return JSONResponse(content={"status": "ok", "snapshot_id": str(datetime.date.today())})
@ -773,20 +615,17 @@ async def get_history(request: Request):
if not db:
return JSONResponse(status_code=500, content={"error": "Firestore is not configured"})
try:
end_date = datetime.date.today()
end_date = datetime.date.today()
start_date = end_date - datetime.timedelta(days=90)
docs = db.collection("daily_snapshots") \
.where("created_at", ">=", start_date.isoformat()) \
.order_by("created_at", direction=firestore.Query.DESCENDING) \
.limit(90) \
.stream()
.limit(90).stream()
history = [{"date": doc.id, **doc.to_dict()} for doc in docs]
history.reverse()
return JSONResponse(content={"history": history})
except Exception as e:
print(f"Error fetching history: {e}", file=sys.stderr)
return JSONResponse(status_code=500, content={"error": str(e)})
app.add_api_route("/billing/history", endpoint=require_auth(get_history), methods=["GET"])
@ -823,43 +662,31 @@ def subscribe_for_push(sub: PushSubscription):
if not db:
raise HTTPException(status_code=500, detail="Firestore is not configured")
try:
doc_ref = db.collection("push_subscribers").document(sub.token)
doc_ref.set({"budget_nok": sub.budget_nok, "subscribed_at": firestore.SERVER_TIMESTAMP})
db.collection("push_subscribers").document(sub.token).set({"budget_nok": sub.budget_nok, "subscribed_at": firestore.SERVER_TIMESTAMP})
return {"status": "ok"}
except Exception as e:
print(f"Error subscribing token: {e}", file=sys.stderr)
raise HTTPException(status_code=500, detail=str(e))
@app.post("/billing/budget-webhook")
async def budget_webhook(payload: BudgetWebhookPayload):
if not db:
print("Webhook called but Firestore is not configured. Aborting.", file=sys.stderr)
return {"status": "error", "detail": "Firestore not configured"}
try:
data = base64.b64decode(payload.message.get("data", "")).decode("utf-8")
data_json = json.loads(data)
cost_amount = data_json.get("costAmount", 0)
cost_amount = data_json.get("costAmount", 0)
budget_amount = data_json.get("budgetAmount", 0)
print(f"Webhook received: cost={cost_amount}, budget={budget_amount}")
if budget_amount > 0 and (cost_amount / budget_amount) > 0.8:
percent_used = round((cost_amount / budget_amount) * 100)
subscribers_ref = db.collection("push_subscribers")
tokens = [subscriber.id for subscriber in subscribers_ref.stream()]
if not tokens:
print("Budget threshold exceeded, but no push subscribers found.")
return {"status": "no subscribers"}
notification = messaging.Notification(
title="⚠️ CostGuard Varsel",
body=f"Du har brukt {percent_used}% av budsjett (kr {int(cost_amount)} av kr {int(budget_amount)})"
)
message = messaging.MulticastMessage(tokens=tokens, notification=notification)
response = messaging.send_multicast(message)
print(f"Sent push notification to {response.success_count} subscribers.")
else:
print("Budget threshold not exceeded, no notification sent.")
tokens = [s.id for s in db.collection("push_subscribers").stream()]
if tokens:
notification = messaging.Notification(
title="⚠️ CostGuard Varsel",
body=f"Du har brukt {percent_used}% av budsjett (kr {int(cost_amount)} av kr {int(budget_amount)})"
)
messaging.send_multicast(messaging.MulticastMessage(tokens=tokens, notification=notification))
return {"status": "processed"}
except Exception as e:
print(f"Error in budget webhook: {e}", file=sys.stderr)
return {"status": "error", "detail": str(e)}
@ -871,13 +698,12 @@ async def billing_live(request: Request):
return JSONResponse(status_code=401, content={"error": "Not authenticated"})
if ALLOWED_EMAILS and user.get('email') not in ALLOWED_EMAILS:
return JSONResponse(status_code=403, content={"error": "Email not allowed"})
print("Billing Budget API not implemented, falling back to BigQuery", file=sys.stderr)
bq_forecast = BillingAgent().get_forecast()
bq_forecast["data_source"] = "BigQuery Fallback"
return bq_forecast
# ── AWS BILLING ENDPOINTS ─────────────────────────────────────────────────────
# ── AWS BILLING ───────────────────────────────────────────────────────────────
async def authenticated_aws_billing_summary(request: Request):
try:
return AWSBillingAgent().get_summary()
@ -902,42 +728,25 @@ def run_agent(req: RunRequest):
raise HTTPException(status_code=400, detail=str(e))
except PermissionError as e:
raise HTTPException(status_code=403, detail=str(e))
store = get_store()
start = time.monotonic()
error_msg = None
response = None
response = None
try:
response = run(
message=req.message,
user_id=req.user_id,
session_id=req.session_id,
mode=req.mode,
)
response = run(message=req.message, user_id=req.user_id, session_id=req.session_id, mode=req.mode)
except Exception as e:
error_msg = str(e)
raise HTTPException(status_code=500, detail=error_msg)
finally:
duration = round(time.monotonic() - start, 3)
success = error_msg is None
model = "gemini-2.5-flash" if req.mode in ("light", "A") else "gemini-2.5-pro"
log_agent_call(
agent_id=AGENT_ID,
input_payload={"message": req.message, "mode": req.mode},
output=response,
model_used=model,
mode=req.mode,
duration_s=duration,
success=success,
error=error_msg,
)
success = error_msg is None
model = "gemini-2.5-flash" if req.mode in ("light", "A") else "gemini-2.5-pro"
log_agent_call(agent_id=AGENT_ID, input_payload={"message": req.message, "mode": req.mode}, output=response, model_used=model, mode=req.mode, duration_s=duration, success=success, error=error_msg)
store.push(AGENT_ID, "last_duration_s", duration)
store.push(AGENT_ID, "last_mode", req.mode)
store.push(AGENT_ID, "last_success", success)
if response:
store.push(AGENT_ID, "last_result_preview", response[:200])
return {"response": response}
@ -949,46 +758,26 @@ def run_dag(req: DagRequest):
raise HTTPException(status_code=400, detail=str(e))
except PermissionError as e:
raise HTTPException(status_code=403, detail=str(e))
def make_agent_fn(msg: str, idx: int):
def _agent_fn(payload: dict):
return run(
message=msg,
user_id=payload["user_id"],
session_id=f"{payload['session_id']}-dag-{idx}",
mode=payload["mode"],
)
return run(message=msg, user_id=payload["user_id"], session_id=f"{payload['session_id']}-dag-{idx}", mode=payload["mode"])
_agent_fn.__name__ = f"opax-dag-{idx}"
return _agent_fn
payload = {"user_id": req.user_id, "session_id": req.session_id, "mode": req.mode}
payload = {"user_id": req.user_id, "session_id": req.session_id, "mode": req.mode}
agent_fns = [make_agent_fn(msg, i) for i, msg in enumerate(req.messages)]
agent_ids = [f"opax-dag-{i}" for i in range(len(req.messages))]
start = time.monotonic()
tasks = build_agent_dag(agent_fns, payload, agent_ids)
start = time.monotonic()
tasks = build_agent_dag(agent_fns, payload, agent_ids)
results = execute_dag(tasks, scheduler=req.scheduler)
total_dur = round(time.monotonic() - start, 3)
store = get_store()
log_dag_execution(dag_id=req.session_id, agent_results=results, total_duration_s=total_dur)
store.push(AGENT_ID, "last_dag_total_duration_s", total_dur)
store.push(AGENT_ID, "last_dag_agent_count", len(agent_ids))
store.push(AGENT_ID, "last_dag_success_count", sum(1 for r in results if r["success"]))
return {
"total_duration_s": total_dur,
"results": [
{
"index": i,
"message": req.messages[i],
"response": r["result"],
"success": r["success"],
"duration_s": r["duration_s"],
"error": r["error"],
}
for i, r in enumerate(results)
],
"results": [{"index": i, "message": req.messages[i], "response": r["result"], "success": r["success"], "duration_s": r["duration_s"], "error": r["error"]} for i, r in enumerate(results)],
}