添加colg金价工具

backend/api/core/Redis.py

@@ -2,7 +2,8 @@ import json
 from api.core.Gzip import gzip_zip, gzip_unzip
 from config.main import config
 
-def get_redis_info(redis, key, fun,without_gzip=False):
+
+def get_redis_info(redis, key, fun, expire = 0, without_gzip=False):
     info = None
     if redis and not config.DEBUG_MODE:
         try:
@@ -19,4 +20,6 @@ def get_redis_info(redis, key, fun,without_gzip=False):
                 redis.set(key, json.dumps(info))
             else:
                 redis.set(key, gzip_zip(json.dumps(info)))
+            if expire > 0:
+                redis.expire(key, expire)
     return info

backend/api/main.py

@@ -3,8 +3,10 @@ from fastapi import APIRouter
 from api.routes import info
 from api.routes import calc
 from api.routes import openapi
+from api.routes import colg
 
 api_router = APIRouter()
 api_router.include_router(info.router,prefix="/api",tags=["info"])
 api_router.include_router(calc.router,prefix="/api",tags=["calc"])
 api_router.include_router(openapi.router,prefix="/api/open",tags=["openapi"])
+api_router.include_router(colg.router,prefix="/api/open",tags=["colg"])

backend/api/routes/colg.py

@@ -0,0 +1,128 @@
+from api.types.colg import HistoryGold, RealTimeGold
+from fastapi.params import Param
+from api.core.Gzip import register_gzip_request
+import json
+from fastapi import APIRouter, Path, Query
+from api.core.Response import response
+from api.core.Redis import get_redis_info
+from api.dp import RedisDep
+from typing import Annotated
+import requests
+from config.main import config
+from datetime import datetime, timezone, timedelta
+from api.core.Response import Return as Response
+
+router = APIRouter()
+
+register_gzip_request(router)
+
+regionMap = {
+    '跨1': 1,
+    '跨2': 2,
+    '跨3A': 3,
+    '跨3B': 4,
+    '跨4': 5,
+    '跨5': 6,
+    '跨6': 7,
+    '跨7': 8,
+}
+
+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'}
+
+
+def get_date_str(timestamp: int) -> str:
+    utc_time = datetime.fromtimestamp(timestamp, tz=timezone.utc)
+    beijing_time = utc_time + timedelta(hours=8)
+    return beijing_time.strftime('%Y-%m-%d %H:%M:%S')
+
+
+@router.get('/gold/realtime/', operation_id='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():
+                payload['area'] = region_id
+                payload['start_time'] = datetime.now().strftime('%Y-%m-%d')
+                payload['end_time'] = datetime.now().strftime('%Y-%m-%d')
+                response = requests.post(url, headers={'Content-Type': 'application/json'}, data=json.dumps(payload))
+                response.raise_for_status()
+                resp = response.json().get('data', {})
+                res.append({'updataTime': get_date_str(resp['x'][-1]), 'region': region_name, 'goldPrice': resp['y'][-1]})
+            return res
+    except Exception as e:
+        # print(f"File not found: {e}")
+        return response(data=[])
+    # 设置半小时过期一次
+    data = get_redis_info(redis, key, get_gold_data, 30 * 60)
+    return response(data=data)
+
+
+@router.get('/gold/history/', operation_id='historyGold')
+async def get_history_gold(region: Annotated[int, Query(..., description='跨区id')], timeRange: Annotated[int, Query(..., description='时间跨度天数')] = 7) -> Response[HistoryGold] | Response[None]:
+    """
+    获取历史金价数据
+
+    Args:
+        region (int): 跨区id，必填参数
+        timeRange (int, optional): 时间跨度天数，默认为7天
+
+    Returns:
+        Response[HistoryGold]: 返回历史金价数据的响应对象
+
+    Raises:
+        无显式异常，但会在跨区ID无效时返回404错误码
+
+    Note:
+        - region参数必须在regionMap的值范围内
+        - timeRange参数是可选的，使用默认值7表示最近7天的数据
+    """
+    print(region, timeRange)
+    if region not in regionMap.values():
+        return response(
+            code=404,
+            message='跨区ID无效',
+            data=None,
+        )
+    region_name = [k for k, v in regionMap.items() if v == region][0]
+    end_date = datetime.now()
+    start_date = end_date - timedelta(days=timeRange - 1)
+    payload['area'] = region
+    payload['start_time'] = start_date.strftime('%Y-%m-%d')
+    payload['end_time'] = end_date.strftime('%Y-%m-%d')
+    data = []
+    echartsData = {}
+    try:
+        req = requests.post(url, headers={'Content-Type': 'application/json'}, data=json.dumps(payload))
+        req.raise_for_status()
+        resp = req.json().get('data', {})
+        for i in range(len(resp['x'])):
+            data.append(
+                {
+                    'date': get_date_str(resp['x'][i]).split(' ')[0],
+                    'avg': resp['y'][i],
+                    'zero': resp['y1'][i],
+                }
+            )
+    except Exception as e:
+        return response(data=None)
+    echartsData = {
+        'title': {'text': f'DNF{region_name}近{timeRange}天金币价格趋势分析', 'left': 'center'},
+        'toolbox': {'feature': {'saveAsImage': {}, 'dataView': {'readOnly': True}, 'magicType': {'type': ['line', 'bar']}}},
+        'tooltip': {'trigger': 'axis', 'axisPointer': {'type': 'cross', 'label': {'backgroundColor': '#6a7985'}}},
+        'legend': {'data': ['平均金价', '12时金价'], 'bottom': 0},
+        'xAxis': {'type': 'category', 'name': '日期', 'boundaryGap': False, 'data': [item['date'] for item in data]},
+        'yAxis': {'type': 'value', 'name': '价格 (万金币/1人民币)', 'min': 65, 'max': 75},
+        'series': [
+            {'name': '平均金价', 'type': 'line', 'smooth': True, 'data': [item['avg'] for item in data], 'markPoint': {'data': [{'type': 'max', 'name': '最大值'}, {'type': 'min', 'name': '最小值'}]}},
+            {'name': '峰值金价', 'type': 'line', 'smooth': True, 'data': [item['zero'] for item in data], 'markLine': {'data': [{'type': 'average', 'name': '平均值'}]}},
+        ],
+    }
+    return response(data={'region': region_name, 'data': data, 'echartsData': echartsData})

backend/api/routes/openapi.py

@@ -219,7 +219,6 @@ async def get_skill_data_summary(
     skills = skills.split(",") if skills else []
     weapons = weapons.split(",") if weapons else []
     try:
-
         def get_skill_data_summary():
             # This function should retrieve the skill data summary based on job
             # For now, we return a placeholder dictionary

backend/api/types/colg.py

@@ -0,0 +1,29 @@
+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 RealTimeGold(BaseModel):
+    updataTime: str = Field(..., description='更新时间')
+    """更新时间"""
+    region: str = Field(..., description='跨区')
+    """跨区"""
+    goldPrice: float = Field(..., description='金价')
+    """金价"""
+
+class DayGold(BaseModel):
+    date: str = Field(..., description='日期')
+    """日期"""
+    avg: float = Field(..., description='峰值')
+    """峰值"""
+    zero: float = Field(..., description='12时值')
+    """12时值"""
+
+class HistoryGold(BaseModel):
+    region: str = Field(..., description='跨区')
+    """跨区"""
+    data: list[DayGold] = Field(..., description='历史数据')
+    echartsData : dict = Field(..., description='echarts数据')
+    """echarts数据"""