video_rewriting/data_channel/douyin.py

import json
import random
import time

import requests
import urllib3
from requests.adapters import HTTPAdapter
from common import Material, Common, Feishu
from common.sql_help import sqlCollect
from data_channel.douyin_help import DouYinHelper


class DY:

    @classmethod
    def get_dy_url(cls, task_mark, url_id, number, mark, feishu_id, cookie_sheet, channel_id, name):
        list = []
        next_cursor = 0
        for i in range(3):
            cookie = Material.get_cookie_data(feishu_id, cookie_sheet, channel_id)
            time.sleep(random.randint(5, 10))
            url = 'https://www.douyin.com/aweme/v1/web/aweme/post/'
            headers = {
                'Accept': 'application/json, text/plain, */*',
                'Accept-Language': 'zh-CN,zh;q=0.9',
                'Cache-Control': 'no-cache',
                'Cookie': cookie,
                'Pragma': 'no-cache',
                'Referer': f'https://www.douyin.com/user/{url_id}',
                'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) '
                              'Chrome/118.0.0.0 Safari/537.36',
            }
            query = DouYinHelper.get_full_query(ua=headers['User-Agent'], extra_data={
                'sec_user_id': url_id,
                'max_cursor': next_cursor,
                'locate_query': 'false',
                'show_live_replay_strategy': '1',
                'need_time_list': '1',
                'time_list_query': '0',
                'whale_cut_token': '',
                'cut_version': '1',
                'count': '18',
                'publish_video_strategy_type': '2',
            })
            urllib3.disable_warnings()
            s = requests.session()
            s.mount('http://', HTTPAdapter(max_retries=3))
            s.mount('https://', HTTPAdapter(max_retries=3))
            response = requests.request(method='GET', url=url, headers=headers, params=query)
            body = response.content.decode()
            obj = json.loads(body)
            has_more = True if obj.get('has_more', 0) == 1 else False
            next_cursor = str(obj.get('max_cursor')) if has_more else None
            data = obj.get('aweme_list', [])
            if data == [] and len(data) == 0:
                return list
            if not data[0].get('search_impr').get('entity_type'):
                Feishu.bot(mark, '机器自动改造消息通知', f'今日任务为空，请关注', name)
                time.sleep(900)
                continue
            response.close()
            for i in range(len(data)):
                entity_type = data[i].get('search_impr').get('entity_type')
                if entity_type == 'GENERAL':
                    # is_top = data[i].get('is_top')  # 是否置顶
                    video_id = data[i].get('aweme_id')  # 文章id
                    status = sqlCollect.is_used(task_mark, video_id, mark, channel_id)
                    if status:
                        video_uri = data[i].get('video', {}).get('play_addr', {}).get('uri')
                        ratio = f'{data[i].get("video", {}).get("height")}p'
                        video_url = f'https://www.iesdouyin.com/aweme/v1/play/?video_id={video_uri}&ratio={ratio}&line=0'  # 视频链接
                        cover_url = data[i].get('video').get('cover').get('url_list')[0]  # 视频封面
                        all_data = {"video_id": video_id, "cover": cover_url, "video_url": video_url}
                        list.append(all_data)
                        if len(list) == int(number):
                            Common.logger("log").info(f"获取抖音视频总数:{len(list)}\n")
                            return list
        return list