Cheng Zhou
564 lines
21 KiB
Python
564 lines
21 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import base64
|
||
import hashlib
|
||
import random
|
||
import secrets
|
||
import smtplib
|
||
import ssl
|
||
import uuid
|
||
from datetime import datetime, timedelta, timezone
|
||
from email.message import EmailMessage
|
||
from email.utils import formataddr
|
||
|
||
from cryptography.fernet import Fernet, InvalidToken
|
||
from fastapi import HTTPException, status
|
||
from sqlalchemy import delete, desc, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.config import settings
|
||
from app.core.security import hash_password, verify_password
|
||
from app.crud import user as user_crud
|
||
from app.models.email_settings import EmailVerificationCode, EmailVerificationPurpose, SmtpSecurity, SystemEmailSettings
|
||
from app.schemas.email_settings import EmailSettingsRead, EmailSettingsUpdate
|
||
|
||
|
||
def _normalize_email(email: str) -> str:
|
||
return email.strip().lower()
|
||
|
||
|
||
def _parse_allowed_domains(value: str) -> list[str]:
|
||
domains: list[str] = []
|
||
for item in (value or "").replace("，", ",").split(","):
|
||
domain = item.strip().lower().lstrip("@")
|
||
if domain and domain not in domains:
|
||
domains.append(domain)
|
||
return domains or ["huapont.cn", "qq.com"]
|
||
|
||
|
||
def _normalize_domain(value: str) -> str:
|
||
return value.strip().lower().lstrip("@")
|
||
|
||
|
||
def _domain_from_email(email: str) -> str:
|
||
parts = _normalize_email(email).rsplit("@", 1)
|
||
return parts[1] if len(parts) == 2 else ""
|
||
|
||
|
||
def _as_aware(value: datetime) -> datetime:
|
||
return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
|
||
|
||
|
||
def _get_fernet() -> Fernet:
|
||
source = settings.SETTINGS_ENCRYPTION_KEY or settings.JWT_SECRET_KEY
|
||
try:
|
||
return Fernet(source.encode("utf-8"))
|
||
except Exception:
|
||
key = base64.urlsafe_b64encode(hashlib.sha256(source.encode("utf-8")).digest())
|
||
return Fernet(key)
|
||
|
||
|
||
def encrypt_secret(value: str) -> str:
|
||
return _get_fernet().encrypt(value.encode("utf-8")).decode("utf-8")
|
||
|
||
|
||
def decrypt_secret(value: str | None) -> str | None:
|
||
if not value:
|
||
return None
|
||
try:
|
||
return _get_fernet().decrypt(value.encode("utf-8")).decode("utf-8")
|
||
except InvalidToken as exc:
|
||
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="邮件授权码解密失败") from exc
|
||
|
||
|
||
async def list_email_settings(db: AsyncSession) -> list[SystemEmailSettings]:
|
||
result = await db.execute(select(SystemEmailSettings).order_by(SystemEmailSettings.register_domain.asc()))
|
||
return list(result.scalars().all())
|
||
|
||
|
||
async def get_email_settings(db: AsyncSession, register_domain: str | None = None) -> SystemEmailSettings | None:
|
||
if register_domain:
|
||
result = await db.execute(
|
||
select(SystemEmailSettings).where(SystemEmailSettings.register_domain == _normalize_domain(register_domain))
|
||
)
|
||
return result.scalar_one_or_none()
|
||
result = await db.execute(select(SystemEmailSettings).order_by(SystemEmailSettings.created_at.asc()))
|
||
return result.scalars().first()
|
||
|
||
|
||
async def get_email_settings_for_email(db: AsyncSession, email: str) -> SystemEmailSettings | None:
|
||
domain = _domain_from_email(email)
|
||
return await get_email_settings(db, domain) if domain else None
|
||
|
||
|
||
def to_email_settings_read(settings_row: SystemEmailSettings | None) -> EmailSettingsRead:
|
||
if not settings_row:
|
||
return EmailSettingsRead()
|
||
return EmailSettingsRead(
|
||
register_domain=settings_row.register_domain,
|
||
smtp_host=settings_row.smtp_host,
|
||
smtp_port=settings_row.smtp_port,
|
||
smtp_security=settings_row.smtp_security.value,
|
||
smtp_username=settings_row.smtp_username,
|
||
smtp_password_configured=bool(settings_row.smtp_password_encrypted),
|
||
sender_email=settings_row.sender_email,
|
||
sender_name=settings_row.sender_name,
|
||
allowed_register_domain=settings_row.allowed_register_domain,
|
||
verification_code_ttl_minutes=settings_row.verification_code_ttl_minutes,
|
||
send_cooldown_seconds=settings_row.send_cooldown_seconds,
|
||
max_verify_attempts=settings_row.max_verify_attempts,
|
||
updated_at=settings_row.updated_at,
|
||
)
|
||
|
||
|
||
async def upsert_email_settings(
|
||
db: AsyncSession,
|
||
register_domain: str,
|
||
payload: EmailSettingsUpdate,
|
||
*,
|
||
updated_by: uuid.UUID,
|
||
) -> SystemEmailSettings:
|
||
register_domain = _normalize_domain(register_domain)
|
||
settings_row = await get_email_settings(db, register_domain)
|
||
if not settings_row:
|
||
settings_row = SystemEmailSettings(
|
||
register_domain=register_domain,
|
||
smtp_host=payload.smtp_host,
|
||
smtp_port=payload.smtp_port,
|
||
smtp_security=SmtpSecurity(payload.smtp_security),
|
||
smtp_username=payload.smtp_username,
|
||
smtp_password_encrypted=encrypt_secret(payload.smtp_password) if payload.smtp_password else None,
|
||
sender_email=str(payload.sender_email),
|
||
sender_name=payload.sender_name,
|
||
allowed_register_domain=register_domain,
|
||
verification_code_ttl_minutes=payload.verification_code_ttl_minutes,
|
||
send_cooldown_seconds=payload.send_cooldown_seconds,
|
||
max_verify_attempts=payload.max_verify_attempts,
|
||
updated_by=updated_by,
|
||
)
|
||
db.add(settings_row)
|
||
else:
|
||
settings_row.smtp_host = payload.smtp_host
|
||
settings_row.smtp_port = payload.smtp_port
|
||
settings_row.smtp_security = SmtpSecurity(payload.smtp_security)
|
||
settings_row.smtp_username = payload.smtp_username
|
||
if payload.smtp_password:
|
||
settings_row.smtp_password_encrypted = encrypt_secret(payload.smtp_password)
|
||
settings_row.sender_email = str(payload.sender_email)
|
||
settings_row.sender_name = payload.sender_name
|
||
settings_row.allowed_register_domain = register_domain
|
||
settings_row.verification_code_ttl_minutes = payload.verification_code_ttl_minutes
|
||
settings_row.send_cooldown_seconds = payload.send_cooldown_seconds
|
||
settings_row.max_verify_attempts = payload.max_verify_attempts
|
||
settings_row.updated_by = updated_by
|
||
await db.commit()
|
||
await db.refresh(settings_row)
|
||
return settings_row
|
||
|
||
|
||
async def create_email_domain(db: AsyncSession, register_domain: str, *, updated_by: uuid.UUID) -> SystemEmailSettings:
|
||
register_domain = _normalize_domain(register_domain)
|
||
existing = await get_email_settings(db, register_domain)
|
||
if existing:
|
||
return existing
|
||
settings_row = SystemEmailSettings(
|
||
register_domain=register_domain,
|
||
smtp_host="",
|
||
smtp_port=465,
|
||
smtp_security=SmtpSecurity.SSL,
|
||
smtp_username="",
|
||
smtp_password_encrypted=None,
|
||
sender_email="",
|
||
sender_name="CTMS 系统",
|
||
allowed_register_domain=register_domain,
|
||
verification_code_ttl_minutes=10,
|
||
send_cooldown_seconds=60,
|
||
max_verify_attempts=5,
|
||
updated_by=updated_by,
|
||
)
|
||
db.add(settings_row)
|
||
await db.commit()
|
||
await db.refresh(settings_row)
|
||
return settings_row
|
||
|
||
|
||
async def delete_email_domain(db: AsyncSession, register_domain: str) -> None:
|
||
await db.execute(
|
||
delete(SystemEmailSettings).where(SystemEmailSettings.register_domain == _normalize_domain(register_domain))
|
||
)
|
||
await db.commit()
|
||
|
||
|
||
def ensure_email_settings_ready(settings_row: SystemEmailSettings | None) -> SystemEmailSettings:
|
||
if not settings_row:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="邮件服务尚未配置")
|
||
if not settings_row.smtp_host or not settings_row.smtp_username or not settings_row.sender_email:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="邮件服务配置不完整")
|
||
if not settings_row.smtp_password_encrypted:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="SMTP 授权码尚未配置")
|
||
return settings_row
|
||
|
||
|
||
def ensure_allowed_register_email(email: str, settings_row: SystemEmailSettings) -> None:
|
||
domain = settings_row.register_domain
|
||
if not _normalize_email(email).endswith(f"@{domain}"):
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"仅允许 @{domain} 邮箱注册")
|
||
|
||
|
||
def _send_email_sync(
|
||
settings_row: SystemEmailSettings,
|
||
*,
|
||
to_email: str,
|
||
subject: str,
|
||
body: str,
|
||
) -> None:
|
||
password = decrypt_secret(settings_row.smtp_password_encrypted)
|
||
if not password:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="SMTP 授权码尚未配置")
|
||
|
||
message = EmailMessage()
|
||
message["Subject"] = subject
|
||
message["From"] = formataddr((settings_row.sender_name or "", settings_row.sender_email))
|
||
message["To"] = to_email
|
||
message.set_content(body)
|
||
|
||
if settings_row.smtp_security == SmtpSecurity.SSL:
|
||
with smtplib.SMTP_SSL(settings_row.smtp_host, settings_row.smtp_port, timeout=15) as smtp:
|
||
smtp.login(settings_row.smtp_username, password)
|
||
smtp.send_message(message)
|
||
else:
|
||
with smtplib.SMTP(settings_row.smtp_host, settings_row.smtp_port, timeout=15) as smtp:
|
||
if settings_row.smtp_security == SmtpSecurity.STARTTLS:
|
||
smtp.starttls()
|
||
smtp.login(settings_row.smtp_username, password)
|
||
smtp.send_message(message)
|
||
|
||
|
||
async def send_email(
|
||
settings_row: SystemEmailSettings,
|
||
*,
|
||
to_email: str,
|
||
subject: str,
|
||
body: str,
|
||
) -> None:
|
||
try:
|
||
await asyncio.to_thread(_send_email_sync, settings_row, to_email=to_email, subject=subject, body=body)
|
||
except HTTPException:
|
||
raise
|
||
except ssl.SSLError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||
detail=(
|
||
"邮件发送失败：SMTP 安全协议或端口不匹配，"
|
||
"请确认当前后缀的 SMTP 地址、端口和安全协议。"
|
||
"SSL 通常使用 465，STARTTLS 通常使用 587。"
|
||
),
|
||
) from exc
|
||
except smtplib.SMTPAuthenticationError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||
detail="邮件发送失败：SMTP 账号或授权码不正确",
|
||
) from exc
|
||
except (smtplib.SMTPRecipientsRefused, smtplib.SMTPDataError) as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="邮箱不存在或无法接收邮件，请检查邮箱地址",
|
||
) from exc
|
||
except (smtplib.SMTPConnectError, TimeoutError, OSError) as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||
detail="邮件发送失败：无法连接 SMTP 服务器，请检查 SMTP 地址、端口和网络连通性",
|
||
) from exc
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"邮件发送失败：{exc}") from exc
|
||
|
||
|
||
async def send_test_email(db: AsyncSession, register_domain: str, recipient_email: str) -> None:
|
||
settings_row = ensure_email_settings_ready(await get_email_settings(db, register_domain))
|
||
await send_email(
|
||
settings_row,
|
||
to_email=recipient_email,
|
||
subject="CTMS 邮件服务测试",
|
||
body="这是一封来自 CTMS 系统的邮件服务测试邮件。收到此邮件说明 SMTP 配置可用。",
|
||
)
|
||
|
||
|
||
async def send_register_code(db: AsyncSession, email: str) -> None:
|
||
email = _normalize_email(email)
|
||
settings_row = ensure_email_settings_ready(await get_email_settings_for_email(db, email))
|
||
ensure_allowed_register_email(email, settings_row)
|
||
if await user_crud.get_by_email(db, email):
|
||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="邮箱已注册")
|
||
|
||
await _send_verification_code(
|
||
db,
|
||
settings_row=settings_row,
|
||
email=email,
|
||
purpose=EmailVerificationPurpose.REGISTER,
|
||
subject="CTMS 注册邮箱验证码",
|
||
body_factory=lambda code: (
|
||
f"您的 CTMS 注册验证码是：{code}\n\n"
|
||
f"验证码 {settings_row.verification_code_ttl_minutes} 分钟内有效。"
|
||
"如非本人操作，请忽略此邮件。"
|
||
),
|
||
)
|
||
|
||
|
||
async def send_password_reset_code(db: AsyncSession, email: str) -> None:
|
||
email = _normalize_email(email)
|
||
db_user = await user_crud.get_by_email(db, email)
|
||
if not db_user:
|
||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="该邮箱未注册")
|
||
settings_row = ensure_email_settings_ready(await get_email_settings_for_email(db, email))
|
||
await _send_verification_code(
|
||
db,
|
||
settings_row=settings_row,
|
||
email=email,
|
||
purpose=EmailVerificationPurpose.PASSWORD_RESET,
|
||
subject="CTMS 密码重置验证码",
|
||
body_factory=lambda code: (
|
||
f"您的 CTMS 密码重置验证码是：{code}\n\n"
|
||
f"验证码 {settings_row.verification_code_ttl_minutes} 分钟内有效。"
|
||
"如非本人操作，请尽快联系系统管理员。"
|
||
),
|
||
)
|
||
|
||
|
||
async def send_password_reset_link(db: AsyncSession, email: str, *, frontend_origin: str) -> None:
|
||
email = _normalize_email(email)
|
||
db_user = await user_crud.get_by_email(db, email)
|
||
if not db_user:
|
||
return
|
||
settings_row = ensure_email_settings_ready(await get_email_settings_for_email(db, email))
|
||
now = datetime.now(timezone.utc)
|
||
latest_result = await db.execute(
|
||
select(EmailVerificationCode)
|
||
.where(
|
||
EmailVerificationCode.email == email,
|
||
EmailVerificationCode.purpose == EmailVerificationPurpose.PASSWORD_RESET_LINK,
|
||
)
|
||
.order_by(desc(EmailVerificationCode.created_at))
|
||
.limit(1)
|
||
)
|
||
latest = latest_result.scalar_one_or_none()
|
||
if latest and latest.created_at and _as_aware(latest.created_at) + timedelta(seconds=settings_row.send_cooldown_seconds) > now:
|
||
raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="重置邮件发送过于频繁，请稍后再试")
|
||
|
||
token = secrets.token_urlsafe(32)
|
||
token_hash = _hash_reset_token(token)
|
||
record = EmailVerificationCode(
|
||
email=email,
|
||
purpose=EmailVerificationPurpose.PASSWORD_RESET_LINK,
|
||
code_hash=token_hash,
|
||
expires_at=now + timedelta(minutes=settings_row.verification_code_ttl_minutes),
|
||
)
|
||
db.add(record)
|
||
await db.commit()
|
||
|
||
origin = frontend_origin.rstrip("/") if frontend_origin else ""
|
||
reset_link = f"{origin}/forgot-password?token={token}" if origin else f"/forgot-password?token={token}"
|
||
await send_email(
|
||
settings_row,
|
||
to_email=email,
|
||
subject="CTMS 密码重置链接",
|
||
body=(
|
||
"请点击以下链接重置您的 CTMS 登录密码：\n\n"
|
||
f"{reset_link}\n\n"
|
||
f"链接 {settings_row.verification_code_ttl_minutes} 分钟内有效，且只能使用一次。"
|
||
"如非本人操作，请忽略此邮件并尽快联系系统管理员。"
|
||
),
|
||
)
|
||
|
||
|
||
def _hash_reset_token(token: str) -> str:
|
||
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
||
|
||
|
||
async def _send_verification_code(
|
||
db: AsyncSession,
|
||
*,
|
||
settings_row: SystemEmailSettings,
|
||
email: str,
|
||
purpose: EmailVerificationPurpose,
|
||
subject: str,
|
||
body_factory,
|
||
) -> None:
|
||
now = datetime.now(timezone.utc)
|
||
latest_result = await db.execute(
|
||
select(EmailVerificationCode)
|
||
.where(
|
||
EmailVerificationCode.email == email,
|
||
EmailVerificationCode.purpose == purpose,
|
||
)
|
||
.order_by(desc(EmailVerificationCode.created_at))
|
||
.limit(1)
|
||
)
|
||
latest = latest_result.scalar_one_or_none()
|
||
if latest and latest.created_at and _as_aware(latest.created_at) + timedelta(seconds=settings_row.send_cooldown_seconds) > now:
|
||
raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="验证码发送过于频繁，请稍后再试")
|
||
|
||
code = f"{random.SystemRandom().randint(0, 999999):06d}"
|
||
record = EmailVerificationCode(
|
||
email=email,
|
||
purpose=purpose,
|
||
code_hash=hash_password(code),
|
||
expires_at=now + timedelta(minutes=settings_row.verification_code_ttl_minutes),
|
||
)
|
||
db.add(record)
|
||
await db.commit()
|
||
|
||
await send_email(
|
||
settings_row,
|
||
to_email=email,
|
||
subject=subject,
|
||
body=body_factory(code),
|
||
)
|
||
|
||
|
||
async def verify_register_code(db: AsyncSession, email: str, code: str) -> bool:
|
||
email = _normalize_email(email)
|
||
settings_row = ensure_email_settings_ready(await get_email_settings_for_email(db, email))
|
||
ensure_allowed_register_email(email, settings_row)
|
||
await _verify_email_code(
|
||
db,
|
||
settings_row=settings_row,
|
||
email=email,
|
||
code=code,
|
||
purpose=EmailVerificationPurpose.REGISTER,
|
||
mark_verified=True,
|
||
)
|
||
return True
|
||
|
||
|
||
async def verify_password_reset_code(db: AsyncSession, email: str, code: str) -> str:
|
||
email = _normalize_email(email)
|
||
if not await user_crud.get_by_email(db, email):
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="验证码错误或已过期")
|
||
settings_row = ensure_email_settings_ready(await get_email_settings_for_email(db, email))
|
||
await _verify_email_code(
|
||
db,
|
||
settings_row=settings_row,
|
||
email=email,
|
||
code=code,
|
||
purpose=EmailVerificationPurpose.PASSWORD_RESET,
|
||
mark_verified=True,
|
||
commit=False,
|
||
)
|
||
now = datetime.now(timezone.utc)
|
||
token = secrets.token_urlsafe(32)
|
||
db.add(
|
||
EmailVerificationCode(
|
||
email=email,
|
||
purpose=EmailVerificationPurpose.PASSWORD_RESET_LINK,
|
||
code_hash=_hash_reset_token(token),
|
||
expires_at=now + timedelta(minutes=settings_row.verification_code_ttl_minutes),
|
||
)
|
||
)
|
||
await db.commit()
|
||
return token
|
||
|
||
|
||
async def reset_password_with_code(db: AsyncSession, email: str, code: str, password: str) -> None:
|
||
email = _normalize_email(email)
|
||
db_user = await user_crud.get_by_email(db, email)
|
||
if not db_user:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="验证码错误或已过期")
|
||
settings_row = ensure_email_settings_ready(await get_email_settings_for_email(db, email))
|
||
await _verify_email_code(
|
||
db,
|
||
settings_row=settings_row,
|
||
email=email,
|
||
code=code,
|
||
purpose=EmailVerificationPurpose.PASSWORD_RESET,
|
||
mark_verified=True,
|
||
commit=False,
|
||
)
|
||
db_user.password_hash = hash_password(password)
|
||
await db.commit()
|
||
|
||
|
||
async def reset_password_with_token(db: AsyncSession, token: str, password: str) -> None:
|
||
now = datetime.now(timezone.utc)
|
||
token_hash = _hash_reset_token(token)
|
||
result = await db.execute(
|
||
select(EmailVerificationCode)
|
||
.where(
|
||
EmailVerificationCode.code_hash == token_hash,
|
||
EmailVerificationCode.purpose == EmailVerificationPurpose.PASSWORD_RESET_LINK,
|
||
EmailVerificationCode.verified_at.is_(None),
|
||
)
|
||
.order_by(desc(EmailVerificationCode.created_at))
|
||
.limit(1)
|
||
.with_for_update()
|
||
)
|
||
record = result.scalar_one_or_none()
|
||
if not record or record.verified_at is not None or _as_aware(record.expires_at) < now:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="重置链接无效或已过期")
|
||
|
||
db_user = await user_crud.get_by_email(db, record.email)
|
||
if not db_user:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="重置链接无效或已过期")
|
||
|
||
record.verified_at = now
|
||
db_user.password_hash = hash_password(password)
|
||
await db.commit()
|
||
|
||
|
||
async def _verify_email_code(
|
||
db: AsyncSession,
|
||
*,
|
||
settings_row: SystemEmailSettings,
|
||
email: str,
|
||
code: str,
|
||
purpose: EmailVerificationPurpose,
|
||
mark_verified: bool,
|
||
commit: bool = True,
|
||
) -> EmailVerificationCode:
|
||
now = datetime.now(timezone.utc)
|
||
result = await db.execute(
|
||
select(EmailVerificationCode)
|
||
.where(
|
||
EmailVerificationCode.email == email,
|
||
EmailVerificationCode.purpose == purpose,
|
||
)
|
||
.order_by(desc(EmailVerificationCode.created_at))
|
||
.limit(1)
|
||
.with_for_update()
|
||
)
|
||
record = result.scalar_one_or_none()
|
||
if (
|
||
not record
|
||
or record.verified_at is not None
|
||
or _as_aware(record.expires_at) < now
|
||
or record.attempt_count >= settings_row.max_verify_attempts
|
||
):
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="验证码错误或已过期")
|
||
record.attempt_count += 1
|
||
if not verify_password(code, record.code_hash):
|
||
await db.commit()
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="验证码错误或已过期")
|
||
if mark_verified:
|
||
record.verified_at = now
|
||
if commit:
|
||
await db.commit()
|
||
return record
|
||
|
||
|
||
async def ensure_register_email_verified(db: AsyncSession, email: str) -> None:
|
||
email = _normalize_email(email)
|
||
now = datetime.now(timezone.utc)
|
||
result = await db.execute(
|
||
select(EmailVerificationCode)
|
||
.where(
|
||
EmailVerificationCode.email == email,
|
||
EmailVerificationCode.purpose == EmailVerificationPurpose.REGISTER,
|
||
EmailVerificationCode.verified_at.is_not(None),
|
||
EmailVerificationCode.expires_at >= now,
|
||
)
|
||
.order_by(desc(EmailVerificationCode.verified_at))
|
||
.limit(1)
|
||
)
|
||
if not result.scalar_one_or_none():
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请先完成邮箱验证码校验")
|