single_video_crawler/single_video/kuaishou/kuaishou_video.py

import json
import re

import requests
from bs4 import BeautifulSoup

from common import Oss, Common
from common.pq_utility import PQ
from common.url_manage import urlManage
from common.userAgent import get_random_user_agent


class kuaishouVideo():
    # 获取快手 标题+视频链接
    @classmethod
    def get_videoList(cls, vx_message, channel):
        data_link = vx_message[1]
        content_id = ''
        for i in range(3):
            content_id = urlManage.url_manage(data_link, channel)
            if content_id:
                break
        if content_id == '':
            return "无法获取到视频ID"
        Common.logger().info(f'渠道:{channel},视频ID:{content_id}')
        for i in range(3):
            try:
                url = f"https://www.kuaishou.com/short-video/{content_id}"
                headers = {
                  'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
                  'Accept-Language': 'zh-CN,zh;q=0.9',
                  'Cache-Control': 'no-cache',
                  'Connection': 'keep-alive',
                  'Cookie': 'kpf=PC_WEB; clientid=3; did=web_9c6a04a4004fdb7c95a658a56ed275b6; userId=1299331643; kuaishou.server.web_st=ChZrdWFpc2hvdS5zZXJ2ZXIud2ViLnN0EqAB-mmmFeafQwQfZLqo6IV4sqTuqIs-bOdErRSFOm82rvJlBMe7mKsCpUL4ROkEaBb8GWLY7K5YD8ChJx1TH-hcY-GwVQCZzs3LS0RgD2-Kbu0VLoFK51YJ3FXI0Nj3kzmR3D15g3tqngkgxfMc58JnUcdQYr9ed8L0QIj2p3n3LT33M3MTdn_TYFTcQhQW9mwkTCSu8rg5APzZirdeUABfnRoSnIqSq99L0mk4jolsseGdcwiNIiDEUt4eoMq2rp70rCoXLaEZTI3KhjdlR8gZ_1Vc9CzhMygFMAE; kuaishou.server.web_ph=91777c1e75e3c21302c35753faccdbb28d85; kpn=KUAISHOU_VISION; did=web_9c6a04a4004fdb7c95a658a56ed275b6; clientid=3; kpf=PC_WEB; kpn=KUAISHOU_VISION',
                  'Pragma': 'no-cache',
                  'Sec-Fetch-Dest': 'document',
                  'Sec-Fetch-Mode': 'navigate',
                  'Sec-Fetch-User': '?1',
                  'Upgrade-Insecure-Requests': '1',
                  'User-Agent': get_random_user_agent(),
                }

                response = requests.get(url, headers=headers)
                if "VisionVideoDetailPhoto" not in response.text:
                    return "cookie过期"
                status_code = response.status_code
                if status_code != 200:
                    return "cookie过期"
                data = response.text
                soup = BeautifulSoup(data, 'html.parser')
                script_tags = soup.find_all('script', text=lambda x: x and "__APOLLO_STATE__" in x)
                pattern = re.compile(r'window\.__APOLLO_STATE__\s*=\s*({.*?});', re.DOTALL)
                for script in script_tags:
                    # 获取<script>标签内容
                    script_content = script.string
                    # 在<script>标签内容中查找匹配的模式
                    match = pattern.search(script_content)
                    if match:
                        # 提取JSON数据
                        json_data = match.group(1)
                        json_data = json.loads(json_data)
                        data = json_data["defaultClient"][f"VisionVideoDetailPhoto:{content_id}"]
                        caption = data["caption"]
                        photoUrl = data["photoUrl"]
                        photoUrl = bytes(photoUrl, "utf-8").decode("unicode_escape")
                        print(f"标题:{caption},视频链接:{photoUrl}")
                        # 随机生成视频oss_id
                        video_id = urlManage.random_id()
                        oss_object_key = Oss.video_url_upload_oss(photoUrl, video_id)
                        # 获取 oss 视频地址
                        oss_object_key = oss_object_key.get("oss_object_key")
                        Common.logger().info(f'准备发送站内参数:{oss_object_key},{caption},{vx_message[3]}')
                        piaoquantv = PQ.insert_piaoquantv(oss_object_key, caption, vx_message[3])
                        if piaoquantv == False:
                            return "视频发送到站内失败"
            except Exception as e:
                Common.logger().info(f'报错信息:{e}')