title_with_video/applications/functions/aigc.py

"""
@author: luojunhui
"""
import asyncio

import aiohttp

TIMEOUT_CODE = 0


async def async_record(url, params) -> int:
    """
    异步请求函数
    """
    try:
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10)) as session:
            async with session.get(url, params=params) as response:
                return response.status
    except asyncio.TimeoutError:
        return TIMEOUT_CODE


async def record_trace_id(trace_id, status) -> bool:
    """
    记录成功的trace_id
    """
    OK_STATUS = 200
    RETRY_TIMES = 3
    url = "http://aigc-api.cybertogether.net//aigc/publish/api/notifyMatchMiniprogramResult"
    params = {"traceId": trace_id, "status": status}
    # response_status_code = await async_record(url, params)
    for _ in range(RETRY_TIMES):
        response_status_code = await async_record(url, params)
        if response_status_code == OK_STATUS:
            return True
        await asyncio.sleep(1)
    return False