""" Xero API service for fetching live revenue data. Handles authentication, token refresh, and department revenue extraction. Supports multiple branches with different department configurations. """ import os import sys from pathlib import Path from datetime import datetime, date, timedelta from typing import Dict, List, Optional import logging import time from dotenv import load_dotenv from app.config.xero_config import get_xero_config, is_xero_enabled, get_department_tracking_config, get_cash_in_bank_account_names, CASH_IN_BANK_ACCOUNT_NAMES from app.services.xero_cache import xero_cache backend_dir = Path(__file__).parent.parent.parent _root_env = backend_dir.parent / ".env" if _root_env.exists(): load_dotenv(_root_env) else: load_dotenv(backend_dir / ".env") logger = logging.getLogger(__name__) # Lazy imports - only import when actually needed def _get_xero_modules(): """Lazy import of Xero API modules to avoid startup issues.""" try: # Add the xero_api directory to the path so we can import from it xero_api_path = backend_dir / "data" / "xero_api" if str(xero_api_path) not in sys.path: sys.path.insert(0, str(xero_api_path)) from token_manager import refresh_access_token from main import ( get_profit_and_loss, get_balance_sheet, get_tenant_id, extract_accounts_by_name_patterns, extract_gross_profit_metrics, extract_total_trading_income, extract_business_groups_revenue, extract_business_groups_gross_profit, extract_business_groups_net_profit, extract_consolidated_net_profit, extract_cash_in_bank_from_balance_sheet, resolve_tracking_ids_by_name, TENANT_ID as XERO_TENANT_ID ) return { 'refresh_access_token': refresh_access_token, 'get_profit_and_loss': get_profit_and_loss, 'get_tenant_id': get_tenant_id, 'extract_accounts_by_name_patterns': extract_accounts_by_name_patterns, 'extract_gross_profit_metrics': extract_gross_profit_metrics, 'extract_total_trading_income': extract_total_trading_income, 'extract_business_groups_revenue': extract_business_groups_revenue, 'extract_business_groups_gross_profit': extract_business_groups_gross_profit, 'extract_business_groups_net_profit': extract_business_groups_net_profit, 'extract_consolidated_net_profit': extract_consolidated_net_profit, 'extract_cash_in_bank_from_balance_sheet': extract_cash_in_bank_from_balance_sheet, 'get_balance_sheet': get_balance_sheet, 'resolve_tracking_ids_by_name': resolve_tracking_ids_by_name, 'TENANT_ID': XERO_TENANT_ID } except Exception as e: logger.error(f"Failed to import Xero modules: {e}", exc_info=True) raise ImportError(f"Xero API modules not available: {e}") from e class XeroService: """Service for fetching revenue data from Xero API for multiple branches.""" def __init__(self, django_client_id: Optional[int] = None): """ Args: django_client_id: When set, Xero OAuth may be loaded from integrations.Client (xero_oauth_* fields). When unset, legacy .env / token_manager behaviour is used. """ self._django_client_id = django_client_id self._access_token: Optional[str] = None self._token_refreshed_at: Optional[float] = None # Timestamp when token was last refreshed self._tenant_ids: Dict[str, Optional[str]] = {} # Cache tenant IDs per branch self._tracking_ids_cache: Dict[str, tuple] = {} # cache_key -> (cat_id, dept_to_option_id) self._xero_modules: Optional[Dict] = None # Xero access tokens expire after 30 minutes - refresh proactively after 25 minutes self._token_refresh_interval = 25 * 60 # 25 minutes in seconds def _cache_prefix(self) -> str: return f"c{self._django_client_id}" if self._django_client_id is not None else "c0" def _ck(self, tail: str) -> str: """Tenant-scoped cache key so in-memory xero_cache cannot leak between clients.""" return f"{self._cache_prefix()}:{tail}" def _db_xero_oauth_credentials(self) -> Optional[Dict[str, str]]: if not self._django_client_id: return None try: from integrations.models import Client except Exception: return None row = Client.objects.filter(pk=self._django_client_id, is_active=True).first() if not row: return None cid = (getattr(row, "xero_oauth_client_id", None) or "").strip() sec = (getattr(row, "xero_oauth_client_secret", None) or "").strip() ref = (getattr(row, "xero_oauth_refresh_token", None) or "").strip() if cid and sec and ref: return {"client_id": cid, "client_secret": sec, "refresh_token": ref} return None def _persist_db_refresh_token(self, new_refresh: Optional[str]) -> None: if not new_refresh or not self._django_client_id: return try: from integrations.models import Client Client.objects.filter(pk=self._django_client_id).update(xero_oauth_refresh_token=new_refresh) except Exception as exc: logger.warning("Could not persist rotated Xero refresh token to DB: %s", exc) def _get_xero_modules(self): """Get Xero API modules (lazy loaded).""" if self._xero_modules is None: self._xero_modules = _get_xero_modules() return self._xero_modules def _get_access_token(self, force_refresh: bool = False) -> str: """ Get a valid access token, refreshing if necessary. When this XeroService is scoped to a Django Client with xero_oauth_* fields set, refresh uses DB-stored credentials and persists rotated refresh tokens to the DB. Otherwise uses legacy token_manager.refresh_access_token() + .env. """ current_time = time.time() should_refresh = ( force_refresh or not self._access_token or not self._token_refreshed_at or (current_time - self._token_refreshed_at) > self._token_refresh_interval ) if not should_refresh: return self._access_token logger.info("Refreshing Xero access token...") db_creds = self._db_xero_oauth_credentials() if db_creds: try: from integrations.xero_oauth import refresh_xero_access_token access, new_refresh = refresh_xero_access_token( db_creds["client_id"], db_creds["client_secret"], db_creds["refresh_token"], ) self._access_token = access self._token_refreshed_at = current_time self._persist_db_refresh_token(new_refresh) logger.info("✅ Xero access token refreshed successfully (DB OAuth)") return self._access_token except ValueError as e: logger.error("❌ Xero authentication failed (DB OAuth): %s", e) self._clear_access_token() raise except Exception as e: logger.error("Failed to refresh Xero access token (DB OAuth): %s", e, exc_info=True) self._clear_access_token() raise try: modules = self._get_xero_modules() self._access_token = modules["refresh_access_token"]() self._token_refreshed_at = current_time logger.info("✅ Xero access token refreshed successfully") except ValueError as e: error_msg = str(e) logger.error( "❌ Xero authentication failed: %s\n" "The refresh token has expired. Revenue data will be unavailable until re-authentication.", error_msg, ) self._clear_access_token() raise except Exception as e: logger.error("Failed to refresh Xero access token: %s", e, exc_info=True) self._clear_access_token() raise return self._access_token def _clear_access_token(self): """Clear the cached access token to force refresh on next request.""" self._access_token = None self._token_refreshed_at = None def _get_tenant_id(self, branch_id: str = "branch2") -> str: """ Get the Xero tenant ID for a specific branch. Args: branch_id: Branch identifier (branch1, branch2, branch3) Returns: Tenant ID string """ # Check cache first if branch_id in self._tenant_ids and self._tenant_ids[branch_id]: return self._tenant_ids[branch_id] modules = self._get_xero_modules() access_token = self._get_access_token() # Get tenant ID from config or environment xero_config = get_xero_config(branch_id) if xero_config: tenant_id = xero_config.get("tenant_id") or "" tenant_id_env = xero_config.get("tenant_id_env", "XERO_TENANT_ID") if not tenant_id: tenant_id = os.getenv(tenant_id_env) # Fallback: if branch-specific tenant not set, use XERO_TENANT_ID (common for single-org setups) if (not tenant_id or (isinstance(tenant_id, str) and tenant_id.lower() in ["your_tenant_id", "placeholder", "xxx", ""])) and tenant_id_env != "XERO_TENANT_ID": if not tenant_id or (isinstance(tenant_id, str) and not tenant_id.strip()): logger.warning( f"Xero {branch_id}: {tenant_id_env} is not set. Set it in .env for correct org. Falling back to XERO_TENANT_ID." ) tenant_id = os.getenv("XERO_TENANT_ID") else: tenant_id = os.getenv("XERO_TENANT_ID") or modules.get('TENANT_ID') # If not found or invalid, fetch from Xero connections API if not tenant_id or (isinstance(tenant_id, str) and tenant_id.lower() in ["your_tenant_id", "placeholder", "xxx", ""]): logger.info(f"Fetching Xero tenant ID for {branch_id}...") tenant_id = modules['get_tenant_id'](access_token) # Cache it self._tenant_ids[branch_id] = tenant_id return tenant_id def get_revenue_for_month(self, year: int, month: int, branch_id: str = "branch2") -> Dict[int, float]: """ Get revenue data for a specific month from Xero for a specific branch. Args: year: Year (e.g., 2026) month: Month (1-12) branch_id: Branch identifier (branch1, branch2, branch3) Returns: Dictionary mapping company_id to revenue amount {company_id: revenue} """ try: # Get branch-specific configuration xero_config = get_xero_config(branch_id) if not xero_config: raise ValueError(f"No Xero configuration found for {branch_id}") if not xero_config.get("enabled", False): raise ValueError(f"Xero is not enabled for {branch_id}") # Calculate date range for the month from_date = date(year, month, 1) # Calculate last day of the month if month == 12: to_date = date(year + 1, 1, 1) # First day of next month else: to_date = date(year, month + 1, 1) # First day of next month # Subtract one day to get last day of current month to_date = to_date - timedelta(days=1) from_date_str = from_date.strftime("%Y-%m-%d") to_date_str = to_date.strftime("%Y-%m-%d") # Check cache first cache_key = self._ck(f"revenue:{branch_id}:{year}:{month}") cached_data = xero_cache.get(cache_key) if cached_data is not None: logger.info(f"✅ Using cached revenue for {branch_id} - {month}/{year}") return cached_data logger.info(f"Fetching Xero revenue for {branch_id} - {month}/{year} ({from_date_str} to {to_date_str})") # Get access token and tenant ID modules = self._get_xero_modules() access_token = self._get_access_token() tenant_id = self._get_tenant_id(branch_id) revenue_by_company = {} tracking_config = get_department_tracking_config(branch_id) if not tracking_config: raise ValueError(f"No department_tracking config for {branch_id} - Business Groups required") dept_to_option = tracking_config["department_to_option"] department_to_company_id = xero_config.get("department_to_company_id", {}) tracking_depts = set() cat_id, dept_to_option_names = None, {} # Flatten for resolve: use first option when dept maps to list resolve_map = {d: (o[0] if isinstance(o, list) else o) for d, o in dept_to_option.items()} tracking_cache_key = self._ck(f"{branch_id}:{tenant_id}") if tracking_cache_key not in self._tracking_ids_cache: try: cat_id, dept_to_option_ids = modules['resolve_tracking_ids_by_name']( access_token, tenant_id, [tracking_config["tracking_category_name"]], resolve_map, ) self._tracking_ids_cache[tracking_cache_key] = (cat_id, dept_to_option_ids, dept_to_option) except Exception as e: logger.warning(f"Could not resolve tracking IDs for {branch_id}: {e}") self._tracking_ids_cache[tracking_cache_key] = (None, {}, dept_to_option) cat_id, dept_to_option_ids, dept_to_option_names = self._tracking_ids_cache[tracking_cache_key] tracking_depts = set(dept_to_option_ids.keys()) if not cat_id: raise ValueError(f"Could not resolve Business Groups tracking for {branch_id}") logger.info(f"Fetching all Business Groups in one call ({len(dept_to_option)} depts)") try: pl_data = modules['get_profit_and_loss']( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id ) except Exception as e: if "401" in str(e) or "Unauthorized" in str(e) or "TokenExpired" in str(e): self._clear_access_token() access_token = self._get_access_token(force_refresh=True) pl_data = modules['get_profit_and_loss']( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id ) else: raise dept_revenues = modules['extract_business_groups_revenue'](pl_data, dept_to_option_names) for dept_name, revenue in dept_revenues.items(): company_id = department_to_company_id.get(dept_name) if company_id: revenue_by_company[company_id] = revenue logger.info(f" {dept_name} (company_id={company_id}): ${revenue:,.2f} (Business Groups)") # Only cache when we got some revenue - avoid caching all-zeros (e.g. wrong tenant or no matching P&L columns) total = sum(revenue_by_company.values()) if total > 0: xero_cache.set(cache_key, revenue_by_company) else: logger.warning( f"⚠️ Xero returned $0 revenue for {branch_id} ({month}/{year}) - not caching. " "Check tenant ID and Business Groups option names in Xero." ) return revenue_by_company except Exception as e: logger.error(f"Error fetching Xero revenue for {branch_id}: {e}", exc_info=True) raise def get_revenue_for_date_range_by_company( self, from_date: date, to_date: date, branch_id: str = "branch2" ) -> Dict[int, float]: """ Get revenue per company for a date range (same structure as get_revenue_for_month). Used when dashboard uses custom date range instead of full month. """ try: xero_config = get_xero_config(branch_id) if not xero_config or not xero_config.get("enabled", False): return {} tracking_config = get_department_tracking_config(branch_id) if not tracking_config: return {} from_date_str = from_date.strftime("%Y-%m-%d") to_date_str = to_date.strftime("%Y-%m-%d") cache_key = self._ck(f"revenue_range_by_co:{branch_id}:{from_date_str}:{to_date_str}") cached = xero_cache.get(cache_key) if cached is not None: return cached modules = self._get_xero_modules() access_token = self._get_access_token() tenant_id = self._get_tenant_id(branch_id) dept_to_option = tracking_config["department_to_option"] department_to_company_id = xero_config.get("department_to_company_id", {}) resolve_map = {d: (o[0] if isinstance(o, list) else o) for d, o in dept_to_option.items()} tracking_cache_key = self._ck(f"{branch_id}:{tenant_id}") if tracking_cache_key not in self._tracking_ids_cache: try: cat_id, dept_to_option_ids = modules["resolve_tracking_ids_by_name"]( access_token, tenant_id, [tracking_config["tracking_category_name"]], resolve_map, ) self._tracking_ids_cache[tracking_cache_key] = (cat_id, dept_to_option_ids, dept_to_option) except Exception as e: logger.warning(f"Could not resolve tracking IDs for {branch_id}: {e}") self._tracking_ids_cache[tracking_cache_key] = (None, {}, dept_to_option) cat_id, _, dept_to_option_names = self._tracking_ids_cache[tracking_cache_key] if not cat_id: return {} pl_data = modules["get_profit_and_loss"]( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id, ) dept_revenues = modules["extract_business_groups_revenue"](pl_data, dept_to_option_names) revenue_by_company = {} for dept_name, revenue in dept_revenues.items(): company_id = department_to_company_id.get(dept_name) if company_id is not None: revenue_by_company[company_id] = revenue xero_cache.set(cache_key, revenue_by_company) return revenue_by_company except Exception as e: logger.error(f"Error fetching revenue for date range by company {branch_id}: {e}", exc_info=True) return {} def get_revenue_for_date_range( self, from_date: date, to_date: date, branch_id: str = "branch2" ) -> float: """ Get total branch revenue (sum of all departments) for a date range. Used for directors: this week, MTD, FYTD. """ try: xero_config = get_xero_config(branch_id) if not xero_config or not xero_config.get("enabled", False): return 0.0 tracking_config = get_department_tracking_config(branch_id) if not tracking_config: return 0.0 from_date_str = from_date.strftime("%Y-%m-%d") to_date_str = to_date.strftime("%Y-%m-%d") cache_key = self._ck(f"revenue_range:{branch_id}:{from_date_str}:{to_date_str}") cached = xero_cache.get(cache_key) if cached is not None: return cached modules = self._get_xero_modules() access_token = self._get_access_token() tenant_id = self._get_tenant_id(branch_id) dept_to_option = tracking_config["department_to_option"] resolve_map = {d: (o[0] if isinstance(o, list) else o) for d, o in dept_to_option.items()} tracking_cache_key = self._ck(f"{branch_id}:{tenant_id}") if tracking_cache_key not in self._tracking_ids_cache: try: cat_id, dept_to_option_ids = modules["resolve_tracking_ids_by_name"]( access_token, tenant_id, [tracking_config["tracking_category_name"]], resolve_map, ) self._tracking_ids_cache[tracking_cache_key] = (cat_id, dept_to_option_ids, dept_to_option) except Exception as e: logger.warning(f"Could not resolve tracking IDs for {branch_id}: {e}") self._tracking_ids_cache[tracking_cache_key] = (None, {}, dept_to_option) cat_id, _, dept_to_option_names = self._tracking_ids_cache[tracking_cache_key] if not cat_id: return 0.0 pl_data = modules["get_profit_and_loss"]( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id, ) dept_revenues = modules["extract_business_groups_revenue"](pl_data, dept_to_option_names) total = sum(dept_revenues.values()) if dept_revenues else 0.0 xero_cache.set(cache_key, total) return total except Exception as e: logger.error(f"Error fetching revenue for date range {branch_id}: {e}", exc_info=True) return 0.0 def get_cash_in_bank(self, branch_id: str) -> Optional[float]: """ Get cash in bank for a branch: sum of Business Trans Acct, Tax & Reserve Account, Operating Reserve. Live balance from Balance Sheet (no date filter = as at today). """ try: xero_config = get_xero_config(branch_id) if not xero_config or not xero_config.get("enabled", False): return None account_names = get_cash_in_bank_account_names(branch_id) date_str = date.today().strftime("%Y-%m-%d") cache_key = self._ck(f"cash_in_bank:{branch_id}:{date_str}") cached = xero_cache.get(cache_key) if cached is not None: return cached modules = self._get_xero_modules() access_token = self._get_access_token() tenant_id = self._get_tenant_id(branch_id) bs_data = modules["get_balance_sheet"]( access_token=access_token, tenant_id=tenant_id, date_str=date_str, ) total = modules["extract_cash_in_bank_from_balance_sheet"](bs_data, account_names) if total == 0.0 and bs_data and bs_data.get("Reports"): logger.warning( f"Cash in bank for {branch_id} returned $0 (Balance Sheet has data). " f"Check that account names in Xero match: {account_names}" ) xero_cache.set(cache_key, total) return total except Exception as e: logger.error(f"Error fetching cash in bank for {branch_id}: {e}", exc_info=True) return None def get_cash_in_bank_by_tenant_id( self, tenant_id: str, account_names: Optional[List[str]] = None, ) -> Optional[float]: """ Get cash in bank for a tenant by ID (used for Nixon Management, Nixon Assets). Uses standard account names if not specified. """ if not tenant_id or (isinstance(tenant_id, str) and tenant_id.lower() in ["your_tenant_id", "placeholder", "xxx", ""]): return None try: names = account_names or CASH_IN_BANK_ACCOUNT_NAMES date_str = date.today().strftime("%Y-%m-%d") cache_key = self._ck(f"cash_in_bank:tenant:{tenant_id[:8]}:{date_str}") cached = xero_cache.get(cache_key) if cached is not None: return cached modules = self._get_xero_modules() access_token = self._get_access_token() bs_data = modules["get_balance_sheet"]( access_token=access_token, tenant_id=tenant_id, date_str=date_str, ) total = modules["extract_cash_in_bank_from_balance_sheet"](bs_data, names) xero_cache.set(cache_key, total) return total except Exception as e: logger.error(f"Error fetching cash in bank for tenant {tenant_id[:8]}...: {e}", exc_info=True) return None def get_financials_for_tenant_id( self, from_date: date, to_date: date, tenant_id: str ) -> Dict[str, float]: """ Get consolidated revenue, gross profit, and net profit for a tenant (e.g. Nixon Management, Nixon Assets). Uses P&L without tracking. Returns {revenue, gross_profit, gp_pct, net_profit, np_pct}. """ if not tenant_id or (isinstance(tenant_id, str) and tenant_id.lower() in ["your_tenant_id", "placeholder", "xxx", ""]): return {"revenue": 0.0, "gross_profit": 0.0, "gp_pct": 0.0, "net_profit": 0.0, "np_pct": 0.0} try: from_date_str = from_date.strftime("%Y-%m-%d") to_date_str = to_date.strftime("%Y-%m-%d") cache_key = self._ck(f"financials_tenant:{tenant_id[:8]}:{from_date_str}:{to_date_str}") cached = xero_cache.get(cache_key) if cached is not None: return cached modules = self._get_xero_modules() access_token = self._get_access_token() pl_data = modules["get_profit_and_loss"]( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, ) gp_metrics = modules["extract_gross_profit_metrics"](pl_data) revenue = gp_metrics.get("total_revenue", 0.0) or 0.0 gross_profit = gp_metrics.get("gross_profit", 0.0) or 0.0 net_profit = modules["extract_consolidated_net_profit"](pl_data) gp_pct = (gross_profit / revenue * 100) if revenue > 0 else 0.0 np_pct = (net_profit / revenue * 100) if revenue > 0 else 0.0 result = { "revenue": revenue, "gross_profit": gross_profit, "gp_pct": round(gp_pct, 1), "net_profit": net_profit, "np_pct": round(np_pct, 1), } xero_cache.set(cache_key, result) return result except Exception as e: logger.warning(f"Error fetching financials for tenant {tenant_id[:8]}...: {e}", exc_info=True) return {"revenue": 0.0, "gross_profit": 0.0, "gp_pct": 0.0, "net_profit": 0.0, "np_pct": 0.0} def get_gross_profit_margin(self, year: int, month: int, branch_id: str = "branch2") -> Optional[float]: """ Get gross profit margin percentage for a specific month from Xero for a specific branch. Args: year: Year (e.g., 2026) month: Month (1-12) branch_id: Branch identifier (branch1, branch2, branch3) Returns: Gross profit margin percentage (0-100), or None if not available """ try: # Prefer Business Group Gross Profit + Total Trading Income metrics_by_company = self.get_business_group_gp_metrics(year, month, branch_id) total_gp = 0.0 total_revenue = 0.0 for metrics in metrics_by_company.values(): if not metrics: continue total_gp += metrics.get("gross_profit", 0.0) or 0.0 total_revenue += metrics.get("total_revenue", 0.0) or 0.0 if total_revenue > 0: gp_margin = (total_gp / total_revenue) * 100 return gp_margin return None except Exception as e: logger.error(f"Error fetching Xero GP margin for {branch_id}: {e}", exc_info=True) return None def get_business_group_gp_metrics( self, year: int, month: int, branch_id: str = "branch2" ) -> Dict[int, Dict[str, Optional[float]]]: """ Get GP metrics per Business Group (department) from a single P&L call. Uses "Gross Profit" and "Total Trading Income" line items per tracking option. Returns: {company_id: {"gross_profit": float, "total_revenue": float, "gross_profit_margin": float|None}} """ try: xero_config = get_xero_config(branch_id) if not xero_config: raise ValueError(f"No Xero configuration found for {branch_id}") if not xero_config.get("enabled", False): raise ValueError(f"Xero is not enabled for {branch_id}") tracking_config = get_department_tracking_config(branch_id) if not tracking_config: raise ValueError(f"No department_tracking config for {branch_id} - Business Groups required") # Calculate date range for the month from_date = date(year, month, 1) if month == 12: to_date = date(year + 1, 1, 1) else: to_date = date(year, month + 1, 1) to_date = to_date - timedelta(days=1) from_date_str = from_date.strftime("%Y-%m-%d") to_date_str = to_date.strftime("%Y-%m-%d") cache_key = self._ck(f"gp_metrics_bg:{branch_id}:{year}:{month}") cached_data = xero_cache.get(cache_key) if cached_data is not None: return cached_data modules = self._get_xero_modules() access_token = self._get_access_token() tenant_id = self._get_tenant_id(branch_id) dept_to_option = tracking_config["department_to_option"] department_to_company_id = xero_config.get("department_to_company_id", {}) # Flatten for resolve: use first option when dept maps to list resolve_map = {d: (o[0] if isinstance(o, list) else o) for d, o in dept_to_option.items()} tracking_cache_key = self._ck(f"{branch_id}:{tenant_id}") if tracking_cache_key not in self._tracking_ids_cache: try: cat_id, dept_to_option_ids = modules['resolve_tracking_ids_by_name']( access_token, tenant_id, [tracking_config["tracking_category_name"]], resolve_map, ) self._tracking_ids_cache[tracking_cache_key] = (cat_id, dept_to_option_ids, dept_to_option) except Exception as e: logger.warning(f"Could not resolve tracking IDs for {branch_id} GP metrics: {e}") self._tracking_ids_cache[tracking_cache_key] = (None, {}, dept_to_option) cat_id, dept_to_option_ids, dept_to_option_names = self._tracking_ids_cache[tracking_cache_key] if not cat_id: raise ValueError(f"Could not resolve Business Groups tracking for {branch_id} GP metrics") logger.info(f"Fetching Business Group GP metrics in one call ({len(dept_to_option)} depts)") try: pl_data = modules['get_profit_and_loss']( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id ) except Exception as e: if "401" in str(e) or "Unauthorized" in str(e) or "TokenExpired" in str(e): self._clear_access_token() access_token = self._get_access_token(force_refresh=True) pl_data = modules['get_profit_and_loss']( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id ) else: raise revenue_by_dept = modules['extract_business_groups_revenue'](pl_data, dept_to_option_names) gp_by_dept = modules['extract_business_groups_gross_profit'](pl_data, dept_to_option_names) metrics_by_company: Dict[int, Dict[str, Optional[float]]] = {} for dept_name, revenue in revenue_by_dept.items(): company_id = department_to_company_id.get(dept_name) if not company_id: continue gross_profit = gp_by_dept.get(dept_name, 0.0) # Keep negative GP — margin can be negative (e.g. -7.5%) and must display as such margin = (gross_profit / revenue) * 100 if revenue != 0 else None metrics_by_company[company_id] = { "gross_profit": gross_profit, "total_revenue": revenue, "gross_profit_margin": margin } xero_cache.set(cache_key, metrics_by_company) return metrics_by_company except Exception as e: logger.error(f"Error fetching Business Group GP metrics for {branch_id}: {e}", exc_info=True) return {} def get_business_group_gp_metrics_for_date_range( self, from_date: date, to_date: date, branch_id: str = "branch2" ) -> Dict[int, Dict[str, Optional[float]]]: """ Get GP metrics per Business Group for a date range (same structure as get_business_group_gp_metrics). """ try: xero_config = get_xero_config(branch_id) if not xero_config or not xero_config.get("enabled", False): return {} tracking_config = get_department_tracking_config(branch_id) if not tracking_config: return {} from_date_str = from_date.strftime("%Y-%m-%d") to_date_str = to_date.strftime("%Y-%m-%d") cache_key = self._ck(f"gp_metrics_bg_range:{branch_id}:{from_date_str}:{to_date_str}") cached_data = xero_cache.get(cache_key) if cached_data is not None: return cached_data modules = self._get_xero_modules() access_token = self._get_access_token() tenant_id = self._get_tenant_id(branch_id) dept_to_option = tracking_config["department_to_option"] department_to_company_id = xero_config.get("department_to_company_id", {}) resolve_map = {d: (o[0] if isinstance(o, list) else o) for d, o in dept_to_option.items()} tracking_cache_key = self._ck(f"{branch_id}:{tenant_id}") if tracking_cache_key not in self._tracking_ids_cache: try: cat_id, dept_to_option_ids = modules['resolve_tracking_ids_by_name']( access_token, tenant_id, [tracking_config["tracking_category_name"]], resolve_map, ) self._tracking_ids_cache[tracking_cache_key] = (cat_id, dept_to_option_ids, dept_to_option) except Exception as e: logger.warning(f"Could not resolve tracking IDs for {branch_id}: {e}") self._tracking_ids_cache[tracking_cache_key] = (None, {}, dept_to_option) cat_id, _, dept_to_option_names = self._tracking_ids_cache[tracking_cache_key] if not cat_id: return {} try: pl_data = modules['get_profit_and_loss']( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id ) except Exception as e: if "401" in str(e) or "Unauthorized" in str(e) or "TokenExpired" in str(e): self._clear_access_token() access_token = self._get_access_token(force_refresh=True) pl_data = modules['get_profit_and_loss']( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id ) else: raise revenue_by_dept = modules['extract_business_groups_revenue'](pl_data, dept_to_option_names) gp_by_dept = modules['extract_business_groups_gross_profit'](pl_data, dept_to_option_names) metrics_by_company: Dict[int, Dict[str, Optional[float]]] = {} for dept_name, revenue in revenue_by_dept.items(): company_id = department_to_company_id.get(dept_name) if not company_id: continue gross_profit = gp_by_dept.get(dept_name, 0.0) margin = (gross_profit / revenue) * 100 if revenue != 0 else None metrics_by_company[company_id] = { "gross_profit": gross_profit, "total_revenue": revenue, "gross_profit_margin": margin } xero_cache.set(cache_key, metrics_by_company) return metrics_by_company except Exception as e: logger.error(f"Error fetching Business Group GP metrics for date range {branch_id}: {e}", exc_info=True) return {} def get_branch_financials_for_date_range( self, from_date: date, to_date: date, branch_id: str = "branch2" ) -> Dict[str, float]: """ Get revenue, gross profit, and net profit for a branch in a single P&L call. Returns {revenue, gross_profit, gp_pct, net_profit, np_pct}. """ try: xero_config = get_xero_config(branch_id) if not xero_config or not xero_config.get("enabled", False): return {"revenue": 0.0, "gross_profit": 0.0, "gp_pct": 0.0, "net_profit": 0.0, "np_pct": 0.0} tracking_config = get_department_tracking_config(branch_id) if not tracking_config: return {"revenue": 0.0, "gross_profit": 0.0, "gp_pct": 0.0, "net_profit": 0.0, "np_pct": 0.0} from_date_str = from_date.strftime("%Y-%m-%d") to_date_str = to_date.strftime("%Y-%m-%d") cache_key = self._ck(f"financials_range:{branch_id}:{from_date_str}:{to_date_str}") cached = xero_cache.get(cache_key) if cached is not None: return cached modules = self._get_xero_modules() access_token = self._get_access_token() tenant_id = self._get_tenant_id(branch_id) dept_to_option = tracking_config["department_to_option"] resolve_map = {d: (o[0] if isinstance(o, list) else o) for d, o in dept_to_option.items()} tracking_cache_key = self._ck(f"{branch_id}:{tenant_id}") if tracking_cache_key not in self._tracking_ids_cache: try: cat_id, dept_to_option_ids = modules["resolve_tracking_ids_by_name"]( access_token, tenant_id, [tracking_config["tracking_category_name"]], resolve_map, ) self._tracking_ids_cache[tracking_cache_key] = (cat_id, dept_to_option_ids, dept_to_option) except Exception as e: logger.warning(f"Could not resolve tracking IDs for {branch_id}: {e}") self._tracking_ids_cache[tracking_cache_key] = (None, {}, dept_to_option) cat_id, _, dept_to_option_names = self._tracking_ids_cache[tracking_cache_key] if not cat_id: return {"revenue": 0.0, "gross_profit": 0.0, "gp_pct": 0.0, "net_profit": 0.0, "np_pct": 0.0} try: pl_data = modules["get_profit_and_loss"]( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id, ) except Exception as e: if "401" in str(e) or "Unauthorized" in str(e) or "TokenExpired" in str(e): self._clear_access_token() access_token = self._get_access_token(force_refresh=True) pl_data = modules["get_profit_and_loss"]( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id, ) else: raise dept_revenues = modules["extract_business_groups_revenue"](pl_data, dept_to_option_names) dept_gp = modules["extract_business_groups_gross_profit"](pl_data, dept_to_option_names) revenue = sum(dept_revenues.values()) if dept_revenues else 0.0 gross_profit = sum(dept_gp.values()) if dept_gp else 0.0 # Fetch consolidated P&L (no tracking) for true Net Profit. # Per-business-group NP misses unallocated operating expenses. try: try: consolidated_pl = modules["get_profit_and_loss"]( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, ) except Exception as e: if "401" in str(e) or "Unauthorized" in str(e) or "TokenExpired" in str(e): self._clear_access_token() access_token = self._get_access_token(force_refresh=True) consolidated_pl = modules["get_profit_and_loss"]( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, ) else: raise net_profit = modules["extract_consolidated_net_profit"](consolidated_pl) except Exception as e: logger.warning(f"Could not fetch consolidated NP for {branch_id}, falling back to per-group: {e}") dept_np = modules["extract_business_groups_net_profit"](pl_data, dept_to_option_names) net_profit = sum(dept_np.values()) if dept_np else 0.0 gp_pct = (gross_profit / revenue * 100) if revenue > 0 else 0.0 np_pct = (net_profit / revenue * 100) if revenue > 0 else 0.0 result = { "revenue": revenue, "gross_profit": gross_profit, "gp_pct": round(gp_pct, 1), "net_profit": net_profit, "np_pct": round(np_pct, 1), } xero_cache.set(cache_key, result) return result except Exception as e: logger.error(f"Error fetching branch financials for {branch_id}: {e}", exc_info=True) return {"revenue": 0.0, "gross_profit": 0.0, "gp_pct": 0.0, "net_profit": 0.0, "np_pct": 0.0} def get_revenue_ytd(self, year: int, month: int, branch_id: str = "branch2") -> Dict[int, float]: """ Get year-to-date revenue data from Xero for a specific branch. Gets revenue from Jan 1 to the end of the specified month. Args: year: Year (e.g., 2026) month: Month (1-12) branch_id: Branch identifier (branch1, branch2, branch3) Returns: Dictionary mapping company_id to revenue amount {company_id: revenue} """ try: # Get branch-specific configuration xero_config = get_xero_config(branch_id) if not xero_config: raise ValueError(f"No Xero configuration found for {branch_id}") if not xero_config.get("enabled", False): raise ValueError(f"Xero is not enabled for {branch_id}") tracking_config = get_department_tracking_config(branch_id) if not tracking_config: raise ValueError(f"No department_tracking config for {branch_id} - Business Groups required") dept_to_option = tracking_config["department_to_option"] department_to_company_id = xero_config.get("department_to_company_id", {}) # Calculate date range: Jan 1 to end of specified month from_date = date(year, 1, 1) # Calculate last day of the specified month if month == 12: to_date = date(year + 1, 1, 1) # First day of next month else: to_date = date(year, month + 1, 1) # First day of next month # Subtract one day to get last day of current month to_date = to_date - timedelta(days=1) from_date_str = from_date.strftime("%Y-%m-%d") to_date_str = to_date.strftime("%Y-%m-%d") # Check cache first cache_key = self._ck(f"revenue_ytd:{branch_id}:{year}:{month}") cached_data = xero_cache.get(cache_key) if cached_data is not None: logger.info(f"✅ Using cached YTD revenue for {branch_id} (Jan 1, {year} to {to_date_str})") return cached_data logger.info(f"Fetching Xero YTD revenue for {branch_id} (Jan 1, {year} to {to_date_str})") modules = self._get_xero_modules() access_token = self._get_access_token() tenant_id = self._get_tenant_id(branch_id) revenue_by_company = {} resolve_map = {d: (o[0] if isinstance(o, list) else o) for d, o in dept_to_option.items()} tracking_cache_key = self._ck(f"{branch_id}:{tenant_id}") if tracking_cache_key not in self._tracking_ids_cache: try: resolved_cat, resolved_opts = modules['resolve_tracking_ids_by_name']( access_token, tenant_id, [tracking_config["tracking_category_name"]], resolve_map, ) self._tracking_ids_cache[tracking_cache_key] = (resolved_cat, resolved_opts, dept_to_option) except Exception as e: logger.warning(f"Could not resolve tracking IDs for {branch_id} YTD: {e}") self._tracking_ids_cache[tracking_cache_key] = (None, {}, dept_to_option) cat_id, dept_to_option_ids, dept_to_option_names = self._tracking_ids_cache[tracking_cache_key] tracking_depts = set(dept_to_option_ids.keys()) if not cat_id: raise ValueError(f"Could not resolve Business Groups tracking for {branch_id} YTD") logger.info(f"Fetching all Business Groups in one call YTD ({len(dept_to_option)} depts)") try: pl_data = modules['get_profit_and_loss']( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id ) except Exception as e: if "401" in str(e) or "Unauthorized" in str(e) or "TokenExpired" in str(e): self._clear_access_token() access_token = self._get_access_token(force_refresh=True) pl_data = modules['get_profit_and_loss']( access_token=access_token, tenant_id=tenant_id, from_date=from_date_str, to_date=to_date_str, tracking_category_id=cat_id ) else: raise dept_revenues = modules['extract_business_groups_revenue'](pl_data, dept_to_option_names) for dept_name, revenue in dept_revenues.items(): company_id = department_to_company_id.get(dept_name) if company_id: revenue_by_company[company_id] = revenue logger.info(f" {dept_name} (company_id={company_id}) YTD: ${revenue:,.2f} (Business Groups)") xero_cache.set(cache_key, revenue_by_company) return revenue_by_company except Exception as e: logger.error(f"Error fetching Xero YTD revenue for {branch_id}: {e}", exc_info=True) raise # Create a singleton instance xero_service = XeroService()