LongArticlesJob/applications/aiditApi.py

"""
@author: luojunhui
通过抓包 aigc 平台，自动化一些操作
"""
import requests
import json

from applications.decoratorApi import retryOnTimeout
from applications.denetMysql import DeNetMysql

HEADERS = {
    'Accept': 'application/json',
    'Accept-Language': 'zh,zh-CN;q=0.9',
    'Content-Type': 'application/json',
    'Origin': 'http://admin.cybertogether.net',
    'Proxy-Connection': 'keep-alive',
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36'
}
PERSON_COOKIE = {
    "token": "af54cdc404c3464d896745df389b2dce",
    "appType": 9,
    "platform": "pc",
    "appVersionCode": 1000,
    "clientTimestamp": 1,
    "fid": 1,
    "loginUid": 1,
    "pageSource": 1,
    "requestId": 1,
    "rid": 1,
    "uid": 1
}

def get_generated_article_list(plan_id):
    db = DeNetMysql()
    sql = f"""
        SELECT 
            account.wx_gh,
            content.title,
            content.content_link,
            content.view_count,
            content.like_count,
            from_unixtime(cprr.create_timestamp / 1000) AS 抓取时间,
            from_unixtime(content.publish_timestamp / 1000) AS 发布时间
        FROM crawler_plan_result_rel cprr
        JOIN crawler_plan plan ON cprr.plan_id = plan.id
        JOIN crawler_content content ON cprr.channel_source_id = content.channel_content_id
        JOIN crawler_account account ON content.channel_account_id = account.channel_account_id
        WHERE plan_id IN (
            SELECT
                input_source_value
            FROM
                produce_plan_input_source
            WHERE plan_id = '{plan_id}'
            );
                """
    article_list = db.select(sql)
    return article_list

def get_generated_article_title(generate_task_id):
    """
    生成计划 id 获取该生成计划已经生成过的文章标题
    :param generate_task_id:
    :return: title_set
    """
    db = DeNetMysql()
    sql = f"""
        SELECT DISTINCT output.output
        FROM produce_plan_exe_record planExeRecord
        JOIN produce_plan_module_output output ON output.plan_exe_id = planExeRecord.plan_exe_id AND output.produce_module_type = 3
        WHERE planExeRecord.plan_id = '{generate_task_id}';
    """
    title_tuple = db.select(sql)
    title_set = set([i[0] for i in title_tuple])
    return title_set


def get_publish_account_from_aigc():
    """
    从 aigc 系统中获取正在发布的账号
    :return:
    name: 公众号名称
    gh_id: 公众号 gh_id
    follower_count: 粉丝数量
    service_type_info： '公众号类型：0-订阅号，1-由历史老账号升级后的订阅号，2-服务号',
    verify_type_info：'公众号认证类型：-1-未认证，0-微信认证，1-新浪微博认证，3-已资质认证通过但还未通过名称认证，4-已资质认证通过、还未通过名称认证，但通过了新浪微博认证'
    """
    db = DeNetMysql()
    sql = """
        SELECT DISTINCT
            t3.`name`,
            t3.gh_id,
            t3.follower_count,
            t3.create_timestamp,
            t4.service_type_info,
            t4.verify_type_info
        FROM
            publish_plan t1
            JOIN publish_plan_account t2 ON t1.id = t2.plan_id
            JOIN publish_account t3 ON t2.account_id = t3.id
            LEFT JOIN publish_account_wx_type t4 on t3.id = t4.account_id
        WHERE
            t1.plan_status = 1
            AND t3.channel = 5
            GROUP BY t3.id
            ORDER BY t3.create_timestamp DESC
    """
    info_tuple = db.select(sql)
    info_list = [
        {
            "name": line[0],
            "ghId": line[1],
            "follower_count": line[2],
            "account_init_timestamp": int(line[3] / 1000),
            "account_type": line[4],
            "account_auth": line[5]
        } for line in info_tuple
    ]
    return info_list

def auto_create_crawler_task(plan_id, plan_name, plan_tag, url_list):
    """
    通过 url 自动创建抓取计划
    :param plan_id: 计划 id， 若往已经存在的 plan_id 中加文章则需要传，否则会新生成一个 id
    :param plan_name: 计划名称
    :param plan_tag: 计划标签
    :param url_list: 输入的 url_list
    :return:
    """
    url = "http://aigc-api.cybertogether.net/aigc/crawler/plan/save"
    payload = json.dumps({
        "params": {
            "contentFilters": [],
            "accountFilters": [],
            "filterAccountMatchMode": 1,
            "filterContentMatchMode": 1,
            "selectModeValues": [],
            "searchModeValues": [],
            "contentModal": 3,
            "analyze": {},
            "crawlerComment": 0,
            "inputGroup": None,
            "inputSourceGroups": [],
            "modePublishTime": [],
            "planType": 2,
            "frequencyType": 2,
            "planTag": plan_tag,
            "tagPenetrateFlag": 0,
            "id": plan_id,
            "name": plan_name,
            "channel": 5,
            "crawlerMode": 5,
            "inputModeValues": url_list,
            "modePublishTimeStart": None,
            "modePublishTimeEnd": None,
            "executeRate": None,
            "executeDate": None,
            "executeWindowStart": None,
            "executeWindowEnd": None,
            "executeTimeInterval": None,
            "executeNum": None,
            "addModal": None,
            "addChannel": None,
            "fileUpload": None,
            "prompt": None,
            "acelFlag": None,
            "tasks": []
        },
        "baseInfo": PERSON_COOKIE
    })
    response = requests.request("POST", url, headers=HEADERS, data=payload)
    return response.json()


def bind_crawler_task_to_generate_task(crawler_task_list, generate_task_id):
    """
    将抓取计划绑定至生成计划
    生成计划已经存在
    :crawler_task_list: 要输入的抓取计划List
    :generate_task_id： 目标生成计划 id
    :return: response
    """
    url = "http://aigc-api.cybertogether.net/aigc/produce/plan/save"
    plan_info = get_generate_task_detail(generate_task_id)
    input_source_groups = plan_info.get("inputSourceGroups")
    existed_crawler_task = input_source_groups[0].get("inputSources")
    new_task_list = existed_crawler_task + crawler_task_list
    input_source_group_0 = input_source_groups[0]
    input_source_group_0['inputSources'] = new_task_list
    payload = json.dumps({
        "params": {
            "contentFilters": [],
            "produceModal": plan_info.get("produceModal"),
            "inputModal": plan_info.get("inputModal"),
            "tasks": plan_info.get("tasks", []),
            "modules": [],
            "moduleGroups": plan_info.get("moduleGroups"),
            "inputSourceGroups": [input_source_group_0],
            "layoutType": plan_info.get("layoutType"),
            "activeManualReview": plan_info.get("activeManualReview"),
            "totalProduceNum": plan_info.get("totalProduceNum"),
            "dailyProduceNum": plan_info.get("dailyProduceNum"),
            "maxConcurrentNum": plan_info.get("maxConcurrentNum"),
            "id": generate_task_id,
            "name": plan_info.get("name"),
            "planTag": plan_info.get("planTag"),
            "tagPenetrateFlag": plan_info.get("tagPenetrateFlag"),
            "inputType": plan_info.get("inputType"),
            "inputChannel": plan_info.get("inputChannel"),
            "activeManualReviewCount": plan_info.get("activeManualReviewCount"),
            "autoComposite": plan_info.get("autoComposite")
        },
        "baseInfo": PERSON_COOKIE
    })
    response = requests.request("POST", url, headers=HEADERS, data=payload)
    return response.json()


@retryOnTimeout()
def get_generate_task_detail(generate_task_id):
    """
    通过生成计划的 id，获取该生成计划已有的抓取计划 list
    :param generate_task_id:
    :return:
    """
    url = "http://aigc-api.cybertogether.net/aigc/produce/plan/detail"
    payload = json.dumps({
        "params": {
            "id": generate_task_id
        },
        "baseInfo": PERSON_COOKIE
    })
    response = requests.request("POST", url, headers=HEADERS, data=payload, timeout=10)
    result = response.json()
    if result['msg'] == 'success':
        return result['data']
    else:
        return {}