dnftools/dcalc
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Activity

dev

Browse Source
This commit is contained in:
imelon
2026-03-09 18:30:23 +08:00
parent fffee149ec
commit 621023bf34
10 changed files with 955 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/api/core/Response.py
+5 -4
View File
@@ -1,14 +1,13 @@
import base64
import json
from typing import Generic, TypeVar
from pydantic import Field
from pydantic.generics import GenericModel
from pydantic import Field, BaseModel
import gzip # 导入 gzip 模块
from io import BytesIO # 从 io 模块导入 BytesIO 类
T = TypeVar('T')
class Return(GenericModel, Generic[T]):
class Return(BaseModel, Generic[T]):
code: int = Field(..., description='响应状态码')
"""响应状态码"""
message: str = Field(..., description='响应消息')
@@ -16,6 +15,7 @@ class Return(GenericModel, Generic[T]):
data: T = Field(..., description='响应数据')
"""响应数据"""
# 定义一个函数，将字符串压缩为 gzip 格式，并进行 base64 编码后返回结果
def gzip_str(to_gzip: str) -> str:
out = BytesIO() # 创建一个 BytesIO 对象
@@ -24,9 +24,10 @@ def gzip_str(to_gzip: str) -> str:
return base64.b64encode(out.getvalue()).decode() # 对压缩后的数据使用 base64 编码并返回结果
def response(*, code=200, data: list | dict | str, message='Success', zip=False)-> Return:
def response(*, code=200, data: list | dict | str, message='Success', zip=False) -> Return:
if zip:
import gzip
data = base64.b64encode(gzip.compress(json.dumps(data).encode()))
return Return(code=code, message=message, data=data)
backend/api/routes/colg.py
+35 -29
View File
@@ -34,6 +34,12 @@ url = 'https://poservice.colg.cn/api/v1/trend/getCoinTrendDataKol'
payload = {'token': config.COLG_TOKEN, 'area': 0, 'time_type': 'other', 'start_time': '2026-01-06', 'end_time': '2026-01-06'}
payload_helper = {
'r': 'godRank/godDealRank',
'userId': config.HELPER_ID,
'token': config.HELPER_TOKEN,
}
def get_date_str(timestamp: int) -> str:
utc_time = datetime.fromtimestamp(timestamp, tz=timezone.utc)
@@ -41,13 +47,14 @@ def get_date_str(timestamp: int) -> str:
return beijing_time.strftime('%Y-%m-%d %H:%M:%S')
@router.get('/gold/realtime/', operation_id='realTimeGold')
@router.get('/gold/realtime/', operation_id='realtime_gold',response_model=Response[list[RealTimeGold]])
async def get_real_time_gold(redis: RedisDep) -> Response[list[RealTimeGold]]:
"""
返回所有跨区的实时金价信息
"""
key = 'openapi:colg:real_time_gold'
try:
def get_gold_data():
res = []
for region_name, region_id in regionMap.items():
@@ -67,7 +74,7 @@ async def get_real_time_gold(redis: RedisDep) -> Response[list[RealTimeGold]]:
return response(data=data)
@router.get('/gold/history/', operation_id='historyGold')
@router.get('/gold/history/', operation_id='history_gold',response_model=Response[HistoryGold] | Response[None])
async def get_history_gold(region: Annotated[int, Query(..., description='跨区id')], timeRange: Annotated[int, Query(..., description='时间跨度天数')] = 7) -> Response[HistoryGold] | Response[None]:
"""
获取历史金价数据
@@ -139,46 +146,45 @@ async def get_history_gold(region: Annotated[int, Query(..., description='跨区
return response(data={'region': region_name, 'data': data, 'echartsData': echartsData})
@router.get('/gold/official/', operation_id='officialGold')
@router.get('/gold/official/', operation_id='official_gold',response_model=Response[list[OfficialGold]])
async def get_official_gold(redis: RedisDep) -> Response[list[OfficialGold]]:
"""
获取官方金币寄售实时数据
获取DNF官方金币寄售实时数据
Returns:
Response[list[OfficialGold]]: 包含官方金币寄售数据的响应对象
Note:
- 该接口会返回所有跨区的官方金币寄售数据，包括最低成交价排名前5的价格和最高成交价排名前5的价格
- 数据来源于DNF官方提供的接口，可能会有一定的延迟和不稳定性
"""
official_url = 'https://dzhu.qq.com/zangyi/activity/app'
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
params = {
'r': 'godRank/godDealRank',
'userId': config.HELPER_ID,
'token': config.HELPER_TOKEN,
}
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
key = 'openapi:colg:real_time_gold_officical'
list = ['1', '2', '3A', '3B', '4', '5', '6', '7']
try:
async def get_data():
datas = []
async def fetch_data():
async with aiohttp.ClientSession() as session:
tasks = []
for i in list:
params['serverNo'] = i
query = '&'.join(f'{k}={v}' for k, v in params.items())
tasks.append(session.post(official_url, data=query, headers=headers, timeout=10))
responses = await asyncio.gather(*tasks)
for index, resp in enumerate(responses):
res = json.loads(await resp.read())
datas.append({
async with aiohttp.ClientSession() as session:
tasks = []
for i in list:
payload_helper['serverNo'] = i
query = '&'.join(f'{k}={v}' for k, v in payload_helper.items())
tasks.append(session.post(official_url, data=query, headers=headers, timeout=10))
responses = await asyncio.gather(*tasks)
for index, resp in enumerate(responses):
res = json.loads(await resp.read())
datas.append(
{
'updataTime': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'region': f'跨{list[index]}',
'godDealRank': [int(i.get('price')) for i in res['data']['godDealRank']],
'godTopDealRank': [int(i.get('price')) for i in res['data']['godTopDealRank']],
})
return datas
return await fetch_data()
'goldDealRank': [int(i.get('price')) for i in res['data']['godDealRank']],
'goldTopDealRank': [int(i.get('price')) for i in res['data']['godTopDealRank']],
}
)
return datas
except Exception as e:
print(e)
return response(data=[])
# 设置半小时过期一次
data = await get_redis_info(redis, key, get_data, 30 * 60)
return response(data=data)
return response(data=data)
backend/api/routes/openapi.py
+5 -5
View File
@@ -37,7 +37,7 @@ def replace_placeholders(template):
return pattern.sub(repl, template)
@router.get("/skills/", operation_id="skillsList")
@router.get("/skills/", operation_id="skills_list", response_model=SearchSkillResponse)
async def get_skills_list(
redis: RedisDep,
skillName: Annotated[str, Param(..., description="技能名称，支持模糊查询")] = None,
@@ -69,7 +69,7 @@ async def get_skills_list(
return response(data=data)
@router.get("/jobInfo/", operation_id="jobsInfo")
@router.get("/jobInfo/", operation_id="jobs_info", response_model=JobInfoResponse)
async def get_jobs_info(
redis: RedisDep,
jobName: Annotated[str, Query(..., description="职业名称，支持模糊查询")] = None,
@@ -101,7 +101,7 @@ async def get_jobs_info(
return response(data=result)
@router.get("/{jobId}/{jobGrowId}/skills/", operation_id="skillsByJob")
@router.get("/{jobId}/{jobGrowId}/skills/", operation_id="skills_by_job",response_model=SkillsListResponse)
async def get_cn_skills_by_job(
redis: RedisDep,
jobId: Annotated[str, Path(..., description="职业")],
@@ -125,7 +125,7 @@ async def get_cn_skills_by_job(
return response(data=skills_list)
@router.get("/{jobId}/{jobGrowId}/{skillId}/", operation_id="skillDetail")
@router.get("/{jobId}/{jobGrowId}/{skillId}/", operation_id="skill_detail", response_model=SkillDetailResponse)
async def get_cn_skill_info(
redis: RedisDep,
jobId: Annotated[str, Path(..., description="职业")],
@@ -192,7 +192,7 @@ async def get_cn_skill_info(
return response(data=skill_info)
@router.get("/skillData/summary/", operation_id="skillDataSummary")
@router.get("/skillData/summary/", operation_id="skill_data_summary",response_model=SkillDataSummaryResponse)
async def get_skill_data_summary(
redis: RedisDep,
job: Annotated[str, Param(..., description="职业")],
backend/api/types/colg.py
+3 -4
View File
@@ -1,7 +1,6 @@
from enum import Enum
from pydantic import BaseModel, Field
from typing import Generic, TypeVar
from pydantic.generics import GenericModel
from api.core.Response import Return as Response
@@ -10,7 +9,7 @@ class RealTimeGold(BaseModel):
"""更新时间"""
region: str = Field(..., description='跨区')
"""跨区"""
goldPrice: float = Field(..., description='金价')
goldPrice: float|str = Field(..., description='金价')
"""金价"""
class DayGold(BaseModel):
@@ -33,7 +32,7 @@ class OfficialGold(BaseModel):
"""更新时间"""
region: str = Field(..., description='跨区')
"""跨区"""
godDealRank: list[float] = Field(..., description='金币寄售最低成交价排名前5的价格')
goldDealRank: list[float|str] = Field(..., description='金币寄售最低成交价排名前5的价格')
"""金币寄售最低成交价排名前5的价格"""
godTopDealRank: list[float] = Field(..., description='金币寄售最高成交价排名前5的价格')
goldTopDealRank: list[float|str] = Field(..., description='金币寄售最高成交价排名前5的价格')
"""金币寄售最高成交价排名前5的价格"""
backend/api/types/open.py
-3
View File
@@ -1,10 +1,7 @@
from enum import Enum
from pydantic import BaseModel, Field
from typing import Generic, TypeVar
from pydantic.generics import GenericModel
from api.core.Response import Return as Response
class SkillInfo(BaseModel):
skillId: str = Field(..., description='技能ID')
"""技能ID"""