359 lines
13 KiB
Python
359 lines
13 KiB
Python
"""系统功能配置（数据库存储 + 内存缓存）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from dataclasses import asdict, dataclass, fields
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from models.models import AppConfig
|
||
|
||
CONFIG_ROW_ID = 1
|
||
PASSWORD_PLACEHOLDER = "******"
|
||
APP_NAME = "抖音回复助手"
|
||
|
||
DEFAULT_EMAIL_VERIFY_SUBJECT = f"【{APP_NAME}】请验证您的邮箱"
|
||
DEFAULT_EMAIL_VERIFY_BODY = (
|
||
"您好 {username}，\n\n"
|
||
"感谢注册{app_name}。请点击以下链接完成邮箱验证：\n"
|
||
"{link}\n\n"
|
||
"如非本人操作，请忽略此邮件。\n"
|
||
)
|
||
DEFAULT_EMAIL_VERIFY_HTML = (
|
||
"<p>您好 <strong>{username}</strong>，</p>"
|
||
"<p>感谢注册{app_name}。请点击下方按钮完成邮箱验证：</p>"
|
||
'<p><a href="{link}" style="display:inline-block;padding:10px 18px;'
|
||
'background:#aa3bff;color:#fff;text-decoration:none;border-radius:6px;">'
|
||
"验证邮箱</a></p>"
|
||
'<p>或复制链接到浏览器：<br><a href="{link}">{link}</a></p>'
|
||
'<p style="color:#888;font-size:12px;">如非本人操作，请忽略此邮件。</p>'
|
||
)
|
||
DEFAULT_EMAIL_TEST_SUBJECT = f"【{APP_NAME}】SMTP 测试邮件"
|
||
DEFAULT_EMAIL_TEST_BODY = "这是一封 SMTP 配置测试邮件。若您收到此邮件，说明邮件服务已配置正确。"
|
||
DEFAULT_EMAIL_TEST_HTML = (
|
||
"<p>这是一封 <strong>SMTP 配置测试</strong> 邮件。</p>"
|
||
"<p>若您收到此邮件，说明邮件服务已配置正确。</p>"
|
||
)
|
||
DEFAULT_PASSWORD_RESET_SUBJECT = f"【{APP_NAME}】重置您的登录密码"
|
||
DEFAULT_PASSWORD_RESET_BODY = (
|
||
"您好 {username}，\n\n"
|
||
"我们收到了重置 {app_name} 账号密码的请求。请点击以下链接设置新密码：\n"
|
||
"{link}\n\n"
|
||
"链接有效期 {expire_hours} 小时。如非本人操作，请忽略此邮件。\n"
|
||
)
|
||
DEFAULT_PASSWORD_RESET_HTML = (
|
||
"<p>您好 <strong>{username}</strong>，</p>"
|
||
"<p>我们收到了重置 {app_name} 账号密码的请求。请点击下方按钮设置新密码：</p>"
|
||
'<p><a href="{link}" style="display:inline-block;padding:10px 18px;'
|
||
'background:#aa3bff;color:#fff;text-decoration:none;border-radius:6px;">'
|
||
"重置密码</a></p>"
|
||
'<p>或复制链接到浏览器：<br><a href="{link}">{link}</a></p>'
|
||
'<p style="color:#888;font-size:12px;">链接有效期 {expire_hours} 小时。如非本人操作，请忽略此邮件。</p>'
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class SystemSettingsData:
|
||
registration_enabled: bool = True
|
||
email_verification_required: bool = True
|
||
email_binding_required: bool = False
|
||
email_verify_token_hours: int = 24
|
||
auto_reply_cooldown_seconds: int = 60
|
||
app_url: str = "http://localhost:8800"
|
||
smtp_host: str = ""
|
||
smtp_port: int = 587
|
||
smtp_user: str = ""
|
||
smtp_password: str = ""
|
||
smtp_from: str = ""
|
||
smtp_use_tls: bool = True
|
||
smtp_use_ssl: bool = False
|
||
debug_show_verify_link: bool = False
|
||
default_register_max_accounts: int = 3
|
||
payment_enabled: bool = False
|
||
payment_demo_mode: bool = True
|
||
wechat_pay_enabled: bool = False
|
||
alipay_pay_enabled: bool = False
|
||
account_slot_unit_price: float = 9.9
|
||
account_slot_purchase_min: int = 1
|
||
account_slot_purchase_max: int = 20
|
||
wechat_app_id: str = ""
|
||
wechat_mch_id: str = ""
|
||
wechat_api_v3_key: str = ""
|
||
wechat_cert_serial: str = ""
|
||
wechat_private_key: str = ""
|
||
alipay_app_id: str = ""
|
||
alipay_private_key: str = ""
|
||
alipay_public_key: str = ""
|
||
alipay_sandbox: bool = False
|
||
email_verify_subject: str = DEFAULT_EMAIL_VERIFY_SUBJECT
|
||
email_verify_body: str = DEFAULT_EMAIL_VERIFY_BODY
|
||
email_verify_html: str = DEFAULT_EMAIL_VERIFY_HTML
|
||
email_test_subject: str = DEFAULT_EMAIL_TEST_SUBJECT
|
||
email_test_body: str = DEFAULT_EMAIL_TEST_BODY
|
||
email_test_html: str = DEFAULT_EMAIL_TEST_HTML
|
||
|
||
def smtp_configured(self) -> bool:
|
||
sender = (self.smtp_from or self.smtp_user or "").strip()
|
||
return bool(self.smtp_host.strip() and sender)
|
||
|
||
def app_url_normalized(self) -> str:
|
||
return (self.app_url or "http://localhost:8800").rstrip("/")
|
||
|
||
def wechat_pay_configured(self) -> bool:
|
||
return bool(
|
||
self.wechat_app_id.strip()
|
||
and self.wechat_mch_id.strip()
|
||
and self.wechat_api_v3_key.strip()
|
||
and self.wechat_cert_serial.strip()
|
||
and self.wechat_private_key.strip()
|
||
)
|
||
|
||
def alipay_configured(self) -> bool:
|
||
return bool(
|
||
self.alipay_app_id.strip()
|
||
and self.alipay_private_key.strip()
|
||
and self.alipay_public_key.strip()
|
||
)
|
||
|
||
def payment_channel_available(self, channel: str) -> bool:
|
||
if channel == "wechat":
|
||
return self.wechat_pay_enabled and self.wechat_pay_configured()
|
||
if channel == "alipay":
|
||
return self.alipay_pay_enabled and self.alipay_configured()
|
||
return False
|
||
|
||
def payment_channel_selectable(self, channel: str) -> bool:
|
||
"""用户可选的支付渠道（含演示模式）。"""
|
||
if channel == "wechat":
|
||
if not self.wechat_pay_enabled:
|
||
return False
|
||
return self.wechat_pay_configured() or (
|
||
self.payment_demo_mode and self.payment_enabled
|
||
)
|
||
if channel == "alipay":
|
||
if not self.alipay_pay_enabled:
|
||
return False
|
||
return self.alipay_configured() or (
|
||
self.payment_demo_mode and self.payment_enabled
|
||
)
|
||
return False
|
||
|
||
def available_payment_channels(self) -> list[str]:
|
||
channels = []
|
||
if self.payment_channel_selectable("wechat"):
|
||
channels.append("wechat")
|
||
if self.payment_channel_selectable("alipay"):
|
||
channels.append("alipay")
|
||
return channels
|
||
|
||
|
||
_settings_cache: SystemSettingsData | None = None
|
||
|
||
|
||
def _coerce_bool(value: Any, default: bool) -> bool:
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, str):
|
||
return value.lower() in ("1", "true", "yes", "on")
|
||
if value is None:
|
||
return default
|
||
return bool(value)
|
||
|
||
|
||
def _parse_settings(raw: dict[str, Any]) -> SystemSettingsData:
|
||
base = SystemSettingsData()
|
||
allowed = {f.name for f in fields(SystemSettingsData)}
|
||
merged: dict[str, Any] = {}
|
||
for key in allowed:
|
||
if key in raw:
|
||
merged[key] = raw[key]
|
||
if "smtp_port" in merged:
|
||
try:
|
||
merged["smtp_port"] = int(merged["smtp_port"])
|
||
except (TypeError, ValueError):
|
||
merged["smtp_port"] = base.smtp_port
|
||
if "email_verify_token_hours" in merged:
|
||
try:
|
||
merged["email_verify_token_hours"] = max(
|
||
1, min(168, int(merged["email_verify_token_hours"]))
|
||
)
|
||
except (TypeError, ValueError):
|
||
merged["email_verify_token_hours"] = base.email_verify_token_hours
|
||
if "auto_reply_cooldown_seconds" in merged:
|
||
try:
|
||
merged["auto_reply_cooldown_seconds"] = max(
|
||
0, min(86400, int(merged["auto_reply_cooldown_seconds"]))
|
||
)
|
||
except (TypeError, ValueError):
|
||
merged["auto_reply_cooldown_seconds"] = base.auto_reply_cooldown_seconds
|
||
if "default_register_max_accounts" in merged:
|
||
try:
|
||
merged["default_register_max_accounts"] = max(
|
||
0, min(999, int(merged["default_register_max_accounts"]))
|
||
)
|
||
except (TypeError, ValueError):
|
||
merged["default_register_max_accounts"] = base.default_register_max_accounts
|
||
if "account_slot_unit_price" in merged:
|
||
try:
|
||
merged["account_slot_unit_price"] = max(
|
||
0.01, min(99999.0, float(merged["account_slot_unit_price"]))
|
||
)
|
||
except (TypeError, ValueError):
|
||
merged["account_slot_unit_price"] = base.account_slot_unit_price
|
||
for int_key, lo, hi in (
|
||
("account_slot_purchase_min", 1, 100),
|
||
("account_slot_purchase_max", 1, 100),
|
||
):
|
||
if int_key in merged:
|
||
try:
|
||
merged[int_key] = max(lo, min(hi, int(merged[int_key])))
|
||
except (TypeError, ValueError):
|
||
merged[int_key] = getattr(base, int_key)
|
||
for bool_key in (
|
||
"registration_enabled",
|
||
"email_verification_required",
|
||
"email_binding_required",
|
||
"smtp_use_tls",
|
||
"smtp_use_ssl",
|
||
"debug_show_verify_link",
|
||
"payment_enabled",
|
||
"payment_demo_mode",
|
||
"wechat_pay_enabled",
|
||
"alipay_pay_enabled",
|
||
"alipay_sandbox",
|
||
):
|
||
if bool_key in merged:
|
||
merged[bool_key] = _coerce_bool(merged[bool_key], getattr(base, bool_key))
|
||
return SystemSettingsData(**{**asdict(base), **merged})
|
||
|
||
|
||
def get_cached_settings() -> SystemSettingsData:
|
||
global _settings_cache
|
||
if _settings_cache is None:
|
||
_settings_cache = SystemSettingsData()
|
||
return _settings_cache
|
||
|
||
|
||
def set_cached_settings(data: SystemSettingsData) -> None:
|
||
global _settings_cache
|
||
_settings_cache = data
|
||
|
||
|
||
async def load_settings(db: AsyncSession) -> SystemSettingsData:
|
||
result = await db.execute(select(AppConfig).where(AppConfig.id == CONFIG_ROW_ID))
|
||
row = result.scalar_one_or_none()
|
||
if not row or not row.data:
|
||
data = SystemSettingsData()
|
||
set_cached_settings(data)
|
||
return data
|
||
try:
|
||
payload = json.loads(row.data)
|
||
except json.JSONDecodeError:
|
||
payload = {}
|
||
data = _parse_settings(payload if isinstance(payload, dict) else {})
|
||
set_cached_settings(data)
|
||
return data
|
||
|
||
|
||
async def ensure_default_settings(db: AsyncSession) -> SystemSettingsData:
|
||
result = await db.execute(select(AppConfig).where(AppConfig.id == CONFIG_ROW_ID))
|
||
row = result.scalar_one_or_none()
|
||
if row:
|
||
return await load_settings(db)
|
||
data = SystemSettingsData()
|
||
row = AppConfig(
|
||
id=CONFIG_ROW_ID,
|
||
data=json.dumps(asdict(data), ensure_ascii=False),
|
||
updated_at=datetime.utcnow(),
|
||
)
|
||
db.add(row)
|
||
await db.commit()
|
||
set_cached_settings(data)
|
||
return data
|
||
|
||
|
||
async def save_settings(db: AsyncSession, updates: dict[str, Any]) -> SystemSettingsData:
|
||
current = await load_settings(db)
|
||
merged = asdict(current)
|
||
for key, value in updates.items():
|
||
if key not in merged or value is None:
|
||
continue
|
||
if key == "smtp_password":
|
||
pwd = str(value).strip()
|
||
if not pwd or pwd == PASSWORD_PLACEHOLDER:
|
||
continue
|
||
merged[key] = pwd
|
||
continue
|
||
if key in ("wechat_api_v3_key", "wechat_private_key", "alipay_private_key"):
|
||
secret = str(value).strip()
|
||
if not secret or secret == PASSWORD_PLACEHOLDER:
|
||
continue
|
||
merged[key] = secret
|
||
continue
|
||
merged[key] = value
|
||
data = _parse_settings(merged)
|
||
|
||
result = await db.execute(select(AppConfig).where(AppConfig.id == CONFIG_ROW_ID))
|
||
row = result.scalar_one_or_none()
|
||
if not row:
|
||
row = AppConfig(id=CONFIG_ROW_ID, data="{}", updated_at=datetime.utcnow())
|
||
db.add(row)
|
||
row.data = json.dumps(asdict(data), ensure_ascii=False)
|
||
row.updated_at = datetime.utcnow()
|
||
await db.commit()
|
||
set_cached_settings(data)
|
||
return data
|
||
|
||
|
||
def settings_to_public(data: SystemSettingsData) -> dict[str, bool]:
|
||
return {
|
||
"registration_enabled": data.registration_enabled,
|
||
"email_verification_required": data.email_verification_required,
|
||
"email_binding_required": data.email_binding_required,
|
||
}
|
||
|
||
|
||
def settings_to_admin_response(data: SystemSettingsData) -> dict[str, Any]:
|
||
payload = asdict(data)
|
||
payload["smtp_password"] = PASSWORD_PLACEHOLDER if data.smtp_password else ""
|
||
payload["smtp_password_configured"] = bool(data.smtp_password)
|
||
return payload
|
||
|
||
|
||
PAYMENT_SETTING_KEYS = (
|
||
"payment_enabled",
|
||
"payment_demo_mode",
|
||
"wechat_pay_enabled",
|
||
"alipay_pay_enabled",
|
||
"account_slot_unit_price",
|
||
"account_slot_purchase_min",
|
||
"account_slot_purchase_max",
|
||
"wechat_app_id",
|
||
"wechat_mch_id",
|
||
"wechat_api_v3_key",
|
||
"wechat_cert_serial",
|
||
"wechat_private_key",
|
||
"alipay_app_id",
|
||
"alipay_private_key",
|
||
"alipay_public_key",
|
||
"alipay_sandbox",
|
||
)
|
||
|
||
|
||
def settings_to_payment_response(data: SystemSettingsData) -> dict[str, Any]:
|
||
payload = {key: getattr(data, key) for key in PAYMENT_SETTING_KEYS}
|
||
payload["app_url"] = data.app_url
|
||
payload["wechat_api_v3_key"] = PASSWORD_PLACEHOLDER if data.wechat_api_v3_key else ""
|
||
payload["wechat_api_v3_key_configured"] = bool(data.wechat_api_v3_key)
|
||
payload["wechat_private_key"] = PASSWORD_PLACEHOLDER if data.wechat_private_key else ""
|
||
payload["wechat_private_key_configured"] = bool(data.wechat_private_key)
|
||
payload["alipay_private_key"] = PASSWORD_PLACEHOLDER if data.alipay_private_key else ""
|
||
payload["alipay_private_key_configured"] = bool(data.alipay_private_key)
|
||
payload["wechat_pay_configured"] = data.wechat_pay_configured()
|
||
payload["alipay_configured"] = data.alipay_configured()
|
||
return payload
|