claw/insightflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1d55ae8f1e7da779fe96c1b83aa6bf34f436fd49
insightflow/backend/localization_manager.py
OpenClaw Bot 2aded2de48 Phase 8: 完成 AI 能力增强、运营与增长工具、开发者生态、运维与监控 
- Task 4: AI 能力增强 (ai_manager.py)
  - 自定义模型训练（领域特定实体识别）
  - 多模态大模型集成（GPT-4V、Claude 3、Gemini、Kimi-VL）
  - 知识图谱 RAG 智能问答
  - 智能摘要（提取式/生成式/关键点/时间线）
  - 预测性分析（趋势/异常/增长/演变预测）

- Task 5: 运营与增长工具 (growth_manager.py)
  - 用户行为分析（Mixpanel/Amplitude 集成）
  - A/B 测试框架
  - 邮件营销自动化
  - 推荐系统（邀请返利、团队升级激励）

- Task 6: 开发者生态 (developer_ecosystem_manager.py)
  - SDK 发布管理（Python/JavaScript/Go）
  - 模板市场
  - 插件市场
  - 开发者文档与示例代码

- Task 8: 运维与监控 (ops_manager.py)
  - 实时告警系统（PagerDuty/Opsgenie 集成）
  - 容量规划与自动扩缩容
  - 灾备与故障转移
  - 成本优化

Phase 8 全部 8 个任务已完成！
2026-02-27 00:01:40 +08:00

1286 lines
59 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
InsightFlow Phase 8 - 全球化与本地化管理模块
功能：
1. 多语言支持（i18n，支持10+语言）
2. 区域数据中心配置（北美、欧洲、亚太）
3. 本地化支付方式管理
4. 时区与日历本地化
作者: InsightFlow Team
"""
import sqlite3
import json
import uuid
import re
from datetime import datetime, timedelta
from typing import Optional, List, Dict, Any, Tuple
from dataclasses import dataclass, asdict
from enum import Enum
import logging
try:
import pytz
PYTZ_AVAILABLE = True
except ImportError:
PYTZ_AVAILABLE = False
try:
from babel import Locale, dates, numbers
BABEL_AVAILABLE = True
except ImportError:
BABEL_AVAILABLE = False
logger = logging.getLogger(__name__)
class LanguageCode(str, Enum):
"""支持的语言代码"""
EN = "en"
ZH_CN = "zh_CN"
ZH_TW = "zh_TW"
JA = "ja"
KO = "ko"
DE = "de"
FR = "fr"
ES = "es"
PT = "pt"
RU = "ru"
AR = "ar"
HI = "hi"
class RegionCode(str, Enum):
"""区域代码"""
GLOBAL = "global"
NORTH_AMERICA = "na"
EUROPE = "eu"
ASIA_PACIFIC = "apac"
CHINA = "cn"
LATIN_AMERICA = "latam"
MIDDLE_EAST = "me"
class DataCenterRegion(str, Enum):
"""数据中心区域"""
US_EAST = "us-east"
US_WEST = "us-west"
EU_WEST = "eu-west"
EU_CENTRAL = "eu-central"
AP_SOUTHEAST = "ap-southeast"
AP_NORTHEAST = "ap-northeast"
AP_SOUTH = "ap-south"
CN_NORTH = "cn-north"
CN_EAST = "cn-east"
class PaymentProvider(str, Enum):
"""支付提供商"""
STRIPE = "stripe"
ALIPAY = "alipay"
WECHAT_PAY = "wechat_pay"
PAYPAL = "paypal"
APPLE_PAY = "apple_pay"
GOOGLE_PAY = "google_pay"
KLARNA = "klarna"
IDEAL = "ideal"
BANCONTACT = "bancontact"
GIROPAY = "giropay"
SEPA = "sepa"
UNIONPAY = "unionpay"
class CalendarType(str, Enum):
"""日历类型"""
GREGORIAN = "gregorian"
CHINESE_LUNAR = "chinese_lunar"
ISLAMIC = "islamic"
HEBREW = "hebrew"
INDIAN = "indian"
PERSIAN = "persian"
BUDDHIST = "buddhist"
@dataclass
class Translation:
id: str
key: str
language: str
value: str
namespace: str
context: Optional[str]
created_at: datetime
updated_at: datetime
is_reviewed: bool
reviewed_by: Optional[str]
reviewed_at: Optional[datetime]
@dataclass
class LanguageConfig:
code: str
name: str
name_local: str
is_rtl: bool
is_active: bool
is_default: bool
fallback_language: Optional[str]
date_format: str
time_format: str
datetime_format: str
number_format: str
currency_format: str
first_day_of_week: int
calendar_type: str
@dataclass
class DataCenter:
id: str
region_code: str
name: str
location: str
endpoint: str
status: str
priority: int
supported_regions: List[str]
capabilities: Dict[str, Any]
created_at: datetime
updated_at: datetime
@dataclass
class TenantDataCenterMapping:
id: str
tenant_id: str
primary_dc_id: str
secondary_dc_id: Optional[str]
region_code: str
data_residency: str
created_at: datetime
updated_at: datetime
@dataclass
class LocalizedPaymentMethod:
id: str
provider: str
name: str
name_local: Dict[str, str]
supported_countries: List[str]
supported_currencies: List[str]
is_active: bool
config: Dict[str, Any]
icon_url: Optional[str]
display_order: int
min_amount: Optional[float]
max_amount: Optional[float]
created_at: datetime
updated_at: datetime
@dataclass
class CountryConfig:
code: str
code3: str
name: str
name_local: Dict[str, str]
region: str
default_language: str
supported_languages: List[str]
default_currency: str
supported_currencies: List[str]
timezone: str
calendar_type: str
date_format: Optional[str]
time_format: Optional[str]
number_format: Optional[str]
address_format: Optional[str]
phone_format: Optional[str]
vat_rate: Optional[float]
is_active: bool
@dataclass
class TimezoneConfig:
id: str
timezone: str
utc_offset: str
dst_offset: Optional[str]
country_code: str
region: str
is_active: bool
@dataclass
class CurrencyConfig:
code: str
name: str
name_local: Dict[str, str]
symbol: str
decimal_places: int
decimal_separator: str
thousands_separator: str
is_active: bool
@dataclass
class LocalizationSettings:
id: str
tenant_id: str
default_language: str
supported_languages: List[str]
default_currency: str
supported_currencies: List[str]
default_timezone: str
default_date_format: Optional[str]
default_time_format: Optional[str]
default_number_format: Optional[str]
calendar_type: str
first_day_of_week: int
region_code: str
data_residency: str
created_at: datetime
updated_at: datetime
class LocalizationManager:
DEFAULT_LANGUAGES = {
LanguageCode.EN: {
"name": "English",
"name_local": "English",
"is_rtl": False,
"date_format": "MM/dd/yyyy",
"time_format": "h:mm a",
"datetime_format": "MM/dd/yyyy h:mm a",
"number_format": "#,##0.##",
"currency_format": "$#,##0.00",
"first_day_of_week": 0,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.ZH_CN: {
"name": "Chinese (Simplified)",
"name_local": "简体中文",
"is_rtl": False,
"date_format": "yyyy-MM-dd",
"time_format": "HH:mm",
"datetime_format": "yyyy-MM-dd HH:mm",
"number_format": "#,##0.##",
"currency_format": "¥#,##0.00",
"first_day_of_week": 1,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.ZH_TW: {
"name": "Chinese (Traditional)",
"name_local": "繁體中文",
"is_rtl": False,
"date_format": "yyyy/MM/dd",
"time_format": "HH:mm",
"datetime_format": "yyyy/MM/dd HH:mm",
"number_format": "#,##0.##",
"currency_format": "NT$#,##0.00",
"first_day_of_week": 0,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.JA: {
"name": "Japanese",
"name_local": "日本語",
"is_rtl": False,
"date_format": "yyyy/MM/dd",
"time_format": "HH:mm",
"datetime_format": "yyyy/MM/dd HH:mm",
"number_format": "#,##0.##",
"currency_format": "¥#,##0",
"first_day_of_week": 0,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.KO: {
"name": "Korean",
"name_local": "한국어",
"is_rtl": False,
"date_format": "yyyy. MM. dd",
"time_format": "HH:mm",
"datetime_format": "yyyy. MM. dd HH:mm",
"number_format": "#,##0.##",
"currency_format": "₩#,##0",
"first_day_of_week": 0,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.DE: {
"name": "German",
"name_local": "Deutsch",
"is_rtl": False,
"date_format": "dd.MM.yyyy",
"time_format": "HH:mm",
"datetime_format": "dd.MM.yyyy HH:mm",
"number_format": "#,##0.##",
"currency_format": "#,##0.00 €",
"first_day_of_week": 1,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.FR: {
"name": "French",
"name_local": "Français",
"is_rtl": False,
"date_format": "dd/MM/yyyy",
"time_format": "HH:mm",
"datetime_format": "dd/MM/yyyy HH:mm",
"number_format": "#,##0.##",
"currency_format": "#,##0.00 €",
"first_day_of_week": 1,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.ES: {
"name": "Spanish",
"name_local": "Español",
"is_rtl": False,
"date_format": "dd/MM/yyyy",
"time_format": "HH:mm",
"datetime_format": "dd/MM/yyyy HH:mm",
"number_format": "#,##0.##",
"currency_format": "#,##0.00 €",
"first_day_of_week": 1,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.PT: {
"name": "Portuguese",
"name_local": "Português",
"is_rtl": False,
"date_format": "dd/MM/yyyy",
"time_format": "HH:mm",
"datetime_format": "dd/MM/yyyy HH:mm",
"number_format": "#,##0.##",
"currency_format": "R$#,##0.00",
"first_day_of_week": 0,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.RU: {
"name": "Russian",
"name_local": "Русский",
"is_rtl": False,
"date_format": "dd.MM.yyyy",
"time_format": "HH:mm",
"datetime_format": "dd.MM.yyyy HH:mm",
"number_format": "#,##0.##",
"currency_format": "#,##0.00 ₽",
"first_day_of_week": 1,
"calendar_type": CalendarType.GREGORIAN.value
},
LanguageCode.AR: {
"name": "Arabic",
"name_local": "العربية",
"is_rtl": True,
"date_format": "dd/MM/yyyy",
"time_format": "hh:mm a",
"datetime_format": "dd/MM/yyyy hh:mm a",
"number_format": "#,##0.##",
"currency_format": "#,##0.00 ر.س",
"first_day_of_week": 6,
"calendar_type": CalendarType.ISLAMIC.value
},
LanguageCode.HI: {
"name": "Hindi",
"name_local": "हिन्दी",
"is_rtl": False,
"date_format": "dd/MM/yyyy",
"time_format": "hh:mm a",
"datetime_format": "dd/MM/yyyy hh:mm a",
"number_format": "#,##0.##",
"currency_format": "₹#,##0.00",
"first_day_of_week": 0,
"calendar_type": CalendarType.INDIAN.value
}
}
DEFAULT_DATA_CENTERS = {
DataCenterRegion.US_EAST: {
"name": "US East (Virginia)",
"location": "Virginia, USA",
"endpoint": "https://api-us-east.insightflow.io",
"priority": 1,
"supported_regions": [RegionCode.NORTH_AMERICA.value, RegionCode.GLOBAL.value],
"capabilities": {"storage": True, "compute": True, "ml": True}
},
DataCenterRegion.US_WEST: {
"name": "US West (California)",
"location": "California, USA",
"endpoint": "https://api-us-west.insightflow.io",
"priority": 2,
"supported_regions": [RegionCode.NORTH_AMERICA.value, RegionCode.GLOBAL.value],
"capabilities": {"storage": True, "compute": True, "ml": False}
},
DataCenterRegion.EU_WEST: {
"name": "EU West (Ireland)",
"location": "Dublin, Ireland",
"endpoint": "https://api-eu-west.insightflow.io",
"priority": 1,
"supported_regions": [RegionCode.EUROPE.value, RegionCode.GLOBAL.value],
"capabilities": {"storage": True, "compute": True, "ml": True}
},
DataCenterRegion.EU_CENTRAL: {
"name": "EU Central (Frankfurt)",
"location": "Frankfurt, Germany",
"endpoint": "https://api-eu-central.insightflow.io",
"priority": 2,
"supported_regions": [RegionCode.EUROPE.value, RegionCode.GLOBAL.value],
"capabilities": {"storage": True, "compute": True, "ml": False}
},
DataCenterRegion.AP_SOUTHEAST: {
"name": "Asia Pacific (Singapore)",
"location": "Singapore",
"endpoint": "https://api-ap-southeast.insightflow.io",
"priority": 1,
"supported_regions": [RegionCode.ASIA_PACIFIC.value, RegionCode.GLOBAL.value],
"capabilities": {"storage": True, "compute": True, "ml": True}
},
DataCenterRegion.AP_NORTHEAST: {
"name": "Asia Pacific (Tokyo)",
"location": "Tokyo, Japan",
"endpoint": "https://api-ap-northeast.insightflow.io",
"priority": 2,
"supported_regions": [RegionCode.ASIA_PACIFIC.value, RegionCode.GLOBAL.value],
"capabilities": {"storage": True, "compute": True, "ml": False}
},
DataCenterRegion.AP_SOUTH: {
"name": "Asia Pacific (Mumbai)",
"location": "Mumbai, India",
"endpoint": "https://api-ap-south.insightflow.io",
"priority": 3,
"supported_regions": [RegionCode.ASIA_PACIFIC.value, RegionCode.GLOBAL.value],
"capabilities": {"storage": True, "compute": True, "ml": False}
},
DataCenterRegion.CN_NORTH: {
"name": "China (Beijing)",
"location": "Beijing, China",
"endpoint": "https://api-cn-north.insightflow.cn",
"priority": 1,
"supported_regions": [RegionCode.CHINA.value],
"capabilities": {"storage": True, "compute": True, "ml": True}
},
DataCenterRegion.CN_EAST: {
"name": "China (Shanghai)",
"location": "Shanghai, China",
"endpoint": "https://api-cn-east.insightflow.cn",
"priority": 2,
"supported_regions": [RegionCode.CHINA.value],
"capabilities": {"storage": True, "compute": True, "ml": False}
}
}
DEFAULT_PAYMENT_METHODS = {
PaymentProvider.STRIPE: {
"name": "Credit Card",
"name_local": {
"en": "Credit Card",
"zh_CN": "信用卡",
"zh_TW": "信用卡",
"ja": "クレジットカード",
"ko": "신용카드",
"de": "Kreditkarte",
"fr": "Carte de crédit",
"es": "Tarjeta de crédito",
"pt": "Cartão de crédito",
"ru": "Кредитная карта"
},
"supported_countries": ["*"],
"supported_currencies": ["USD", "EUR", "GBP", "CAD", "AUD", "JPY"],
"display_order": 1
},
PaymentProvider.ALIPAY: {
"name": "Alipay",
"name_local": {"en": "Alipay", "zh_CN": "支付宝", "zh_TW": "支付寶"},
"supported_countries": ["CN", "HK", "MO", "TW", "SG", "MY", "TH"],
"supported_currencies": ["CNY", "HKD", "USD"],
"display_order": 2
},
PaymentProvider.WECHAT_PAY: {
"name": "WeChat Pay",
"name_local": {"en": "WeChat Pay", "zh_CN": "微信支付", "zh_TW": "微信支付"},
"supported_countries": ["CN", "HK", "MO"],
"supported_currencies": ["CNY", "HKD"],
"display_order": 3
},
PaymentProvider.PAYPAL: {
"name": "PayPal",
"name_local": {"en": "PayPal"},
"supported_countries": ["*"],
"supported_currencies": ["USD", "EUR", "GBP", "CAD", "AUD", "JPY"],
"display_order": 4
},
PaymentProvider.APPLE_PAY: {
"name": "Apple Pay",
"name_local": {"en": "Apple Pay"},
"supported_countries": ["US", "CA", "GB", "AU", "JP", "DE", "FR"],
"supported_currencies": ["USD", "EUR", "GBP", "CAD", "AUD", "JPY"],
"display_order": 5
},
PaymentProvider.GOOGLE_PAY: {
"name": "Google Pay",
"name_local": {"en": "Google Pay"},
"supported_countries": ["US", "CA", "GB", "AU", "JP", "DE", "FR"],
"supported_currencies": ["USD", "EUR", "GBP", "CAD", "AUD", "JPY"],
"display_order": 6
},
PaymentProvider.KLARNA: {
"name": "Klarna",
"name_local": {"en": "Klarna", "de": "Klarna", "fr": "Klarna"},
"supported_countries": ["DE", "AT", "NL", "BE", "FI", "SE", "NO", "DK", "GB"],
"supported_currencies": ["EUR", "GBP"],
"display_order": 7
},
PaymentProvider.IDEAL: {
"name": "iDEAL",
"name_local": {"en": "iDEAL", "de": "iDEAL"},
"supported_countries": ["NL"],
"supported_currencies": ["EUR"],
"display_order": 8
},
PaymentProvider.BANCONTACT: {
"name": "Bancontact",
"name_local": {"en": "Bancontact", "de": "Bancontact"},
"supported_countries": ["BE"],
"supported_currencies": ["EUR"],
"display_order": 9
},
PaymentProvider.GIROPAY: {
"name": "giropay",
"name_local": {"en": "giropay", "de": "giropay"},
"supported_countries": ["DE"],
"supported_currencies": ["EUR"],
"display_order": 10
},
PaymentProvider.SEPA: {
"name": "SEPA Direct Debit",
"name_local": {"en": "SEPA Direct Debit", "de": "SEPA-Lastschrift"},
"supported_countries": ["DE", "AT", "NL", "BE", "FR", "ES", "IT"],
"supported_currencies": ["EUR"],
"display_order": 11
},
PaymentProvider.UNIONPAY: {
"name": "UnionPay",
"name_local": {"en": "UnionPay", "zh_CN": "银联", "zh_TW": "銀聯"},
"supported_countries": ["CN", "HK", "MO", "TW"],
"supported_currencies": ["CNY", "USD"],
"display_order": 12
}
}
DEFAULT_COUNTRIES = {
"US": {"name": "United States", "name_local": {"en": "United States"}, "region": RegionCode.NORTH_AMERICA.value, "default_language": LanguageCode.EN.value, "supported_languages": [LanguageCode.EN.value], "default_currency": "USD", "supported_currencies": ["USD"], "timezone": "America/New_York", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": None},
"CN": {"name": "China", "name_local": {"zh_CN": "中国"}, "region": RegionCode.CHINA.value, "default_language": LanguageCode.ZH_CN.value, "supported_languages": [LanguageCode.ZH_CN.value, LanguageCode.EN.value], "default_currency": "CNY", "supported_currencies": ["CNY", "USD"], "timezone": "Asia/Shanghai", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.13},
"JP": {"name": "Japan", "name_local": {"ja": "日本"}, "region": RegionCode.ASIA_PACIFIC.value, "default_language": LanguageCode.JA.value, "supported_languages": [LanguageCode.JA.value, LanguageCode.EN.value], "default_currency": "JPY", "supported_currencies": ["JPY", "USD"], "timezone": "Asia/Tokyo", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.10},
"DE": {"name": "Germany", "name_local": {"de": "Deutschland"}, "region": RegionCode.EUROPE.value, "default_language": LanguageCode.DE.value, "supported_languages": [LanguageCode.DE.value, LanguageCode.EN.value], "default_currency": "EUR", "supported_currencies": ["EUR", "USD"], "timezone": "Europe/Berlin", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.19},
"GB": {"name": "United Kingdom", "name_local": {"en": "United Kingdom"}, "region": RegionCode.EUROPE.value, "default_language": LanguageCode.EN.value, "supported_languages": [LanguageCode.EN.value], "default_currency": "GBP", "supported_currencies": ["GBP", "EUR", "USD"], "timezone": "Europe/London", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.20},
"FR": {"name": "France", "name_local": {"fr": "France"}, "region": RegionCode.EUROPE.value, "default_language": LanguageCode.FR.value, "supported_languages": [LanguageCode.FR.value, LanguageCode.EN.value], "default_currency": "EUR", "supported_currencies": ["EUR", "USD"], "timezone": "Europe/Paris", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.20},
"SG": {"name": "Singapore", "name_local": {"en": "Singapore"}, "region": RegionCode.ASIA_PACIFIC.value, "default_language": LanguageCode.EN.value, "supported_languages": [LanguageCode.EN.value, LanguageCode.ZH_CN.value], "default_currency": "SGD", "supported_currencies": ["SGD", "USD"], "timezone": "Asia/Singapore", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.08},
"AU": {"name": "Australia", "name_local": {"en": "Australia"}, "region": RegionCode.ASIA_PACIFIC.value, "default_language": LanguageCode.EN.value, "supported_languages": [LanguageCode.EN.value], "default_currency": "AUD", "supported_currencies": ["AUD", "USD"], "timezone": "Australia/Sydney", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.10},
"CA": {"name": "Canada", "name_local": {"en": "Canada", "fr": "Canada"}, "region": RegionCode.NORTH_AMERICA.value, "default_language": LanguageCode.EN.value, "supported_languages": [LanguageCode.EN.value, LanguageCode.FR.value], "default_currency": "CAD", "supported_currencies": ["CAD", "USD"], "timezone": "America/Toronto", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.05},
"BR": {"name": "Brazil", "name_local": {"pt": "Brasil"}, "region": RegionCode.LATIN_AMERICA.value, "default_language": LanguageCode.PT.value, "supported_languages": [LanguageCode.PT.value, LanguageCode.EN.value], "default_currency": "BRL", "supported_currencies": ["BRL", "USD"], "timezone": "America/Sao_Paulo", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.17},
"IN": {"name": "India", "name_local": {"hi": "भारत"}, "region": RegionCode.ASIA_PACIFIC.value, "default_language": LanguageCode.EN.value, "supported_languages": [LanguageCode.EN.value, LanguageCode.HI.value], "default_currency": "INR", "supported_currencies": ["INR", "USD"], "timezone": "Asia/Kolkata", "calendar_type": CalendarType.GREGORIAN.value, "vat_rate": 0.18},
"AE": {"name": "United Arab Emirates", "name_local": {"ar": "الإمارات العربية المتحدة"}, "region": RegionCode.MIDDLE_EAST.value, "default_language": LanguageCode.EN.value, "supported_languages": [LanguageCode.EN.value, LanguageCode.AR.value], "default_currency": "AED", "supported_currencies": ["AED", "USD"], "timezone": "Asia/Dubai", "calendar_type": CalendarType.ISLAMIC.value, "vat_rate": 0.05}
}
def __init__(self, db_path: str = "insightflow.db"):
self.db_path = db_path
self._is_memory_db = db_path == ":memory:"
self._conn = None
self._init_db()
self._init_default_data()
def _get_connection(self) -> sqlite3.Connection:
if self._is_memory_db:
if self._conn is None:
self._conn = sqlite3.connect(self.db_path)
self._conn.row_factory = sqlite3.Row
return self._conn
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def _close_if_file_db(self, conn):
if not self._is_memory_db:
conn.close()
def _init_db(self):
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS translations (
id TEXT PRIMARY KEY, key TEXT NOT NULL, language TEXT NOT NULL, value TEXT NOT NULL,
namespace TEXT DEFAULT 'common', context TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, is_reviewed INTEGER DEFAULT 0,
reviewed_by TEXT, reviewed_at TIMESTAMP, UNIQUE(key, language, namespace)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS language_configs (
code TEXT PRIMARY KEY, name TEXT NOT NULL, name_local TEXT NOT NULL, is_rtl INTEGER DEFAULT 0,
is_active INTEGER DEFAULT 1, is_default INTEGER DEFAULT 0, fallback_language TEXT,
date_format TEXT, time_format TEXT, datetime_format TEXT, number_format TEXT,
currency_format TEXT, first_day_of_week INTEGER DEFAULT 1, calendar_type TEXT DEFAULT 'gregorian'
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS data_centers (
id TEXT PRIMARY KEY, region_code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, location TEXT NOT NULL,
endpoint TEXT NOT NULL, status TEXT DEFAULT 'active', priority INTEGER DEFAULT 1,
supported_regions TEXT DEFAULT '[]', capabilities TEXT DEFAULT '{}',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS tenant_data_center_mappings (
id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL UNIQUE, primary_dc_id TEXT NOT NULL,
secondary_dc_id TEXT, region_code TEXT NOT NULL, data_residency TEXT DEFAULT 'regional',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE,
FOREIGN KEY (primary_dc_id) REFERENCES data_centers(id),
FOREIGN KEY (secondary_dc_id) REFERENCES data_centers(id)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS localized_payment_methods (
id TEXT PRIMARY KEY, provider TEXT NOT NULL UNIQUE, name TEXT NOT NULL, name_local TEXT DEFAULT '{}',
supported_countries TEXT DEFAULT '[]', supported_currencies TEXT DEFAULT '[]',
is_active INTEGER DEFAULT 1, config TEXT DEFAULT '{}', icon_url TEXT, display_order INTEGER DEFAULT 0,
min_amount REAL, max_amount REAL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS country_configs (
code TEXT PRIMARY KEY, code3 TEXT NOT NULL, name TEXT NOT NULL, name_local TEXT DEFAULT '{}',
region TEXT NOT NULL, default_language TEXT NOT NULL, supported_languages TEXT DEFAULT '[]',
default_currency TEXT NOT NULL, supported_currencies TEXT DEFAULT '[]', timezone TEXT NOT NULL,
calendar_type TEXT DEFAULT 'gregorian', date_format TEXT, time_format TEXT, number_format TEXT,
address_format TEXT, phone_format TEXT, vat_rate REAL, is_active INTEGER DEFAULT 1
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS timezone_configs (
id TEXT PRIMARY KEY, timezone TEXT NOT NULL UNIQUE, utc_offset TEXT NOT NULL, dst_offset TEXT,
country_code TEXT NOT NULL, region TEXT NOT NULL, is_active INTEGER DEFAULT 1
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS currency_configs (
code TEXT PRIMARY KEY, name TEXT NOT NULL, name_local TEXT DEFAULT '{}', symbol TEXT NOT NULL,
decimal_places INTEGER DEFAULT 2, decimal_separator TEXT DEFAULT '.',
thousands_separator TEXT DEFAULT ',', is_active INTEGER DEFAULT 1
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS localization_settings (
id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL UNIQUE, default_language TEXT DEFAULT 'en',
supported_languages TEXT DEFAULT '["en"]', default_currency TEXT DEFAULT 'USD',
supported_currencies TEXT DEFAULT '["USD"]', default_timezone TEXT DEFAULT 'UTC',
default_date_format TEXT, default_time_format TEXT, default_number_format TEXT,
calendar_type TEXT DEFAULT 'gregorian', first_day_of_week INTEGER DEFAULT 1,
region_code TEXT DEFAULT 'global', data_residency TEXT DEFAULT 'regional',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE
)
""")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_translations_key ON translations(key)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_translations_lang ON translations(language)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_translations_ns ON translations(namespace)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_dc_region ON data_centers(region_code)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_dc_status ON data_centers(status)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_tenant_dc ON tenant_data_center_mappings(tenant_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_payment_provider ON localized_payment_methods(provider)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_payment_active ON localized_payment_methods(is_active)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_country_region ON country_configs(region)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_tz_country ON timezone_configs(country_code)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_locale_settings_tenant ON localization_settings(tenant_id)")
conn.commit()
logger.info("Localization tables initialized successfully")
except Exception as e:
logger.error(f"Error initializing localization tables: {e}")
raise
finally:
self._close_if_file_db(conn)
def _init_default_data(self):
conn = self._get_connection()
try:
cursor = conn.cursor()
for code, config in self.DEFAULT_LANGUAGES.items():
cursor.execute("""
INSERT OR IGNORE INTO language_configs
(code, name, name_local, is_rtl, is_active, is_default, fallback_language,
date_format, time_format, datetime_format, number_format, currency_format,
first_day_of_week, calendar_type)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (code.value, config["name"], config["name_local"], int(config["is_rtl"]), 1,
1 if code == LanguageCode.EN else 0, "en" if code != LanguageCode.EN else None,
config["date_format"], config["time_format"], config["datetime_format"],
config["number_format"], config["currency_format"],
config["first_day_of_week"], config["calendar_type"]))
for region_code, config in self.DEFAULT_DATA_CENTERS.items():
dc_id = str(uuid.uuid4())
cursor.execute("""
INSERT OR IGNORE INTO data_centers
(id, region_code, name, location, endpoint, priority, supported_regions, capabilities)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (dc_id, region_code.value, config["name"], config["location"], config["endpoint"],
config["priority"], json.dumps(config["supported_regions"]), json.dumps(config["capabilities"])))
for provider, config in self.DEFAULT_PAYMENT_METHODS.items():
pm_id = str(uuid.uuid4())
cursor.execute("""
INSERT OR IGNORE INTO localized_payment_methods
(id, provider, name, name_local, supported_countries, supported_currencies, is_active, display_order)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (pm_id, provider.value, config["name"], json.dumps(config["name_local"]),
json.dumps(config["supported_countries"]), json.dumps(config["supported_currencies"]),
1, config["display_order"]))
for code, config in self.DEFAULT_COUNTRIES.items():
cursor.execute("""
INSERT OR IGNORE INTO country_configs
(code, code3, name, name_local, region, default_language, supported_languages,
default_currency, supported_currencies, timezone, calendar_type, vat_rate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (code, code, config["name"], json.dumps(config["name_local"]), config["region"],
config["default_language"], json.dumps(config["supported_languages"]),
config["default_currency"], json.dumps(config["supported_currencies"]),
config["timezone"], config["calendar_type"], config["vat_rate"]))
conn.commit()
logger.info("Default localization data initialized")
except Exception as e:
logger.error(f"Error initializing default localization data: {e}")
finally:
self._close_if_file_db(conn)
def get_translation(self, key: str, language: str, namespace: str = "common", fallback: bool = True) -> Optional[str]:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT value FROM translations WHERE key = ? AND language = ? AND namespace = ?",
(key, language, namespace))
row = cursor.fetchone()
if row:
return row['value']
if fallback:
lang_config = self.get_language_config(language)
if lang_config and lang_config.fallback_language:
return self.get_translation(key, lang_config.fallback_language, namespace, False)
if language != "en":
return self.get_translation(key, "en", namespace, False)
return None
finally:
self._close_if_file_db(conn)
def set_translation(self, key: str, language: str, value: str, namespace: str = "common", context: Optional[str] = None) -> Translation:
conn = self._get_connection()
try:
translation_id = str(uuid.uuid4())
now = datetime.now()
cursor = conn.cursor()
cursor.execute("""
INSERT INTO translations (id, key, language, value, namespace, context, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(key, language, namespace) DO UPDATE SET
value = excluded.value, context = excluded.context, updated_at = excluded.updated_at, is_reviewed = 0
""", (translation_id, key, language, value, namespace, context, now, now))
conn.commit()
return self._get_translation_internal(conn, key, language, namespace)
finally:
self._close_if_file_db(conn)
def _get_translation_internal(self, conn: sqlite3.Connection, key: str, language: str, namespace: str) -> Optional[Translation]:
cursor = conn.cursor()
cursor.execute("SELECT * FROM translations WHERE key = ? AND language = ? AND namespace = ?",
(key, language, namespace))
row = cursor.fetchone()
if row:
return self._row_to_translation(row)
return None
def delete_translation(self, key: str, language: str, namespace: str = "common") -> bool:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("DELETE FROM translations WHERE key = ? AND language = ? AND namespace = ?",
(key, language, namespace))
conn.commit()
return cursor.rowcount > 0
finally:
self._close_if_file_db(conn)
def list_translations(self, language: Optional[str] = None, namespace: Optional[str] = None,
limit: int = 1000, offset: int = 0) -> List[Translation]:
conn = self._get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM translations WHERE 1=1"
params = []
if language:
query += " AND language = ?"
params.append(language)
if namespace:
query += " AND namespace = ?"
params.append(namespace)
query += " ORDER BY namespace, key LIMIT ? OFFSET ?"
params.extend([limit, offset])
cursor.execute(query, params)
rows = cursor.fetchall()
return [self._row_to_translation(row) for row in rows]
finally:
self._close_if_file_db(conn)
def get_language_config(self, code: str) -> Optional[LanguageConfig]:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM language_configs WHERE code = ?", (code,))
row = cursor.fetchone()
if row:
return self._row_to_language_config(row)
return None
finally:
self._close_if_file_db(conn)
def list_language_configs(self, active_only: bool = True) -> List[LanguageConfig]:
conn = self._get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM language_configs"
if active_only:
query += " WHERE is_active = 1"
query += " ORDER BY name"
cursor.execute(query)
rows = cursor.fetchall()
return [self._row_to_language_config(row) for row in rows]
finally:
self._close_if_file_db(conn)
def get_data_center(self, dc_id: str) -> Optional[DataCenter]:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM data_centers WHERE id = ?", (dc_id,))
row = cursor.fetchone()
if row:
return self._row_to_data_center(row)
return None
finally:
self._close_if_file_db(conn)
def get_data_center_by_region(self, region_code: str) -> Optional[DataCenter]:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM data_centers WHERE region_code = ?", (region_code,))
row = cursor.fetchone()
if row:
return self._row_to_data_center(row)
return None
finally:
self._close_if_file_db(conn)
def list_data_centers(self, status: Optional[str] = None, region: Optional[str] = None) -> List[DataCenter]:
conn = self._get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM data_centers WHERE 1=1"
params = []
if status:
query += " AND status = ?"
params.append(status)
if region:
query += " AND supported_regions LIKE ?"
params.append(f'%"{region}"%')
query += " ORDER BY priority"
cursor.execute(query, params)
rows = cursor.fetchall()
return [self._row_to_data_center(row) for row in rows]
finally:
self._close_if_file_db(conn)
def get_tenant_data_center(self, tenant_id: str) -> Optional[TenantDataCenterMapping]:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM tenant_data_center_mappings WHERE tenant_id = ?", (tenant_id,))
row = cursor.fetchone()
if row:
return self._row_to_tenant_dc_mapping(row)
return None
finally:
self._close_if_file_db(conn)
def set_tenant_data_center(self, tenant_id: str, region_code: str, data_residency: str = "regional") -> TenantDataCenterMapping:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM data_centers WHERE supported_regions LIKE ? AND status = 'active'
ORDER BY priority LIMIT 1
""", (f'%"{region_code}"%',))
row = cursor.fetchone()
if not row:
cursor.execute("""
SELECT * FROM data_centers WHERE supported_regions LIKE '%"global"%' AND status = 'active'
ORDER BY priority LIMIT 1
""")
row = cursor.fetchone()
if not row:
raise ValueError(f"No data center available for region: {region_code}")
primary_dc_id = row['id']
cursor.execute("""
SELECT * FROM data_centers WHERE id != ? AND status = 'active' ORDER BY priority LIMIT 1
""", (primary_dc_id,))
secondary_row = cursor.fetchone()
secondary_dc_id = secondary_row['id'] if secondary_row else None
mapping_id = str(uuid.uuid4())
now = datetime.now()
cursor.execute("""
INSERT INTO tenant_data_center_mappings
(id, tenant_id, primary_dc_id, secondary_dc_id, region_code, data_residency, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(tenant_id) DO UPDATE SET
primary_dc_id = excluded.primary_dc_id, secondary_dc_id = excluded.secondary_dc_id,
region_code = excluded.region_code, data_residency = excluded.data_residency, updated_at = excluded.updated_at
""", (mapping_id, tenant_id, primary_dc_id, secondary_dc_id, region_code, data_residency, now, now))
conn.commit()
return self.get_tenant_data_center(tenant_id)
finally:
self._close_if_file_db(conn)
def get_payment_method(self, provider: str) -> Optional[LocalizedPaymentMethod]:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM localized_payment_methods WHERE provider = ?", (provider,))
row = cursor.fetchone()
if row:
return self._row_to_payment_method(row)
return None
finally:
self._close_if_file_db(conn)
def list_payment_methods(self, country_code: Optional[str] = None, currency: Optional[str] = None,
active_only: bool = True) -> List[LocalizedPaymentMethod]:
conn = self._get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM localized_payment_methods WHERE 1=1"
params = []
if active_only:
query += " AND is_active = 1"
if country_code:
query += " AND (supported_countries LIKE ? OR supported_countries LIKE '%\"*\"%')"
params.append(f'%"{country_code}"%')
if currency:
query += " AND supported_currencies LIKE ?"
params.append(f'%"{currency}"%')
query += " ORDER BY display_order"
cursor.execute(query, params)
rows = cursor.fetchall()
return [self._row_to_payment_method(row) for row in rows]
finally:
self._close_if_file_db(conn)
def get_localized_payment_methods(self, country_code: str, language: str = "en") -> List[Dict[str, Any]]:
methods = self.list_payment_methods(country_code=country_code)
result = []
for method in methods:
name_local = method.name_local.get(language, method.name)
result.append({
"id": method.id, "provider": method.provider, "name": name_local,
"icon_url": method.icon_url, "min_amount": method.min_amount,
"max_amount": method.max_amount, "supported_currencies": method.supported_currencies
})
return result
def get_country_config(self, code: str) -> Optional[CountryConfig]:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM country_configs WHERE code = ?", (code,))
row = cursor.fetchone()
if row:
return self._row_to_country_config(row)
return None
finally:
self._close_if_file_db(conn)
def list_country_configs(self, region: Optional[str] = None, active_only: bool = True) -> List[CountryConfig]:
conn = self._get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM country_configs WHERE 1=1"
params = []
if active_only:
query += " AND is_active = 1"
if region:
query += " AND region = ?"
params.append(region)
query += " ORDER BY name"
cursor.execute(query, params)
rows = cursor.fetchall()
return [self._row_to_country_config(row) for row in rows]
finally:
self._close_if_file_db(conn)
def format_datetime(self, dt: datetime, language: str = "en", timezone: Optional[str] = None,
format_type: str = "datetime") -> str:
try:
if timezone and PYTZ_AVAILABLE:
tz = pytz.timezone(timezone)
if dt.tzinfo is None:
dt = pytz.UTC.localize(dt)
dt = dt.astimezone(tz)
lang_config = self.get_language_config(language)
if not lang_config:
lang_config = self.get_language_config("en")
if format_type == "date":
fmt = lang_config.date_format if lang_config else "%Y-%m-%d"
elif format_type == "time":
fmt = lang_config.time_format if lang_config else "%H:%M"
else:
fmt = lang_config.datetime_format if lang_config else "%Y-%m-%d %H:%M"
if BABEL_AVAILABLE:
try:
locale = Locale.parse(language.replace('_', '-'))
if format_type == "date":
return dates.format_date(dt, locale=locale)
elif format_type == "time":
return dates.format_time(dt, locale=locale)
else:
return dates.format_datetime(dt, locale=locale)
except:
pass
return dt.strftime(fmt)
except Exception as e:
logger.error(f"Error formatting datetime: {e}")
return dt.strftime("%Y-%m-%d %H:%M")
def format_number(self, number: float, language: str = "en", decimal_places: Optional[int] = None) -> str:
try:
if BABEL_AVAILABLE:
try:
locale = Locale.parse(language.replace('_', '-'))
return numbers.format_decimal(number, locale=locale, decimal_quantization=(decimal_places is not None))
except:
pass
if decimal_places is not None:
return f"{number:,.{decimal_places}f}"
return f"{number:,}"
except Exception as e:
logger.error(f"Error formatting number: {e}")
return str(number)
def format_currency(self, amount: float, currency: str, language: str = "en") -> str:
try:
if BABEL_AVAILABLE:
try:
locale = Locale.parse(language.replace('_', '-'))
return numbers.format_currency(amount, currency, locale=locale)
except:
pass
return f"{currency} {amount:,.2f}"
except Exception as e:
logger.error(f"Error formatting currency: {e}")
return f"{currency} {amount:.2f}"
def convert_timezone(self, dt: datetime, from_tz: str, to_tz: str) -> datetime:
try:
if PYTZ_AVAILABLE:
from_zone = pytz.timezone(from_tz)
to_zone = pytz.timezone(to_tz)
if dt.tzinfo is None:
dt = from_zone.localize(dt)
return dt.astimezone(to_zone)
return dt
except Exception as e:
logger.error(f"Error converting timezone: {e}")
return dt
def get_calendar_info(self, calendar_type: str, year: int, month: int) -> Dict[str, Any]:
import calendar
cal = calendar.Calendar()
month_days = cal.monthdayscalendar(year, month)
return {
"calendar_type": calendar_type, "year": year, "month": month,
"month_name": calendar.month_name[month], "days_in_month": calendar.monthrange(year, month)[1],
"first_day_of_week": calendar.monthrange(year, month)[0], "weeks": month_days
}
def get_localization_settings(self, tenant_id: str) -> Optional[LocalizationSettings]:
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM localization_settings WHERE tenant_id = ?", (tenant_id,))
row = cursor.fetchone()
if row:
return self._row_to_localization_settings(row)
return None
finally:
self._close_if_file_db(conn)
def create_localization_settings(self, tenant_id: str, default_language: str = "en",
supported_languages: Optional[List[str]] = None,
default_currency: str = "USD",
supported_currencies: Optional[List[str]] = None,
default_timezone: str = "UTC", region_code: str = "global",
data_residency: str = "regional") -> LocalizationSettings:
conn = self._get_connection()
try:
settings_id = str(uuid.uuid4())
now = datetime.now()
supported_languages = supported_languages or [default_language]
supported_currencies = supported_currencies or [default_currency]
lang_config = self.get_language_config(default_language)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO localization_settings
(id, tenant_id, default_language, supported_languages, default_currency, supported_currencies,
default_timezone, default_date_format, default_time_format, default_number_format, calendar_type,
first_day_of_week, region_code, data_residency, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (settings_id, tenant_id, default_language, json.dumps(supported_languages), default_currency,
json.dumps(supported_currencies), default_timezone,
lang_config.date_format if lang_config else "%Y-%m-%d",
lang_config.time_format if lang_config else "%H:%M",
lang_config.number_format if lang_config else "#,##0.##",
lang_config.calendar_type if lang_config else CalendarType.GREGORIAN.value,
lang_config.first_day_of_week if lang_config else 1, region_code, data_residency, now, now))
conn.commit()
return self.get_localization_settings(tenant_id)
finally:
self._close_if_file_db(conn)
def update_localization_settings(self, tenant_id: str, **kwargs) -> Optional[LocalizationSettings]:
conn = self._get_connection()
try:
settings = self.get_localization_settings(tenant_id)
if not settings:
return None
updates = []
params = []
allowed_fields = ['default_language', 'supported_languages', 'default_currency', 'supported_currencies',
'default_timezone', 'default_date_format', 'default_time_format', 'default_number_format',
'calendar_type', 'first_day_of_week', 'region_code', 'data_residency']
for key, value in kwargs.items():
if key in allowed_fields:
updates.append(f"{key} = ?")
if key in ['supported_languages', 'supported_currencies']:
params.append(json.dumps(value) if value else '[]')
elif key == 'first_day_of_week':
params.append(int(value))
else:
params.append(value)
if not updates:
return settings
updates.append("updated_at = ?")
params.append(datetime.now())
params.append(tenant_id)
cursor = conn.cursor()
cursor.execute(f"UPDATE localization_settings SET {', '.join(updates)} WHERE tenant_id = ?", params)
conn.commit()
return self.get_localization_settings(tenant_id)
finally:
self._close_if_file_db(conn)
def detect_user_preferences(self, accept_language: Optional[str] = None, ip_country: Optional[str] = None) -> Dict[str, str]:
preferences = {"language": "en", "country": "US", "timezone": "UTC", "currency": "USD"}
if accept_language:
langs = accept_language.split(',')
for lang in langs:
lang_code = lang.split(';')[0].strip().replace('-', '_')
lang_config = self.get_language_config(lang_code)
if lang_config and lang_config.is_active:
preferences["language"] = lang_code
break
if ip_country:
country = self.get_country_config(ip_country)
if country:
preferences["country"] = ip_country
preferences["currency"] = country.default_currency
preferences["timezone"] = country.timezone
if country.default_language not in preferences["language"]:
preferences["language"] = country.default_language
return preferences
def _row_to_translation(self, row: sqlite3.Row) -> Translation:
return Translation(
id=row['id'], key=row['key'], language=row['language'], value=row['value'],
namespace=row['namespace'], context=row['context'],
created_at=datetime.fromisoformat(row['created_at']) if isinstance(row['created_at'], str) else row['created_at'],
updated_at=datetime.fromisoformat(row['updated_at']) if isinstance(row['updated_at'], str) else row['updated_at'],
is_reviewed=bool(row['is_reviewed']), reviewed_by=row['reviewed_by'],
reviewed_at=datetime.fromisoformat(row['reviewed_at']) if row['reviewed_at'] and isinstance(row['reviewed_at'], str) else row['reviewed_at']
)
def _row_to_language_config(self, row: sqlite3.Row) -> LanguageConfig:
return LanguageConfig(
code=row['code'], name=row['name'], name_local=row['name_local'], is_rtl=bool(row['is_rtl']),
is_active=bool(row['is_active']), is_default=bool(row['is_default']), fallback_language=row['fallback_language'],
date_format=row['date_format'], time_format=row['time_format'], datetime_format=row['datetime_format'],
number_format=row['number_format'], currency_format=row['currency_format'],
first_day_of_week=row['first_day_of_week'], calendar_type=row['calendar_type']
)
def _row_to_data_center(self, row: sqlite3.Row) -> DataCenter:
return DataCenter(
id=row['id'], region_code=row['region_code'], name=row['name'], location=row['location'],
endpoint=row['endpoint'], status=row['status'], priority=row['priority'],
supported_regions=json.loads(row['supported_regions'] or '[]'),
capabilities=json.loads(row['capabilities'] or '{}'),
created_at=datetime.fromisoformat(row['created_at']) if isinstance(row['created_at'], str) else row['created_at'],
updated_at=datetime.fromisoformat(row['updated_at']) if isinstance(row['updated_at'], str) else row['updated_at']
)
def _row_to_tenant_dc_mapping(self, row: sqlite3.Row) -> TenantDataCenterMapping:
return TenantDataCenterMapping(
id=row['id'], tenant_id=row['tenant_id'], primary_dc_id=row['primary_dc_id'],
secondary_dc_id=row['secondary_dc_id'], region_code=row['region_code'], data_residency=row['data_residency'],
created_at=datetime.fromisoformat(row['created_at']) if isinstance(row['created_at'], str) else row['created_at'],
updated_at=datetime.fromisoformat(row['updated_at']) if isinstance(row['updated_at'], str) else row['updated_at']
)
def _row_to_payment_method(self, row: sqlite3.Row) -> LocalizedPaymentMethod:
return LocalizedPaymentMethod(
id=row['id'], provider=row['provider'], name=row['name'], name_local=json.loads(row['name_local'] or '{}'),
supported_countries=json.loads(row['supported_countries'] or '[]'),
supported_currencies=json.loads(row['supported_currencies'] or '[]'), is_active=bool(row['is_active']),
config=json.loads(row['config'] or '{}'), icon_url=row['icon_url'], display_order=row['display_order'],
min_amount=row['min_amount'], max_amount=row['max_amount'],
created_at=datetime.fromisoformat(row['created_at']) if isinstance(row['created_at'], str) else row['created_at'],
updated_at=datetime.fromisoformat(row['updated_at']) if isinstance(row['updated_at'], str) else row['updated_at']
)
def _row_to_country_config(self, row: sqlite3.Row) -> CountryConfig:
return CountryConfig(
code=row['code'], code3=row['code3'], name=row['name'], name_local=json.loads(row['name_local'] or '{}'),
region=row['region'], default_language=row['default_language'],
supported_languages=json.loads(row['supported_languages'] or '[]'), default_currency=row['default_currency'],
supported_currencies=json.loads(row['supported_currencies'] or '[]'), timezone=row['timezone'],
calendar_type=row['calendar_type'], date_format=row['date_format'], time_format=row['time_format'],
number_format=row['number_format'], address_format=row['address_format'], phone_format=row['phone_format'],
vat_rate=row['vat_rate'], is_active=bool(row['is_active'])
)
def _row_to_localization_settings(self, row: sqlite3.Row) -> LocalizationSettings:
return LocalizationSettings(
id=row['id'], tenant_id=row['tenant_id'], default_language=row['default_language'],
supported_languages=json.loads(row['supported_languages'] or '["en"]'),
default_currency=row['default_currency'], supported_currencies=json.loads(row['supported_currencies'] or '["USD"]'),
default_timezone=row['default_timezone'], default_date_format=row['default_date_format'],
default_time_format=row['default_time_format'], default_number_format=row['default_number_format'],
calendar_type=row['calendar_type'], first_day_of_week=row['first_day_of_week'], region_code=row['region_code'],
data_residency=row['data_residency'],
created_at=datetime.fromisoformat(row['created_at']) if isinstance(row['created_at'], str) else row['created_at'],
updated_at=datetime.fromisoformat(row['updated_at']) if isinstance(row['updated_at'], str) else row['updated_at']
)
_localization_manager = None
def get_localization_manager(db_path: str = "insightflow.db") -> LocalizationManager:
global _localization_manager
if _localization_manager is None:
_localization_manager = LocalizationManager(db_path)
return _localization_manager
Reference in New Issue View Git Blame Copy Permalink