title_with_video/applications/process.py

"""
@author: luojunhui
对请求进行操作
"""

import os
import json
import uuid
import requests
import urllib.parse

from applications.functions.ask_kimi import ask_kimi
from applications.functions.calculate import title_mix


class ProcessParams(object):
    """
    Params Analysis
    """

    @classmethod
    def get_params(cls, data):
        """
        "accountName": "公众号名称",
        "content": "文章正文",
        "title": "文章标题",
        "cover": "封面链接"
        :param data:
        :return: title
        """
        return data['title']

    @classmethod
    def ask_kimi_and_save_to_local(cls, title):
        """
        save file to local
        :param title:
        :return:
        """
        save_path = os.path.join(os.getcwd(), 'applications', 'static', "titles", "{}.json".format(title))
        if os.path.exists(save_path):
            return
        else:
            os.makedirs(os.path.dirname(save_path), exist_ok=True)
            if not title:
                result = "{}"
            else:
                result = ask_kimi(title)
            with open(save_path, "w", encoding="utf-8") as f:
                f.write(json.dumps(result, ensure_ascii=False))

    @classmethod
    def create_gzh_path(cls, video_id, shared_uid):
        """
        :param video_id: 视频 id
        :param shared_uid: 分享 id
        """
        root_share_id = str(uuid.uuid4())
        url = f"pages/user-videos?id={video_id}&su={shared_uid}&fromGzh=1&rootShareId={root_share_id}&shareId={root_share_id}"
        return root_share_id, f"pages/category?jumpPage={urllib.parse.quote(url)}"

    @classmethod
    def request_for_info(cls, video_id):
        """
        请求数据
        :param video_id:
        :return:
        """
        url = "https://longvideoapi.piaoquantv.com/longvideoapi/openapi/video/batchSelectVideoInfo"
        data = {
            "videoIdList": [video_id]
        }
        header = {
            "Content-Type": "application/json",
        }
        response = requests.post(url, headers=header, data=json.dumps(data))
        return response.json()

    @classmethod
    def process(cls, data):
        """执行代码"""
        title = cls.get_params(data)
        title_p = os.path.join(os.getcwd(), 'applications', 'static', "titles", "{}.json".format(title))
        if os.path.exists(title_p):
            result = title_mix(title_p=title_p, dt="20240417")
        else:
            cls.ask_kimi_and_save_to_local(title)
            result = title_mix(title_p=title_p, dt="20240417")
        uid, video_id = result['s1_uid'], result['s1_vid']
        root_share_id, productionPath = cls.create_gzh_path(uid, video_id)
        response = cls.request_for_info(video_id)
        productionCover = response["data"][0]['coverImg']['coverImgPath']
        productionName = response["data"][0]['title']
        programAvatar = "Avatar"
        programId = "wx89e7eb06478361d7"
        programName = "票圈vlog"
        source = "Web"
        result = {
            "productionCover": productionCover,
            "productionName": productionName,
            "programAvatar": programAvatar,
            "programId": programId,
            "programName": programName,
            "source": source,
            "rootShareId": root_share_id,
            "productionPath": productionPath
        }
        return result