xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 15 Pull Requests 16 Actions 34.2k Packages Projects Releases 7 Wiki Activity

style: auto-format with black + isort + prettier [ci skip]

Browse Source
This commit is contained in:
CI Bot
2026-07-21 08:41:56 +08:00
parent 100e054896
commit 78bb3c11a8
7 changed files with 62 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/app/api/routes/auth.py
+8 -6
View File
@@ -509,17 +509,18 @@ async def send_verification_code(
request: SendVerificationCodeRequest,
) -> SendVerificationCodeResponse:
"""发送验证码（手机或邮箱）"""
from app.dependencies import get_db
from packages.adapters.sms.sms_service import get_sms_service
from packages.adapters.smtp import get_email_service
from packages.adapters.sqlalchemy_impl.verification_code_repository import (
SQLAlchemyVerificationCodeRepository,
)
from packages.application.auth.bind_contact_use_case import SendVerificationCodeRequest as UseCaseRequest
from packages.application.auth.bind_contact_use_case import (
SendVerificationCodeRequest as UseCaseRequest,
SendVerificationCodeUseCase,
)
from packages.application.auth.bind_contact_use_case import SendVerificationCodeUseCase
from packages.application.auth.verification_code_service import VerificationCodeService
from app.dependencies import get_db
db = next(get_db())
repo = SQLAlchemyVerificationCodeRepository(db)
@@ -554,15 +555,16 @@ async def bind_contact(
user_repository: UserRepository = Depends(get_user_repository),
) -> BindContactResponse:
"""绑定手机号和/或邮箱（需登录态）"""
from app.dependencies import get_db
from packages.adapters.sqlalchemy_impl.verification_code_repository import (
SQLAlchemyVerificationCodeRepository,
)
from packages.application.auth.bind_contact_use_case import BindContactRequest as UseCaseRequest
from packages.application.auth.bind_contact_use_case import (
BindContactRequest as UseCaseRequest,
BindContactUseCase,
)
from packages.application.auth.bind_contact_use_case import BindContactUseCase
from packages.application.auth.verification_code_service import VerificationCodeService
from app.dependencies import get_db
db = next(get_db())
vc_repo = SQLAlchemyVerificationCodeRepository(db)
packages/adapters/sms/sms_service.py
+4 -5
View File
@@ -35,19 +35,18 @@ class AliyunSmsService:
self.access_key_id = access_key_id or os.environ.get("ALIYUN_SMS_ACCESS_KEY_ID", "")
self.access_key_secret = access_key_secret or os.environ.get("ALIYUN_SMS_ACCESS_KEY_SECRET", "")
self.sign_name = sign_name or os.environ.get("ALIYUN_SMS_SIGN_NAME", "小应剪辑")
self.verify_template_id = verify_template_id or os.environ.get(
"ALIYUN_SMS_VERIFY_TEMPLATE_ID", "SMS_123456789"
)
self.verify_template_id = verify_template_id or os.environ.get("ALIYUN_SMS_VERIFY_TEMPLATE_ID", "SMS_123456789")
def send_verification_code(self, phone: str, code: str) -> bool:
return self.send_template_sms(phone, self.verify_template_id, {"code": code})
def send_template_sms(self, phone: str, template_id: str, params: dict) -> bool:
try:
import json
from alibabacloud_dysmsapi20170525 import models as dysmsapi_models
from alibabacloud_dysmsapi20170525.client import Client as DysmsapiClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_dysmsapi20170525 import models as dysmsapi_models
import json
config = open_api_models.Config(
access_key_id=self.access_key_id,
packages/adapters/sqlalchemy_impl/models.py
+1 -1
View File
@@ -573,4 +573,4 @@ class VerificationCodeModel(Base):
expires_at = Column(DateTime, nullable=False)
used_at = Column(DateTime, nullable=True)
attempts = Column(Integer, nullable=False, default=0)
created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
packages/application/auth/bind_contact_use_case.py
+10 -5
View File
@@ -189,7 +189,9 @@ class SendVerificationCodeUseCase:
self.sms_service = sms_service
self.email_service = email_service
def execute(self, request: SendVerificationCodeRequest) -> tuple[Optional[SendVerificationCodeResponse], Optional[str]]:
def execute(
self, request: SendVerificationCodeRequest
) -> tuple[Optional[SendVerificationCodeResponse], Optional[str]]:
try:
# 1. 确定 code_type
if request.target == "phone":
@@ -222,10 +224,13 @@ class SendVerificationCodeUseCase:
# 4. 返回
ttl = (code_obj.expires_at - code_obj.created_at).total_seconds()
return SendVerificationCodeResponse(
expires_in=int(ttl),
resend_after=60,
), None
return (
SendVerificationCodeResponse(
expires_in=int(ttl),
resend_after=60,
),
None,
)
except Exception as e:
logger.error("发送验证码失败: %s", e, exc_info=True)
packages/application/auth/verification_code_service.py
+4 -4
View File
@@ -18,10 +18,10 @@ from packages.ports.verification_code_repository import VerificationCodeReposito
logger = logging.getLogger(__name__)
# 频控参数
RESEND_COOLDOWN_SECONDS = 60 # 重发冷却时间
DAILY_LIMIT = 10 # 每日发送上限
MAX_ATTEMPTS = 5 # 单验证码最大尝试次数
DEFAULT_TTL_SECONDS = 300 # 默认有效期 5 分钟
RESEND_COOLDOWN_SECONDS = 60 # 重发冷却时间
DAILY_LIMIT = 10 # 每日发送上限
MAX_ATTEMPTS = 5 # 单验证码最大尝试次数
DEFAULT_TTL_SECONDS = 300 # 默认有效期 5 分钟
# 验证码类型
CODE_TYPE_EMAIL_BIND = "email_bind"
packages/application/auth/wechat_oauth_service.py
+28 -21
View File
@@ -21,6 +21,7 @@ logger = logging.getLogger(__name__)
@dataclass
class WechatUserInfo:
"""微信用户信息"""
openid: str
unionid: str = ""
nickname: str = ""
@@ -39,9 +40,7 @@ class WechatOAuthService:
):
self.app_id = app_id or os.environ.get("WECHAT_OPEN_APP_ID", "")
self.app_secret = app_secret or os.environ.get("WECHAT_OPEN_APP_SECRET", "")
self.redirect_uri = redirect_uri or os.environ.get(
"WECHAT_OPEN_REDIRECT_URI", ""
)
self.redirect_uri = redirect_uri or os.environ.get("WECHAT_OPEN_REDIRECT_URI", "")
self._state_store = state_store # 可选：state 存储（Redis/内存），用于 CSRF 防护
def is_configured(self) -> bool:
@@ -59,12 +58,14 @@ class WechatOAuthService:
if not self.is_configured():
# 未配置时返回 mock URL，方便前端联调
mock_params = urllib.parse.urlencode({
"app_id": "mock",
"redirect_uri": self.redirect_uri,
"scope": scope,
"state": state,
})
mock_params = urllib.parse.urlencode(
{
"app_id": "mock",
"redirect_uri": self.redirect_uri,
"scope": scope,
"state": state,
}
)
return f"/mock/wechat/auth?{mock_params}", state
params = {
@@ -94,12 +95,15 @@ class WechatOAuthService:
if not self.is_configured():
# 开发模式：返回 mock 用户信息
logger.info("微信未配置，使用 mock 用户信息")
return WechatUserInfo(
openid=f"mock_{code[:20]}",
unionid=f"mock_union_{code[:16]}",
nickname="微信测试用户",
avatar_url="",
), None
return (
WechatUserInfo(
openid=f"mock_{code[:20]}",
unionid=f"mock_union_{code[:16]}",
nickname="微信测试用户",
avatar_url="",
),
None,
)
try:
# 1. 用 code 换 access_token
@@ -135,12 +139,15 @@ class WechatOAuthService:
logger.error("微信获取用户信息失败: %s", user_data)
return None, f"获取用户信息失败: {user_data.get('errmsg', '未知错误')}"
return WechatUserInfo(
openid=openid,
unionid=unionid,
nickname=user_data.get("nickname", ""),
avatar_url=user_data.get("headimgurl", ""),
), None
return (
WechatUserInfo(
openid=openid,
unionid=unionid,
nickname=user_data.get("nickname", ""),
avatar_url=user_data.get("headimgurl", ""),
),
None,
)
except requests.RequestException as e:
logger.error("微信 OAuth 请求异常: %s", e, exc_info=True)
tests/unit/test_wechat_login_and_verification.py
+7 -10
View File
@@ -2,9 +2,10 @@
验证码服务 + 绑定流程单元测试
"""
import pytest
from datetime import datetime, timedelta, timezone
import pytest
from packages.adapters.in_memory.user_repository import InMemoryUserRepository
from packages.application.auth.bind_contact_use_case import (
BindContactRequest,
@@ -16,9 +17,9 @@ from packages.application.auth.verification_code_service import (
CODE_TYPE_EMAIL_BIND,
CODE_TYPE_PHONE_BIND,
VerificationCodeService,
normalize_phone,
validate_email,
validate_phone,
normalize_phone,
)
from packages.domain.entities import User
from packages.domain.verification_code import VerificationCode
@@ -34,10 +35,7 @@ class InMemoryVerificationCodeRepository:
self._codes[code.id] = code
def find_latest(self, recipient, code_type):
candidates = [
c for c in self._codes.values()
if c.recipient == recipient and c.code_type == code_type
]
candidates = [c for c in self._codes.values() if c.recipient == recipient and c.code_type == code_type]
if not candidates:
return None
return max(candidates, key=lambda c: c.created_at)
@@ -49,10 +47,9 @@ class InMemoryVerificationCodeRepository:
now = datetime.now(timezone.utc)
start_of_day = now.replace(hour=0, minute=0, second=0, microsecond=0)
return sum(
1 for c in self._codes.values()
if c.recipient == recipient
and c.code_type == code_type
and c.created_at >= start_of_day
1
for c in self._codes.values()
if c.recipient == recipient and c.code_type == code_type and c.created_at >= start_of_day
)