Spaces:
Sleeping
Sleeping
File size: 7,618 Bytes
f0ba3c6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | import asyncio
from datetime import datetime, timedelta
import logging
import os
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from integrations.microsoft365_service import microsoft365_service
try:
from integrations.stripe_service import stripe_service
HAS_STRIPE = True
except ImportError:
# Stripe is SaaS-specific billing integration
stripe_service = None
HAS_STRIPE = False
from integrations.xero_service import XeroService
from integrations.zoho_books_service import ZohoBooksService
router = APIRouter(prefix="/api/atom/finance/live", tags=["finance-live"])
logger = logging.getLogger(__name__)
# --- Data Models ---
class UnifiedTransaction(BaseModel):
id: str
description: str
amount: float
currency: str
date: str
status: str
platform: str # 'stripe', 'xero', 'quickbooks', 'zoho', 'dynamics'
customer_name: Optional[str] = None
url: Optional[str] = None
class FinanceStats(BaseModel):
total_revenue: float
pending_revenue: float
transaction_count: int
platform_breakdown: Dict[str, float]
class LiveFinanceResponse(BaseModel):
ok: bool = True
stats: FinanceStats
transactions: List[UnifiedTransaction]
providers: Dict[str, bool]
# --- Helper Functions ---
def map_stripe_payment(payment: Dict[str, Any]) -> UnifiedTransaction:
amount = float(payment.get("amount", 0)) / 100.0 # Stripe is in cents
return UnifiedTransaction(
id=payment.get("id"),
description=payment.get("description") or "Stripe Payment",
amount=amount,
currency=payment.get("currency", "usd"),
date=datetime.fromtimestamp(payment.get("created", 0)).isoformat(),
status=payment.get("status", "unknown"),
platform="stripe",
customer_name=None, # Requires extra fetch or expansion
url=f"https://dashboard.stripe.com/payments/{payment.get('id')}"
)
def map_xero_invoice(invoice: Dict[str, Any]) -> UnifiedTransaction:
return UnifiedTransaction(
id=invoice.get("InvoiceID"),
description=f"Invoice #{invoice.get('InvoiceNumber')}",
amount=float(invoice.get("Total", 0.0)),
currency=invoice.get("CurrencyCode", "USD"),
date=invoice.get("DateString", "") or datetime.now().isoformat(),
status=invoice.get("Status", "unknown"),
platform="xero",
customer_name=invoice.get("Contact", {}).get("Name"),
url=None # Would need organization ID to construct Deep Link
)
def map_zoho_invoice(invoice: Dict[str, Any]) -> UnifiedTransaction:
return UnifiedTransaction(
id=invoice.get("invoice_id"),
description=f"Invoice {invoice.get('invoice_number')}",
amount=float(invoice.get("total", 0.0)),
currency=invoice.get("currency_code", "USD"),
date=invoice.get("date"),
status=invoice.get("status"),
customer_name=invoice.get("customer_name")
)
def map_dynamics_invoice(invoice: Dict[str, Any]) -> UnifiedTransaction:
# Placeholder mapping for Dynamics 365 via MS Graph trending/insights
return UnifiedTransaction(
id=invoice.get("id", "dynamics_invoice"),
description=invoice.get("resourceVisualization", {}).get("title") or "Dynamics Invoice",
amount=0.0,
currency="USD",
date=datetime.now().isoformat(),
status="active",
platform="dynamics",
url=invoice.get("resourceReference", {}).get("webUrl")
)
# --- Endpoints ---
@router.get("/overview", response_model=LiveFinanceResponse)
async def get_live_financial_overview(
limit: int = 50,
):
"""
Fetch live financial data from connected providers (Stripe, Xero)
and aggregate into a unified view.
"""
transactions = []
providers_status = {"stripe": False, "xero": False, "zoho": False, "dynamics": False}
# 1. Fetch Stripe Data
try:
# Check env token for now, similar to other live APIs
stripe_token = os.getenv("STRIPE_SECRET_KEY")
if stripe_token:
raw_payments = stripe_service.list_payments(stripe_token, limit=limit)
charges = raw_payments.get("data", [])
transactions.extend([map_stripe_payment(p) for p in charges])
providers_status["stripe"] = True
except Exception as e:
logger.warning(f"Failed to fetch live Stripe data: {e}")
# 2. Fetch Xero Data
# Fetching Xero requires a valid access_token from environment or user context
try:
xero_token = os.getenv("XERO_ACCESS_TOKEN")
if not xero_token:
logger.warning("XERO_ACCESS_TOKEN not configured, skipping Xero fetch")
else:
# Use xero_service to fetch invoices/transactions
from integrations.xero_service import xero_service
xero_invoices = xero_service.get_invoices(access_token=xero_token, limit=limit)
transactions.extend([map_xero_invoice(i) for i in xero_invoices])
providers_status["xero"] = True
except Exception as e:
logger.warning(f"Failed to fetch live Xero data: {e}")
# 3. Fetch Zoho Books Data
try:
zoho_token = os.getenv("ZOHO_CRM_ACCESS_TOKEN") # Reusing token
org_id = os.getenv("ZOHO_BOOKS_ORG_ID")
if zoho_token and org_id:
zoho = ZohoBooksService()
# Fetching invoices as proxy for transactions
headers = zoho._get_headers(zoho_token, org_id)
url = f"{zoho.base_url}/invoices"
async with httpx.AsyncClient() as client:
res = await client.get(url, headers=headers, params={"organization_id": org_id})
if res.status_code == 200:
raw_invoices = res.json().get("invoices", [])
transactions.extend([map_zoho_invoice(i) for i in raw_invoices])
providers_status["zoho"] = True
except Exception as e:
logger.warning(f"Failed to fetch live Zoho Books data: {e}")
# 4. Fetch Dynamics 365 Finance Data
try:
ms_token = os.getenv("MICROSOFT_365_ACCESS_TOKEN")
if ms_token:
res = await microsoft365_service.get_dynamics_invoices(access_token=ms_token, top=limit)
if res.get("status") == "success":
raw_invoices = res.get("data", {}).get("value", [])
transactions.extend([map_dynamics_invoice(i) for i in raw_invoices])
providers_status["dynamics"] = True
except Exception as e:
logger.warning(f"Failed to fetch live Dynamics 365 data: {e}")
# Calculate Stats
total_rev = sum(t.amount for t in transactions if t.status in ['succeeded', 'paid', 'paid'])
pending_rev = sum(t.amount for t in transactions if t.status in ['pending', 'open'])
breakdown = {
"stripe": sum(t.amount for t in transactions if t.platform == 'stripe'),
"xero": sum(t.amount for t in transactions if t.platform == 'xero'),
"zoho": sum(t.amount for t in transactions if t.platform == 'zoho'),
"dynamics": sum(t.amount for t in transactions if t.platform == 'dynamics')
}
return LiveFinanceResponse(
ok=True,
stats=FinanceStats(
total_revenue=total_rev,
pending_revenue=pending_rev,
transaction_count=len(transactions),
platform_breakdown=breakdown
),
transactions=sorted(transactions, key=lambda x: x.date, reverse=True)[:limit],
providers=providers_status
)
|