title_with_video/test_code/publish_by_async.py

"""
@author: luojunhui
"""
import asyncio
import json
import os.path

import aiohttp
import aiofiles


async def write_json_file(old_video_id, new_obj):
    """
    异步写文件
    :param old_video_id:
    :param new_obj:
    """
    filename = "vv/{}.json".format(old_video_id)
    async with aiofiles.open(filename, mode='w', encoding='utf-8') as f:
        await f.write(json.dumps(new_obj, ensure_ascii=False, indent=4))


async def fetch_data(url, headers, payload):
    """
    :param url:
    :param headers:
    :param payload:
    :return:
    """
    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, data=payload) as response:
            return await response.json()


async def publish(obj):
    """

    :param obj:
    :return:
    """
    cover = obj['cover']
    uid = obj['uid']
    title = obj['title']
    video_path = obj['oss_path']
    old_video_id = obj['vid']
    path = "vv/{}.json".format(old_video_id)
    if os.path.exists(path):
        pass
    else:
        try:
            url = "https://vlogapi.piaoquantv.com/longvideoapi/crawler/video/send"
            headers = {
                "User-Agent": "PQSpeed/486 CFNetwork/1410.1 Darwin/22.6.0",
                "cookie": "JSESSIONID=4DEA2B5173BB9A9E82DB772C0ACDBC9F; JSESSIONID=D02C334150025222A0B824A98B539B78",
                "referer": "http://appspeed.piaoquantv.com",
                "token": "524a8bc871dbb0f4d4717895083172ab37c02d2f",
                "accept-language": "zh-CN,zh-Hans;q=0.9",
                "Content-Type": "application/x-www-form-urlencoded",
            }
            payload = {
                "coverImgPath": cover,
                "deviceToken": "9ef064f2f7869b3fd67d6141f8a899175dddc91240971172f1f2a662ef891408",
                "fileExtensions": "MP4",
                "loginUid": uid,
                "networkType": "Wi-Fi",
                "platform": "iOS",
                "requestId": "fb972cbd4f390afcfd3da1869cd7d001",
                "sessionId": "362290597725ce1fa870d7be4f46dcc2",
                "subSessionId": "362290597725ce1fa870d7be4f46dcc2",
                "title": title,
                "token": "524a8bc871dbb0f4d4717895083172ab37c02d2f",
                "uid": uid,
                "versionCode": "486",
                "versionName": "3.4.12",
                "videoFromScene": "1",
                "videoPath": video_path,
                "viewStatus": "1",
            }
            new_obj = await fetch_data(url, headers, payload)
            await write_json_file(old_video_id, new_obj)
        except:
            pass
        # return new_obj['data']['id']


# 批量处理函数
async def process_in_batches(task_list, batch_size):
    """

    :param task_list:
    :param batch_size:
    :return:
    """
    for i in range(0, len(task_list), batch_size):
        batch = task_list[i:i + batch_size]  # 拆分批次
        tasks = [publish(params) for params in batch]
        results = await asyncio.gather(*tasks)
        print("批次{}结果:".format(i), results)


async def main():
    with open("id_dict.json", encoding="utf-8") as f:
        data = json.loads(f.read())
    L = []
    for key in data:
        detail = data[key]
        detail['vid'] = key
        L.append(detail)
    task_list = L
    tasks = [publish(params) for params in task_list]
    await process_in_batches(task_list, batch_size=30)
    await asyncio.gather(*tasks)

asyncio.run(main())