add generate_a_bogus api

api.py

+import execjs
 from loguru import logger
 
 
@@ -34,6 +35,10 @@ class DoudianLoginRequest(BaseModel):
 class DoudianUploadVideoRequest(BaseModel):
     task: dict
 
+class GenerateABogusRequest(BaseModel):
+    params: str | dict
+    data: dict = {}
+    ua: str
 
 
 @app.post(
@@ -159,6 +164,26 @@ def doudian_upload_video(req: DoudianUploadVideoRequest):
             'error_type': 'TimeoutError'
         }
 
+@sync_router.post(
+    "/generate_a_bogus",
+    status_code=status.HTTP_200_OK,
+    summary="生成加密参数",
+)
+def generate_a_bogus(req: GenerateABogusRequest):
+    try:
+        task = celery_app.send_task('generate_a_bogus', kwargs=req.dict())
+        result = task.get(timeout=60)
+        logger.info(f"任务结果: {result}")
+        return result
+    except Exception as e:
+        logger.error(f"任务执行失败: {str(e)}")
+        return {
+            'code': 504,
+            'msg': '请求超时',
+            'data': None,
+            'error_type': 'TimeoutError'
+        }
+
 # 注册路由
 app.include_router(sync_router)
 
@@ -187,3 +212,7 @@ def get_task_result(task_id: str):
     except Exception as e:
         logger.error(f"查询任务状态失败: {str(e)}")
         raise AppError
+
+
+
+

task_worker.py

+from urllib.parse import urlencode
+
+import execjs
 from celery import shared_task
 
-from upload_video import upload_video
-from login import login
 from errors import *
 from hub_ import *
+from login import login
 from main import syncShop, syncShopInfo, createTemplate
+from upload_video import upload_video
 
 
 @shared_task(name='sync_shop')
@@ -153,3 +156,30 @@ def execute_doudian_upload_video(task: dict):
             'data': None,
             'error_type': 'InternalError'
         }
+
+@shared_task(name='generate_a_bogus')
+def execute_generate_a_bogus(params:str | dict,data: dict,ua: str):
+    """生成加密参数"""
+    try:
+        # 直接读取JS文件
+        with open('./js/a_bogus.js', 'r', encoding='gbk') as f:
+            js_code = f.read()
+        # 执行加密算法
+        ctx = execjs.compile(js_code)
+        if type(params) == dict:
+            params = urlencode(params)
+        result = ctx.call('get_ab', params, data, ua)
+        return {
+            'code': 200,
+            'msg': 'success',
+            'data': result,
+            'error_type': ''
+        }
+    except Exception as e:
+        logger.error(f'生成加密异常:{e}')
+        return {
+            'code': 500,
+            'msg': f'生成加密异常:{e}',
+            'data': None,
+            'error_type': 'InternalError'
+        }
\ No newline at end of file