Spaces:
Running
Running
| """ | |
| Shared currency detection and formatting utilities | |
| Enterprise-grade multi-currency support for $50K product | |
| Detects: USD, EUR, GBP, INR, JPY, AUD, CAD, CHF, CNY, SGD | |
| """ | |
| import pandas as pd | |
| from pathlib import Path | |
| import re | |
| import json | |
| from typing import Optional, Dict, Any, List | |
| # Comprehensive currency configuration | |
| CURRENCY_CONFIG = { | |
| 'USD': {'symbol': '$', 'name': 'US Dollar', 'locale': 'en-US', 'keywords': ['usd', 'dollar', 'us$', 'united states']}, | |
| 'EUR': {'symbol': '€', 'name': 'Euro', 'locale': 'de-DE', 'keywords': ['eur', 'euro', '€']}, | |
| 'GBP': {'symbol': '£', 'name': 'British Pound', 'locale': 'en-GB', 'keywords': ['gbp', 'pound', 'sterling', '£', 'british']}, | |
| 'INR': {'symbol': '₹', 'name': 'Indian Rupee', 'locale': 'en-IN', 'keywords': ['inr', 'rupee', 'rupees', '₹', 'indian', 'india']}, | |
| 'JPY': {'symbol': '¥', 'name': 'Japanese Yen', 'locale': 'ja-JP', 'keywords': ['jpy', 'yen', '¥', 'japan']}, | |
| 'AUD': {'symbol': 'A$', 'name': 'Australian Dollar', 'locale': 'en-AU', 'keywords': ['aud', 'australian', 'australia']}, | |
| 'CAD': {'symbol': 'C$', 'name': 'Canadian Dollar', 'locale': 'en-CA', 'keywords': ['cad', 'canadian', 'canada']}, | |
| 'CHF': {'symbol': 'CHF', 'name': 'Swiss Franc', 'locale': 'de-CH', 'keywords': ['chf', 'franc', 'swiss']}, | |
| 'CNY': {'symbol': '¥', 'name': 'Chinese Yuan', 'locale': 'zh-CN', 'keywords': ['cny', 'yuan', 'rmb', 'chinese', 'china']}, | |
| 'SGD': {'symbol': 'S$', 'name': 'Singapore Dollar', 'locale': 'en-SG', 'keywords': ['sgd', 'singapore']}, | |
| } | |
| # Currency symbol to code mapping | |
| SYMBOL_TO_CODE = { | |
| '$': 'USD', | |
| '€': 'EUR', | |
| '£': 'GBP', | |
| '₹': 'INR', | |
| '¥': 'JPY', # Could be JPY or CNY - needs context | |
| 'A$': 'AUD', | |
| 'C$': 'CAD', | |
| 'S$': 'SGD', | |
| } | |
| def detect_currency_from_value(value: str) -> Optional[str]: | |
| """Detect currency from a single value string.""" | |
| if not isinstance(value, str): | |
| value = str(value) | |
| # Check for currency symbols at start | |
| value = value.strip() | |
| # Check specific multi-char symbols first | |
| if value.startswith('A$') or 'AUD' in value.upper(): | |
| return 'AUD' | |
| if value.startswith('C$') or 'CAD' in value.upper(): | |
| return 'CAD' | |
| if value.startswith('S$') or 'SGD' in value.upper(): | |
| return 'SGD' | |
| if value.startswith('CHF') or 'CHF' in value.upper(): | |
| return 'CHF' | |
| # Check single char symbols | |
| if '€' in value: | |
| return 'EUR' | |
| if '£' in value: | |
| return 'GBP' | |
| if '₹' in value: | |
| return 'INR' | |
| if '$' in value and not any(x in value for x in ['A$', 'C$', 'S$']): | |
| return 'USD' | |
| if '¥' in value: | |
| # Distinguish between JPY and CNY based on value magnitude | |
| # JPY typically has no decimal, CNY does | |
| return 'JPY' if '.' not in value else 'CNY' | |
| return None | |
| def detect_currency_from_column_names(columns: List[str]) -> Optional[str]: | |
| """Detect currency from column names.""" | |
| col_text = ' '.join(columns).lower() | |
| for code, config in CURRENCY_CONFIG.items(): | |
| if any(kw in col_text for kw in config['keywords']): | |
| return code | |
| return None | |
| def detect_currency_from_filename(filename: str) -> Optional[str]: | |
| """Detect currency from filename.""" | |
| filename_lower = filename.lower() | |
| for code, config in CURRENCY_CONFIG.items(): | |
| if any(kw in filename_lower for kw in config['keywords']): | |
| return code | |
| return None | |
| def detect_currency( | |
| df: pd.DataFrame = None, | |
| files_path: Path = None, | |
| filename: str = None, | |
| metadata_path: Path = None, | |
| skip_metadata: bool = False | |
| ) -> str: | |
| """ | |
| Enterprise-grade currency detection using multiple strategies. | |
| Priority order (FIXED - data values take precedence): | |
| 1. Stored metadata (from previous detection) - only if skip_metadata=False | |
| 2. Currency column in DataFrame (explicit currency field) | |
| 3. Currency SYMBOLS in amount values (₹, $, €, £, ¥) - MOST RELIABLE | |
| 4. Column name hints (amount_usd, price_eur, etc.) | |
| 5. Filename hints (only as fallback) | |
| 6. Files in directory (only as fallback) | |
| When skip_metadata=True (during file upload/rebuild): | |
| - Skip stored metadata, always detect fresh from data | |
| Args: | |
| df: DataFrame with amount data | |
| files_path: Path to user's files directory | |
| filename: Specific filename being processed | |
| metadata_path: Path to stored currency metadata | |
| skip_metadata: If True, skip reading stored metadata (for fresh detection) | |
| Returns: | |
| Currency code: 'USD', 'EUR', 'GBP', 'INR', 'JPY', etc. | |
| """ | |
| detected_currencies = [] | |
| # STRATEGY 1: Check stored metadata (skip during upload/rebuild) | |
| if not skip_metadata and metadata_path and metadata_path.exists(): | |
| try: | |
| with open(metadata_path, 'r') as f: | |
| meta = json.load(f) | |
| if 'currency' in meta: | |
| print(f"💰 Using stored currency from metadata: {meta['currency']}") | |
| return meta['currency'] | |
| except: | |
| pass | |
| # STRATEGY 2: Check for explicit currency column | |
| if df is not None and not df.empty: | |
| currency_cols = [c for c in df.columns if 'currency' in c.lower()] | |
| if currency_cols: | |
| currency_val = df[currency_cols[0]].iloc[0] | |
| if isinstance(currency_val, str) and currency_val.upper() in CURRENCY_CONFIG: | |
| print(f"💰 Detected from currency column: {currency_val.upper()}") | |
| return currency_val.upper() | |
| # STRATEGY 3: Detect from amount values - THIS IS THE HIGHEST PRIORITY FOR ACTUAL DATA | |
| # Currency symbols in data are the most reliable indicator | |
| if df is not None and not df.empty: | |
| amount_cols = [c for c in df.columns if any( | |
| term in c.lower() for term in ['amount', 'total', 'price', 'revenue', 'sales', 'value', 'cost'] | |
| )] | |
| for col in amount_cols: | |
| # Sample first 100 non-null values | |
| sample = df[col].dropna().head(100).astype(str) | |
| for val in sample: | |
| detected = detect_currency_from_value(val) | |
| if detected: | |
| detected_currencies.append(detected) | |
| print(f"💰 Detected {detected} from value '{val[:20]}' in column '{col}'") | |
| # If we found a currency symbol in the data, this is highly reliable | |
| # Return immediately - data symbols are the most trustworthy | |
| return detected | |
| # STRATEGY 4: Check column names for currency hints | |
| if df is not None and not df.empty: | |
| col_currency = detect_currency_from_column_names(list(df.columns)) | |
| if col_currency: | |
| print(f"💰 Detected from column names: {col_currency}") | |
| detected_currencies.append(col_currency) | |
| # STRATEGY 5: Check specific filename (only as fallback) | |
| if filename and not detected_currencies: | |
| file_currency = detect_currency_from_filename(filename) | |
| if file_currency: | |
| print(f"💰 Detected from filename '{filename}': {file_currency}") | |
| detected_currencies.append(file_currency) | |
| # STRATEGY 6: Check all files in directory (only as fallback) | |
| if files_path and files_path.exists() and not detected_currencies: | |
| try: | |
| for f in files_path.glob('*'): | |
| if f.is_file(): | |
| file_currency = detect_currency_from_filename(f.name) | |
| if file_currency: | |
| detected_currencies.append(file_currency) | |
| except: | |
| pass | |
| # Return most common detected currency, or USD as fallback (most common global currency) | |
| if detected_currencies: | |
| from collections import Counter | |
| most_common = Counter(detected_currencies).most_common(1)[0][0] | |
| print(f"💰 Final detected currency: {most_common}") | |
| return most_common | |
| print(f"💰 No currency detected, defaulting to USD") | |
| return 'USD' # Default fallback - USD is most common globally | |
| def detect_currency_from_files(files_path: Path) -> Dict[str, str]: | |
| """ | |
| Detect currency for each file in a directory. | |
| Returns a mapping of filename -> currency code. | |
| """ | |
| result = {} | |
| if not files_path or not files_path.exists(): | |
| return result | |
| for file in files_path.glob('*'): | |
| if file.is_file() and file.suffix.lower() in ['.csv', '.xlsx', '.xls']: | |
| try: | |
| if file.suffix.lower() == '.csv': | |
| df = pd.read_csv(file, nrows=100) | |
| else: | |
| df = pd.read_excel(file, nrows=100) | |
| currency = detect_currency(df=df, filename=file.name) | |
| result[file.name] = currency | |
| except: | |
| result[file.name] = detect_currency_from_filename(file.name) or 'USD' | |
| return result | |
| def format_currency(amount: float, currency: str = 'INR', include_code: bool = False) -> str: | |
| """ | |
| Format amount with proper currency symbol and locale formatting. | |
| Args: | |
| amount: Numeric amount | |
| currency: Currency code | |
| include_code: Whether to include currency code after symbol | |
| Returns: | |
| Formatted string with currency symbol | |
| """ | |
| config = CURRENCY_CONFIG.get(currency, CURRENCY_CONFIG['INR']) | |
| symbol = config['symbol'] | |
| # Format number with thousands separator | |
| if currency == 'INR': | |
| # Indian numbering system (lakhs, crores) | |
| formatted = format_indian_number(amount) | |
| elif currency == 'JPY': | |
| # Yen typically has no decimal places | |
| formatted = f"{amount:,.0f}" | |
| else: | |
| formatted = f"{amount:,.2f}" | |
| result = f"{symbol}{formatted}" | |
| if include_code: | |
| result = f"{result} {currency}" | |
| return result | |
| def format_indian_number(num: float) -> str: | |
| """Format number in Indian numbering system (lakhs, crores).""" | |
| if num < 1000: | |
| return f"{num:.2f}" | |
| # Convert to string with 2 decimal places | |
| num_str = f"{num:.2f}" | |
| parts = num_str.split('.') | |
| integer_part = parts[0] | |
| decimal_part = parts[1] if len(parts) > 1 else "00" | |
| # Apply Indian formatting | |
| result = "" | |
| length = len(integer_part) | |
| if length <= 3: | |
| result = integer_part | |
| else: | |
| # Last 3 digits | |
| result = integer_part[-3:] | |
| remaining = integer_part[:-3] | |
| # Group remaining in pairs | |
| while remaining: | |
| if len(remaining) <= 2: | |
| result = remaining + "," + result | |
| break | |
| else: | |
| result = remaining[-2:] + "," + result | |
| remaining = remaining[:-2] | |
| return f"{result}.{decimal_part}" | |
| def get_currency_symbol(currency: str = 'USD') -> str: | |
| """Get currency symbol for code.""" | |
| config = CURRENCY_CONFIG.get(currency, CURRENCY_CONFIG['USD']) | |
| return config['symbol'] | |
| def get_currency_info(currency: str = 'USD') -> Dict[str, Any]: | |
| """Get full currency information.""" | |
| return CURRENCY_CONFIG.get(currency, CURRENCY_CONFIG['USD']) | |
| def parse_currency_value(value: str, detected_currency: str = None) -> tuple: | |
| """ | |
| Parse a currency string into numeric value and currency code. | |
| Args: | |
| value: String like "$1,234.56" or "€1.234,56" or "₹1,23,456" | |
| detected_currency: Already detected currency code | |
| Returns: | |
| Tuple of (numeric_value, currency_code) | |
| """ | |
| if not isinstance(value, str): | |
| try: | |
| return (float(value), detected_currency or 'USD') | |
| except: | |
| return (0.0, detected_currency or 'USD') | |
| # Detect currency from value | |
| currency = detect_currency_from_value(value) or detected_currency or 'USD' | |
| # Remove currency symbols and whitespace | |
| clean = re.sub(r'[₹$€£¥A$C$S$CHF\s]', '', value) | |
| # Handle European format (1.234,56) vs US format (1,234.56) | |
| if ',' in clean and '.' in clean: | |
| # If comma comes after dot, it's European format | |
| if clean.rfind(',') > clean.rfind('.'): | |
| clean = clean.replace('.', '').replace(',', '.') | |
| else: | |
| clean = clean.replace(',', '') | |
| elif ',' in clean and '.' not in clean: | |
| # Could be European decimal or US thousands | |
| # If single comma with 2 digits after, treat as decimal | |
| parts = clean.split(',') | |
| if len(parts) == 2 and len(parts[1]) == 2: | |
| clean = clean.replace(',', '.') | |
| else: | |
| clean = clean.replace(',', '') | |
| try: | |
| return (float(clean), currency) | |
| except: | |
| return (0.0, currency) | |
| def detect_and_save_user_currency(user_id: str, files_path: Path, storage_base: Path) -> str: | |
| """ | |
| Consolidated function to detect currency from all user files and save to metadata. | |
| ALWAYS detects fresh from data - ignores stored metadata. | |
| Args: | |
| user_id: User identifier | |
| files_path: Path to user's files directory | |
| storage_base: Base storage path for metadata | |
| Returns: | |
| Detected currency code (defaults to INR) | |
| """ | |
| from collections import Counter | |
| if not files_path or not files_path.exists(): | |
| return 'INR' | |
| detected_currencies = [] | |
| for file in files_path.glob('*'): | |
| if file.is_file(): | |
| ext = file.suffix.lower() | |
| if ext not in ['.csv', '.xlsx', '.xls']: | |
| continue | |
| try: | |
| df = None | |
| if ext == '.csv': | |
| df = pd.read_csv(file, nrows=100) | |
| elif ext in ['.xlsx', '.xls']: | |
| df = pd.read_excel(file, nrows=100) | |
| if df is not None and not df.empty: | |
| # skip_metadata=True to always detect fresh from actual data | |
| currency = detect_currency(df=df, filename=file.name, skip_metadata=True) | |
| detected_currencies.append(currency) | |
| print(f"💰 Detected {currency} from {file.name}") | |
| except Exception as e: | |
| print(f"⚠️ Error detecting currency from {file.name}: {e}") | |
| if detected_currencies: | |
| most_common = Counter(detected_currencies).most_common(1)[0][0] | |
| save_currency_metadata(user_id, most_common, storage_base) | |
| print(f"✅ Saved currency {most_common} for user {user_id}") | |
| return most_common | |
| print(f"⚠️ No currency detected from files, defaulting to USD") | |
| return 'USD' # Default to USD for global compatibility | |
| def save_currency_metadata(user_id: str, currency: str, storage_base: Path): | |
| """Save detected currency to user metadata.""" | |
| meta_path = storage_base / user_id / "metadata.json" | |
| meta_path.parent.mkdir(parents=True, exist_ok=True) | |
| metadata = {} | |
| if meta_path.exists(): | |
| try: | |
| with open(meta_path, 'r') as f: | |
| metadata = json.load(f) | |
| except: | |
| pass | |
| metadata['currency'] = currency | |
| metadata['updated_at'] = pd.Timestamp.now().isoformat() | |
| with open(meta_path, 'w') as f: | |
| json.dump(metadata, f) | |
| def load_currency_metadata(user_id: str, storage_base: Path) -> Optional[str]: | |
| """Load stored currency from user metadata.""" | |
| meta_path = storage_base / user_id / "metadata.json" | |
| if meta_path.exists(): | |
| try: | |
| with open(meta_path, 'r') as f: | |
| metadata = json.load(f) | |
| return metadata.get('currency') | |
| except: | |
| pass | |
| return None | |
| # ============================================================================ | |
| # MULTI-CURRENCY SUPPORT - Exchange Rates and Breakdown | |
| # ============================================================================ | |
| # Exchange rates to USD (approximate rates - should be updated for production) | |
| # Last updated: December 2024 | |
| EXCHANGE_RATES_TO_USD = { | |
| 'USD': 1.0, | |
| 'EUR': 1.05, # 1 EUR = 1.05 USD | |
| 'GBP': 1.27, # 1 GBP = 1.27 USD | |
| 'INR': 0.01136, # 1 INR = 0.01136 USD (₹88 = $1 - Dec 2024) | |
| 'JPY': 0.0067, # 1 JPY = 0.0067 USD (1 USD = ~149 JPY) | |
| 'AUD': 0.65, # 1 AUD = 0.65 USD | |
| 'CAD': 0.74, # 1 CAD = 0.74 USD | |
| 'CHF': 1.13, # 1 CHF = 1.13 USD | |
| 'CNY': 0.14, # 1 CNY = 0.14 USD | |
| 'SGD': 0.74, # 1 SGD = 0.74 USD | |
| } | |
| def convert_to_usd(amount: float, from_currency: str) -> float: | |
| """ | |
| Convert amount from any supported currency to USD. | |
| Args: | |
| amount: Amount in source currency | |
| from_currency: Source currency code (e.g., 'INR', 'EUR') | |
| Returns: | |
| Amount in USD | |
| """ | |
| rate = EXCHANGE_RATES_TO_USD.get(from_currency, 1.0) | |
| return round(amount * rate, 2) | |
| def convert_from_usd(amount: float, to_currency: str) -> float: | |
| """ | |
| Convert amount from USD to any supported currency. | |
| Args: | |
| amount: Amount in USD | |
| to_currency: Target currency code | |
| Returns: | |
| Amount in target currency | |
| """ | |
| rate = EXCHANGE_RATES_TO_USD.get(to_currency, 1.0) | |
| if rate == 0: | |
| return 0.0 | |
| return round(amount / rate, 2) | |
| def get_exchange_rate(from_currency: str, to_currency: str = 'USD') -> float: | |
| """Get exchange rate between two currencies.""" | |
| from_rate = EXCHANGE_RATES_TO_USD.get(from_currency, 1.0) | |
| to_rate = EXCHANGE_RATES_TO_USD.get(to_currency, 1.0) | |
| if to_rate == 0: | |
| return 0.0 | |
| return from_rate / to_rate | |
| def convert_currency(amount: float, from_currency: str, to_currency: str) -> float: | |
| """ | |
| Convert amount between any two supported currencies. | |
| Args: | |
| amount: Amount to convert | |
| from_currency: Source currency code | |
| to_currency: Target currency code | |
| Returns: | |
| Converted amount | |
| """ | |
| if from_currency == to_currency: | |
| return amount | |
| # Convert to USD first (base currency) | |
| usd_amount = convert_to_usd(amount, from_currency) | |
| # Then convert from USD to target | |
| return convert_from_usd(usd_amount, to_currency) | |
| def calculate_currency_breakdown(amounts_by_currency: Dict[str, float]) -> Dict[str, Any]: | |
| """ | |
| Calculate multi-currency breakdown with USD equivalent. | |
| Args: | |
| amounts_by_currency: Dict mapping currency codes to total amounts | |
| e.g., {'INR': 1000000, 'USD': 5000} | |
| Returns: | |
| Dict with breakdown and USD equivalent: | |
| { | |
| 'breakdown': [ | |
| {'currency': 'INR', 'amount': 1000000, 'symbol': '₹', 'formatted': '₹10,00,000', 'usd_equivalent': 12000}, | |
| {'currency': 'USD', 'amount': 5000, 'symbol': '$', 'formatted': '$5,000', 'usd_equivalent': 5000} | |
| ], | |
| 'total_usd_equivalent': 17000, | |
| 'primary_currency': 'INR', # Currency with highest original amount | |
| 'currencies_count': 2 | |
| } | |
| """ | |
| breakdown = [] | |
| total_usd = 0.0 | |
| primary_currency = 'USD' | |
| max_amount = 0 | |
| for currency, amount in amounts_by_currency.items(): | |
| if amount <= 0: | |
| continue | |
| symbol = get_currency_symbol(currency) | |
| formatted = format_currency(amount, currency) | |
| usd_equivalent = convert_to_usd(amount, currency) | |
| breakdown.append({ | |
| 'currency': currency, | |
| 'amount': round(amount, 2), | |
| 'symbol': symbol, | |
| 'formatted': formatted, | |
| 'usd_equivalent': round(usd_equivalent, 2), | |
| 'name': CURRENCY_CONFIG.get(currency, {}).get('name', currency) | |
| }) | |
| total_usd += usd_equivalent | |
| # Track primary currency (highest amount in original currency) | |
| if amount > max_amount: | |
| max_amount = amount | |
| primary_currency = currency | |
| # Sort breakdown by USD equivalent (highest first) | |
| breakdown.sort(key=lambda x: x['usd_equivalent'], reverse=True) | |
| return { | |
| 'breakdown': breakdown, | |
| 'total_usd_equivalent': round(total_usd, 2), | |
| 'total_usd_formatted': format_currency(total_usd, 'USD'), | |
| 'primary_currency': primary_currency, | |
| 'currencies_count': len(breakdown) | |
| } | |
| def detect_currency_from_amount_string(value: str) -> tuple: | |
| """ | |
| Extract currency and numeric amount from a string like '$100.50' or '₹1,234'. | |
| Args: | |
| value: String containing currency symbol and amount | |
| Returns: | |
| Tuple of (numeric_amount, currency_code) | |
| """ | |
| if not isinstance(value, str): | |
| try: | |
| return (float(value), 'USD') | |
| except: | |
| return (0.0, 'USD') | |
| value = value.strip() | |
| # Detect currency | |
| currency = detect_currency_from_value(value) | |
| if not currency: | |
| currency = 'USD' | |
| # Extract numeric value | |
| clean = re.sub(r'[₹$€£¥A$C$S$CHF\s,]', '', value) | |
| clean = clean.replace(' ', '') | |
| try: | |
| amount = float(clean) | |
| except: | |
| amount = 0.0 | |
| return (amount, currency) | |