title_with_video/applications/functions/common.py

# encoding: utf-8
"""
@author: luojunhui
"""
import random
import aiohttp
import asyncio


def shuffle_list(ori_list):
    """
    随机打乱 list
    :param ori_list:
    :return:
    """
    shuffled_list = ori_list[:]  # 使用切片操作创建副本
    random.shuffle(shuffled_list)
    return shuffled_list


def clean_title(strings):
    """
    :param strings:
    :return:
    """
    return (
        strings.strip()
        .replace("\n", "")
        .replace("/", "")
        .replace("\r", "")
        .replace("#", "")
        .replace(".", "。")
        .replace("\\", "")
        .replace("&NBSP", "")
        .replace(":", "")
        .replace("*", "")
        .replace("？", "")
        .replace("?", "")
        .replace('"', "")
        .replace("<", "")
        .replace(">", "")
        .replace("|", "")
        .replace(" ", "")
        .replace('"', "")
        .replace("'", "")
    )


def sensitive_flag(s_words, ori_title):
    """
    :param s_words:
    :param ori_title:
    :return:
    """
    # for word in s_words:
    #     if str(word) in ori_title:
    #         return False
    return True


async def request_etl(url, headers, json_data, retries=6):
    """
    :param url:
    :param headers:
    :param json_data:
    :param retries:
    :return:
    """
    async with aiohttp.ClientSession() as session:
        for attempt in range(retries):
            try:
                async with session.post(url, headers=headers, json=json_data, timeout=120) as response:
                    return await response.json()
            except asyncio.TimeoutError:
                if attempt < retries - 1:
                    await asyncio.sleep(2)  # 等待一段时间后重试
                else:
                    raise


async def async_post(url, headers, payload):
    """
    :param url:
    :param headers:
    :param payload:
    :return:
    """
    retries = 3
    async with aiohttp.ClientSession() as session:
        for attempt in range(3):
            try:
                async with session.post(url, headers=headers, data=payload, timeout=60) as response:
                    return await response.json()
            except asyncio.TimeoutError:
                if attempt < retries - 1:
                    await asyncio.sleep(2)  # 等待一段时间后重试
                else:
                    raise