Archived
snowy-yang
18c77296c2
- web/routers/index.py：GET / 首页，用 Jinja2 以 Flask 同口径渲染 （keep_trailing_newline=False）——输出与 Flask render_template 逐字节一致 - web/app.py：挂载 /static 静态资源（site_auth.js 等，与 Flask 同目录） - web/routers/auth_status.py：/api/v1/auth/me 认证状态 · user/duration_text/announcement/smtp_configured/register_credits 逐字段对齐 - core/repo.py：补 get_announcement / get_smtp_config / smtp_configured 异步读 契约奇偶校验：28 项全 OK（新增 index 逐字节、auth:me）； 验证：pytest 51/51；ruff All checks passed；pyright 0 errors； uvicorn 实跑 / 与 /static 与 /api/v1/* 均 200。
79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""认证状态端点（FastAPI 栈）——与 Flask /auth/me 响应契约逐字段对齐。
|
|
|
|
- GET/POST /api/v1/auth/me ↔ Flask /auth/me
|
|
· user：站点用户公开字段（未登录为 None）
|
|
· duration_text：次数文案（不限次数 / 已用完 / N 次）
|
|
· announcement、smtp_configured、register_credits 逐字段一致
|
|
"""
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
|
|
from core import repo
|
|
from web import ratelimit
|
|
from web.deps import SiteUser, current_user
|
|
from web.response import error, success
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _format_credits(count: int) -> str:
|
|
if count < 0:
|
|
return '不限次数'
|
|
if count <= 0:
|
|
return '已用完'
|
|
return f'{count} 次'
|
|
|
|
|
|
def _user_public(row: dict) -> dict:
|
|
return {
|
|
'id': row['id'],
|
|
'username': row['username'],
|
|
'email': row.get('email'),
|
|
'role': row['role'],
|
|
'duration_seconds': row['duration_seconds'],
|
|
'created_at': row.get('created_at'),
|
|
'last_login': row.get('last_login'),
|
|
'ann_seen': row.get('ann_seen'),
|
|
}
|
|
|
|
|
|
@router.api_route('/auth/me', methods=['GET', 'POST'])
|
|
async def auth_me(
|
|
user: Annotated[SiteUser | None, Depends(current_user)],
|
|
request: Request,
|
|
):
|
|
"""当前会话与站点公开配置（未登录也返回，与 Flask 同口径）。"""
|
|
await ratelimit.guard(request, bucket='free', limit=30, window=60)
|
|
try:
|
|
ann = await repo.get_announcement()
|
|
smtp_ok = await repo.smtp_configured()
|
|
try:
|
|
credits = int(await repo.get_setting('register_credits', '3') or 0)
|
|
except (TypeError, ValueError):
|
|
credits = 3
|
|
credits = max(0, credits)
|
|
|
|
pub = None
|
|
duration_text = ''
|
|
if user is not None:
|
|
row = await repo.get_user_by_id(user.id)
|
|
if row is not None:
|
|
pub = _user_public(row)
|
|
duration_text = _format_credits(int(row.get('duration_seconds') or 0))
|
|
return success({
|
|
'user': pub,
|
|
'duration_text': duration_text,
|
|
'announcement': {
|
|
'title': ann.get('title', ''),
|
|
'content': ann.get('content', ''),
|
|
'popup_mode': ann.get('popup_mode', 'off'),
|
|
'updated_at': ann.get('updated_at', ''),
|
|
},
|
|
'smtp_configured': smtp_ok,
|
|
'register_credits': credits,
|
|
})
|
|
except Exception:
|
|
return error('服务暂时不可用，请稍后再试', 500)
|