refactor(userpage): move APIs into g0v0 private API

This commit is contained in:
MingxuanGame
2025-10-04 04:57:24 +00:00
parent 2bfde24b84
commit 7c18fc5fb6
8 changed files with 186 additions and 206 deletions

View File

@@ -1,47 +0,0 @@
"""
用户页面相关的异常类
"""
class UserpageError(Exception):
"""用户页面处理错误基类"""
def __init__(self, message: str, code: str = "userpage_error"):
self.message = message
self.code = code
super().__init__(message)
class ContentTooLongError(UserpageError):
"""内容过长错误"""
def __init__(self, current_length: int, max_length: int):
message = f"Content too long. Maximum {max_length} characters allowed, got {current_length}."
super().__init__(message, "content_too_long")
self.current_length = current_length
self.max_length = max_length
class ContentEmptyError(UserpageError):
"""内容为空错误"""
def __init__(self):
super().__init__("Content cannot be empty.", "content_empty")
class BBCodeValidationError(UserpageError):
"""BBCode验证错误"""
def __init__(self, errors: list[str]):
message = f"BBCode validation failed: {'; '.join(errors)}"
super().__init__(message, "bbcode_validation_error")
self.errors = errors
class ForbiddenTagError(UserpageError):
"""禁止标签错误"""
def __init__(self, tag: str):
message = f"Forbidden tag '{tag}' is not allowed."
super().__init__(message, "forbidden_tag")
self.tag = tag

View File

@@ -54,3 +54,47 @@ class ValidateBBCodeResponse(BaseModel):
valid: bool = Field(description="BBCode是否有效")
errors: list[str] = Field(default_factory=list, description="错误列表")
preview: dict[str, str] = Field(description="预览内容")
class UserpageError(Exception):
"""用户页面处理错误基类"""
def __init__(self, message: str, code: str = "userpage_error"):
self.message = message
self.code = code
super().__init__(message)
class ContentTooLongError(UserpageError):
"""内容过长错误"""
def __init__(self, current_length: int, max_length: int):
message = f"Content too long. Maximum {max_length} characters allowed, got {current_length}."
super().__init__(message, "content_too_long")
self.current_length = current_length
self.max_length = max_length
class ContentEmptyError(UserpageError):
"""内容为空错误"""
def __init__(self):
super().__init__("Content cannot be empty.", "content_empty")
class BBCodeValidationError(UserpageError):
"""BBCode验证错误"""
def __init__(self, errors: list[str]):
message = f"BBCode validation failed: {'; '.join(errors)}"
super().__init__(message, "bbcode_validation_error")
self.errors = errors
class ForbiddenTagError(UserpageError):
"""禁止标签错误"""
def __init__(self, tag: str):
message = f"Forbidden tag '{tag}' is not allowed."
super().__init__(message, "forbidden_tag")
self.tag = tag

View File

@@ -1,6 +1,6 @@
from app.config import settings
from . import admin, audio_proxy, avatar, beatmapset, cover, oauth, relationship, score, team, username # noqa: F401
from . import admin, audio_proxy, avatar, beatmapset, cover, oauth, relationship, score, team, user # noqa: F401
from .router import router as private_router
if settings.enable_totp_verification:

130
app/router/private/user.py Normal file
View File

@@ -0,0 +1,130 @@
from typing import Annotated
from app.auth import validate_username
from app.config import settings
from app.database import User
from app.database.events import Event, EventType
from app.dependencies.database import Database
from app.dependencies.user import ClientUser
from app.models.user import Page
from app.models.userpage import (
UpdateUserpageRequest,
UpdateUserpageResponse,
UserpageError,
ValidateBBCodeRequest,
ValidateBBCodeResponse,
)
from app.service.bbcode_service import bbcode_service
from app.utils import utcnow
from .router import router
from fastapi import Body, HTTPException
from sqlmodel import exists, select
@router.post("/rename", name="修改用户名", tags=["用户", "g0v0 API"])
async def user_rename(
session: Database,
new_name: Annotated[str, Body(..., description="新的用户名")],
current_user: ClientUser,
):
"""修改用户名
为指定用户修改用户名,并将原用户名添加到历史用户名列表中
错误情况:
- 404: 找不到指定用户
- 409: 新用户名已被占用
返回:
- 成功: None
"""
samename_user = (await session.exec(select(exists()).where(User.username == new_name))).first()
if samename_user:
raise HTTPException(409, "Username Exisits")
errors = validate_username(new_name)
if errors:
raise HTTPException(403, "\n".join(errors))
previous_username = []
previous_username.extend(current_user.previous_usernames)
previous_username.append(current_user.username)
current_user.username = new_name
current_user.previous_usernames = previous_username
rename_event = Event(
created_at=utcnow(),
type=EventType.USERNAME_CHANGE,
user_id=current_user.id,
user=current_user,
)
rename_event.event_payload["user"] = {
"username": new_name,
"url": settings.web_url + "users/" + str(current_user.id),
"previous_username": current_user.previous_usernames[-1],
}
session.add(rename_event)
await session.commit()
return None
@router.put(
"/user/page",
response_model=UpdateUserpageResponse,
name="更新用户页面",
description="更新指定用户的个人页面内容支持BBCode。匹配官方osu-web API格式。",
tags=["用户", "g0v0 API"],
)
async def update_userpage(
request: UpdateUserpageRequest,
session: Database,
current_user: ClientUser,
):
"""更新用户页面内容"""
try:
# 处理BBCode内容
processed_page = bbcode_service.process_userpage_content(request.body)
# 更新数据库 - 直接更新用户对象
current_user.page = Page(html=processed_page["html"], raw=processed_page["raw"])
session.add(current_user)
await session.commit()
await session.refresh(current_user)
# 返回官方格式的响应只包含html
return UpdateUserpageResponse(html=processed_page["html"])
except UserpageError as e:
# 使用官方格式的错误响应:{'error': message}
raise HTTPException(status_code=422, detail={"error": e.message})
except Exception:
raise HTTPException(status_code=500, detail={"error": "Failed to update user page"})
@router.post(
"/user/validate-bbcode",
response_model=ValidateBBCodeResponse,
name="验证BBCode",
description="验证BBCode语法并返回预览。",
tags=["用户", "g0v0 API"],
)
async def validate_bbcode(
request: ValidateBBCodeRequest,
):
"""验证BBCode语法"""
try:
# 验证BBCode语法
errors = bbcode_service.validate_bbcode(request.content)
# 生成预览(如果没有严重错误)
if len(errors) == 0:
preview = bbcode_service.process_userpage_content(request.content)
else:
preview = {"raw": request.content, "html": ""}
return ValidateBBCodeResponse(valid=len(errors) == 0, errors=errors, preview=preview)
except UserpageError as e:
return ValidateBBCodeResponse(valid=False, errors=[e.message], preview={"raw": request.content, "html": ""})
except Exception:
raise HTTPException(status_code=500, detail={"error": "Failed to validate BBCode"})

View File

@@ -1,58 +0,0 @@
from typing import Annotated
from app.auth import validate_username
from app.config import settings
from app.database.events import Event, EventType
from app.database.user import User
from app.dependencies.database import Database
from app.dependencies.user import ClientUser
from app.utils import utcnow
from .router import router
from fastapi import Body, HTTPException
from sqlmodel import exists, select
@router.post("/rename", name="修改用户名", tags=["用户", "g0v0 API"])
async def user_rename(
session: Database,
new_name: Annotated[str, Body(..., description="新的用户名")],
current_user: ClientUser,
):
"""修改用户名
为指定用户修改用户名,并将原用户名添加到历史用户名列表中
错误情况:
- 404: 找不到指定用户
- 409: 新用户名已被占用
返回:
- 成功: None
"""
samename_user = (await session.exec(select(exists()).where(User.username == new_name))).first()
if samename_user:
raise HTTPException(409, "Username Exisits")
errors = validate_username(new_name)
if errors:
raise HTTPException(403, "\n".join(errors))
previous_username = []
previous_username.extend(current_user.previous_usernames)
previous_username.append(current_user.username)
current_user.username = new_name
current_user.previous_usernames = previous_username
rename_event = Event(
created_at=utcnow(),
type=EventType.USERNAME_CHANGE,
user_id=current_user.id,
user=current_user,
)
rename_event.event_payload["user"] = {
"username": new_name,
"url": settings.web_url + "users/" + str(current_user.id),
"previous_username": current_user.previous_usernames[-1],
}
session.add(rename_event)
await session.commit()
return None

View File

@@ -1,22 +1,14 @@
from typing import Annotated
from app.database import MeResp, User
from app.database import MeResp
from app.dependencies.database import Database
from app.dependencies.user import UserAndToken, get_current_user, get_current_user_and_token
from app.exceptions.userpage import UserpageError
from app.dependencies.user import UserAndToken, get_current_user_and_token
from app.models.score import GameMode
from app.models.user import Page
from app.models.userpage import (
UpdateUserpageRequest,
UpdateUserpageResponse,
ValidateBBCodeRequest,
ValidateBBCodeResponse,
)
from app.service.bbcode_service import bbcode_service
from .router import router
from fastapi import HTTPException, Path, Security
from fastapi import Path, Security
from fastapi.responses import RedirectResponse
@router.get(
@@ -50,92 +42,11 @@ async def get_user_info_default(
return user_resp
# @router.get(
# "/users/{user_id}/page",
# response_model=UserpageResponse,
# name="获取用户页面",
# description="获取指定用户的个人页面内容。匹配官方osu-web API格式。",
# tags=["用户"],
# )
# async def get_userpage(
# session: Database,
# user_id: int = Path(description="用户ID"),
# ):
# """获取用户页面内容"""
# # 查找用户
# user = await session.get(User, user_id)
# if not user:
# raise HTTPException(status_code=404, detail={"error": "User not found"})
# # 返回页面内容
# if user.page:
# return UserpageResponse(html=user.page.get("html", ""), raw=user.page.get("raw", ""))
# else:
# return UserpageResponse(html="", raw="")
@router.put("/users/{user_id}/page", include_in_schema=False)
async def update_userpage():
return RedirectResponse(url="/api/private/user/page", status_code=307)
@router.put(
"/users/{user_id}/page",
response_model=UpdateUserpageResponse,
name="更新用户页面",
description="更新指定用户的个人页面内容支持BBCode。匹配官方osu-web API格式。",
tags=["用户"],
)
async def update_userpage(
request: UpdateUserpageRequest,
session: Database,
user_id: Annotated[int, Path(description="用户ID")],
current_user: Annotated[User, Security(get_current_user, scopes=["edit"])],
):
"""更新用户页面内容匹配官方osu-web实现"""
# 检查权限:只能编辑自己的页面(除非是管理员)
if user_id != current_user.id:
raise HTTPException(status_code=403, detail={"error": "Access denied"})
try:
# 处理BBCode内容
processed_page = bbcode_service.process_userpage_content(request.body)
# 更新数据库 - 直接更新用户对象
current_user.page = Page(html=processed_page["html"], raw=processed_page["raw"])
session.add(current_user)
await session.commit()
await session.refresh(current_user)
# 返回官方格式的响应只包含html
return UpdateUserpageResponse(html=processed_page["html"])
except UserpageError as e:
# 使用官方格式的错误响应:{'error': message}
raise HTTPException(status_code=422, detail={"error": e.message})
except Exception:
raise HTTPException(status_code=500, detail={"error": "Failed to update user page"})
@router.post(
"/me/validate-bbcode",
response_model=ValidateBBCodeResponse,
name="验证BBCode",
description="验证BBCode语法并返回预览。",
tags=["用户"],
)
async def validate_bbcode(
request: ValidateBBCodeRequest,
):
"""验证BBCode语法"""
try:
# 验证BBCode语法
errors = bbcode_service.validate_bbcode(request.content)
# 生成预览(如果没有严重错误)
if len(errors) == 0:
preview = bbcode_service.process_userpage_content(request.content)
else:
preview = {"raw": request.content, "html": ""}
return ValidateBBCodeResponse(valid=len(errors) == 0, errors=errors, preview=preview)
except UserpageError as e:
return ValidateBBCodeResponse(valid=False, errors=[e.message], preview={"raw": request.content, "html": ""})
except Exception:
raise HTTPException(status_code=500, detail={"error": "Failed to validate BBCode"})
@router.post("/me/validate-bbcode", include_in_schema=False)
async def validate_bbcode():
return RedirectResponse(url="/api/private/user/validate-bbcode", status_code=307)

View File

@@ -8,7 +8,7 @@ import html
import re
from typing import ClassVar
from app.exceptions.userpage import (
from app.models.userpage import (
ContentEmptyError,
ContentTooLongError,
ForbiddenTagError,