新增翻译接口
md5.py:
翻译接口密钥处理函数
settings.py
Settings类依赖.env文件写入
trans_schemas.py:
提供翻译请求类和翻译回应类(请求类尚未使用)
translator.py:
提供翻译接口和管理debug接口
This commit is contained in:
Miyamizu-MitsuhaSang 2025-09-15 01:25:18 +08:00
parent 08eada223b
commit 2c121c9a1e
5 changed files with 161 additions and 2 deletions

118
app/api/translator.py Normal file
View File

@ -0,0 +1,118 @@
from typing import Tuple
from wsgiref import headers
import httpx
import random
import json
from hashlib import md5
import requests
from fastapi import APIRouter, Depends, HTTPException
from app.models import User
from app.schemas.trans_schemas import TransResponse
from app.utils.security import is_admin_user, get_current_user
from scripts.md5 import make_md5
from settings import settings
translator_router = APIRouter()
# For list of language codes, please refer to `https://api.fanyi.baidu.com/doc/21`
from_lang = 'en'
to_lang = 'zh'
# endpoint = 'https://api.fanyi.baidu.com'
# path = '/api/trans/vip/translate'
# url = endpoint + path
#
# query = 'Hello World! This is 1st paragraph.\nThis is 2nd paragraph.'
#
# salt = random.randint(32768, 65536)
# sign = make_md5(appid + query + str(salt) + appkey)
#
# headers = {'Content-Type': 'application/x-www-form-urlencoded'}
# payload = {'appid': appid, 'q': query, 'from': from_lang, 'to': to_lang, 'salt': salt, 'sign': sign}
#
#
# # Send request
# r = requests.post(url, params=payload, headers=headers)
# result = r.json()
#
# # Show response
# print(json.dumps(result, indent=4, ensure_ascii=False))
async def baidu_translation(query: str, from_lang: str, to_lang: str):
url = "http://api.fanyi.baidu.com/api/trans/vip/translate"
appid = settings.BAIDU_APPID
appkey = settings.BAIDU_APPKEY
salt = str(random.randint(32768, 65536))
sign = make_md5(appid + query + salt + appkey)
payload = {
"q": query,
"from": from_lang,
"to": to_lang,
"appid": appid,
"salt": salt,
"sign": sign,
}
print(payload)
request = httpx.Request(
"POST",
url,
data=payload,
headers={"Content-Type": "application/x-www-form-urlencoded"}
)
print("完整请求内容:")
print("URL:", request.url)
print("Headers:", request.headers)
print("Body:", request.content.decode("utf-8"))
async with httpx.AsyncClient(timeout=10) as client:
response = await client.post(
url,
data=payload,
headers={'Content-Type': 'application/x-www-form-urlencoded'}
)
if response.status_code != 200:
raise HTTPException(status_code=500, detail=response.text)
data = response.json()
print(json.dumps(data, indent=2, ensure_ascii=False))
if "trans_result" not in data:
raise HTTPException(status_code=500, detail=data.get("error_msg", "Unknown error"))
return "\n".join([item["dst"] for item in data["trans_result"]])
@translator_router.post('/translate', response_model=TransResponse)
async def translate(
query: str,
from_lang: str = 'auto',
to_lang: str = 'zh',
user=Depends(get_current_user)
):
text = await baidu_translation(
query=query,
from_lang=from_lang,
to_lang=to_lang
)
return TransResponse(translated_text=text)
@translator_router.post('/translate/debug')
async def test_translate(
query: str,
from_lang: str = "auto",
to_lang: str = 'zh',
admin_user: Tuple[User, dict] = Depends(is_admin_user)
):
raw = await baidu_translation(query, from_lang, to_lang)
return TransResponse(translated_text=raw)

View File

@ -0,0 +1,28 @@
from typing import Literal
from pydantic import BaseModel, field_validator, model_validator
class TransRequest(BaseModel):
from_lang: Literal['fr', 'jp', 'zh'] = 'fr'
to_lang: Literal['fr', 'jp', 'zh'] = 'zh'
@field_validator('from_lang', 'to_lang')
@classmethod
def validate_lang(cls, v):
allowed_langs = {'auto', 'fr', 'jp', 'zh'}
if v not in allowed_langs:
raise ValueError(f'Unsupported language: {v}')
return v
@model_validator(mode="after")
def check_lang(self):
if self.from_lang == self.to_lang:
raise ValueError("from_lang and to_lang cannot be the same")
elif self.to_lang == 'auto':
raise ValueError("to_lang cannot be 'auto'")
return self
class TransResponse(BaseModel):
translated_text: str

View File

@ -6,6 +6,7 @@ import uvicorn
from tortoise.contrib.fastapi import register_tortoise from tortoise.contrib.fastapi import register_tortoise
from app.api.redis_test import redis_test_router from app.api.redis_test import redis_test_router
from app.api.translator import translator_router
from app.utils import redis_client from app.utils import redis_client
from settings import TORTOISE_ORM,ONLINE_SETTINGS from settings import TORTOISE_ORM,ONLINE_SETTINGS
from app.api.users import users_router from app.api.users import users_router
@ -54,5 +55,7 @@ app.include_router(dict_search, tags=["Dictionary Search API"])
app.include_router(redis_test_router, tags=["Redis Test-Only API"]) app.include_router(redis_test_router, tags=["Redis Test-Only API"])
app.include_router(translator_router, tags=["Translation API"])
if __name__ == "__main__": if __name__ == "__main__":
uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True) uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

5
scripts/md5.py Normal file
View File

@ -0,0 +1,5 @@
from hashlib import md5
def make_md5(s, encoding='utf-8'):
return md5(s.encode(encoding)).hexdigest()

View File

@ -43,8 +43,13 @@ ONLINE_SETTINGS = {
class Settings(BaseSettings): class Settings(BaseSettings):
USE_OAUTH = False USE_OAUTH: bool = False
SECRET_KEY = "asdasdasd-odjfnsodfnosidnfdf-0oq2j01j0jf0i1ej0fij10fd" SECRET_KEY: str
BAIDU_APPID: str
BAIDU_APPKEY: str
class Config:
env_file = '.env'
settings = Settings() settings = Settings()