piaoquan_crawler/gongzhonghao/gongzhonghao_main/gzh_bug_review.py

import json
import time
import requests
import urllib3
from common.public import task_fun_mq, get_consumer, ack_message
from common.scheduling_db import MysqlHelper
from common.common import Common
import asyncio
from gongzhonghao.gongzhonghao_author.gongzhonghao_author import GongzhonghaoAuthor

token_d = {
    "token": "883406306",
    "cookie": "appmsglist_action_3524986952=card; ua_id=j6t2xNuC0mv6dLVbAAAAAMPRLKj1sVGSlMDwNFJKE3s=; wxuin=93278011749821; mm_lang=zh_CN; pgv_pvid=6815195556; noticeLoginFlag=1; remember_acct=2071735594%40qq.com; rewardsn=; wxtokenkey=777; _clck=3930572231|1|ff1|0; uuid=680bd7f128bf80058bc62dd82ff85c96; rand_info=CAESIBtaIUDyVXWwBRD33d7CafRp3rV5rXK7mcvYCy4Yvnn+; slave_bizuin=3236647229; data_bizuin=3236647229; bizuin=3236647229; data_ticket=Dx0Yxt5o9JJuMyndtyu3+JZBym0Dcjy6QqjPcfp+xwsLHf3Y+L9ZmP+kDX6o4t9r; slave_sid=WjV0MXhZZXlrcG9BTGVOZjBEOUlyUFptMWEyN2JNcXlpeU5kcGIyVm9IZUZOV3J1RElKb29KTDJIRHRYaGZtNnVSbklua1FOdUNsX3NoQWE4RFVKM0lKbDkzU25wblRGTDhDWFJteExtMHBjZGwyanZKOVVCWmE1UmNxT3FaZWNsd0VrVm52eEpLakFocGVz; slave_user=gh_d284c09295eb; xid=675798a4e148cb559bed6bb65681ebf9; _clsk=1a6iklq|1694746372692|2|1|mp.weixin.qq.com/weheat-agent/payload/record"
}


def get_user_info(token_dict):
    url = "https://mp.weixin.qq.com/cgi-bin/searchbiz?"
    headers = {
        "accept": "*/*",
        "accept-encoding": "gzip, deflate, br",
        "accept-language": "zh-CN,zh;q=0.9",
        "referer": "https://mp.weixin.qq.com/cgi-bin/appmsg?"
                   "t=media/appmsg_edit_v2&action=edit&isNew=1"
                   "&type=77&createType=5&token=1011071554&lang=zh_CN",
        "sec-ch-ua": '" Not A;Brand";v="99", "Chromium";v="100", "Google Chrome";v="100"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
                      " (KHTML, like Gecko) Chrome/100.0.4896.127 Safari/537.36",
        "x-requested-with": "XMLHttpRequest",
        "cookie": token_dict["cookie"],
    }
    params = {
        "action": "search_biz",
        "begin": "0",
        "count": "5",
        "query": "生活小妙招小助手",
        "token": token_dict["token"],
        "lang": "zh_CN",
        "f": "json",
        "ajax": "1",
    }
    # proxies = Common.tunnel_proxies()
    # print(proxies)
    urllib3.disable_warnings()
    r = requests.get(url=url, headers=headers, params=params, verify=False)
    r.close()
    print(r.json())
    # if r.json()["base_resp"]["err_msg"] == "invalid session":
    #     Common.logger(log_type, crawler).warning(
    #         f"status_code:{r.status_code}, get_fakeid:{r.text}\n"
    #     )
    #     # Common.logging(
    #     #     log_type,
    #     #     crawler,
    #     #     env,
    #     #     f"status_code:{r.status_code}, get_fakeid:{r.text}\n",
    #     # )
    #     cls.release_token(log_type, crawler, env, token_dict["token_id"], -2)
    #     if 20 >= datetime.datetime.now().hour >= 10:
    #         Feishu.bot(
    #             log_type,
    #             crawler,
    #             f"{token_dict['title']}\n操作人:{token_dict['operator']}\n更换日期:{token_dict['update_time']} \n过期啦，请扫码更换token\nhttps://mp.weixin.qq.com/",
    #         )
    #     time.sleep(60 * 15)
    #     continue
    # if r.json()["base_resp"]["err_msg"] == "freq control":
    #     Common.logger(log_type, crawler).warning(
    #         f"status_code:{r.status_code}, get_fakeid:{r.text}\n"
    #     )
    #     # Common.logging(
    #     #     log_type,
    #     #     crawler,
    #     #     env,
    #     #     f"status_code:{r.status_code}, get_fakeid:{r.text}\n",
    #     # )
    #     cls.release_token(log_type, crawler, env, token_dict["token_id"], -2)
    #     if 20 >= datetime.datetime.now().hour >= 10:
    #         Feishu.bot(
    #             log_type,
    #             crawler,
    #             f"{token_dict['title']}\n操作人:{token_dict['operator']}\n更换日期:{token_dict['update_time']} \n频控啦，请扫码更换其他公众号token\nhttps://mp.weixin.qq.com/",
    #         )
    #     time.sleep(60 * 15)
    #     continue
    # if r.json()["base_resp"]["err_msg"] == "ok" and len(r.json()["list"]) == 0:
    #     Common.logger(log_type, crawler).warning(
    #         f"status_code:{r.status_code}, get_fakeid:{r.text}\n"
    #     )
    #     # Common.logging(
    #     #     log_type,
    #     #     crawler,
    #     #     env,
    #     #     f"status_code:{r.status_code}, get_fakeid:{r.text}\n",
    #     # )
    #     unbind_msg = task_unbind(
    #         log_type=log_type,
    #         crawler=crawler,
    #         taskid=task_dict["id"],
    #         uids=str(user_dict["uid"]),
    #         env=env,
    #     )
    #     if unbind_msg == "success":
    #         if 20 >= datetime.datetime.now().hour >= 10:
    #             Feishu.bot(
    #                 log_type,
    #                 crawler,
    #                 f"公众号:{user_dict['link']}, 站内昵称:{user_dict['nick_name']}\n抓取异常, 已取消抓取该公众号\n",
    #             )
    #         # Common.logging(
    #         #     log_type,
    #         #     crawler,
    #         #     env,
    #         #     f"公众号:{user_dict['link']}, 站内昵称:{user_dict['nick_name']}\n抓取异常, 已取消抓取该公众号\n",
    #         # )
    #     else:
    #         Common.logger(log_type, crawler).warning(f"unbind_msg:{unbind_msg}")
    #         # Common.logging(log_type, crawler, env, f"unbind_msg:{unbind_msg}")
    #     return None
    user_info_dict = {
        "user_name": r.json()["list"][0]["nickname"],
        "user_id": r.json()["list"][0]["fakeid"],
        "avatar_url": r.json()["list"][0]["round_head_img"],
    }
    return user_info_dict


def get_videoList(token_dict, user_dict):
    begin = 0
    url = "https://mp.weixin.qq.com/cgi-bin/appmsg?"
    headers = {
        "accept": "*/*",
        "accept-encoding": "gzip, deflate, br",
        "accept-language": "zh-CN,zh;q=0.9",
        "referer": "https://mp.weixin.qq.com/cgi-bin/appmsg?"
                   "t=media/appmsg_edit_v2&action=edit&isNew=1"
                   "&type=77&createType=5&token="
                   + str(token_dict["token"])
                   + "&lang=zh_CN",
        "sec-ch-ua": '" Not A;Brand";v="99", "Chromium";v="100", "Google Chrome";v="100"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
                      " (KHTML, like Gecko) Chrome/100.0.4896.127 Safari/537.36",
        "x-requested-with": "XMLHttpRequest",
        "cookie": token_dict["cookie"],
    }
    params = {
        "action": "list_ex",
        "begin": str(begin),
        "count": "5",
        "fakeid": user_dict["user_id"],
        "type": "9",
        "query": "",
        "token": str(token_dict["token"]),
        "lang": "zh_CN",
        "f": "json",
        "ajax": "1",
    }
    urllib3.disable_warnings()
    r = requests.get(url=url, headers=headers, params=params, verify=False)
    print(r.url)
    r.close()
    print(r.json())
    if r.json()["base_resp"]["err_msg"] == "invalid session":
        time.sleep(60 * 15)
        print("invalid session")
    if r.json()["base_resp"]["err_msg"] == "freq control":
        print("freq control")
    if (
            r.json()["base_resp"]["err_msg"] == "invalid args"
            and r.json()["base_resp"]["ret"] == 200002
    ):
        print("invalid args")
    if "app_msg_list" not in r.json():
        print("no app_msg_list")
    if len(r.json()["app_msg_list"]) == 0:
        print("没有更多视频了\n")
        return
    else:
        begin += 5
        app_msg_list = r.json()["app_msg_list"]
        for article in app_msg_list:
            # try:
            create_time = article.get("create_time", 0)
            publish_time_stamp = int(create_time)
            publish_time_str = time.strftime(
                "%Y-%m-%d %H:%M:%S", time.localtime(publish_time_stamp)
            )
            article_url = article.get("link", "")
            video_dict = {
                "video_id": article.get("aid", ""),
                "video_title": article.get("title", "")
                .replace(" ", "")
                .replace('"', "")
                .replace("'", ""),
                "publish_time_stamp": publish_time_stamp,
                "publish_time_str": publish_time_str,
                "user_name": user_dict["user_name"],
                "play_cnt": 0,
                "comment_cnt": 0,
                "like_cnt": 0,
                "share_cnt": 0,
                "user_id": user_dict["user_id"],
                "avatar_url": user_dict["avatar_url"],
                "cover_url": article.get("cover", ""),
                "article_url": article.get("link", ""),
                # "video_url": cls.get_video_url(article_url, env),
                "video_url": "url",
                "session": f"gongzhonghao-author1-{int(time.time())}",
            }
            print(video_dict)
            # for k, v in video_dict.items():
            #     Common.logger(log_type, crawler).info(f"{k}:{v}")
            # Common.logging(
            #     log_type, crawler, env, f"video_dict:{video_dict}"
            # )

        #         if int(time.time()) - publish_time_stamp > 3600 * 24 * int(
        #                 rule_dict.get("period", {}).get("max", 1000)
        #         ):
        #             Common.logger(log_type, crawler).info(
        #                 f"发布时间超过{int(rule_dict.get('period', {}).get('max', 1000))}天\n"
        #             )
        #             # Common.logging(
        #             #     log_type,
        #             #     crawler,
        #             #     env,
        #             #     f"发布时间超过{int(rule_dict.get('period', {}).get('max', 1000))}天\n",
        #             # )
        #             return
        #
        #         if (
        #                 video_dict["article_url"] == 0
        #                 or video_dict["video_url"] == 0
        #         ):
        #             Common.logger(log_type, crawler).info("文章涉嫌违反相关法律法规和政策\n")
        #             # Common.logging(log_type, crawler, env, "文章涉嫌违反相关法律法规和政策\n")
        #         # 标题敏感词过滤
        #         elif (
        #                 any(
        #                     str(word)
        #                     if str(word) in video_dict["video_title"]
        #                     else False
        #                     for word in get_config_from_mysql(
        #                         log_type=log_type,
        #                         source=crawler,
        #                         env=env,
        #                         text="filter",
        #                         action="",
        #                     )
        #                 )
        #                 is True
        #         ):
        #             Common.logger(log_type, crawler).info("标题已中过滤词\n")
        #             # Common.logging(log_type, crawler, env, "标题已中过滤词\n")
        #         # 已下载判断
        #         elif (
        #                 cls.repeat_video(
        #                     log_type, crawler, video_dict["video_id"], env
        #                 )
        #                 != 0
        #         ):
        #             Common.logger(log_type, crawler).info("视频已下载\n")
        #             # Common.logging(log_type, crawler, env, "视频已下载\n")
        #         # 标题相似度
        #         elif (
        #                 title_like(
        #                     log_type,
        #                     crawler,
        #                     video_dict["video_title"],
        #                     cls.platform,
        #                     env,
        #                 )
        #                 is True
        #         ):
        #             Common.logger(log_type, crawler).info(
        #                 f'标题相似度>=80%:{video_dict["video_title"]}\n'
        #             )
        #             # Common.logging(
        #             #     log_type,
        #             #     crawler,
        #             #     env,
        #             #     f'标题相似度>=80%:{video_dict["video_title"]}\n',
        #             # )
        #         else:
        #             video_dict["out_user_id"] = video_dict["user_id"]
        #             video_dict["platform"] = crawler
        #             video_dict["strategy"] = log_type
        #             video_dict["out_video_id"] = video_dict["video_id"]
        #             video_dict["width"] = 0
        #             video_dict["height"] = 0
        #             video_dict["crawler_rule"] = json.dumps(rule_dict)
        #             video_dict["user_id"] = user_dict["uid"]
        #             video_dict["publish_time"] = video_dict["publish_time_str"]
        #             mq.send_msg(video_dict)
        #     except Exception as e:
        #         Common.logger(log_type, crawler).error(f"抓取单条视频异常:{e}\n")
        #         Common.logging(log_type, crawler, env, f"抓取单条视频异常:{e}\n")
        # Common.logger(log_type, crawler).info("休眠 60 秒\n")
        # Common.logging(log_type, crawler, env, "休眠 60 秒\n")
        time.sleep(60)


# 分割列表
def chunks(data_list, chunk_size):
    """
    :param data_list: 列表
    :param chunk_size: 每个子列表的长度
    :return: 大列表包小列表[[], [], [], []......]
    """
    for i in range(0, len(data_list), chunk_size):
        yield data_list[i: i + chunk_size]


async def get_author_videos(args):
    await asyncio.sleep(1)
    print(args['log_type'])
    await GongzhonghaoAuthor.get_all_videos(
        log_type=args['log_type'],
        crawler=args['crawler'],
        task_dict=args['task_dict'],
        token_index=args['token_index'],
        rule_dict=args['rule_dict'],
        user_list=args['user_list'],
        env=args['env']
    )


if __name__ == "__main__":
    mess = {
        "createTime": 1684500378438,
        "id": 27,
        "interval": 86400,
        "machine": "aliyun",
        "mode": "author",
        "operator": "xxl",
        "rule": "[{\"duration\":{\"min\":20,\"max\":2700}},{\"period\":{\"min\":1,\"max\":2}}]",
        "source": "gongzhonghao",
        "spiderName": "run_gzh2_author",
        "startTime": 1693493854438,
        "status": 0,
        "taskName": "公众号_2",
        "updateTime": 1688572800179
    }

    # 解析 task_dict
    rule_list = json.loads(mess['rule'])
    rule_dict = {}
    for item in rule_list:
        for key, val in item.items():
            rule_dict[key] = val
    mess['rule'] = rule_dict
    task_dict = mess
    # 解析 user_list
    task_id = task_dict["id"]
    select_user_sql = (
        f"""select * from crawler_user_v3 where task_id={task_id}"""
    )
    user_list = MysqlHelper.get_values(
        "author", "gongzhonghao", select_user_sql, "prod", action=""
    )
    print(len(user_list))
    user_list = chunks(user_list, 250)
    print(user_list)
    for index, i in enumerate(user_list):
        with open("/Users/luojunhui/cyber/gzh_spider/test_AB/200/user_list_{}.json".format(index + 1), "w", encoding="utf-8") as f:
            f.write(json.dumps(i, ensure_ascii=False, indent=4))
    # print(user_list)
    # loop = asyncio.get_event_loop()
    # arg_list = []
    # for index, sub_list in enumerate(user_list):
    #     arg = {'log_type': "author{}".format(index + 1), 'crawler': "gongzhonghao", 'token_index': index + 1,
    #            'task_dict': task_dict, 'rule_dict': rule_dict, 'user_list': sub_list, 'env': 'prod'}
    #     arg_list.append(arg)
    #
    # coroutines_list = [get_author_videos(arg) for arg in arg_list]
    #
    #
    # async def test():
    #     await asyncio.gather(*coroutines_list)
    # asyncio.run(test())

    user_d = get_user_info(token_d)
    # print(user_d)
    # # #
    # get_videoList(token_d, user_d)