Files
insightflow/backend/subscription_manager.py
AutoFix Bot e46c938b40 fix: auto-fix code issues (cron)
- 修复重复导入/字段
- 修复异常处理
- 修复PEP8格式问题 (E302, E305, E501)
- 修复行长度超过100字符的问题
- 修复F821未定义名称错误
2026-03-01 18:19:06 +08:00

2214 lines
72 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. 多层级订阅计划Free/Pro/Enterprise
2. 按量计费转录时长、存储空间、API 调用次数)
3. 支付集成Stripe、支付宝、微信支付
4. 发票管理、退款处理、账单历史
作者: InsightFlow Team
"""
import json
import logging
import sqlite3
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import StrEnum
from typing import Any
logger = logging.getLogger(__name__)
class SubscriptionStatus(StrEnum):
"""订阅状态"""
ACTIVE = "active" # 活跃
CANCELLED = "cancelled" # 已取消
EXPIRED = "expired" # 已过期
PAST_DUE = "past_due" # 逾期
TRIAL = "trial" # 试用中
PENDING = "pending" # 待支付
class PaymentProvider(StrEnum):
"""支付提供商"""
STRIPE = "stripe" # Stripe
ALIPAY = "alipay" # 支付宝
WECHAT = "wechat" # 微信支付
BANK_TRANSFER = "bank_transfer" # 银行转账
class PaymentStatus(StrEnum):
"""支付状态"""
PENDING = "pending" # 待支付
PROCESSING = "processing" # 处理中
COMPLETED = "completed" # 已完成
FAILED = "failed" # 失败
REFUNDED = "refunded" # 已退款
PARTIAL_REFUNDED = "partial_refunded" # 部分退款
class InvoiceStatus(StrEnum):
"""发票状态"""
DRAFT = "draft" # 草稿
ISSUED = "issued" # 已开具
PAID = "paid" # 已支付
OVERDUE = "overdue" # 逾期
VOID = "void" # 作废
CREDIT_NOTE = "credit_note" # 贷项通知单
class RefundStatus(StrEnum):
"""退款状态"""
PENDING = "pending" # 待处理
APPROVED = "approved" # 已批准
REJECTED = "rejected" # 已拒绝
COMPLETED = "completed" # 已完成
FAILED = "failed" # 失败
@dataclass
class SubscriptionPlan:
"""订阅计划数据类"""
id: str
name: str
tier: str # free/pro/enterprise
description: str
price_monthly: float # 月付价格
price_yearly: float # 年付价格
currency: str # CNY/USD
features: list[str] # 功能列表
limits: dict[str, Any] # 资源限制
is_active: bool
created_at: datetime
updated_at: datetime
metadata: dict[str, Any]
@dataclass
class Subscription:
"""订阅数据类"""
id: str
tenant_id: str
plan_id: str
status: str
current_period_start: datetime
current_period_end: datetime
cancel_at_period_end: bool
canceled_at: datetime | None
trial_start: datetime | None
trial_end: datetime | None
payment_provider: str | None
provider_subscription_id: str | None # 支付提供商的订阅ID
created_at: datetime
updated_at: datetime
metadata: dict[str, Any]
@dataclass
class UsageRecord:
"""用量记录数据类"""
id: str
tenant_id: str
resource_type: str # transcription/storage/api_call
quantity: float # 使用量
unit: str # minutes/mb/count
recorded_at: datetime
cost: float # 费用
description: str | None
metadata: dict[str, Any]
@dataclass
class Payment:
"""支付记录数据类"""
id: str
tenant_id: str
subscription_id: str | None
invoice_id: str | None
amount: float
currency: str
provider: str
provider_payment_id: str | None
status: str
payment_method: str | None
payment_details: dict[str, Any]
paid_at: datetime | None
failed_at: datetime | None
failure_reason: str | None
created_at: datetime
updated_at: datetime
@dataclass
class Invoice:
"""发票数据类"""
id: str
tenant_id: str
subscription_id: str | None
invoice_number: str
status: str
amount_due: float
amount_paid: float
currency: str
period_start: datetime
period_end: datetime
description: str
line_items: list[dict[str, Any]]
due_date: datetime
paid_at: datetime | None
voided_at: datetime | None
void_reason: str | None
created_at: datetime
updated_at: datetime
@dataclass
class Refund:
"""退款数据类"""
id: str
tenant_id: str
payment_id: str
invoice_id: str | None
amount: float
currency: str
reason: str
status: str
requested_by: str
requested_at: datetime
approved_by: str | None
approved_at: str | None
completed_at: datetime | None
provider_refund_id: str | None
metadata: dict[str, Any]
created_at: datetime
updated_at: datetime
@dataclass
class BillingHistory:
"""账单历史数据类"""
id: str
tenant_id: str
type: str # subscription/usage/payment/refund
amount: float
currency: str
description: str
reference_id: str # 关联的订阅/支付/退款ID
balance_after: float # 操作后余额
created_at: datetime
metadata: dict[str, Any]
class SubscriptionManager:
"""订阅与计费管理器"""
# 默认订阅计划配置
DEFAULT_PLANS = {
"free": {
"name": "Free",
"tier": "free",
"description": "免费版,适合个人用户试用",
"price_monthly": 0.0,
"price_yearly": 0.0,
"currency": "CNY",
"features": [
"basic_analysis",
"export_png",
"3_projects",
"100_mb_storage",
"60_min_transcription",
],
"limits": {
"max_projects": 3,
"max_storage_mb": 100,
"max_transcription_minutes": 60,
"max_api_calls_per_day": 100,
"max_team_members": 2,
"max_entities": 100,
},
},
"pro": {
"name": "Pro",
"tier": "pro",
"description": "专业版,适合小型团队",
"price_monthly": 99.0,
"price_yearly": 990.0,
"currency": "CNY",
"features": [
"all_free_features",
"advanced_analysis",
"export_all_formats",
"api_access",
"webhooks",
"collaboration",
"20_projects",
"10_gb_storage",
"600_min_transcription",
],
"limits": {
"max_projects": 20,
"max_storage_mb": 10240,
"max_transcription_minutes": 600,
"max_api_calls_per_day": 10000,
"max_team_members": 10,
"max_entities": 1000,
},
},
"enterprise": {
"name": "Enterprise",
"tier": "enterprise",
"description": "企业版,适合大型企业",
"price_monthly": 999.0,
"price_yearly": 9990.0,
"currency": "CNY",
"features": [
"all_pro_features",
"unlimited_projects",
"unlimited_storage",
"unlimited_transcription",
"priority_support",
"custom_integration",
"sla_guarantee",
"dedicated_manager",
],
"limits": {
"max_projects": -1,
"max_storage_mb": -1,
"max_transcription_minutes": -1,
"max_api_calls_per_day": -1,
"max_team_members": -1,
"max_entities": -1,
},
},
}
# 按量计费单价CNY
USAGE_PRICING = {
"transcription": {
"unit": "minute",
"price": 0.5,
"free_quota": 60,
}, # 0.5元/分钟 # 每月免费额度
"storage": {"unit": "gb", "price": 10.0, "free_quota": 0.1}, # 10元/GB/月 # 100MB免费
"api_call": {
"unit": "1000_calls",
"price": 5.0,
"free_quota": 1000,
}, # 5元/1000次 # 每月免费1000次
"export": {"unit": "page", "price": 0.1, "free_quota": 100}, # 0.1元/页PDF导出
}
def __init__(self, db_path: str = "insightflow.db"):
self.db_path = db_path
self._init_db()
self._init_default_plans()
def _get_connection(self) -> sqlite3.Connection:
"""获取数据库连接"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def _init_db(self) -> None:
"""初始化数据库表"""
conn = self._get_connection()
try:
cursor = conn.cursor()
# 订阅计划表
cursor.execute("""
CREATE TABLE IF NOT EXISTS subscription_plans (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
tier TEXT UNIQUE NOT NULL,
description TEXT,
price_monthly REAL DEFAULT 0,
price_yearly REAL DEFAULT 0,
currency TEXT DEFAULT 'CNY',
features TEXT DEFAULT '[]',
limits TEXT DEFAULT '{}',
is_active INTEGER DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
metadata TEXT DEFAULT '{}'
)
""")
# 订阅表
cursor.execute("""
CREATE TABLE IF NOT EXISTS subscriptions (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
plan_id TEXT NOT NULL,
status TEXT DEFAULT 'pending',
current_period_start TIMESTAMP,
current_period_end TIMESTAMP,
cancel_at_period_end INTEGER DEFAULT 0,
canceled_at TIMESTAMP,
trial_start TIMESTAMP,
trial_end TIMESTAMP,
payment_provider TEXT,
provider_subscription_id TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
metadata TEXT DEFAULT '{}',
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE,
FOREIGN KEY (plan_id) REFERENCES subscription_plans(id)
)
""")
# 用量记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS usage_records (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
resource_type TEXT NOT NULL,
quantity REAL DEFAULT 0,
unit TEXT NOT NULL,
recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
cost REAL DEFAULT 0,
description TEXT,
metadata TEXT DEFAULT '{}',
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE
)
""")
# 支付记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS payments (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
subscription_id TEXT,
invoice_id TEXT,
amount REAL NOT NULL,
currency TEXT DEFAULT 'CNY',
provider TEXT NOT NULL,
provider_payment_id TEXT,
status TEXT DEFAULT 'pending',
payment_method TEXT,
payment_details TEXT DEFAULT '{}',
paid_at TIMESTAMP,
failed_at TIMESTAMP,
failure_reason TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE,
FOREIGN KEY (subscription_id) REFERENCES subscriptions(id) ON DELETE SET NULL,
FOREIGN KEY (invoice_id) REFERENCES invoices(id) ON DELETE SET NULL
)
""")
# 发票表
cursor.execute("""
CREATE TABLE IF NOT EXISTS invoices (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
subscription_id TEXT,
invoice_number TEXT UNIQUE NOT NULL,
status TEXT DEFAULT 'draft',
amount_due REAL DEFAULT 0,
amount_paid REAL DEFAULT 0,
currency TEXT DEFAULT 'CNY',
period_start TIMESTAMP,
period_end TIMESTAMP,
description TEXT,
line_items TEXT DEFAULT '[]',
due_date TIMESTAMP,
paid_at TIMESTAMP,
voided_at TIMESTAMP,
void_reason TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE,
FOREIGN KEY (subscription_id) REFERENCES subscriptions(id) ON DELETE SET NULL
)
""")
# 退款表
cursor.execute("""
CREATE TABLE IF NOT EXISTS refunds (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
payment_id TEXT NOT NULL,
invoice_id TEXT,
amount REAL NOT NULL,
currency TEXT DEFAULT 'CNY',
reason TEXT,
status TEXT DEFAULT 'pending',
requested_by TEXT NOT NULL,
requested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
approved_by TEXT,
approved_at TIMESTAMP,
completed_at TIMESTAMP,
provider_refund_id TEXT,
metadata TEXT DEFAULT '{}',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE,
FOREIGN KEY (payment_id) REFERENCES payments(id) ON DELETE CASCADE,
FOREIGN KEY (invoice_id) REFERENCES invoices(id) ON DELETE SET NULL
)
""")
# 账单历史表
cursor.execute("""
CREATE TABLE IF NOT EXISTS billing_history (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
type TEXT NOT NULL,
amount REAL NOT NULL,
currency TEXT DEFAULT 'CNY',
description TEXT,
reference_id TEXT,
balance_after REAL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
metadata TEXT DEFAULT '{}',
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE
)
""")
# 创建索引
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_subscriptions_tenant ON subscriptions(tenant_id)"
)
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_subscriptions_status ON subscriptions(status)"
)
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_subscriptions_plan ON subscriptions(plan_id)"
)
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_usage_tenant ON usage_records(tenant_id)"
)
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_usage_type ON usage_records(resource_type)"
)
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_usage_recorded ON usage_records(recorded_at)"
)
cursor.execute("CREATE INDEX IF NOT EXISTS idx_payments_tenant ON payments(tenant_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_payments_status ON payments(status)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_invoices_tenant ON invoices(tenant_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_invoices_status ON invoices(status)")
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_invoices_number ON invoices(invoice_number)"
)
cursor.execute("CREATE INDEX IF NOT EXISTS idx_refunds_tenant ON refunds(tenant_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_refunds_status ON refunds(status)")
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_billing_tenant ON billing_history(tenant_id)"
)
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_billing_created ON billing_history(created_at)"
)
conn.commit()
logger.info("Subscription tables initialized successfully")
except Exception as e:
logger.error(f"Error initializing subscription tables: {e}")
raise
finally:
conn.close()
def _init_default_plans(self) -> None:
"""初始化默认订阅计划"""
conn = self._get_connection()
try:
cursor = conn.cursor()
for tier, plan_data in self.DEFAULT_PLANS.items():
cursor.execute(
"""
INSERT OR IGNORE INTO subscription_plans
(id, name, tier, description, price_monthly, price_yearly, currency,
features, limits, is_active, created_at, updated_at, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
str(uuid.uuid4()),
plan_data["name"],
plan_data["tier"],
plan_data["description"],
plan_data["price_monthly"],
plan_data["price_yearly"],
plan_data["currency"],
json.dumps(plan_data["features"]),
json.dumps(plan_data["limits"]),
1,
datetime.now(),
datetime.now(),
json.dumps({}),
),
)
conn.commit()
logger.info("Default subscription plans initialized")
except Exception as e:
logger.error(f"Error initializing default plans: {e}")
finally:
conn.close()
# ==================== 订阅计划管理 ====================
def get_plan(self, plan_id: str) -> SubscriptionPlan | None:
"""获取订阅计划"""
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM subscription_plans WHERE id = ?", (plan_id,))
row = cursor.fetchone()
if row:
return self._row_to_plan(row)
return None
finally:
conn.close()
def get_plan_by_tier(self, tier: str) -> SubscriptionPlan | None:
"""通过层级获取订阅计划"""
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM subscription_plans WHERE tier = ? AND is_active = 1", (tier,)
)
row = cursor.fetchone()
if row:
return self._row_to_plan(row)
return None
finally:
conn.close()
def list_plans(self, include_inactive: bool = False) -> list[SubscriptionPlan]:
"""列出所有订阅计划"""
conn = self._get_connection()
try:
cursor = conn.cursor()
if include_inactive:
cursor.execute("SELECT * FROM subscription_plans ORDER BY price_monthly")
else:
cursor.execute(
"SELECT * FROM subscription_plans WHERE is_active = 1 ORDER BY price_monthly"
)
rows = cursor.fetchall()
return [self._row_to_plan(row) for row in rows]
finally:
conn.close()
def create_plan(
self,
name: str,
tier: str,
description: str,
price_monthly: float,
price_yearly: float,
currency: str = "CNY",
features: list[str] = None,
limits: dict[str, Any] = None,
) -> SubscriptionPlan:
"""创建新订阅计划"""
conn = self._get_connection()
try:
plan_id = str(uuid.uuid4())
plan = SubscriptionPlan(
id=plan_id,
name=name,
tier=tier,
description=description,
price_monthly=price_monthly,
price_yearly=price_yearly,
currency=currency,
features=features or [],
limits=limits or {},
is_active=True,
created_at=datetime.now(),
updated_at=datetime.now(),
metadata={},
)
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO subscription_plans
(id, name, tier, description, price_monthly, price_yearly, currency,
features, limits, is_active, created_at, updated_at, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
plan.id,
plan.name,
plan.tier,
plan.description,
plan.price_monthly,
plan.price_yearly,
plan.currency,
json.dumps(plan.features),
json.dumps(plan.limits),
int(plan.is_active),
plan.created_at,
plan.updated_at,
json.dumps(plan.metadata),
),
)
conn.commit()
logger.info(f"Subscription plan created: {plan_id} ({name})")
return plan
except Exception as e:
conn.rollback()
logger.error(f"Error creating plan: {e}")
raise
finally:
conn.close()
def update_plan(self, plan_id: str, **kwargs) -> SubscriptionPlan | None:
"""更新订阅计划"""
conn = self._get_connection()
try:
plan = self.get_plan(plan_id)
if not plan:
return None
updates = []
params = []
allowed_fields = [
"name",
"description",
"price_monthly",
"price_yearly",
"currency",
"features",
"limits",
"is_active",
]
for key, value in kwargs.items():
if key in allowed_fields:
updates.append(f"{key} = ?")
if key in ["features", "limits"]:
params.append(json.dumps(value) if value else "{}")
elif key == "is_active":
params.append(int(value))
else:
params.append(value)
if not updates:
return plan
updates.append("updated_at = ?")
params.append(datetime.now())
params.append(plan_id)
cursor = conn.cursor()
cursor.execute(
f"""
UPDATE subscription_plans SET {", ".join(updates)}
WHERE id = ?
""",
params,
)
conn.commit()
return self.get_plan(plan_id)
finally:
conn.close()
# ==================== 订阅管理 ====================
def create_subscription(
self,
tenant_id: str,
plan_id: str,
payment_provider: str | None = None,
trial_days: int = 0,
billing_cycle: str = "monthly",
) -> Subscription:
"""创建新订阅"""
conn = self._get_connection()
try:
# 检查是否已有活跃订阅
cursor = conn.cursor()
cursor.execute(
"""
SELECT * FROM subscriptions
WHERE tenant_id = ? AND status IN ('active', 'trial', 'pending')
""",
(tenant_id,),
)
existing = cursor.fetchone()
if existing:
raise ValueError(f"Tenant {tenant_id} already has an active subscription")
# 获取计划信息
plan = self.get_plan(plan_id)
if not plan:
raise ValueError(f"Plan {plan_id} not found")
subscription_id = str(uuid.uuid4())
now = datetime.now()
# 计算周期
if billing_cycle == "yearly":
period_end = now + timedelta(days=365)
else:
period_end = now + timedelta(days=30)
# 试用处理
trial_start = None
trial_end = None
if trial_days > 0:
trial_start = now
trial_end = now + timedelta(days=trial_days)
status = SubscriptionStatus.TRIAL.value
else:
status = SubscriptionStatus.PENDING.value
subscription = Subscription(
id=subscription_id,
tenant_id=tenant_id,
plan_id=plan_id,
status=status,
current_period_start=now,
current_period_end=period_end,
cancel_at_period_end=False,
canceled_at=None,
trial_start=trial_start,
trial_end=trial_end,
payment_provider=payment_provider,
provider_subscription_id=None,
created_at=now,
updated_at=now,
metadata={"billing_cycle": billing_cycle},
)
cursor.execute(
"""
INSERT INTO subscriptions
(id, tenant_id, plan_id, status, current_period_start, current_period_end,
cancel_at_period_end, canceled_at, trial_start, trial_end,
payment_provider, provider_subscription_id, created_at, updated_at, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
subscription.id,
subscription.tenant_id,
subscription.plan_id,
subscription.status,
subscription.current_period_start,
subscription.current_period_end,
int(subscription.cancel_at_period_end),
subscription.canceled_at,
subscription.trial_start,
subscription.trial_end,
subscription.payment_provider,
subscription.provider_subscription_id,
subscription.created_at,
subscription.updated_at,
json.dumps(subscription.metadata),
),
)
# 创建发票
amount = plan.price_yearly if billing_cycle == "yearly" else plan.price_monthly
if amount > 0 and trial_days == 0:
self._create_invoice_internal(
conn,
tenant_id,
subscription_id,
amount,
plan.currency,
now,
period_end,
f"{plan.name} Subscription ({billing_cycle})",
)
# 记录账单历史
self._add_billing_history_internal(
conn,
tenant_id,
"subscription",
0,
plan.currency,
f"Subscription created: {plan.name}",
subscription_id,
0,
)
conn.commit()
logger.info(f"Subscription created: {subscription_id} for tenant {tenant_id}")
return subscription
except Exception as e:
conn.rollback()
logger.error(f"Error creating subscription: {e}")
raise
finally:
conn.close()
def get_subscription(self, subscription_id: str) -> Subscription | None:
"""获取订阅信息"""
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM subscriptions WHERE id = ?", (subscription_id,))
row = cursor.fetchone()
if row:
return self._row_to_subscription(row)
return None
finally:
conn.close()
def get_tenant_subscription(self, tenant_id: str) -> Subscription | None:
"""获取租户的当前订阅"""
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute(
"""
SELECT * FROM subscriptions
WHERE tenant_id = ? AND status IN ('active', 'trial', 'past_due', 'pending')
ORDER BY created_at DESC LIMIT 1
""",
(tenant_id,),
)
row = cursor.fetchone()
if row:
return self._row_to_subscription(row)
return None
finally:
conn.close()
def update_subscription(self, subscription_id: str, **kwargs) -> Subscription | None:
"""更新订阅"""
conn = self._get_connection()
try:
subscription = self.get_subscription(subscription_id)
if not subscription:
return None
updates = []
params = []
allowed_fields = [
"status",
"current_period_start",
"current_period_end",
"cancel_at_period_end",
"canceled_at",
"trial_end",
"payment_provider",
"provider_subscription_id",
]
for key, value in kwargs.items():
if key in allowed_fields:
updates.append(f"{key} = ?")
if key == "cancel_at_period_end":
params.append(int(value))
else:
params.append(value)
if not updates:
return subscription
updates.append("updated_at = ?")
params.append(datetime.now())
params.append(subscription_id)
cursor = conn.cursor()
cursor.execute(
f"""
UPDATE subscriptions SET {", ".join(updates)}
WHERE id = ?
""",
params,
)
conn.commit()
return self.get_subscription(subscription_id)
finally:
conn.close()
def cancel_subscription(
self, subscription_id: str, at_period_end: bool = True
) -> Subscription | None:
"""取消订阅"""
conn = self._get_connection()
try:
subscription = self.get_subscription(subscription_id)
if not subscription:
return None
now = datetime.now()
if at_period_end:
# 在周期结束时取消
cursor = conn.cursor()
cursor.execute(
"""
UPDATE subscriptions
SET cancel_at_period_end = 1, canceled_at = ?, updated_at = ?
WHERE id = ?
""",
(now, now, subscription_id),
)
else:
# 立即取消
cursor = conn.cursor()
cursor.execute(
"""
UPDATE subscriptions
SET status = 'cancelled', canceled_at = ?, updated_at = ?
WHERE id = ?
""",
(now, now, subscription_id),
)
# 记录账单历史
self._add_billing_history_internal(
conn,
subscription.tenant_id,
"subscription",
0,
"CNY",
f"Subscription cancelled{' (at period end)' if at_period_end else ''}",
subscription_id,
0,
)
conn.commit()
logger.info(f"Subscription cancelled: {subscription_id}")
return self.get_subscription(subscription_id)
finally:
conn.close()
def change_plan(
self, subscription_id: str, new_plan_id: str, prorate: bool = True
) -> Subscription | None:
"""更改订阅计划"""
conn = self._get_connection()
try:
subscription = self.get_subscription(subscription_id)
if not subscription:
return None
old_plan = self.get_plan(subscription.plan_id)
new_plan = self.get_plan(new_plan_id)
if not new_plan:
raise ValueError(f"Plan {new_plan_id} not found")
now = datetime.now()
# 按比例计算差价(简化实现)
if prorate and old_plan:
# 这里应该实现实际的按比例计算逻辑
pass
cursor = conn.cursor()
cursor.execute(
"""
UPDATE subscriptions
SET plan_id = ?, updated_at = ?
WHERE id = ?
""",
(new_plan_id, now, subscription_id),
)
# 记录账单历史
self._add_billing_history_internal(
conn,
subscription.tenant_id,
"subscription",
0,
new_plan.currency,
f"Plan changed from {old_plan.name if old_plan else 'unknown'} to {new_plan.name}",
subscription_id,
0,
)
conn.commit()
logger.info(f"Subscription plan changed: {subscription_id} -> {new_plan_id}")
return self.get_subscription(subscription_id)
finally:
conn.close()
# ==================== 用量计费 ====================
def record_usage(
self,
tenant_id: str,
resource_type: str,
quantity: float,
unit: str,
description: str | None = None,
metadata: dict | None = None,
) -> UsageRecord:
"""记录用量"""
conn = self._get_connection()
try:
# 计算费用
cost = self._calculate_usage_cost(resource_type, quantity)
record_id = str(uuid.uuid4())
record = UsageRecord(
id=record_id,
tenant_id=tenant_id,
resource_type=resource_type,
quantity=quantity,
unit=unit,
recorded_at=datetime.now(),
cost=cost,
description=description,
metadata=metadata or {},
)
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO usage_records
(id, tenant_id, resource_type, quantity, unit, recorded_at, cost, description, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
record.id,
record.tenant_id,
record.resource_type,
record.quantity,
record.unit,
record.recorded_at,
record.cost,
record.description,
json.dumps(record.metadata),
),
)
conn.commit()
return record
finally:
conn.close()
def get_usage_summary(
self, tenant_id: str, start_date: datetime | None = None, end_date: datetime | None = None
) -> dict[str, Any]:
"""获取用量汇总"""
conn = self._get_connection()
try:
cursor = conn.cursor()
query = """
SELECT
resource_type,
SUM(quantity) as total_quantity,
SUM(cost) as total_cost,
COUNT(*) as record_count
FROM usage_records
WHERE tenant_id = ?
"""
params = [tenant_id]
if start_date:
query += " AND recorded_at >= ?"
params.append(start_date)
if end_date:
query += " AND recorded_at <= ?"
params.append(end_date)
query += " GROUP BY resource_type"
cursor.execute(query, params)
rows = cursor.fetchall()
summary = {}
total_cost = 0
for row in rows:
summary[row["resource_type"]] = {
"quantity": row["total_quantity"],
"cost": row["total_cost"],
"records": row["record_count"],
}
total_cost += row["total_cost"]
return {
"tenant_id": tenant_id,
"period": {
"start": start_date.isoformat() if start_date else None,
"end": end_date.isoformat() if end_date else None,
},
"breakdown": summary,
"total_cost": total_cost,
}
finally:
conn.close()
def _calculate_usage_cost(self, resource_type: str, quantity: float) -> float:
"""计算用量费用"""
pricing = self.USAGE_PRICING.get(resource_type)
if not pricing:
return 0.0
# 扣除免费额度
chargeable = max(0, quantity - pricing.get("free_quota", 0))
# 计算费用
if pricing["unit"] == "1000_calls":
return (chargeable / 1000) * pricing["price"]
else:
return chargeable * pricing["price"]
# ==================== 支付管理 ====================
def create_payment(
self,
tenant_id: str,
amount: float,
currency: str,
provider: str,
subscription_id: str | None = None,
invoice_id: str | None = None,
payment_method: str | None = None,
payment_details: dict | None = None,
) -> Payment:
"""创建支付记录"""
conn = self._get_connection()
try:
payment_id = str(uuid.uuid4())
now = datetime.now()
payment = Payment(
id=payment_id,
tenant_id=tenant_id,
subscription_id=subscription_id,
invoice_id=invoice_id,
amount=amount,
currency=currency,
provider=provider,
provider_payment_id=None,
status=PaymentStatus.PENDING.value,
payment_method=payment_method,
payment_details=payment_details or {},
paid_at=None,
failed_at=None,
failure_reason=None,
created_at=now,
updated_at=now,
)
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO payments
(id, tenant_id, subscription_id, invoice_id, amount, currency,
provider, provider_payment_id, status, payment_method, payment_details,
paid_at, failed_at, failure_reason, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
payment.id,
payment.tenant_id,
payment.subscription_id,
payment.invoice_id,
payment.amount,
payment.currency,
payment.provider,
payment.provider_payment_id,
payment.status,
payment.payment_method,
json.dumps(payment.payment_details),
payment.paid_at,
payment.failed_at,
payment.failure_reason,
payment.created_at,
payment.updated_at,
),
)
conn.commit()
return payment
finally:
conn.close()
def confirm_payment(
self, payment_id: str, provider_payment_id: str | None = None
) -> Payment | None:
"""确认支付完成"""
conn = self._get_connection()
try:
payment = self._get_payment_internal(conn, payment_id)
if not payment:
return None
now = datetime.now()
cursor = conn.cursor()
cursor.execute(
"""
UPDATE payments
SET status = 'completed', provider_payment_id = ?, paid_at = ?, updated_at = ?
WHERE id = ?
""",
(provider_payment_id, now, now, payment_id),
)
# 如果有关联发票,更新发票状态
if payment.invoice_id:
cursor.execute(
"""
UPDATE invoices
SET status = 'paid', amount_paid = amount_due, paid_at = ?
WHERE id = ?
""",
(now, payment.invoice_id),
)
# 如果有关联订阅,激活订阅
if payment.subscription_id:
cursor.execute(
"""
UPDATE subscriptions
SET status = 'active', updated_at = ?
WHERE id = ? AND status = 'pending'
""",
(now, payment.subscription_id),
)
# 记录账单历史
self._add_billing_history_internal(
conn,
payment.tenant_id,
"payment",
payment.amount,
payment.currency,
f"Payment completed via {payment.provider}",
payment_id,
0, # 余额更新应该在账户管理中处理
)
conn.commit()
logger.info(f"Payment confirmed: {payment_id}")
return self._get_payment_internal(conn, payment_id)
finally:
conn.close()
def fail_payment(self, payment_id: str, reason: str) -> Payment | None:
"""标记支付失败"""
conn = self._get_connection()
try:
now = datetime.now()
cursor = conn.cursor()
cursor.execute(
"""
UPDATE payments
SET status = 'failed', failure_reason = ?, failed_at = ?, updated_at = ?
WHERE id = ?
""",
(reason, now, now, payment_id),
)
conn.commit()
return self._get_payment_internal(conn, payment_id)
finally:
conn.close()
def get_payment(self, payment_id: str) -> Payment | None:
"""获取支付记录"""
conn = self._get_connection()
try:
return self._get_payment_internal(conn, payment_id)
finally:
conn.close()
def list_payments(
self, tenant_id: str, status: str | None = None, limit: int = 100, offset: int = 0
) -> list[Payment]:
"""列出支付记录"""
conn = self._get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM payments WHERE tenant_id = ?"
params = [tenant_id]
if status:
query += " AND status = ?"
params.append(status)
query += " ORDER BY created_at DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
cursor.execute(query, params)
rows = cursor.fetchall()
return [self._row_to_payment(row) for row in rows]
finally:
conn.close()
def _get_payment_internal(self, conn: sqlite3.Connection, payment_id: str) -> Payment | None:
"""内部方法:获取支付记录"""
cursor = conn.cursor()
cursor.execute("SELECT * FROM payments WHERE id = ?", (payment_id,))
row = cursor.fetchone()
if row:
return self._row_to_payment(row)
return None
# ==================== 发票管理 ====================
def _create_invoice_internal(
self,
conn: sqlite3.Connection,
tenant_id: str,
subscription_id: str | None,
amount: float,
currency: str,
period_start: datetime,
period_end: datetime,
description: str,
line_items: list[dict] | None = None,
) -> Invoice:
"""内部方法:创建发票"""
invoice_id = str(uuid.uuid4())
invoice_number = self._generate_invoice_number()
now = datetime.now()
due_date = now + timedelta(days=7) # 7天付款期限
invoice = Invoice(
id=invoice_id,
tenant_id=tenant_id,
subscription_id=subscription_id,
invoice_number=invoice_number,
status=InvoiceStatus.DRAFT.value,
amount_due=amount,
amount_paid=0,
currency=currency,
period_start=period_start,
period_end=period_end,
description=description,
line_items=line_items or [{"description": description, "amount": amount}],
due_date=due_date,
paid_at=None,
voided_at=None,
void_reason=None,
created_at=now,
updated_at=now,
)
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO invoices
(id, tenant_id, subscription_id, invoice_number, status, amount_due, amount_paid,
currency, period_start, period_end, description, line_items, due_date,
paid_at, voided_at, void_reason, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
invoice.id,
invoice.tenant_id,
invoice.subscription_id,
invoice.invoice_number,
invoice.status,
invoice.amount_due,
invoice.amount_paid,
invoice.currency,
invoice.period_start,
invoice.period_end,
invoice.description,
json.dumps(invoice.line_items),
invoice.due_date,
invoice.paid_at,
invoice.voided_at,
invoice.void_reason,
invoice.created_at,
invoice.updated_at,
),
)
return invoice
def get_invoice(self, invoice_id: str) -> Invoice | None:
"""获取发票"""
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM invoices WHERE id = ?", (invoice_id,))
row = cursor.fetchone()
if row:
return self._row_to_invoice(row)
return None
finally:
conn.close()
def get_invoice_by_number(self, invoice_number: str) -> Invoice | None:
"""通过发票号获取发票"""
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute("SELECT * FROM invoices WHERE invoice_number = ?", (invoice_number,))
row = cursor.fetchone()
if row:
return self._row_to_invoice(row)
return None
finally:
conn.close()
def list_invoices(
self, tenant_id: str, status: str | None = None, limit: int = 100, offset: int = 0
) -> list[Invoice]:
"""列出发票"""
conn = self._get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM invoices WHERE tenant_id = ?"
params = [tenant_id]
if status:
query += " AND status = ?"
params.append(status)
query += " ORDER BY created_at DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
cursor.execute(query, params)
rows = cursor.fetchall()
return [self._row_to_invoice(row) for row in rows]
finally:
conn.close()
def void_invoice(self, invoice_id: str, reason: str) -> Invoice | None:
"""作废发票"""
conn = self._get_connection()
try:
invoice = self.get_invoice(invoice_id)
if not invoice:
return None
if invoice.status == InvoiceStatus.PAID.value:
raise ValueError("Cannot void a paid invoice")
now = datetime.now()
cursor = conn.cursor()
cursor.execute(
"""
UPDATE invoices
SET status = 'void', voided_at = ?, void_reason = ?, updated_at = ?
WHERE id = ?
""",
(now, reason, now, invoice_id),
)
conn.commit()
return self.get_invoice(invoice_id)
finally:
conn.close()
def _generate_invoice_number(self) -> str:
"""生成发票号"""
now = datetime.now()
prefix = f"INV-{now.strftime('%Y%m')}"
conn = self._get_connection()
try:
cursor = conn.cursor()
cursor.execute(
"""
SELECT COUNT(*) as count FROM invoices
WHERE invoice_number LIKE ?
""",
(f"{prefix}%",),
)
row = cursor.fetchone()
count = row["count"] + 1
return f"{prefix}-{count:06d}"
finally:
conn.close()
# ==================== 退款管理 ====================
def request_refund(
self, tenant_id: str, payment_id: str, amount: float, reason: str, requested_by: str
) -> Refund:
"""申请退款"""
conn = self._get_connection()
try:
# 验证支付记录
payment = self._get_payment_internal(conn, payment_id)
if not payment:
raise ValueError(f"Payment {payment_id} not found")
if payment.tenant_id != tenant_id:
raise ValueError("Payment does not belong to this tenant")
if payment.status != PaymentStatus.COMPLETED.value:
raise ValueError("Can only refund completed payments")
if amount > payment.amount:
raise ValueError("Refund amount cannot exceed payment amount")
refund_id = str(uuid.uuid4())
now = datetime.now()
refund = Refund(
id=refund_id,
tenant_id=tenant_id,
payment_id=payment_id,
invoice_id=payment.invoice_id,
amount=amount,
currency=payment.currency,
reason=reason,
status=RefundStatus.PENDING.value,
requested_by=requested_by,
requested_at=now,
approved_by=None,
approved_at=None,
completed_at=None,
provider_refund_id=None,
metadata={},
created_at=now,
updated_at=now,
)
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO refunds
(id, tenant_id, payment_id, invoice_id, amount, currency, reason, status,
requested_by, requested_at, approved_by, approved_at, completed_at,
provider_refund_id, metadata, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
refund.id,
refund.tenant_id,
refund.payment_id,
refund.invoice_id,
refund.amount,
refund.currency,
refund.reason,
refund.status,
refund.requested_by,
refund.requested_at,
refund.approved_by,
refund.approved_at,
refund.completed_at,
refund.provider_refund_id,
json.dumps(refund.metadata),
refund.created_at,
refund.updated_at,
),
)
conn.commit()
logger.info(f"Refund requested: {refund_id} for payment {payment_id}")
return refund
finally:
conn.close()
def approve_refund(self, refund_id: str, approved_by: str) -> Refund | None:
"""批准退款"""
conn = self._get_connection()
try:
refund = self._get_refund_internal(conn, refund_id)
if not refund:
return None
if refund.status != RefundStatus.PENDING.value:
raise ValueError("Can only approve pending refunds")
now = datetime.now()
cursor = conn.cursor()
cursor.execute(
"""
UPDATE refunds
SET status = 'approved', approved_by = ?, approved_at = ?, updated_at = ?
WHERE id = ?
""",
(approved_by, now, now, refund_id),
)
conn.commit()
return self._get_refund_internal(conn, refund_id)
finally:
conn.close()
def complete_refund(
self, refund_id: str, provider_refund_id: str | None = None
) -> Refund | None:
"""完成退款"""
conn = self._get_connection()
try:
refund = self._get_refund_internal(conn, refund_id)
if not refund:
return None
now = datetime.now()
cursor = conn.cursor()
cursor.execute(
"""
UPDATE refunds
SET status = 'completed', provider_refund_id = ?, completed_at = ?, updated_at = ?
WHERE id = ?
""",
(provider_refund_id, now, now, refund_id),
)
# 更新原支付记录状态
cursor.execute(
"""
UPDATE payments
SET status = 'refunded', updated_at = ?
WHERE id = ?
""",
(now, refund.payment_id),
)
# 记录账单历史
self._add_billing_history_internal(
conn,
refund.tenant_id,
"refund",
-refund.amount,
refund.currency,
f"Refund processed: {refund.reason}",
refund_id,
0,
)
conn.commit()
logger.info(f"Refund completed: {refund_id}")
return self._get_refund_internal(conn, refund_id)
finally:
conn.close()
def reject_refund(self, refund_id: str, reason: str) -> Refund | None:
"""拒绝退款"""
conn = self._get_connection()
try:
refund = self._get_refund_internal(conn, refund_id)
if not refund:
return None
now = datetime.now()
cursor = conn.cursor()
cursor.execute(
"""
UPDATE refunds
SET status = 'rejected', metadata = json_set(metadata, '$.rejection_reason', ?), updated_at = ?
WHERE id = ?
""",
(reason, now, refund_id),
)
conn.commit()
return self._get_refund_internal(conn, refund_id)
finally:
conn.close()
def get_refund(self, refund_id: str) -> Refund | None:
"""获取退款记录"""
conn = self._get_connection()
try:
return self._get_refund_internal(conn, refund_id)
finally:
conn.close()
def list_refunds(
self, tenant_id: str, status: str | None = None, limit: int = 100, offset: int = 0
) -> list[Refund]:
"""列出退款记录"""
conn = self._get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM refunds WHERE tenant_id = ?"
params = [tenant_id]
if status:
query += " AND status = ?"
params.append(status)
query += " ORDER BY created_at DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
cursor.execute(query, params)
rows = cursor.fetchall()
return [self._row_to_refund(row) for row in rows]
finally:
conn.close()
def _get_refund_internal(self, conn: sqlite3.Connection, refund_id: str) -> Refund | None:
"""内部方法:获取退款记录"""
cursor = conn.cursor()
cursor.execute("SELECT * FROM refunds WHERE id = ?", (refund_id,))
row = cursor.fetchone()
if row:
return self._row_to_refund(row)
return None
# ==================== 账单历史 ====================
def _add_billing_history_internal(
self,
conn: sqlite3.Connection,
tenant_id: str,
type: str,
amount: float,
currency: str,
description: str,
reference_id: str,
balance_after: float,
):
"""内部方法:添加账单历史"""
history_id = str(uuid.uuid4())
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO billing_history
(id, tenant_id, type, amount, currency, description, reference_id, balance_after, created_at, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
history_id,
tenant_id,
type,
amount,
currency,
description,
reference_id,
balance_after,
datetime.now(),
json.dumps({}),
),
)
def get_billing_history(
self,
tenant_id: str,
start_date: datetime | None = None,
end_date: datetime | None = None,
limit: int = 100,
offset: int = 0,
) -> list[BillingHistory]:
"""获取账单历史"""
conn = self._get_connection()
try:
cursor = conn.cursor()
query = "SELECT * FROM billing_history WHERE tenant_id = ?"
params = [tenant_id]
if start_date:
query += " AND created_at >= ?"
params.append(start_date)
if end_date:
query += " AND created_at <= ?"
params.append(end_date)
query += " ORDER BY created_at DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
cursor.execute(query, params)
rows = cursor.fetchall()
return [self._row_to_billing_history(row) for row in rows]
finally:
conn.close()
# ==================== 支付提供商集成 ====================
def create_stripe_checkout_session(
self,
tenant_id: str,
plan_id: str,
success_url: str,
cancel_url: str,
billing_cycle: str = "monthly",
) -> dict[str, Any]:
"""创建 Stripe Checkout 会话(占位实现)"""
# 这里应该集成 Stripe SDK
# 简化实现,返回模拟数据
return {
"session_id": f"cs_{uuid.uuid4().hex[:24]}",
"url": f"https://checkout.stripe.com/mock/{uuid.uuid4().hex[:24]}",
"status": "created",
"provider": "stripe",
}
def create_alipay_order(
self, tenant_id: str, plan_id: str, billing_cycle: str = "monthly"
) -> dict[str, Any]:
"""创建支付宝订单(占位实现)"""
# 这里应该集成支付宝 SDK
plan = self.get_plan(plan_id)
amount = plan.price_yearly if billing_cycle == "yearly" else plan.price_monthly
return {
"order_id": f"ALI{datetime.now().strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:8].upper()}",
"amount": amount,
"currency": plan.currency,
"qr_code_url": f"https://qr.alipay.com/mock/{uuid.uuid4().hex[:16]}",
"status": "pending",
"provider": "alipay",
}
def create_wechat_order(
self, tenant_id: str, plan_id: str, billing_cycle: str = "monthly"
) -> dict[str, Any]:
"""创建微信支付订单(占位实现)"""
# 这里应该集成微信支付 SDK
plan = self.get_plan(plan_id)
amount = plan.price_yearly if billing_cycle == "yearly" else plan.price_monthly
return {
"order_id": f"WX{datetime.now().strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:8].upper()}",
"amount": amount,
"currency": plan.currency,
"prepay_id": f"wx{uuid.uuid4().hex[:32]}",
"status": "pending",
"provider": "wechat",
}
def handle_webhook(self, provider: str, payload: dict[str, Any]) -> bool:
"""处理支付提供商的 Webhook占位实现"""
# 这里应该实现实际的 Webhook 处理逻辑
logger.info(f"Received webhook from {provider}: {payload.get('event_type', 'unknown')}")
event_type = payload.get("event_type", "")
if provider == "stripe":
if event_type == "checkout.session.completed":
# 处理支付完成
pass
elif event_type == "invoice.payment_failed":
# 处理支付失败
pass
elif provider in ["alipay", "wechat"]:
if payload.get("trade_status") == "TRADE_SUCCESS":
# 处理支付完成
pass
return True
# ==================== 辅助方法 ====================
def _row_to_plan(self, row: sqlite3.Row) -> SubscriptionPlan:
"""数据库行转换为 SubscriptionPlan 对象"""
return SubscriptionPlan(
id=row["id"],
name=row["name"],
tier=row["tier"],
description=row["description"] or "",
price_monthly=row["price_monthly"],
price_yearly=row["price_yearly"],
currency=row["currency"],
features=json.loads(row["features"] or "[]"),
limits=json.loads(row["limits"] or "{}"),
is_active=bool(row["is_active"]),
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"]
),
metadata=json.loads(row["metadata"] or "{}"),
)
def _row_to_subscription(self, row: sqlite3.Row) -> Subscription:
"""数据库行转换为 Subscription 对象"""
return Subscription(
id=row["id"],
tenant_id=row["tenant_id"],
plan_id=row["plan_id"],
status=row["status"],
current_period_start=(
datetime.fromisoformat(row["current_period_start"])
if row["current_period_start"] and isinstance(row["current_period_start"], str)
else row["current_period_start"]
),
current_period_end=(
datetime.fromisoformat(row["current_period_end"])
if row["current_period_end"] and isinstance(row["current_period_end"], str)
else row["current_period_end"]
),
cancel_at_period_end=bool(row["cancel_at_period_end"]),
canceled_at=(
datetime.fromisoformat(row["canceled_at"])
if row["canceled_at"] and isinstance(row["canceled_at"], str)
else row["canceled_at"]
),
trial_start=(
datetime.fromisoformat(row["trial_start"])
if row["trial_start"] and isinstance(row["trial_start"], str)
else row["trial_start"]
),
trial_end=(
datetime.fromisoformat(row["trial_end"])
if row["trial_end"] and isinstance(row["trial_end"], str)
else row["trial_end"]
),
payment_provider=row["payment_provider"],
provider_subscription_id=row["provider_subscription_id"],
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"]
),
metadata=json.loads(row["metadata"] or "{}"),
)
def _row_to_usage(self, row: sqlite3.Row) -> UsageRecord:
"""数据库行转换为 UsageRecord 对象"""
return UsageRecord(
id=row["id"],
tenant_id=row["tenant_id"],
resource_type=row["resource_type"],
quantity=row["quantity"],
unit=row["unit"],
recorded_at=(
datetime.fromisoformat(row["recorded_at"])
if isinstance(row["recorded_at"], str)
else row["recorded_at"]
),
cost=row["cost"],
description=row["description"],
metadata=json.loads(row["metadata"] or "{}"),
)
def _row_to_payment(self, row: sqlite3.Row) -> Payment:
"""数据库行转换为 Payment 对象"""
return Payment(
id=row["id"],
tenant_id=row["tenant_id"],
subscription_id=row["subscription_id"],
invoice_id=row["invoice_id"],
amount=row["amount"],
currency=row["currency"],
provider=row["provider"],
provider_payment_id=row["provider_payment_id"],
status=row["status"],
payment_method=row["payment_method"],
payment_details=json.loads(row["payment_details"] or "{}"),
paid_at=(
datetime.fromisoformat(row["paid_at"])
if row["paid_at"] and isinstance(row["paid_at"], str)
else row["paid_at"]
),
failed_at=(
datetime.fromisoformat(row["failed_at"])
if row["failed_at"] and isinstance(row["failed_at"], str)
else row["failed_at"]
),
failure_reason=row["failure_reason"],
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_invoice(self, row: sqlite3.Row) -> Invoice:
"""数据库行转换为 Invoice 对象"""
return Invoice(
id=row["id"],
tenant_id=row["tenant_id"],
subscription_id=row["subscription_id"],
invoice_number=row["invoice_number"],
status=row["status"],
amount_due=row["amount_due"],
amount_paid=row["amount_paid"],
currency=row["currency"],
period_start=(
datetime.fromisoformat(row["period_start"])
if row["period_start"] and isinstance(row["period_start"], str)
else row["period_start"]
),
period_end=(
datetime.fromisoformat(row["period_end"])
if row["period_end"] and isinstance(row["period_end"], str)
else row["period_end"]
),
description=row["description"],
line_items=json.loads(row["line_items"] or "[]"),
due_date=(
datetime.fromisoformat(row["due_date"])
if row["due_date"] and isinstance(row["due_date"], str)
else row["due_date"]
),
paid_at=(
datetime.fromisoformat(row["paid_at"])
if row["paid_at"] and isinstance(row["paid_at"], str)
else row["paid_at"]
),
voided_at=(
datetime.fromisoformat(row["voided_at"])
if row["voided_at"] and isinstance(row["voided_at"], str)
else row["voided_at"]
),
void_reason=row["void_reason"],
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_refund(self, row: sqlite3.Row) -> Refund:
"""数据库行转换为 Refund 对象"""
return Refund(
id=row["id"],
tenant_id=row["tenant_id"],
payment_id=row["payment_id"],
invoice_id=row["invoice_id"],
amount=row["amount"],
currency=row["currency"],
reason=row["reason"],
status=row["status"],
requested_by=row["requested_by"],
requested_at=(
datetime.fromisoformat(row["requested_at"])
if isinstance(row["requested_at"], str)
else row["requested_at"]
),
approved_by=row["approved_by"],
approved_at=(
datetime.fromisoformat(row["approved_at"])
if row["approved_at"] and isinstance(row["approved_at"], str)
else row["approved_at"]
),
completed_at=(
datetime.fromisoformat(row["completed_at"])
if row["completed_at"] and isinstance(row["completed_at"], str)
else row["completed_at"]
),
provider_refund_id=row["provider_refund_id"],
metadata=json.loads(row["metadata"] 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_billing_history(self, row: sqlite3.Row) -> BillingHistory:
"""数据库行转换为 BillingHistory 对象"""
return BillingHistory(
id=row["id"],
tenant_id=row["tenant_id"],
type=row["type"],
amount=row["amount"],
currency=row["currency"],
description=row["description"],
reference_id=row["reference_id"],
balance_after=row["balance_after"],
created_at=(
datetime.fromisoformat(row["created_at"])
if isinstance(row["created_at"], str)
else row["created_at"]
),
metadata=json.loads(row["metadata"] or "{}"),
)
# 全局订阅管理器实例
subscription_manager = None
def get_subscription_manager(db_path: str = "insightflow.db") -> SubscriptionManager:
"""获取订阅管理器实例(单例模式)"""
global subscription_manager
if subscription_manager is None:
subscription_manager = SubscriptionManager(db_path)
return subscription_manager