|
import os |
|
import stripe |
|
import requests |
|
import logging |
|
import pytz |
|
from fastapi import APIRouter, HTTPException, Header |
|
from datetime import datetime, timedelta |
|
from dateutil.relativedelta import relativedelta |
|
from typing import List, Dict, Any |
|
|
|
router = APIRouter() |
|
|
|
|
|
stripe.api_key = os.getenv("STRIPE_KEY") |
|
stripe.api_version = "2023-10-16" |
|
SUPABASE_URL = "https://ussxqnifefkgkaumjann.supabase.co" |
|
SUPABASE_KEY = os.getenv("SUPA_KEY") |
|
|
|
if not stripe.api_key or not SUPABASE_KEY: |
|
raise ValueError("❌ STRIPE_KEY ou SUPA_KEY não foram definidos no ambiente!") |
|
|
|
SUPABASE_HEADERS = { |
|
"apikey": SUPABASE_KEY, |
|
"Authorization": f"Bearer {SUPABASE_KEY}", |
|
"Content-Type": "application/json" |
|
} |
|
|
|
logging.basicConfig(level=logging.INFO) |
|
logger = logging.getLogger(__name__) |
|
|
|
def verify_token(user_token: str) -> str: |
|
""" |
|
Valida o token JWT no Supabase e retorna o user_id se for válido. |
|
""" |
|
headers = { |
|
"Authorization": f"Bearer {user_token}", |
|
"apikey": SUPABASE_KEY, |
|
"Content-Type": "application/json" |
|
} |
|
|
|
response = requests.get(f"{SUPABASE_URL}/auth/v1/user", headers=headers) |
|
|
|
if response.status_code == 200: |
|
user_data = response.json() |
|
user_id = user_data.get("id") |
|
if not user_id: |
|
raise HTTPException(status_code=400, detail="Invalid token: User ID not found") |
|
return user_id |
|
else: |
|
raise HTTPException(status_code=401, detail="Invalid or expired token") |
|
|
|
def get_monthly_revenue(account_id: str) -> List[Dict[str, Any]]: |
|
""" |
|
Busca os valores recebidos nos últimos 6 meses para o account_id. |
|
Retorna uma lista com número do mês, nome do mês, indicador se é o mês atual e valor. |
|
""" |
|
|
|
ny_timezone = pytz.timezone('America/New_York') |
|
now_ny = datetime.now(ny_timezone) |
|
current_month = now_ny.month |
|
current_year = now_ny.year |
|
|
|
|
|
monthly_data = {} |
|
|
|
|
|
for i in range(6): |
|
|
|
target_date = now_ny - relativedelta(months=i) |
|
month_num = target_date.month |
|
month_name = target_date.strftime('%b') |
|
year = target_date.year |
|
|
|
|
|
month_key = f"{year}-{month_num}" |
|
|
|
|
|
is_current = (month_num == current_month and year == current_year) |
|
|
|
monthly_data[month_key] = { |
|
"month": month_num, |
|
"name": month_name, |
|
"current": is_current, |
|
"amount": 0 |
|
} |
|
|
|
|
|
start_date = now_ny - relativedelta(months=6) |
|
start_timestamp = int(start_date.timestamp()) |
|
|
|
|
|
try: |
|
transfers = stripe.Transfer.list( |
|
destination=account_id, |
|
created={"gte": start_timestamp}, |
|
limit=100 |
|
) |
|
|
|
|
|
for transfer in transfers.data: |
|
transfer_date = datetime.fromtimestamp(transfer.created, ny_timezone) |
|
month_num = transfer_date.month |
|
year = transfer_date.year |
|
month_key = f"{year}-{month_num}" |
|
|
|
|
|
if month_key in monthly_data: |
|
|
|
monthly_data[month_key]["amount"] += transfer.amount |
|
|
|
|
|
result = list(monthly_data.values()) |
|
result.sort(key=lambda x: (current_month - x["month"]) % 12) |
|
|
|
return result |
|
|
|
except Exception as e: |
|
logger.error(f"❌ Error getting monthly revenue: {str(e)}") |
|
|
|
return list(monthly_data.values()) |
|
|
|
def get_account_balance(account_id: str) -> Dict[str, Any]: |
|
""" |
|
Busca o saldo disponível e pendente da conta Stripe. |
|
""" |
|
try: |
|
balance = stripe.Balance.retrieve(stripe_account=account_id) |
|
|
|
|
|
available_balance = 0 |
|
pending_balance = 0 |
|
currency = "BRL" |
|
|
|
|
|
for balance_item in balance.available: |
|
if balance_item.currency.upper() == "BRL": |
|
available_balance = balance_item.amount |
|
currency = "BRL" |
|
break |
|
|
|
|
|
for balance_item in balance.pending: |
|
if balance_item.currency.upper() == "BRL": |
|
pending_balance = balance_item.amount |
|
break |
|
|
|
return { |
|
"available_balance": available_balance, |
|
"pending_balance": pending_balance, |
|
"currency": currency |
|
} |
|
|
|
except Exception as e: |
|
logger.error(f"❌ Error getting account balance: {str(e)}") |
|
return { |
|
"available_balance": 0, |
|
"pending_balance": 0, |
|
"currency": "BRL" |
|
} |
|
|
|
@router.get("/dashboard") |
|
def get_dashboard(user_token: str = Header(None, alias="User-key")): |
|
try: |
|
if not user_token: |
|
raise HTTPException(status_code=401, detail="Missing User-key header") |
|
|
|
|
|
user_id = verify_token(user_token) |
|
logger.info(f"🔹 User verified. user_id: {user_id}") |
|
|
|
|
|
user_data_url = f"{SUPABASE_URL}/rest/v1/User?id=eq.{user_id}" |
|
response = requests.get( |
|
user_data_url, |
|
headers={ |
|
"Authorization": f"Bearer {user_token}", |
|
"apikey": SUPABASE_KEY, |
|
"Content-Type": "application/json" |
|
} |
|
) |
|
|
|
if response.status_code != 200 or not response.json(): |
|
raise HTTPException(status_code=404, detail="User not found") |
|
|
|
user_data = response.json()[0] |
|
stripe_id = user_data.get("stripe_id") |
|
|
|
if not stripe_id: |
|
return { |
|
"stripe_id": None, |
|
"available_balance": 0, |
|
"pending_balance": 0, |
|
"currency": "BRL", |
|
"monthly_revenue": [] |
|
} |
|
|
|
|
|
balance_info = get_account_balance(stripe_id) |
|
|
|
|
|
monthly_revenue = get_monthly_revenue(stripe_id) |
|
|
|
return { |
|
"stripe_id": stripe_id, |
|
"available_balance": balance_info["available_balance"], |
|
"pending_balance": balance_info["pending_balance"], |
|
"currency": balance_info["currency"], |
|
"monthly_revenue": monthly_revenue |
|
} |
|
|
|
except HTTPException as http_err: |
|
raise http_err |
|
except Exception as e: |
|
error_message = str(e) if str(e) else "An unknown error occurred" |
|
logger.error(f"❌ Error: {error_message}") |
|
raise HTTPException(status_code=500, detail=error_message) |