OPC-Hub/server-core
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sms): 迁移 PineSound 阿里云短信+Redis缓存+真实随机数；强制MySQL禁用SQLite；连接池取消请求shield修复

Browse Source
This commit is contained in:
Pine
2026-09-08 00:10:16 +08:00
parent 539282f18f
commit ff234f7071
9 changed files with 262 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/dependencies.py
+14 -1
View File
@@ -6,6 +6,9 @@
"""
from __future__ import annotations
import asyncio
import contextlib
from fastapi import Depends, HTTPException, Request
from .. import config
@@ -34,7 +37,17 @@ async def get_db():
try:
yield db
finally:
await db.close()
try:
await db.close()
except asyncio.CancelledError:
# 请求任务已被取消（客户端断开 / 超时 / 服务关闭）：取消上下文中
# 的 await 会立即抛 CancelledError，SQLAlchemy 的 rollback 会被
# asyncmy 打断并报 "Cancelled during execution"，导致 session 关不掉、
# 连接无法归还连接池（随后被 GC 回收时产生 non-checked-in 警告）。
# 用 shield 把 close 移到独立任务执行，确保连接一定归还。
with contextlib.suppress(BaseException):
await asyncio.shield(db.close())
raise
async def _compute_capabilities(db: Database, user: dict) -> list[str]:
app/api/routers/auth.py
+3 -3
View File
@@ -479,7 +479,7 @@ async def phone_login(req: PhoneLoginRequest, request: Request, db: Database = D
if not config.AUTH_ENABLED:
raise HTTPException(status_code=403, detail="认证未开启")
try:
sms.verify(req.phone, req.code)
await sms.verify(req.phone, req.code)
except sms.SmsError as exc:
raise HTTPException(status_code=401, detail=str(exc))
@@ -559,7 +559,7 @@ async def wx_phone(req: WxPhoneRequest, db: Database = Depends(get_db)):
if not config.AUTH_ENABLED:
raise HTTPException(status_code=403, detail="认证未开启")
try:
sms.verify(req.phone, req.code)
await sms.verify(req.phone, req.code)
except sms.SmsError as exc:
raise HTTPException(status_code=401, detail=str(exc))
@@ -614,7 +614,7 @@ async def bind_phone(
if not _is_phone(req.phone):
raise HTTPException(status_code=400, detail="手机号需为 11 位（1 开头）")
try:
sms.verify(req.phone, req.code)
await sms.verify(req.phone, req.code)
except sms.SmsError as exc:
raise HTTPException(status_code=401, detail=str(exc))
merged = await _bind_phone_merge(db, user, req.phone)
app/config.py
+5 -2
View File
@@ -42,7 +42,7 @@ UPLOADS_DIR = SERVERDATA_DIR / "uploads"
for _d in (DATA_DIR, LOG_DIR, FILES_DIR, KEYS_DIR, PROMPTS_DIR, UPLOADS_DIR):
_d.mkdir(parents=True, exist_ok=True)
# 数据库连接（异步）。开发默认 SQLite+aiosqlite；生产用 MySQL+asyncmy（环境变量覆盖）。
# 数据库连接（异步）。强制 MySQL+asyncmy；未配置时启动失败，绝不回退 SQLite。
# 支持两种配置方式：
# 1. PINEAGENTS_DEMO_DATABASE_URL 完整连接串（优先级最高）
# 2. PINEAGENTS_MYSQL_HOST/PORT/USER/PASSWORD/DB 分字段配置（与 .env 对齐）
@@ -58,7 +58,10 @@ elif _mysql_host:
_auth = f"{_mysql_user}:{_mysql_password}@" if _mysql_user else ""
DATABASE_URL = f"mysql+asyncmy://{_auth}{_mysql_host}:{_mysql_port}/{_mysql_db}?charset=utf8mb4"
else:
DATABASE_URL = f"sqlite+aiosqlite:///{DATA_DIR / 'app.db'}"
raise RuntimeError(
"数据库未配置！必须在 .env 中设置 PINEAGENTS_DEMO_DATABASE_URL 或 "
"PINEAGENTS_MYSQL_HOST 等 MySQL 连接参数，禁止使用 SQLite。"
)
# Redis（缓存/会话/限流/分布式锁），空则降级内存。
# 支持两种配置方式：
app/infrastructure/seed.py
+14 -10
View File
@@ -235,20 +235,24 @@ _DEFAULT_SYSTEM_CONFIGS: list[tuple[str, str, str]] = [
("platform.fee.task", "0.05", "任务佣金费率（5%）"),
("platform.fee.provider", "0.15", "服务商分成比例（15%）"),
# ── 短信生态配置（admin「系统配置 → 短信生态」在线维护） ──────────────
("sms.provider", "stub", "短信 provider（stub|aliyun）"),
# 阿里云凭据/签名与 PineSoundServer .env 完全一致（ACCESS_KEY_ID /
# ACCESS_KEY_SECRET / REGION_ID=cn-qingdao / SIGN_NAME=派音人工智能）；
# 模板仅迁移本项目需要的 PineSound SMSTemplate：T1 登录（SMS_510980133，
# 变量 code），register 复用 T1（本项目注册=手机号登录一体）。
("sms.provider", "aliyun", "短信 provider（stub|aliyun）"),
("sms.aliyun",
'{"access_key_id":"","access_key_secret":"","sign_name":"","region_id":"cn-hangzhou","endpoint":"dysmsapi.aliyuncs.com"}',
"阿里云短信总配置（密钥/签名/区域）"),
("sms.template.login", '{"template_id":"","variables":["code","minute"]}',
"登录验证码模板（变量：code 验证码, minute 有效分钟）"),
("sms.template.register", '{"template_id":"","variables":["code","minute"]}',
"注册验证码模板（变量：code 验证码, minute 有效分钟）"),
'{"access_key_id":"LTAI5t6ZU2Nf8qbGpEpn7bzN","access_key_secret":"lnNkxiXvNVsCPdc7hVZSXz6goOz8YJ","sign_name":"派音人工智能","region_id":"cn-qingdao","endpoint":"dysmsapi.aliyuncs.com"}',
"阿里云短信总配置（与 PineSoundServer .env 一致）"),
("sms.template.login", '{"template_id":"SMS_510980133","variables":["code"]}',
"登录验证码模板（PineSound T1：验证码 code，5分钟有效，请勿泄露）"),
("sms.template.register", '{"template_id":"SMS_510980133","variables":["code"]}',
"注册验证码模板（本项目注册=手机号登录一体，复用 T1）"),
("sms.template.notification", '{"template_id":"","variables":[]}',
"通知提醒模板（变量按业务定义，如审核结果等）"),
"通知提醒模板（PineSound 无匹配模板，暂留空；变量按业务定义）"),
("sms.template.order_status", '{"template_id":"","variables":["order_no","status"]}',
"订单状态变动模板（变量：order_no 订单号, status 状态）"),
"订单状态变动模板（PineSound 无匹配模板，暂留空；变量：order_no 订单号, status 状态）"),
("sms.template.appointment", '{"template_id":"","variables":["service_name","schedule_at","contact"]}',
"服务预约模板（变量：service_name 服务名, schedule_at 预约时间, contact 联系方式）"),
"服务预约模板（PineSound 无匹配模板，暂留空；变量：service_name 服务名, schedule_at 预约时间, contact 联系方式）"),
]
app/services/sms.py
+79 -40
View File
@@ -18,10 +18,12 @@
场景（scene）：login 登录 / register 注册 / notification 通知 /
order_status 订单状态变动 / appointment 服务预约
验证码存储为内存（带 TTL 与尝试次数），进程重启即失效——生产应换 Redis。
验证码存储优先使用 Redis（``sms:code:{phone}``，带 TTL 与尝试次数，跨进程
共享、重启不丢）；Redis 未连接时降级内存（仅单进程联调可用）。
模块保留无 db 依赖的验证码核心（issue/verify）；配置由调用方从
``system_configs`` 解析后传入（``load_sms_config(db)`` / ``parse_sms_config``）。
**随机数策略**：后端 provider 为 ``aliyun`` 时**强制真实随机 6 位码**，绝不允许
使用演示码 123456；仅 provider 为 ``stub``（未配置阿里云）时才使用演示码
123456 便于联调。
"""
from __future__ import annotations
@@ -38,11 +40,14 @@ from dataclasses import dataclass, field
import httpx
from .. import config
from ..infrastructure.cache import cache as _redis_cache
# ── 验证码存储（内存） ──────────────────────────────────────────────────────
# ── 验证码存储 ─────────────────────────────────────────────────────────────
# phone -> {"code": str, "expires_at": float, "attempts": int}
# 优先 Redis（key: sms:code:{phone}，EX=_CODE_TTL）；Redis 不可用降级内存。
_STORE: dict[str, dict] = {}
_LOCK = threading.Lock()
_STORE_PREFIX = "sms:code:"
_CODE_TTL = config.SMS_CODE_TTL_SECONDS
_MAX_ATTEMPTS = config.SMS_RATE_LIMIT
@@ -134,49 +139,73 @@ async def load_sms_config(db) -> SmsConfig:
return parse_sms_config(rows)
# ── 验证码核心（无 db 依赖） ───────────────────────────────────────────────
# ── 验证码核心（Redis 优先，降级内存） ─────────────────────────────────────
def issue(phone: str) -> str:
async def _store_save(phone: str, rec: dict) -> None:
"""写入验证码记录（Redis EX TTL 自动过期；不可用时降级内存）。"""
if _redis_cache.available:
await _redis_cache.set_json(f"{_STORE_PREFIX}{phone}", rec, ttl=_CODE_TTL)
return
with _LOCK:
_STORE[phone] = rec
async def _store_load(phone: str) -> dict | None:
if _redis_cache.available:
return await _redis_cache.get_json(f"{_STORE_PREFIX}{phone}")
with _LOCK:
return _STORE.get(phone)
async def _store_delete(phone: str) -> None:
if _redis_cache.available:
await _redis_cache.delete(f"{_STORE_PREFIX}{phone}")
return
with _LOCK:
_STORE.pop(phone, None)
async def issue(phone: str) -> str:
"""为手机号生成 6 位验证码并写入存储，返回明文。
真实发送由调用方依据场景模板调用 :func:`send_template` 完成；
若 provider 为 stub 且未指定场景模板，也可直接调用 :func:`send_code`。
随机数策略与 :func:`_pick_code` 一致：仅系统级 stub 配置（env
``PINEAGENTS_SMS_PROVIDER=stub``）使用演示码 123456；其余一律真实随机。
真实发送由调用方依据场景模板调用 :func:`send_template` 完成。
"""
code = DEMO_CODE if config.SMS_PROVIDER == "stub" else f"{secrets.randbelow(1_000_000):06d}"
with _LOCK:
_STORE[phone] = {
"code": code,
"expires_at": time.monotonic() + _CODE_TTL,
"attempts": 0,
}
await _store_save(phone, {
"code": code,
"expires_at": time.monotonic() + _CODE_TTL,
"attempts": 0,
})
return code
def verify(phone: str, code: str) -> bool:
async def verify(phone: str, code: str) -> bool:
"""校验验证码；成功即作废，失败累计次数超限则不可再试。"""
with _LOCK:
rec = _STORE.get(phone)
if not rec:
raise SmsError("verification code not issued")
if time.monotonic() > rec["expires_at"]:
_STORE.pop(phone, None)
raise SmsError("verification code expired")
if rec["attempts"] >= _MAX_ATTEMPTS:
raise SmsError("too many attempts, request a new code")
if rec["code"] != code.strip():
rec["attempts"] += 1
raise SmsError("invalid verification code")
_STORE.pop(phone, None)
return True
rec = await _store_load(phone)
if not rec:
raise SmsError("verification code not issued")
if time.monotonic() > rec["expires_at"]:
await _store_delete(phone)
raise SmsError("verification code expired")
if rec["attempts"] >= _MAX_ATTEMPTS:
raise SmsError("too many attempts, request a new code")
if rec["code"] != code.strip():
rec["attempts"] += 1
await _store_save(phone, rec)
raise SmsError("invalid verification code")
await _store_delete(phone)
return True
# 演示验证码：stub provider 统一用固定码（先跑通流程，生产换真实短信）。
# 演示验证码：仅 stub provider（未配置阿里云）联调用；aliyun 绝不允许使用。
DEMO_CODE = "123456"
def _pick_code(cfg: SmsConfig) -> str:
"""按 provider 决定验证码：stub 固定码，aliyun 随机码。"""
if cfg.provider == "stub" or config.SMS_PROVIDER == "stub":
"""按 provider 决定验证码：仅 stub 用固定码 123456，其余（含 aliyun）真实随机。"""
if cfg.provider == "stub":
return DEMO_CODE
return f"{secrets.randbelow(1_000_000):06d}"
@@ -190,15 +219,25 @@ async def send_code(cfg: SmsConfig, phone: str, scene: str = "login",
"""
scene = scene if scene in SMS_SCENES else "login"
code = _pick_code(cfg)
with _LOCK:
_STORE[phone] = {
"code": code,
"expires_at": time.monotonic() + _CODE_TTL,
"attempts": 0,
}
params = {"code": code, "minute": str(_CODE_TTL // 60)}
await _store_save(phone, {
"code": code,
"expires_at": time.monotonic() + _CODE_TTL,
"attempts": 0,
})
# 只传模板声明过的变量：阿里云对模板未声明的变量名会报
# isv.TEMPLATE_PARAMS_ILLEGAL / 变量不匹配。模板未配置时仍传 code
# （stub 打印用），保证联调可见。
tpl = cfg.template(scene)
declared = set(tpl.variables) if tpl else set()
params: dict[str, str] = {}
if not declared or "code" in declared:
params["code"] = code
if "minute" in declared:
params["minute"] = str(_CODE_TTL // 60)
if extra:
params.update(extra)
for k, v in extra.items():
if k in declared:
params[k] = str(v)
await send_template(cfg, scene, phone, params)
return code
app/training/main.py
+1 -1
View File
@@ -484,7 +484,7 @@ async def bind_phone(req: Request, authorization: str = Header(default="")):
code = str(b.get("code", "")).strip()
if not PHONE_RE.match(phone):
raise HTTPException(400, "请输入正确的 11 位手机号")
if not platform_sms.verify(phone, code):
if not await platform_sms.verify(phone, code):
raise HTTPException(401, "验证码错误或已过期")
u = await _current_user_async(payload)
if not u:
scripts/db/set_sms_config.py
+104
View File
@@ -0,0 +1,104 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""短信生态配置写入脚本（迁移自 PineSoundServer .env）。
把阿里云凭据/签名/区域与 PineSound 短信模板（仅本项目需要的 T1 登录）写入
system_configs 表：
- sms.provider -> aliyun
- sms.aliyun -> 凭据/签名/区域（与 PineSoundServer .env 完全一致：
ACCESS_KEY_ID / ACCESS_KEY_SECRET / REGION_ID=cn-qingdao /
SIGN_NAME=派音人工智能）
- sms.template.login -> SMS_510980133（PineSound T1 登录验证码，变量 code）
- sms.template.register -> SMS_510980133（本项目注册=手机号登录一体，复用 T1）
PineSound 其余模板（修改密码/绑定手机/找回密码/注销/充值/内测）本项目当前业务
未使用，不迁移；notification / order_status / appointment 场景 PineSound 无匹配
模板，保持留空（stub 打印，不阻断业务）。
用法：
python scripts/db/set_sms_config.py # 实际执行
python scripts/db/set_sms_config.py --dry-run # 只预览不写库
"""
from __future__ import annotations
import argparse
import asyncio
from pathlib import Path
from dotenv import load_dotenv
load_dotenv(Path(__file__).resolve().parent.parent.parent / ".env")
from sqlalchemy import text # noqa: E402
from app.infrastructure.db import AsyncSessionLocal # noqa: E402
from app.infrastructure.repositories import utcnow_iso # noqa: E402
# 与 PineSoundServer .env 完全一致的阿里云短信配置
ALIYUN_CONFIG = {
"access_key_id": "LTAI5t6ZU2Nf8qbGpEpn7bzN",
"access_key_secret": "lnNkxiXvNVsCPdc7hVZSXz6goOz8YJ",
"sign_name": "派音人工智能",
"region_id": "cn-qingdao",
"endpoint": "dysmsapi.aliyuncs.com",
}
# key -> (value, description)；幂等覆盖：已存在的键也更新为 PineSound 配置。
SMS_CONFIGS: list[tuple[str, str, str]] = [
("sms.provider", "aliyun", "短信 provider（stub|aliyun）"),
("sms.aliyun",
__import__("json").dumps(ALIYUN_CONFIG, ensure_ascii=False),
"阿里云短信总配置（与 PineSoundServer .env 一致）"),
("sms.template.login",
'{"template_id":"SMS_510980133","variables":["code"]}',
"登录验证码模板（PineSound T1：验证码 code，5分钟有效，请勿泄露）"),
("sms.template.register",
'{"template_id":"SMS_510980133","variables":["code"]}',
"注册验证码模板（本项目注册=手机号登录一体，复用 T1）"),
("sms.template.notification",
'{"template_id":"","variables":[]}',
"通知提醒模板（PineSound 无匹配模板，暂留空）"),
("sms.template.order_status",
'{"template_id":"","variables":["order_no","status"]}',
"订单状态变动模板（PineSound 无匹配模板，暂留空）"),
("sms.template.appointment",
'{"template_id":"","variables":["service_name","schedule_at","contact"]}',
"服务预约模板（PineSound 无匹配模板，暂留空）"),
]
async def main(dry_run: bool) -> int:
now = utcnow_iso()
async with AsyncSessionLocal() as session:
for key, value, description in SMS_CONFIGS:
exists = (
await session.execute(
text("SELECT `key` FROM system_configs WHERE `key` = :k"),
{"k": key},
)
).scalar_one_or_none()
if exists:
sql = text(
"UPDATE system_configs SET `value` = :v, description = :d, "
"updated_at = :now WHERE `key` = :k"
)
else:
sql = text(
"INSERT INTO system_configs (`key`, `value`, `description`, "
"updated_at) VALUES (:k, :v, :d, :now)"
)
if dry_run:
print(f"[dry-run] {'UPDATE' if exists else 'INSERT'} {key} = {value}")
continue
await session.execute(sql, {"k": key, "v": value, "d": description, "now": now})
if not dry_run:
await session.commit()
return 0
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="写入短信生态配置（PineSound 迁移）")
parser.add_argument("--dry-run", action="store_true", help="只预览不写库")
args = parser.parse_args()
raise SystemExit(asyncio.run(main(args.dry_run)))
serverrun/mysql/docker-compose.yml
+2 -2
View File
@@ -1,4 +1,4 @@
# MySQL 服务（业务主库，后续启用；先建编排）
# MySQL 服务（业务主库 + compute 库，首次启动由 init.sql 自动建库）
services:
mysql:
image: mysql:8
@@ -6,7 +6,6 @@ services:
restart: always
environment:
MYSQL_ROOT_PASSWORD: sjnxhyjashaywiuwhaja
MYSQL_DATABASE: opc
MYSQL_USER: opc
MYSQL_PASSWORD: jjjsgysyujkwjwgb
command: --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci
@@ -14,6 +13,7 @@ services:
- "8091:3306"
volumes:
- ../../serverdata/mysql-data:/var/lib/mysql
- ./init.sql:/docker-entrypoint-initdb.d/init.sql:ro
networks:
- opc-network
serverrun/mysql/init.sql
+40
View File
@@ -0,0 +1,40 @@
-- MySQL 首次启动初始化：建两个业务库 + 运营端超级管理员
-- 仅在容器数据目录为空时执行一次（/docker-entrypoint-initdb.d/ 机制）
CREATE DATABASE IF NOT EXISTS opc CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
CREATE DATABASE IF NOT EXISTS opc_compute CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
GRANT ALL PRIVILEGES ON opc.* TO 'opc'@'%';
GRANT ALL PRIVILEGES ON opc_compute.* TO 'opc'@'%';
FLUSH PRIVILEGES;
-- 运营端超级管理员：手机号 17637177199（短信验证码登录，无密码），昵称 Pine
INSERT INTO opc.users
(id, username, nickname, password_hash, password_salt,
wx_openid, wx_unionid, wx_mini_openid, phone, email,
account, company, room, avatar, company_avatar,
gender, birthday, id_card, ethnicity, grad_school_major, grad_time,
role, sub_role, status, token_version, source, auth_type,
register_ip, last_login_ip, last_login_at,
compute_provisioned, compute_username, compute_quota, compute_used_quota,
certification_status, certification_time, affiliation, park_id, park_name,
account_type, opc_status, topics, created_at, updated_at)
VALUES
('u_pine_admin', '17637177199', 'Pine',
'', '',
'', '', '', '17637177199', '',
'', '', '', '', '',
'unknown', '', '', '', '', '',
'operator', 'op_super_admin', 'active', 0, 'seed', 'phone',
'', '', '',
0, '17637177199', 0, 0,
'', '', '', '', '',
'', '', '', '', '')
ON DUPLICATE KEY UPDATE
phone = '17637177199',
role = 'operator',
sub_role = 'op_super_admin',
status = 'active',
nickname = VALUES(nickname),
compute_username = VALUES(compute_username);