233 lines
8.1 KiB
Python
233 lines
8.1 KiB
Python
import tempfile
|
||
from pathlib import Path
|
||
|
||
from fastapi import Depends, HTTPException, Request, Query, UploadFile, File
|
||
from typing import Literal, Tuple
|
||
|
||
from tortoise.exceptions import DoesNotExist
|
||
from tortoise.transactions import in_transaction
|
||
|
||
from app.models.base import User
|
||
from app.utils.security import is_admin_user
|
||
from app.api.admin.router import admin_router
|
||
import app.models.fr as fr
|
||
import app.models.jp as jp
|
||
from app.schemas.admin_schemas import CreateWord, UpdateWordSet, UpdateWord, SearchWordRequest
|
||
from scripts.update_fr import import_wordlist_fr, import_def_fr
|
||
|
||
|
||
@admin_router.get("/dict")
|
||
async def get_wordlist(request: Request,
|
||
page: int = Query(1, ge=1),
|
||
page_size: int = Query(10, le=10),
|
||
lang_code: Literal["fr", "jp"] = "fr",
|
||
admin_user: Tuple[User, dict] = Depends(is_admin_user)):
|
||
"""
|
||
后台管理系统中关于词典部分的初始界面，分页显示
|
||
:param request: 请求头
|
||
:param page: 显示的表格视窗的页数，起始默认为 1
|
||
:param page_size: 控制每页的单词内容条数
|
||
:param lang_code: 查询并显示对应语言的单词表
|
||
:param admin_user: 管理员权限校验（自动完成）
|
||
:return: None
|
||
"""
|
||
offset = (page - 1) * page_size
|
||
if lang_code == "fr":
|
||
total = await fr.DefinitionFr.all().count()
|
||
wordlist = await fr.DefinitionFr.all().offset(offset).limit(page_size).values(
|
||
"word__text",
|
||
"pos",
|
||
"meaning",
|
||
"example",
|
||
"eng_explanation"
|
||
)
|
||
else:
|
||
total = await jp.DefinitionJp.all().count()
|
||
wordlist = await jp.DefinitionJp.all().offset(offset).limit(page_size).values(
|
||
"word__text",
|
||
"pos",
|
||
"meaning",
|
||
"example",
|
||
)
|
||
return {
|
||
"total": total,
|
||
"data": wordlist
|
||
}
|
||
|
||
|
||
@admin_router.post("/dict/search_word")
|
||
async def search_word(
|
||
request: Request,
|
||
search_word: SearchWordRequest,
|
||
admin_user: Tuple[User, dict] = Depends(is_admin_user),
|
||
):
|
||
"""
|
||
查询单词
|
||
:param request: 请求体参数
|
||
:param search_word: Pydantic 模型校验：可提供词性筛选
|
||
:param admin_user:
|
||
:return:
|
||
"""
|
||
if not admin_user[0].is_admin:
|
||
raise HTTPException(status_code=403, detail="非管理员，无权限访问")
|
||
# 筛选参数构造
|
||
filter_kwargs = {}
|
||
if search_word.pos:
|
||
filter_kwargs["pos"] = search_word.pos
|
||
if search_word.language == "fr":
|
||
try:
|
||
word_obj = await fr.WordlistFr.get(text=search_word.word)
|
||
except DoesNotExist:
|
||
raise HTTPException(status_code=400, detail=f"词条 {search_word.word} 不存在于法语词表中")
|
||
definitions = await word_obj.definitions.filter(**filter_kwargs)
|
||
result = [{
|
||
"id": d.id,
|
||
"word": word_obj.text,
|
||
"pos": d.pos,
|
||
"meaning": d.meaning,
|
||
"example": d.example,
|
||
"eng_explanation": d.eng_explanation
|
||
} for d in definitions]
|
||
return result
|
||
else:
|
||
try:
|
||
word_obj = await jp.WordlistJp.get(text=search_word.word)
|
||
except DoesNotExist:
|
||
raise HTTPException(status_code=400, detail=f"词条 {search_word.word} 不存在于日语词表中")
|
||
definitions = await word_obj.definitions.filter(**filter_kwargs)
|
||
result = [{
|
||
"id": d.id,
|
||
"word": word_obj.text,
|
||
"pos": d.pos,
|
||
"meaning": d.meaning,
|
||
"example": d.example
|
||
} for d in definitions]
|
||
return result
|
||
|
||
|
||
@admin_router.put("/dict/adjust")
|
||
async def adjust_dict(
|
||
request: Request,
|
||
updated_contents: UpdateWordSet,
|
||
admin_user: Tuple[User, dict] = Depends(is_admin_user)
|
||
):
|
||
"""
|
||
只关心更新的内容，不关心未改变的内容。
|
||
批量更新 Definition 项，跳过失败项但记录错误。
|
||
:param request:
|
||
:param updated_contents:
|
||
:param admin_user:
|
||
:return:
|
||
"""
|
||
|
||
if not admin_user[0].is_admin:
|
||
raise HTTPException(status_code=403, detail="非管理员，无权限访问")
|
||
|
||
if updated_contents.count() == 0:
|
||
raise HTTPException(status_code=422, detail="无改动信息")
|
||
|
||
errors = []
|
||
|
||
async def update_definition(update_word: UpdateWord) -> None:
|
||
# 检查词条是否存在
|
||
if update_word.language == 'fr':
|
||
word_entry = await fr.WordlistFr.get_or_none(id=update_word.id)
|
||
if not word_entry:
|
||
raise HTTPException(status_code=400, detail=f"词条 ID {update_word.id} 不存在于法语词表中")
|
||
update_obj = await fr.DefinitionFr.get_or_none(id=update_word.id)
|
||
else:
|
||
word_entry = await jp.WordlistJp.get_or_none(id=update_word.id)
|
||
if not word_entry:
|
||
raise HTTPException(status_code=400, detail=f"词条 ID {update_word.id} 不存在于日语词表中")
|
||
update_obj = await jp.DefinitionJp.get_or_none(id=update_word.id)
|
||
|
||
if not update_obj:
|
||
raise HTTPException(status_code=404, detail=f"定义 ID {update_word.id} 不存在")
|
||
|
||
# 获取更新字段
|
||
update_data = update_word.model_dump(exclude_unset=True)
|
||
|
||
for field, value in update_data.items():
|
||
if field != "id":
|
||
setattr(update_obj, field, value)
|
||
|
||
await update_obj.save()
|
||
|
||
for updated_content in updated_contents:
|
||
try:
|
||
await update_definition(updated_content)
|
||
except HTTPException as e:
|
||
errors.append({
|
||
"id": updated_content.id,
|
||
"error": e.detail
|
||
})
|
||
|
||
return {
|
||
"msg": "更新完成",
|
||
"success_count": updated_contents.count() - len(errors),
|
||
"fail_count": len(errors),
|
||
"errors": errors
|
||
}
|
||
|
||
|
||
@admin_router.post("/dict/add", deprecated=False)
|
||
async def add_dict(
|
||
request: Request,
|
||
new_word: CreateWord,
|
||
admin_user: Tuple[User, dict] = Depends(is_admin_user),
|
||
) -> None:
|
||
if not admin_user[0].is_admin:
|
||
raise HTTPException(status_code=403, detail="非管理员，无权限访问")
|
||
if new_word.language == "fr":
|
||
cls_word, _ = await fr.WordlistFr.get_or_create(text=new_word.word)
|
||
new_definition, created = await fr.DefinitionFr.get_or_create(
|
||
word=cls_word,
|
||
pos=new_word.pos,
|
||
meaning=new_word.meaning,
|
||
example=new_word.example,
|
||
eng_explanation=new_word.eng_explanation
|
||
)
|
||
if not created:
|
||
raise HTTPException(status_code=409, detail="释义已存在")
|
||
elif new_word.language == "jp":
|
||
cls_word, _ = await jp.WordlistJp.get_or_create(text=new_word.word)
|
||
new_definition, created = await jp.DefinitionJp.get_or_create(
|
||
word=cls_word,
|
||
pos=new_word.pos,
|
||
meaning=new_word.meaning,
|
||
example=new_word.example,
|
||
)
|
||
if not created:
|
||
raise HTTPException(status_code=409, detail="释义已存在")
|
||
else:
|
||
raise HTTPException(status_code=400, detail="暂不支持语言类型")
|
||
|
||
|
||
@admin_router.post("/dict/update_by_xlsx", deprecated=False)
|
||
async def update_by_xlsx(
|
||
file: UploadFile = File(...),
|
||
admin_user: Tuple[User, dict] = Depends(is_admin_user),
|
||
):
|
||
"""
|
||
上传法语词典Excel文件并导入至数据库
|
||
"""
|
||
|
||
if not file.filename.endswith((".xlsx", ".xls")):
|
||
raise HTTPException(status_code=400, detail="文件格式必须为Excel（.xlsx或.xls）")
|
||
|
||
try:
|
||
suffix = Path(file.filename).suffix
|
||
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
|
||
content = await file.read()
|
||
tmp.write(content)
|
||
tmp_path = Path(tmp.name)
|
||
|
||
async with in_transaction():
|
||
await import_wordlist_fr(path=tmp_path)
|
||
await import_def_fr(path=tmp_path)
|
||
except Exception as e:
|
||
# Tortoise ORM 会自动回滚事务，所以无需手动删除已添加内容
|
||
raise HTTPException(status_code=500, detail=f"导入失败：{str(e)}")
|
||
|
||
return {"message": "导入成功"}
|