LongArticleTaskServer/app/api/v1/routes.py

from quart import Blueprint, jsonify, request
from app.ab_test import GetCoverService
from app.infra.shared.tools import generate_task_trace_id

from app.core.config import GlobalConfigSettings
from app.core.database import DatabaseManager
from app.core.observability import LogService

from app.api.service import TaskScheduler
from app.api.service import TaskManager
from app.api.service import GzhCookieManager


server_blueprint = Blueprint("api", __name__, url_prefix="/api")


def server_routes(
    pools: DatabaseManager, log_service: LogService, config: GlobalConfigSettings
):
    @server_blueprint.route("/get_cover", methods=["POST"])
    async def get_cover():
        params = await request.get_json()
        task = GetCoverService(pools, params)
        return jsonify(await task.deal())

    @server_blueprint.route("/run_task", methods=["POST"])
    async def run_task():
        trace_id = generate_task_trace_id()
        data = await request.get_json()
        task_scheduler = TaskScheduler(data, log_service, pools, trace_id, config)
        response = await task_scheduler.deal()
        return jsonify(response)

    @server_blueprint.route("/health", methods=["GET"])
    async def hello_world():
        # data = await request.get_json()
        return jsonify({"message": "hello world"})

    @server_blueprint.route("/tasks", methods=["POST"])
    async def task_list():
        data = await request.get_json()
        TMS = TaskManager(pool=pools, data=data, config=config)
        res = await TMS.list_tasks()
        return jsonify(res)

    @server_blueprint.route("/save_token", methods=["POST"])
    async def save_token():
        data = await request.get_json()
        GCM = GzhCookieManager(pool=pools, log_client=log_service)
        res = await GCM.deal(data)
        return jsonify(res)

    return server_blueprint