title_with_video/applications/functions/common.py

# encoding: utf-8
"""
@author: luojunhui
"""
import time
import json
import uuid
import random
import hashlib
import requests
import aiohttp
import urllib.parse


def create_gzh_path(video_id, shared_uid):
    """
    :param video_id: 视频 id
    :param shared_uid: 分享 id
    """

    def auto_white(root_share_id_):
        """
        自动加入白名单， 保证公众号百分百出广告
        :param root_share_id_:
        :return:
        """

        def get_cookie():
            """
            获取 cookie
            :return:
            """
            url = "https://admin.piaoquantv.com/manager/login?account=luojunhui&passWd=e10adc3949ba59abbe56e057f20f883e&muid=7"
            payload = {}
            headers = {
                'accept': 'application/json, text/plain, */*',
                'accept-language': 'en',
                'priority': 'u=1, i',
                'sec-ch-ua': '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
                'sec-ch-ua-mobile': '?0',
                'sec-ch-ua-platform': '"macOS"',
                'sec-fetch-dest': 'empty',
                'sec-fetch-mode': 'cors',
                'sec-fetch-site': 'same-origin',
                'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'
            }
            response = requests.request("GET", url, headers=headers, data=payload)
            return response.cookies.values()[0]

        url = "https://admin.piaoquantv.com/manager/ad/own/white/rootShare/save"
        dd = {
            "rootShareId": root_share_id_,
            "commit": "算法自动加入白名单--"
        }
        payload = json.dumps(dd)
        cookie = get_cookie()
        headers = {
            'accept': 'application/json',
            'accept-language': 'en',
            'content-type': 'application/json;',
            'cookie': "SESSION=" + cookie,
            'origin': 'https://admin.piaoquantv.com',
            'priority': 'u=1, i',
            'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'
        }
        response = requests.request("POST", url, headers=headers, data=payload)
        return response.json()['content']

    def generate_source_id():
        """
        generate_source_id
        :return:
        """
        timestamp = str(int(time.time() * 1000))
        random_str = str(random.randint(1000, 9999))
        hash_input = f"{timestamp}-{random_str}"
        return hashlib.md5(hash_input.encode()).hexdigest()

    root_share_id = str(uuid.uuid4())
    source_id = "longArticles_" + generate_source_id()
    url = f"pages/user-videos?id={video_id}&su={shared_uid}&fromGzh=1&rootShareId={root_share_id}&shareId={root_share_id}&rootSourceId={source_id}"
    # 自动把 root_share_id 加入到白名单
    # auto_white(root_share_id)
    return root_share_id, source_id, f"pages/category?jumpPage={urllib.parse.quote(url, safe='')}"


def request_for_info(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()


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


def account_info_map(gh_id):
    """
    根据账号 id 来判断返回哪个小程序账号
    :param gh_id:
    :return:
    """
    pqlh = {
        "name": "票圈乐活",
        "id": "wxe8f8f0e23cecad0f",
        "avatar": "https://rescdn.yishihui.com/0temp/lehuo.png"
    }
    xyfxhyjl = {
        "name": "幸运福星好运锦鲤",
        "id": "wx95dcbfc0753c06a8",
        "avatar": "https://rescdn.yishihui.com/0temp/xyfxhyjl.png"
    }
    pqzf = {
        "name": "票圈祝福",
        "id": "wxf7261ed54f2e450e",
        "avatar": "https://rescdn.yishihui.com/0temp/pqzf.png"
    }
    buy_accounts = [
        "gh_084a485e859a",
        "gh_e24da99dc899",
        "gh_e0eb490115f5",
        "gh_183d80deffb8",
        "gh_5ff48e9fb9ef",
        "gh_9f8dc5b0c74e",
        "gh_6d9f36e3a7be"
    ]
    dyy = [
        "gh_9877c8541764",
        "gh_6d205db62f04",
        "gh_c69776baf2cd",
        "gh_7e5818b2dd83",
        "gh_89ef4798d3ea",
        "gh_a2901d34f75b",
        "gh_b15de7c99912"
    ]
    if gh_id in buy_accounts:
        return ""
    elif gh_id in dyy:
        return ""
    else:
        return ""


async def request_etl(url, headers, json_data):
    """
    异步请求ETL
    :return:
    """
    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, json=json_data, timeout=60) as response:
            response_data = await response.json()
            return response_data