250 lines
9.2 KiB
Python
250 lines
9.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""API-Routen KI-Zusatzguthaben (Phase 1, keine Live-Charges ohne Freigabe)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from aza_ai_budget import resolve_license_for_empfang
|
|
from aza_ai_credit import (
|
|
DEFAULT_SETUP_CANCEL_URL,
|
|
DEFAULT_SETUP_SUCCESS_URL,
|
|
DEFAULT_TOPUP_CANCEL_URL,
|
|
DEFAULT_TOPUP_CHF,
|
|
DEFAULT_TOPUP_SUCCESS_URL,
|
|
MAX_TOPUP_CHF,
|
|
MIN_TOPUP_CHF,
|
|
apply_auto_topup_user_settings,
|
|
auto_topup_settings_status,
|
|
chf_to_internal_usd,
|
|
create_setup_checkout_session,
|
|
create_topup_checkout_session,
|
|
ensure_ai_credit_schema,
|
|
list_user_credit_history,
|
|
)
|
|
from aza_security import require_api_token
|
|
|
|
router = APIRouter(tags=["ai-credit"], dependencies=[Depends(require_api_token)])
|
|
|
|
|
|
def _stripe_db_path() -> Path:
|
|
import os
|
|
|
|
base = Path(__file__).resolve().parent
|
|
return Path(os.environ.get("STRIPE_DB_PATH", str(base / "data" / "stripe_webhook.sqlite")))
|
|
|
|
|
|
def _resolve_practice_license(
|
|
*,
|
|
device_id: Optional[str],
|
|
practice_id: Optional[str],
|
|
):
|
|
db_path = _stripe_db_path()
|
|
if not db_path.exists():
|
|
raise HTTPException(status_code=503, detail="Billing database unavailable")
|
|
import backend_main as bm
|
|
|
|
bm.ensure_license_schema(db_path)
|
|
with sqlite3.connect(str(db_path)) as con:
|
|
ensure_ai_credit_schema(con)
|
|
lic = resolve_license_for_empfang(
|
|
con,
|
|
x_device_id=device_id,
|
|
session_practice_id=practice_id,
|
|
)
|
|
if not lic:
|
|
raise HTTPException(status_code=402, detail="No active license mapping")
|
|
if not (lic.practice_id or "").strip():
|
|
raise HTTPException(status_code=400, detail="practice_id missing on license")
|
|
return con, lic
|
|
|
|
|
|
class TopupCheckoutIn(BaseModel):
|
|
package_chf: Optional[float] = Field(default=30.0)
|
|
custom_amount_chf: Optional[float] = None
|
|
save_payment_method: bool = False
|
|
success_url: str = Field(default=DEFAULT_TOPUP_SUCCESS_URL)
|
|
cancel_url: str = Field(default=DEFAULT_TOPUP_CANCEL_URL)
|
|
|
|
|
|
class AutoTopupSettingsIn(BaseModel):
|
|
enabled: bool
|
|
amount_chf: Optional[float] = Field(default=None, ge=MIN_TOPUP_CHF, le=MAX_TOPUP_CHF)
|
|
trigger_below_percent: Optional[int] = Field(default=None, ge=1, le=50)
|
|
monthly_limit_chf: Optional[float] = Field(default=None, ge=MIN_TOPUP_CHF, le=900.0)
|
|
|
|
|
|
class SetupSessionIn(BaseModel):
|
|
success_url: str = Field(default=DEFAULT_SETUP_SUCCESS_URL)
|
|
cancel_url: str = Field(default=DEFAULT_SETUP_CANCEL_URL)
|
|
|
|
|
|
@router.post("/v1/ai-credit/topup/checkout")
|
|
def ai_credit_topup_checkout(
|
|
body: TopupCheckoutIn,
|
|
request: Request,
|
|
x_device_id: Optional[str] = Header(default=None, alias="X-Device-Id"),
|
|
x_practice_id: Optional[str] = Header(default=None, alias="X-Practice-Id"),
|
|
) -> JSONResponse:
|
|
device_id = (x_device_id or "").strip() or None
|
|
practice_hdr = (x_practice_id or "").strip() or None
|
|
db_path = _stripe_db_path()
|
|
import backend_main as bm
|
|
|
|
bm.ensure_license_schema(db_path)
|
|
with sqlite3.connect(str(db_path)) as con:
|
|
ensure_ai_credit_schema(con)
|
|
lic = resolve_license_for_empfang(
|
|
con,
|
|
x_device_id=device_id,
|
|
session_practice_id=practice_hdr,
|
|
)
|
|
if not lic or not (lic.practice_id or "").strip():
|
|
return JSONResponse(status_code=402, content={"ok": False, "error_code": "NO_LICENSE"})
|
|
paid_chf = float(body.custom_amount_chf or body.package_chf or DEFAULT_TOPUP_CHF)
|
|
if paid_chf < MIN_TOPUP_CHF or paid_chf > MAX_TOPUP_CHF:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"ok": False, "error_code": "INVALID_AMOUNT", "min_chf": MIN_TOPUP_CHF, "max_chf": MAX_TOPUP_CHF},
|
|
)
|
|
internal_usd = chf_to_internal_usd(paid_chf)
|
|
result = create_topup_checkout_session(
|
|
practice_id=lic.practice_id or "",
|
|
subscription_id=lic.subscription_id,
|
|
customer_email=lic.customer_email,
|
|
paid_chf=paid_chf,
|
|
internal_usd=internal_usd,
|
|
save_payment_method=body.save_payment_method,
|
|
success_url=DEFAULT_TOPUP_SUCCESS_URL,
|
|
cancel_url=(body.cancel_url or DEFAULT_TOPUP_CANCEL_URL).strip() or DEFAULT_TOPUP_CANCEL_URL,
|
|
)
|
|
return JSONResponse(content=result)
|
|
|
|
|
|
@router.get("/v1/ai-credit/history")
|
|
def ai_credit_history(
|
|
x_device_id: Optional[str] = Header(default=None, alias="X-Device-Id"),
|
|
x_practice_id: Optional[str] = Header(default=None, alias="X-Practice-Id"),
|
|
limit: int = Query(default=100, ge=1, le=200),
|
|
) -> JSONResponse:
|
|
device_id = (x_device_id or "").strip() or None
|
|
practice_hdr = (x_practice_id or "").strip() or None
|
|
db_path = _stripe_db_path()
|
|
import backend_main as bm
|
|
|
|
bm.ensure_license_schema(db_path)
|
|
with sqlite3.connect(str(db_path)) as con:
|
|
ensure_ai_credit_schema(con)
|
|
lic = resolve_license_for_empfang(
|
|
con,
|
|
x_device_id=device_id,
|
|
session_practice_id=practice_hdr,
|
|
)
|
|
if not lic or not (lic.practice_id or "").strip():
|
|
return JSONResponse(status_code=402, content={"ok": False, "error_code": "NO_LICENSE"})
|
|
pid = lic.practice_id or ""
|
|
items = list_user_credit_history(con, practice_id=pid, limit=limit)
|
|
return JSONResponse(
|
|
content={
|
|
"ok": True,
|
|
"practice_id": pid,
|
|
"subscription_id": lic.subscription_id,
|
|
"items": items,
|
|
}
|
|
)
|
|
|
|
|
|
@router.get("/v1/ai-credit/auto-topup/settings")
|
|
def ai_credit_auto_topup_settings_get(
|
|
x_device_id: Optional[str] = Header(default=None, alias="X-Device-Id"),
|
|
x_practice_id: Optional[str] = Header(default=None, alias="X-Practice-Id"),
|
|
) -> JSONResponse:
|
|
device_id = (x_device_id or "").strip() or None
|
|
practice_hdr = (x_practice_id or "").strip() or None
|
|
db_path = _stripe_db_path()
|
|
import backend_main as bm
|
|
|
|
bm.ensure_license_schema(db_path)
|
|
with sqlite3.connect(str(db_path)) as con:
|
|
ensure_ai_credit_schema(con)
|
|
lic = resolve_license_for_empfang(
|
|
con,
|
|
x_device_id=device_id,
|
|
session_practice_id=practice_hdr,
|
|
)
|
|
if not lic or not (lic.practice_id or "").strip():
|
|
return JSONResponse(status_code=402, content={"ok": False, "error_code": "NO_LICENSE"})
|
|
return JSONResponse(
|
|
content=auto_topup_settings_status(con, practice_id=lic.practice_id or "")
|
|
)
|
|
|
|
|
|
@router.post("/v1/ai-credit/auto-topup/settings")
|
|
def ai_credit_auto_topup_settings(
|
|
body: AutoTopupSettingsIn,
|
|
x_device_id: Optional[str] = Header(default=None, alias="X-Device-Id"),
|
|
x_practice_id: Optional[str] = Header(default=None, alias="X-Practice-Id"),
|
|
) -> JSONResponse:
|
|
device_id = (x_device_id or "").strip() or None
|
|
practice_hdr = (x_practice_id or "").strip() or None
|
|
db_path = _stripe_db_path()
|
|
import backend_main as bm
|
|
|
|
bm.ensure_license_schema(db_path)
|
|
with sqlite3.connect(str(db_path)) as con:
|
|
ensure_ai_credit_schema(con)
|
|
lic = resolve_license_for_empfang(
|
|
con,
|
|
x_device_id=device_id,
|
|
session_practice_id=practice_hdr,
|
|
)
|
|
if not lic or not (lic.practice_id or "").strip():
|
|
return JSONResponse(status_code=402, content={"ok": False, "error_code": "NO_LICENSE"})
|
|
result = apply_auto_topup_user_settings(
|
|
con,
|
|
practice_id=lic.practice_id or "",
|
|
enabled=body.enabled,
|
|
amount_chf=body.amount_chf,
|
|
trigger_below_percent=body.trigger_below_percent,
|
|
monthly_limit_chf=body.monthly_limit_chf,
|
|
)
|
|
if not result.get("ok"):
|
|
return JSONResponse(status_code=400, content=result)
|
|
return JSONResponse(content=result)
|
|
|
|
|
|
@router.post("/v1/ai-credit/auto-topup/setup-session")
|
|
def ai_credit_auto_topup_setup_session(
|
|
body: SetupSessionIn,
|
|
x_device_id: Optional[str] = Header(default=None, alias="X-Device-Id"),
|
|
x_practice_id: Optional[str] = Header(default=None, alias="X-Practice-Id"),
|
|
) -> JSONResponse:
|
|
device_id = (x_device_id or "").strip() or None
|
|
practice_hdr = (x_practice_id or "").strip() or None
|
|
db_path = _stripe_db_path()
|
|
import backend_main as bm
|
|
|
|
bm.ensure_license_schema(db_path)
|
|
with sqlite3.connect(str(db_path)) as con:
|
|
ensure_ai_credit_schema(con)
|
|
lic = resolve_license_for_empfang(
|
|
con,
|
|
x_device_id=device_id,
|
|
session_practice_id=practice_hdr,
|
|
)
|
|
if not lic or not (lic.practice_id or "").strip():
|
|
return JSONResponse(status_code=402, content={"ok": False, "error_code": "NO_LICENSE"})
|
|
result = create_setup_checkout_session(
|
|
practice_id=lic.practice_id or "",
|
|
customer_email=lic.customer_email,
|
|
success_url=body.success_url,
|
|
cancel_url=body.cancel_url,
|
|
)
|
|
return JSONResponse(content=result)
|