import random import time from datetime import datetime import requests import json import re from common import Material, Feishu from common.sql_help import sqlCollect class SphNrxs: """创建票圈账号""" @classmethod def insert_number(cls, mid, tag_id): for i in range(3): url = "https://admin.piaoquantv.com/manager/crawler/v3/user/save" payload = { "source": "jiqizidonggaizao", "mode": "author", "modeValue": "", "modeBoard": "", "recomStatus": -7, "appRecomStatus": -7, "autoAuditStatus": 0, "tag": f"7592,452,8776,{tag_id}", "contentCategory": 0, "link": str(mid) } cookie = Material.get_cookie_data("KsoMsyP2ghleM9tzBfmcEEXBnXg", "U1gySe", "票圈后台-cookie") headers = { 'content-length': '0', 'cookie': cookie, 'origin': 'https://admin.piaoquantv.com', 'priority': 'u=1, i', 'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"macOS"' } response = requests.request("POST", url, headers=headers, json=payload) response = response.json() code = response["code"] if code == 0: print("添加账号成功") time.sleep(1) url = "https://admin.piaoquantv.com/manager/crawler/v3/user/list" payload = { "pageNum": 1, "pageSize": 20 } response = requests.request("POST", url, headers=headers, json=payload) response = response.json() list = response["content"]['list'] link = list[0]["link"] if link == str(mid): print("获取站内账号ID成功") return list[0]["uid"] else: if code == '10010': return None Feishu.bot("xinxin", '票圈后台提醒', f'票圈后台cookie 失效了，请及时更换', 'xinxin') """腾讯互选平台通过appid获取观众画像""" @classmethod def get_hx(cls, aid): url = "http://8.217.190.241:8888/crawler/wei_xin/shi_pin_hao/hu_xuan_detail" cookie = Material.get_cookie_data("KsoMsyP2ghleM9tzBfmcEEXBnXg", "U1gySe", "腾讯互选平台-cookie") account_id = Material.get_cookie_data("KsoMsyP2ghleM9tzBfmcEEXBnXg", "U1gySe", "腾讯互选平台-account_id") payload = json.dumps({ "account_id": aid, "uid": str(account_id), "cookie": cookie }) headers = { 'Content-Type': 'application/json' } response = requests.request("POST", url, headers=headers, data=payload) response = response.json() ret = response['code'] if ret == 0: data = response['data']['data'] if data: age_ranges = ['<18 岁', '18 ~ 24 岁', '25 ~ 29 岁', '30 ~ 39 岁', '40 ~ 49 岁', '>50 岁'] viewer_portrait = data['viewer_portrait'] # 观众画像 # 找到占比最大的项 viewer_max_age_range = max( (item for item in viewer_portrait if item['name'] in age_ranges), key=lambda x: float(x['percentage'].strip('%')) ) if viewer_max_age_range['name'] != '>50 岁': return "2" fans_portrait = data['fans_portrait'] # 粉丝画像 # 找到占比最大的项 fans_max_age_range = max( (item for item in fans_portrait if item['name'] in age_ranges), key=lambda x: float(x['percentage'].strip('%')) ) if fans_max_age_range['name'] != '>50 岁': return "3" return "0" else: Feishu.bot("xinxin", '腾讯互选平台提醒', f'腾讯互选平台cookie 失效了，请及时更换', 'xinxin') return None """腾讯互选平台通过搜索获取appid""" @classmethod def get_hxpt_appid(cls, user): url = "https://huxuan.qq.com/cgi-bin/advertiser/finder_publisher/search" cookie = Material.get_cookie_data("KsoMsyP2ghleM9tzBfmcEEXBnXg", "U1gySe", "腾讯互选平台-cookie") account_id = Material.get_cookie_data("KsoMsyP2ghleM9tzBfmcEEXBnXg", "U1gySe", "腾讯互选平台-account_id") payload = json.dumps({ "keyword": user, "page": { "no": 1, "size": 50 } }) headers = { 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', 'Content-Type': 'application/json', 'Cookie': cookie, 'Origin': 'https://huxuan.qq.com', 'Pragma': 'no-cache', 'Referer': 'https://huxuan.qq.com/trade/selection/46251713/selection_list?type=finder-trade', 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36', 'account_id': str(account_id), 'sec-ch-ua': '"Not)A;Brand";v="99", "Google Chrome";v="127", "Chromium";v="127"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"macOS"' } for i in range(3): response = requests.request("POST", url, headers=headers, data=payload) response = response.json() ret = response['ret'] if ret == 0: try: appid = response['data']['item'][0]['appid'] if appid: return appid except Exception as e: continue else: Feishu.bot("xinxin", '腾讯互选平台提醒', f'腾讯互选平台cookie 失效了，请及时更换', 'xinxin') return None return None """获取用户主页是否符合规则""" @classmethod def get_sph_data(cls, user, uid): url = "http://61.48.133.26:30001/FinderGetUpMasterNextPage" headers = { 'Content-Type': 'application/json' } payload = json.dumps({ "username": user, "last_buffer": "" }) response = requests.request("POST", url, headers=headers, data=payload) time.sleep(random.randint(1, 5)) if response.text == "" or response.text == None: return res_json = response.json() try: if len(res_json["DownloadAddress"]) == 0 or res_json["DownloadAddress"] == "" or res_json[ "DownloadAddress"] == None: return except: pass if "objectId" not in response.text or response.status_code != 200: return if len(res_json["UpMasterHomePage"]) == 0: return if not res_json["UpMasterHomePage"]: return try: for obj in res_json["UpMasterHomePage"]: objectId = obj['objectId'] object_id = sqlCollect.sph_data_info_v_id(objectId, "视频号") if object_id: continue objectNonceId = obj['objectNonceId'] url1 = "http://61.48.133.26:30001/GetFinderDownloadAddress" payload = json.dumps({ "objectId": objectId, "objectNonceId": objectNonceId }) headers = { 'Content-Type': 'text/plain' } response = requests.request("POST", url1, headers=headers, data=payload) time.sleep(random.randint(0, 1)) video_obj = response.json() video_url = video_obj.get('DownloadAddress') if len(video_url) == 0: continue duration = video_obj.get('play_len') # cover = video_obj.get('thumb_url') share_cnt = int(obj['forward_count']) # 分享 like_cnt = int(obj['like_count']) # 点赞 # user_name = obj['username'] # 用户名标示 nick_name = obj['nickname'] # 用户名 # comment_count = obj['comment_count'] # 评论数 # fav_count = obj['fav_count'] # 大拇指点赞数 video_percent = '%.2f' % (share_cnt / like_cnt) special = float(0.25) if share_cnt >= 300 and float(video_percent) >= special and int(duration) >= 30: return nick_name return None except Exception as e: return None """视频号加热平台相似溯源""" @classmethod def get_nrxs_list(cls, uid): cookie = Material.get_cookie_data("KsoMsyP2ghleM9tzBfmcEEXBnXg", "U1gySe", "视频号加热") time.sleep(10) url = "http://8.217.190.241:8888/crawler/wei_xin/shi_pin_hao/jia_re" payload = json.dumps({ "account_name": uid, "category": "推荐", "cookie": cookie }) headers = { 'Content-Type': 'application/json' } try: # time.sleep(2) response = requests.request("POST", url, headers=headers, data=payload) response = response.json() code = response['code'] if code == 0: if response['data'] == None: sqlCollect.update_machine_making_reflux(uid) return status = sqlCollect.select_crawler_uesr_v3(uid) if status: pq_id = re.sub(r'[(),]', '', str(status)) else: pq_id = cls.insert_number(uid, '467') if pq_id == None: return sqlCollect.update_machine_making_reflux(uid) data_list = response['data']['data'] if data_list: for data in data_list: nick_name = data['nickName'] # 用户名 user_name = data['username'] # 用户v2 user = cls.get_sph_data(user_name, uid) if user: # time.sleep(180) # appid = cls.get_hxpt_appid(user) # if appid: # time.sleep(180) # has_used = cls.get_hx(appid) # if has_used: # if has_used == '0': res = sqlCollect.insert_xssy_sph_info(uid, nick_name, "视频号", user_name, "0", "", pq_id) if res == 1: current_time = datetime.now() formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S") values = [ [ "视频号", nick_name, str(pq_id), "5", "通用-分享到群", "AI片尾引导", "zhifeng_emo,sijia", "", "", "AI标题", "", f"溯源账号:{uid}", formatted_time ] ] Feishu.insert_columns("WGIYsSDdxhItBwtJ0xgc0yE7nEg", '0701bd', "ROWS", 1, 2) time.sleep(0.5) Feishu.update_values("WGIYsSDdxhItBwtJ0xgc0yE7nEg", '0701bd', "B2:Z2", values) Feishu.bot("xinxin", '视频号溯源成功提示', f'原账号:{uid}，溯源到的账号:{nick_name},写入账号：{pq_id}', 'xinxin') else: # sqlCollect.insert_xssy_sph_info(uid, nick_name, user_name, has_used, appid) sqlCollect.insert_xssy_sph_info(uid, nick_name, user_name, "1") # else: # sqlCollect.insert_xssy_sph_info(uid, nick_name, user_name, "1") # continue else: return None else: Feishu.bot("xinxin", '视频号加热提醒', f'视频号加热平台 cookie 失效了，请及时更换', 'xinxin') return None except Exception as e: Feishu.bot("xinxin", '视频号加热提醒', f'视频号加热平台 cookie 失效了，请及时更换', 'xinxin') return None """获取需溯源账号""" @classmethod def sph_nrxs_data(cls): user = sqlCollect.get_machine_making_reflux("视频号", "单点视频", "视频号历史", "相似溯源", "视频H品类账号") if user == None: return user = [item[0] for item in user] Feishu.bot("xinxin", '视频号溯源提醒', f'今日需溯源账号共{len(user)}条', 'xinxin') for uid in user: if re.match(r'^[A-Za-z0-9]+$', uid): # 匹配成功，进行下一次循环 continue Feishu.bot("xinxin", '视频号溯源提醒', f'开始溯源账号名称{uid}', 'xinxin') list = cls.get_nrxs_list(uid) print(list) if __name__ == '__main__': SphNrxs.sph_nrxs_data()