import os import datetime from google.cloud import bigquery class BillingAgent: def __init__(self): self.project_id = os.environ.get("GOOGLE_CLOUD_PROJECT") if not self.project_id: raise ValueError("GOOGLE_CLOUD_PROJECT environment variable not set.") self.billing_table = os.environ.get("BILLING_TABLE") if not self.billing_table: raise ValueError("BILLING_TABLE environment variable not set.") self.bq_client = bigquery.Client(project=self.project_id) # ── summary ────────────────────────────────────────────────────────────── def get_summary(self): query = f""" SELECT DATE(usage_start_time) AS usage_date, project.id AS project_id, service.description AS service, SUM(cost) AS daily_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) GROUP BY usage_date, project_id, service ORDER BY usage_date DESC, daily_cost DESC LIMIT 100 """ results = self.bq_client.query(query).result() summary = [ { "usage_date": str(row.usage_date), "project_id": row.project_id, "service": row.service, "daily_cost": row.daily_cost, } for row in results ] if not summary: return { "onboarding_status": { "state": "awaiting_data", "message": "Fakturaeksport er aktiv, men ingen data for siste 30 dager ennå.", } } return {"summary": summary} # ── forecast ────────────────────────────────────────────────────────────── def get_forecast(self): q7 = f""" SELECT SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS total_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY) """ total_7d = list(self.bq_client.query(q7).result())[0].total_cost or 0 daily_average = total_7d / 7 today = datetime.date.today() next_month = datetime.date( today.year + (1 if today.month == 12 else 0), (today.month % 12) + 1, 1 ) remaining_days = (next_month - today).days q_mtd = f""" SELECT SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS total_cost FROM `{self.billing_table}` WHERE EXTRACT(MONTH FROM _PARTITIONTIME) = EXTRACT(MONTH FROM CURRENT_DATE()) AND EXTRACT(YEAR FROM _PARTITIONTIME) = EXTRACT(YEAR FROM CURRENT_DATE()) """ mtd_cost = list(self.bq_client.query(q_mtd).result())[0].total_cost or 0 return { "daily_average_last_7_days": daily_average, "month_to_date_cost": mtd_cost, "forecasted_remaining_cost": daily_average * remaining_days, "total_monthly_forecast": mtd_cost + daily_average * remaining_days, "remaining_days_in_month": remaining_days, "data_note": "Prognose basert på siste 7 dager. BigQuery kan ha 24-48 timers forsinkelse.", } # ── anomalier ───────────────────────────────────────────────────────────── def get_anomalies(self): query = f""" WITH daily_costs AS ( SELECT service.description AS service, DATE(_PARTITIONTIME) AS usage_date, SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS daily_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 14 DAY) GROUP BY 1, 2 ), costs_with_avg AS ( SELECT service, usage_date, daily_cost, AVG(daily_cost) OVER ( PARTITION BY service ORDER BY usage_date ROWS BETWEEN 7 PRECEDING AND 1 PRECEDING ) AS avg_7day FROM daily_costs ) SELECT service, daily_cost AS today_cost, avg_7day, (daily_cost / avg_7day) AS ratio FROM costs_with_avg WHERE usage_date = CURRENT_DATE() AND avg_7day > 0 AND daily_cost > (2.0 * avg_7day) """ results = self.bq_client.query(query).result() return { "anomalies": [ { "service": row.service, "today_cost": row.today_cost, "avg_7d": row.avg_7day, "ratio": row.ratio, } for row in results ] } # ── daglig historikk for bar-chart (CG6) ────────────────────────────────── def get_daily_history(self, days: int = 30): """ Returnerer [{date: str, mtd: float}] for siste `days` dager, sortert ASC. MTD er kumulativ sum fra 1. i måneden til den dato. """ query = f""" SELECT DATE(usage_start_time) AS usage_date, SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS day_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {int(days)} DAY) GROUP BY usage_date ORDER BY usage_date ASC """ results = list(self.bq_client.query(query).result()) history = [] mtd = 0.0 for row in results: if history and row.usage_date.day == 1: mtd = 0.0 mtd += float(row.day_cost or 0) history.append({"date": str(row.usage_date), "mtd": round(mtd, 6)}) return history # ── tjenester gruppert per service (CG5+CG6) ───────────────────────────── def get_service_totals(self, days: int = 30): """ Grupperer kostnader per tjeneste (ikke per SKU). Compute Engine vises som én rad med total, med SKU-detaljer som sub-liste. Unngår duplikate tjeneste-rader i dashbordet. """ query = f""" SELECT service.description AS service, sku.description AS sku, SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) AS sku_cost FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {int(days)} DAY) GROUP BY service, sku HAVING sku_cost > 0 ORDER BY service, sku_cost DESC """ results = self.bq_client.query(query).result() # Grupper SKU-er under tjeneste — én rad per tjeneste i dashbordet services: dict = {} for row in results: svc = row.service if svc not in services: services[svc] = {"service": svc, "total_cost": 0.0, "skus": []} services[svc]["total_cost"] = round( services[svc]["total_cost"] + float(row.sku_cost), 6 ) services[svc]["skus"].append({ "sku": row.sku, "sku_cost": round(float(row.sku_cost), 6), }) return sorted(services.values(), key=lambda x: x["total_cost"], reverse=True) if __name__ == '__main__': agent = BillingAgent() print("Summary:", agent.get_summary()) print("Forecast:", agent.get_forecast()) print("Anomalies:", agent.get_anomalies()) print("By-service:", agent.get_service_totals()) print("History:", agent.get_daily_history())