import os import datetime from google.cloud import bigquery # ── Kreditt-konfig: tildelte totaler per type (NOK) ────────────────────────── # Sett disse i Cloud Run env-vars eller cloudbuild.yaml --set-env-vars # GOOGLE_CREDIT_INFRA_NOK = Free Trial (Cloud Run, BQ, Compute, Artifact Registry) # GOOGLE_CREDIT_VERTEX_NOK = Trial for Gen App Builder (Vertex AI tokens) # GOOGLE_CREDIT_DIALOGFLOW_NOK = Dialogflow CX Trial CREDIT_CONFIG = [ { "env_key": "GOOGLE_CREDIT_INFRA_NOK", "label": "Free Trial (Infrastruktur)", "bq_types": ["PROMOTION"], # BQ credit type "skus": None, # alle SKU-er (generell) "expires_days": 27, # kjent utløp fra Console "currency": "NOK", }, { "env_key": "GOOGLE_CREDIT_VERTEX_NOK", "label": "Trial for Gen App Builder (Vertex AI)", "bq_types": ["PROMOTION"], "skus": ["Vertex AI", "Generative AI", "Cloud AI"], "expires_days": None, "currency": "NOK", }, { "env_key": "GOOGLE_CREDIT_DIALOGFLOW_NOK", "label": "Dialogflow CX Trial", "bq_types": ["PROMOTION"], "skus": ["Dialogflow"], "expires_days": None, "currency": "NOK", }, ] 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.", } # ── discover_credits: aktiv scanner for alle kreditttyper i BQ ─────────────── def discover_credits(self, days: int = 90): """ Scanner BQ-billing for alle unike kreditttyper siste {days} dager. Returnerer liste med type, full_name, total brukt, og forslag til env-var. Nyttig for å oppdage nye kreditter automatisk. """ q = f""" SELECT cr.type AS credit_type, cr.full_name AS full_name, ROUND(SUM(cr.amount), 2) AS total_used, COUNT(DISTINCT DATE(_PARTITIONTIME)) AS active_days FROM `{self.billing_table}`, UNNEST(credits) AS cr WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {int(days)} DAY) GROUP BY credit_type, full_name ORDER BY total_used ASC """ rows = list(self.bq_client.query(q).result()) discovered = [] for row in rows: label = (row.full_name or row.credit_type or "Ukjent").strip() env_suggestion = "GOOGLE_CREDIT_" + label.upper().replace(" ", "_").replace("-", "_")[:30] + "_NOK" discovered.append({ "credit_type": row.credit_type, "full_name": row.full_name, "total_used_nok": abs(float(row.total_used or 0)), "active_days": row.active_days, "env_suggestion": env_suggestion, "configured": any( os.environ.get(c["env_key"]) for c in CREDIT_CONFIG if row.credit_type in c["bq_types"] ), }) return { "discovered": discovered, "period_days": days, "note": "Sett env-vars i Cloud Run for nøyaktig runway-beregning per kreditttype.", } # ── credits-status (oppdatert) ──────────────────────────────────────────────── def get_credits_status(self, days: int = 90): """ Henter faktiske kreditter fra BigQuery og beregner separat runway per kreditttype (INFRA, VERTEX, DIALOGFLOW) basert på env-vars i NOK. Returnerer: - credits_by_type: [{type, full_name, amount}] - credit_pools: [{label, total_nok, used_nok, remaining_nok, runway_days, exhaustion_date, expires_days, warning}] - credits_used_total_nok: total brukt (alle typer) - gross_cost_total_nok: bruttokostnad - net_cost_total_nok: nettokostnad etter kreditter - daily_gross_burn_nok: daglig bruttokostnad (7d snitt) - daily_credit_burn_nok: daglig kredittforbruk (7d snitt) - data_as_of: siste dato med data i BQ - warning: kritisk advarsel (Free Trial 27d) """ q_credits = f""" SELECT cr.type AS credit_type, cr.full_name AS full_name, ROUND(SUM(cr.amount), 4) AS total_amount FROM `{self.billing_table}`, UNNEST(credits) AS cr WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {int(days)} DAY) GROUP BY credit_type, full_name ORDER BY total_amount ASC """ q_burn = f""" SELECT ROUND(SUM(cost) / 7, 6) AS daily_gross, ROUND(SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)) / 7, 6) AS daily_credit FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY) """ q_totals = f""" SELECT ROUND(SUM(cost), 4) AS gross_total, ROUND(SUM(cost) + SUM(IFNULL( (SELECT SUM(c.amount) FROM UNNEST(credits) c), 0)), 4) AS net_total FROM `{self.billing_table}` WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {int(days)} DAY) """ q_latest = f""" SELECT MAX(DATE(usage_start_time)) AS latest_date FROM `{self.billing_table}` """ credits_rows = list(self.bq_client.query(q_credits).result()) burn_row = list(self.bq_client.query(q_burn).result())[0] totals_row = list(self.bq_client.query(q_totals).result())[0] latest_row = list(self.bq_client.query(q_latest).result())[0] credits_by_type = [ { "type": row.credit_type, "full_name": row.full_name, "amount": float(row.total_amount), } for row in credits_rows ] credits_used_total = abs(sum(r["amount"] for r in credits_by_type)) gross_total = float(totals_row.gross_total or 0) net_total = float(totals_row.net_total or 0) daily_gross = float(burn_row.daily_gross or 0) daily_credit = abs(float(burn_row.daily_credit or 0)) data_as_of = str(latest_row.latest_date) if latest_row.latest_date else None # ── Per-pool runway ─────────────────────────────────────────────────────── credit_pools = [] top_warning = None for cfg in CREDIT_CONFIG: total_nok = float(os.environ.get(cfg["env_key"], "0") or "0") # Brukt beregnes som andel av total BQ-kreditter (alle PROMOTION er NOK) used_nok = credits_used_total # samme BQ-total deles; raffineres hvis SKU-filter legges til remaining_nok = round(total_nok - used_nok, 2) if total_nok > 0 else None runway_days = None exhaustion_date = None pool_warning = None if total_nok > 0 and daily_credit > 0 and remaining_nok is not None: runway_days = int(remaining_nok / daily_credit) if remaining_nok > 0 else 0 exhaustion_date = str( datetime.date.today() + datetime.timedelta(days=runway_days) ) if runway_days > 0 else str(datetime.date.today()) # Utløpsadvarsel fra Console (kjent) expires_days = cfg.get("expires_days") if expires_days is not None and expires_days <= 30: pool_warning = f"⚠️ Utløper om {expires_days} dager — kr {total_nok:,.2f} går tapt!" top_warning = pool_warning elif runway_days is not None and runway_days < 30: pool_warning = f"⚠️ Estimert tom om {runway_days} dager ({exhaustion_date})." if top_warning is None: top_warning = pool_warning elif runway_days is not None and runway_days < 60: pool_warning = f"⚠️ Kreditter estimert tom om {runway_days} dager." credit_pools.append({ "env_key": cfg["env_key"], "label": cfg["label"], "total_nok": total_nok if total_nok > 0 else None, "used_nok": round(used_nok, 2), "remaining_nok": remaining_nok, "runway_days": runway_days, "exhaustion_date": exhaustion_date, "expires_days": expires_days, "configured": total_nok > 0, "warning": pool_warning, }) return { "credits_used_total_nok": round(credits_used_total, 4), "gross_cost_total_nok": gross_total, "net_cost_total_nok": net_total, "daily_gross_burn_nok": round(daily_gross, 6), "daily_credit_burn_nok": round(daily_credit, 6), "credit_pools": credit_pools, "credits_by_type": credits_by_type, "data_as_of": data_as_of, "period_days": days, "warning": top_warning, "setup_note": None if any(p["configured"] for p in credit_pools) else ( "Sett GOOGLE_CREDIT_INFRA_NOK, GOOGLE_CREDIT_VERTEX_NOK og " "GOOGLE_CREDIT_DIALOGFLOW_NOK i Cloud Run env for nøyaktig runway per kreditttype." ), } # ── 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): 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): 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() 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("Credits status:", agent.get_credits_status()) print("Credits discover:",agent.discover_credits()) print("Anomalies:", agent.get_anomalies()) print("By-service:", agent.get_service_totals()) print("History:", agent.get_daily_history())