feat(CG5a+CG5b): NotificationService + POST /notify/webhook + POST /notify/email
This commit is contained in:
parent
d4ffae29b8
commit
4a1e23e077
335
main.py
335
main.py
|
|
@ -6,6 +6,8 @@ ML-1: telemetri, state store, DAG.
|
|||
CG1+CG2: billing endpoints + IAP-beskyttelse.
|
||||
CG3: static-mappe serveres fra /static/*.
|
||||
AUTH: /auth/login + /auth/callback for klient OAuth2 onboarding.
|
||||
CG5a: POST /notify/webhook — push til Slack/Teams/Chat/Discord/etc.
|
||||
CG5b: POST /notify/email — SendGrid digest til kunde-e-post.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
|
@ -19,8 +21,9 @@ from fastapi import FastAPI, HTTPException, Request
|
|||
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from pydantic import BaseModel, Field
|
||||
from typing import List
|
||||
from typing import List, Optional, Dict, Any
|
||||
import datetime
|
||||
import httpx
|
||||
from cachetools import cached, TTLCache
|
||||
|
||||
from agent import run, authorize_mode
|
||||
|
|
@ -66,46 +69,33 @@ async def require_iap(request: Request, call_next):
|
|||
# ── FIREBASE & FIRESTORE INIT ───────────────────────────────────────────────
|
||||
db = None
|
||||
try:
|
||||
# When running in Google Cloud, ADC will be used automatically.
|
||||
# For local dev, `gcloud auth application-default login` is required.
|
||||
firebase_admin.initialize_app()
|
||||
db = firestore.Client()
|
||||
print("Firestore client initialized successfully.")
|
||||
except Exception as e:
|
||||
# The app can still run in a limited mode without Firestore.
|
||||
# Endpoints that depend on `db` will return a 500 error.
|
||||
print(f"WARNING: Firestore client failed to initialize: {e}", file=sys.stderr)
|
||||
|
||||
# ── AUTH & SESSION ────────────────────────────────────────────────────────────
|
||||
# Add session middleware for storing auth state
|
||||
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*")
|
||||
app.add_middleware(SessionMiddleware, secret_key=os.environ.get("SESSION_SECRET"))
|
||||
|
||||
# Configure Authlib's OAuth client
|
||||
oauth = OAuth()
|
||||
oauth.register(
|
||||
name='google',
|
||||
client_id=os.environ.get("GOOGLE_CLIENT_ID"),
|
||||
client_secret=os.environ.get("GOOGLE_CLIENT_SECRET"),
|
||||
server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
|
||||
client_kwargs={
|
||||
'scope': 'openid email profile'
|
||||
}
|
||||
client_kwargs={'scope': 'openid email profile'}
|
||||
)
|
||||
|
||||
|
||||
@app.get('/auth/login')
|
||||
async def login(request: Request):
|
||||
"""Redirects user to Google's OAuth 2.0 login page."""
|
||||
redirect_uri = "https://opax.vauco.no/auth/callback"
|
||||
return await oauth.google.authorize_redirect(request, redirect_uri)
|
||||
|
||||
@app.get('/auth/callback', name='auth')
|
||||
async def auth(request: Request):
|
||||
"""
|
||||
Handles the callback from Google's OAuth.
|
||||
Stores user info in session and redirects to the billing dashboard.
|
||||
"""
|
||||
token = await oauth.google.authorize_access_token(request)
|
||||
user = token.get('userinfo')
|
||||
if user:
|
||||
|
|
@ -115,7 +105,6 @@ async def auth(request: Request):
|
|||
|
||||
@app.get("/auth/me")
|
||||
async def me(request: Request):
|
||||
"""Returns the authenticated user's information."""
|
||||
user = request.session.get("user")
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Not authenticated")
|
||||
|
|
@ -124,12 +113,10 @@ async def auth(request: Request):
|
|||
|
||||
@app.get('/auth/logout')
|
||||
async def logout(request: Request):
|
||||
"""Clears the user session and logs them out."""
|
||||
request.session.pop('user', None)
|
||||
return RedirectResponse(url='/static/billing-dashboard.html')
|
||||
|
||||
|
||||
# Allowed emails for login
|
||||
ALLOWED_EMAILS = [email.strip() for email in os.environ.get("ALLOWED_EMAILS", "").split(",") if email.strip()]
|
||||
ALERT_EMAIL = os.environ.get("ALERT_EMAIL")
|
||||
|
||||
|
|
@ -141,10 +128,8 @@ def require_auth(func):
|
|||
user = request.session.get('user')
|
||||
if not user:
|
||||
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"})
|
||||
|
||||
return await func(request, *args, **kwargs)
|
||||
return wrapper
|
||||
|
||||
|
|
@ -181,10 +166,193 @@ class BudgetWebhookPayload(BaseModel):
|
|||
class BudgetUpdateRequest(BaseModel):
|
||||
budget: float
|
||||
|
||||
|
||||
# ── NOTIFICATION MODELS (CG5a + CG5b) ────────────────────────────────────────
|
||||
class WebhookNotifyRequest(BaseModel):
|
||||
url: str = Field(..., description="Webhook-URL (Slack/Teams/Google Chat/Discord/custom)")
|
||||
event: str = Field("custom", description="Event-type: spike | digest | budget | anomaly | custom")
|
||||
title: str = Field("CostGuard varsel", description="Tittel på meldingen")
|
||||
body: str = Field(..., description="Meldingstekst")
|
||||
payload: Optional[Dict[str, Any]] = Field(None, description="Valgfri rådata å inkludere")
|
||||
|
||||
class EmailNotifyRequest(BaseModel):
|
||||
to: Optional[str] = Field(None, description="Mottaker-e-post. Faller tilbake til ALERT_EMAIL hvis tom.")
|
||||
subject: Optional[str] = Field(None, description="Emne. Auto-generert fra event hvis tom.")
|
||||
event: str = Field("digest", description="Event-type: spike | digest | budget | anomaly | custom")
|
||||
body_html: Optional[str] = Field(None, description="HTML-innhold. Auto-generert fra billing-data hvis tom.")
|
||||
payload: Optional[Dict[str, Any]] = Field(None, description="Valgfri rådata å inkludere i e-post")
|
||||
|
||||
|
||||
# ── NOTIFICATION SERVICE (CG5a + CG5b) ───────────────────────────────────────
|
||||
class NotificationService:
|
||||
"""
|
||||
Felles delivery layer for CostGuard-varsler.
|
||||
CG5a: send_webhook() — poster til hvilken som helst webhook-URL
|
||||
CG5b: send_email() — sender via SendGrid
|
||||
Brukt av POST /notify/webhook og POST /notify/email.
|
||||
"""
|
||||
|
||||
# Event-type → emoji + standardtekst
|
||||
EVENT_META = {
|
||||
"spike": {"emoji": "🚨", "label": "Kostnadsspike oppdaget"},
|
||||
"digest": {"emoji": "📊", "label": "Daglig kostnadsoppsummering"},
|
||||
"budget": {"emoji": "⚠️", "label": "Budsjettgrense nærmer seg"},
|
||||
"anomaly": {"emoji": "🔍", "label": "Anomali oppdaget"},
|
||||
"custom": {"emoji": "📢", "label": "CostGuard varsel"},
|
||||
}
|
||||
|
||||
def _event_meta(self, event: str) -> dict:
|
||||
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}",
|
||||
"attachments": [
|
||||
{
|
||||
"color": "#FF6B35" if req.event in ("spike", "budget") else "#4A90D9",
|
||||
"fields": [
|
||||
{"title": k, "value": str(v), "short": True}
|
||||
for k, v in (req.payload or {}).items()
|
||||
]
|
||||
}
|
||||
] if req.payload else []
|
||||
}
|
||||
|
||||
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}")
|
||||
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}
|
||||
except httpx.TimeoutException:
|
||||
return {"status": "error", "detail": "Webhook timeout (>10s)"}
|
||||
except Exception as e:
|
||||
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)"}
|
||||
|
||||
meta = self._event_meta(req.event)
|
||||
subject = req.subject or f"{meta['emoji']} CostGuard: {meta['label']} — {datetime.date.today()}"
|
||||
|
||||
# Auto-generer HTML-innhold fra billing-data hvis ikke oppgitt
|
||||
if req.body_html:
|
||||
html_content = req.body_html
|
||||
else:
|
||||
html_content = 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,
|
||||
)
|
||||
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 _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]
|
||||
)
|
||||
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()
|
||||
)
|
||||
|
||||
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'>
|
||||
<h2 style='color:#fff;margin:0'>📊 CostGuard — Daglig oppsummering</h2>
|
||||
<p style='color:#aaa;margin:4px 0 0'>{datetime.date.today()}</p>
|
||||
</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>
|
||||
{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 · opax.vauco.no</p>
|
||||
</div>
|
||||
</div>
|
||||
"""
|
||||
|
||||
|
||||
# Singleton
|
||||
_notifier = NotificationService()
|
||||
|
||||
|
||||
# ── NOTIFY ENDPOINTS (CG5a + CG5b) ───────────────────────────────────────────
|
||||
|
||||
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)
|
||||
|
||||
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)
|
||||
|
||||
app.add_api_route("/notify/email", endpoint=require_auth(_notify_email), methods=["POST"])
|
||||
|
||||
|
||||
# ── ROOT LANDING PAGE ───────────────────────────────────────────────────
|
||||
@app.get("/")
|
||||
def root():
|
||||
"""Serves the main landing page."""
|
||||
return FileResponse("static/opax.html")
|
||||
|
||||
|
||||
|
|
@ -282,7 +450,7 @@ async def get_budget(request: Request):
|
|||
if doc.exists:
|
||||
return JSONResponse(content={"budget": doc.to_dict().get("limit", 500)})
|
||||
else:
|
||||
return JSONResponse(content={"budget": 500}) # Default
|
||||
return JSONResponse(content={"budget": 500})
|
||||
except Exception as e:
|
||||
return JSONResponse(status_code=500, content={"error": str(e)})
|
||||
|
||||
|
|
@ -303,77 +471,23 @@ app.add_api_route("/billing/budget", endpoint=require_auth(set_budget), methods=
|
|||
@app.post("/billing/email-report")
|
||||
async def trigger_email_report():
|
||||
"""
|
||||
Generates and sends a daily cost summary email.
|
||||
Legacy scheduler-endepunkt. Kaller NotificationService.send_email() internt.
|
||||
Beholdes for bakoverkompatibilitet med Cloud Scheduler.
|
||||
"""
|
||||
if not ALERT_EMAIL or not os.environ.get("SENDGRID_API_KEY"):
|
||||
err_msg = "Email reporting is not configured. Missing ALERT_EMAIL or SENDGRID_API_KEY."
|
||||
print(err_msg, file=sys.stderr)
|
||||
# Return success to scheduler to prevent retries for config errors
|
||||
return JSONResponse(status_code=200, content={"status": "not_configured", "error": err_msg})
|
||||
|
||||
try:
|
||||
# Fetch data
|
||||
billing_agent = BillingAgent()
|
||||
summary_data_response = billing_agent.get_summary()
|
||||
summary_data = summary_data_response.get("summary", [])
|
||||
forecast_data = billing_agent.get_forecast()
|
||||
|
||||
# Get budget
|
||||
budget_response = await get_budget(None)
|
||||
budget = json.loads(budget_response.body).get("budget", 500)
|
||||
|
||||
# Format email content
|
||||
mtd_cost = forecast_data.get("month_to_date_cost", 0)
|
||||
daily_avg = forecast_data.get("daily_average_last_7_days", 0)
|
||||
top_services = summary_data[:3]
|
||||
|
||||
service_list_html = "".join([f"<li>{s['service']}: kr {s.get('total_cost', 0):.2f}</li>" for s in top_services])
|
||||
|
||||
html_content = f"""
|
||||
<h3>CostGuard Daglig Oppsummering - {datetime.date.today()}</h3>
|
||||
<p><strong>Kostnad MTD:</strong> kr {mtd_cost:.2f}</p>
|
||||
<p><strong>Daglig snitt (7d):</strong> kr {daily_avg:.2f}</p>
|
||||
<p><strong>Budsjettstatus:</strong> kr {mtd_cost:.2f} av kr {budget:.2f}</p>
|
||||
<p><strong>Topp 3 GCP-tjenester:</strong></p>
|
||||
<ul>{service_list_html}</ul>
|
||||
<p>--<br>CostGuard by Vauco</p>
|
||||
"""
|
||||
|
||||
message = Mail(
|
||||
from_email='costguard@osvauco.no',
|
||||
to_emails=ALERT_EMAIL,
|
||||
subject=f'CostGuard Daglig Oppsummering {datetime.date.today()}',
|
||||
html_content=html_content
|
||||
)
|
||||
|
||||
sg = sendgrid.SendGridAPIClient(os.environ.get('SENDGRID_API_KEY'))
|
||||
response = sg.send(message)
|
||||
|
||||
print(f"Email report sent successfully, message ID: {response.headers.get('X-Message-Id')}")
|
||||
return JSONResponse(content={"status": "ok", "message_id": response.headers.get('X-Message-Id')})
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error sending email report: {e}", file=sys.stderr)
|
||||
# Return 500 to indicate a transient failure that scheduler might retry
|
||||
return JSONResponse(status_code=500, content={"error": str(e)})
|
||||
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)
|
||||
|
||||
|
||||
@app.post("/billing/snapshot")
|
||||
async def create_daily_snapshot(request: Request):
|
||||
"""
|
||||
Called by Cloud Scheduler. Fetches current billing data and saves a snapshot to Firestore.
|
||||
This endpoint should be secured by Cloud Scheduler's authentication (e.g., OIDC token).
|
||||
"""
|
||||
# ... (snapshot logic remains the same) ...
|
||||
|
||||
# After successfully creating the snapshot, trigger the email report.
|
||||
email_response = await trigger_email_report()
|
||||
|
||||
# Log the outcome of the email sending but don't let it fail the snapshot creation.
|
||||
if email_response.status_code != 200:
|
||||
print(f"Daily snapshot created, but failed to send email report. Status: {email_response.status_code}", file=sys.stderr)
|
||||
|
||||
return JSONResponse(content={"status": "ok", "snapshot_id": date_str})
|
||||
print(f"Daily snapshot: email report failed. Status: {email_response.status_code}", file=sys.stderr)
|
||||
return JSONResponse(content={"status": "ok", "snapshot_id": str(datetime.date.today())})
|
||||
|
||||
|
||||
async def get_history(request: Request):
|
||||
|
|
@ -382,18 +496,13 @@ async def get_history(request: Request):
|
|||
try:
|
||||
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()
|
||||
|
||||
history = [{"date": doc.id, **doc.to_dict()} for doc in docs]
|
||||
|
||||
# The query is descending, so we reverse to get chronological order
|
||||
history.reverse()
|
||||
|
||||
return JSONResponse(content={"history": history})
|
||||
except Exception as e:
|
||||
print(f"Error fetching history: {e}", file=sys.stderr)
|
||||
|
|
@ -402,11 +511,6 @@ async def get_history(request: Request):
|
|||
app.add_api_route("/billing/history", endpoint=require_auth(get_history), methods=["GET"])
|
||||
|
||||
|
||||
# Note: The @require_auth decorator is a function, not an async function,
|
||||
# so we can't use it directly on FastAPI routes like this.
|
||||
# Instead, we will wrap the functions manually for now.
|
||||
# A more robust solution would use Depends.
|
||||
|
||||
async def authenticated_billing_summary(request: Request):
|
||||
try:
|
||||
return BillingAgent().get_summary()
|
||||
|
|
@ -415,7 +519,6 @@ async def authenticated_billing_summary(request: Request):
|
|||
app.add_api_route("/billing/summary", endpoint=require_auth(authenticated_billing_summary), methods=["GET"])
|
||||
|
||||
|
||||
|
||||
async def authenticated_billing_forecast(request: Request):
|
||||
try:
|
||||
return BillingAgent().get_forecast()
|
||||
|
|
@ -434,10 +537,6 @@ app.add_api_route("/billing/anomalies", endpoint=require_auth(authenticated_bill
|
|||
|
||||
@app.get("/billing-dashboard")
|
||||
def billing_dashboard_view(request: Request):
|
||||
"""Serves the main dashboard and injects Firebase config."""
|
||||
# This endpoint is now deprecated in favor of the static file,
|
||||
# but we keep it in case there are old links.
|
||||
# The new auth flow redirects to the static file directly.
|
||||
return RedirectResponse(url='/static/billing-dashboard.html')
|
||||
|
||||
@app.post("/billing/subscribe")
|
||||
|
|
@ -456,76 +555,50 @@ def subscribe_for_push(sub: PushSubscription):
|
|||
async def budget_webhook(payload: BudgetWebhookPayload):
|
||||
if not db:
|
||||
print("Webhook called but Firestore is not configured. Aborting.", file=sys.stderr)
|
||||
# Return 200 to prevent Pub/Sub from retrying
|
||||
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)
|
||||
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")
|
||||
subscribers = subscribers_ref.stream()
|
||||
|
||||
tokens = [subscriber.id for subscriber in 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
|
||||
)
|
||||
|
||||
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.")
|
||||
|
||||
return {"status": "processed"}
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error in budget webhook: {e}", file=sys.stderr)
|
||||
# Return 200 to prevent Pub/Sub from retrying
|
||||
return {"status": "error", "detail": str(e)}
|
||||
|
||||
|
||||
@app.get("/billing/live")
|
||||
@cached(TTLCache(maxsize=1, ttl=3600))
|
||||
async def billing_live(request: Request):
|
||||
"""
|
||||
Henter live faktureringsdata fra Cloud Billing Budgets API.
|
||||
Dette gir raskere, men mindre detaljert, data enn BigQuery-eksporten.
|
||||
Fallback til BigQuery ved feil.
|
||||
"""
|
||||
# This endpoint should also be protected.
|
||||
user = request.session.get('user')
|
||||
if not user:
|
||||
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"})
|
||||
|
||||
# Fallback to BigQuery forecast
|
||||
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 (CG6) ──────────────────────────────────────────────
|
||||
|
||||
# ── AWS BILLING ENDPOINTS ─────────────────────────────────────────────────────
|
||||
async def authenticated_aws_billing_summary(request: Request):
|
||||
try:
|
||||
return AWSBillingAgent().get_summary()
|
||||
|
|
@ -621,7 +694,7 @@ def run_dag(req: DagRequest):
|
|||
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(results))
|
||||
store.push(AGENT_ID, "last_dag_agent_count", len(agents))
|
||||
store.push(AGENT_ID, "last_dag_success_count", sum(1 for r in results if r["success"]))
|
||||
|
||||
return {
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user