# -*- coding: utf-8 -*- # @Author: wangkun # @Time: 2023/2/17 import base64 import json import os import sys import time import requests import urllib3 from selenium.webdriver import DesiredCapabilities from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By from seleniumwire import webdriver sys.path.append(os.getcwd()) from common.common import Common from common.feishu import Feishu from common.publish import Publish proxies = {"http": None, "https": None} class Follow: # 个人主页视频翻页参数 offset = 0 # 下载规则 @staticmethod def download_rule(duration, width, height): if int(duration) >= 60: if int(width) >= 720 or int(height) >= 720: return True else: return False else: return False # 过滤词库 @classmethod def filter_words(cls, log_type, crawler): try: filter_words_sheet = Feishu.get_values_batch(log_type, crawler, 'KGB4Hc') filter_words_list = [] for x in filter_words_sheet: for y in x: if y is None: pass else: filter_words_list.append(y) return filter_words_list except Exception as e: Common.logger(log_type, crawler).error(f'filter_words异常:{e}\n') # 获取用户信息（字典格式）. 注意：部分 user_id 字符类型是 int / str @classmethod def get_user_info_from_feishu(cls, log_type, crawler): try: user_sheet = Feishu.get_values_batch(log_type, crawler, '5tlTYB') user_dict = {} for i in range(1, len(user_sheet)): user_name = user_sheet[i][0] out_id = user_sheet[i][1] our_id = user_sheet[i][3] if user_name is None or out_id is None or our_id is None: pass else: user_dict[user_name] = str(out_id) + ',' + str(our_id) return user_dict except Exception as e: Common.logger(log_type, crawler).error(f'get_user_id_from_feishu异常:{e}\n') @classmethod def get_signature(cls, log_type, crawler, out_uid, machine): try: # 打印请求配置 ca = DesiredCapabilities.CHROME ca["goog:loggingPrefs"] = {"performance": "ALL"} # 不打开浏览器运行 chrome_options = webdriver.ChromeOptions() chrome_options.add_argument("--headless") chrome_options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.79 Safari/537.36') chrome_options.add_argument("--no-sandbox") # driver初始化 if machine == 'aliyun' or machine == 'aliyun_hk': driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options) elif machine == 'macpro': driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options, service=Service('/Users/lieyunye/Downloads/chromedriver_v86/chromedriver')) elif machine == 'macair': driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options, service=Service('/Users/piaoquan/Downloads/chromedriver')) else: driver = webdriver.Chrome(desired_capabilities=ca, options=chrome_options, service=Service('/Users/wangkun/Downloads/chromedriver/chromedriver_v110/chromedriver')) driver.implicitly_wait(10) driver.get(f'https://www.ixigua.com/home/{out_uid}/') time.sleep(3) data_src = driver.find_elements(By.XPATH, '//img[@class="tt-img BU-MagicImage tt-img-loaded"]')[1].get_attribute("data-src") signature = data_src.split("x-signature=")[-1] # print(f"data_src:{data_src}") # print(f"signature:{signature}") return signature except Exception as e: Common.logger(log_type, crawler).error(f'get_signature异常:{e}\n') # 获取视频详情 @classmethod def get_video_url(cls, log_type, crawler, gid): # try: url = 'https://www.ixigua.com/api/mixVideo/information?' headers = { "accept-encoding": "gzip, deflate", "accept-language": "zh-CN,zh-Hans;q=0.9", "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.5 Safari/605.1.15", "referer": "https://www.ixigua.com/7102614741050196520?logTag=0531c88ac04f38ab2c62", } params = { 'mixId': gid, 'msToken': 'IlG0wd0Pylyw9ghcYiB2YseUmTwrsrqqhXrbIcsSaTcLTJyVlbYJzk20zw3UO-CfrfC' 'NVVIOBNjIl7vfBoxnVUwO9ZyzAI3umSKsT5-pef_RRfQCJwmA', 'X-Bogus': 'DFSzswVupYTANCJOSBk0P53WxM-r', '_signature': '_02B4Z6wo0000119LvEwAAIDCuktNZ0y5wkdfS7jAALThuOR8D9yWNZ.EmWHKV0WSn6Px' 'fPsH9-BldyxVje0f49ryXgmn7Tzk-swEHNb15TiGqa6YF.cX0jW8Eds1TtJOIZyfc9s5emH7gdWN94', } cookies = { 'ixigua-a-s': '1', 'msToken': 'IlG0wd0Pylyw9ghcYiB2YseUmTwrsrqqhXrbIcsSaTcLTJyVlbYJzk20zw3UO-CfrfCNVVIOB' 'NjIl7vfBoxnVUwO9ZyzAI3umSKsT5-pef_RRfQCJwmA', 'ttwid': '1%7C_yXQeHWwLZgCsgHClOwTCdYSOt_MjdOkgnPIkpi-Sr8%7C1661241238%7Cf57d0c5ef3f1d7' '6e049fccdca1ac54887c34d1f8731c8e51a49780ff0ceab9f8', 'tt_scid': 'QZ4l8KXDG0YAEaMCSbADdcybdKbUfG4BC6S4OBv9lpRS5VyqYLX2bIR8CTeZeGHR9ee3', 'MONITOR_WEB_ID': '0a49204a-7af5-4e96-95f0-f4bafb7450ad', '__ac_nonce': '06304878000964fdad287', '__ac_signature': '_02B4Z6wo00f017Rcr3AAAIDCUVxeW1tOKEu0fKvAAI4cvoYzV-wBhq7B6D8k0no7lb' 'FlvYoinmtK6UXjRIYPXnahUlFTvmWVtb77jsMkKAXzAEsLE56m36RlvL7ky.M3Xn52r9t1IEb7IR3ke8', 'ttcid': 'e56fabf6e85d4adf9e4d91902496a0e882', '_tea_utm_cache_1300': 'undefined', 'support_avif': 'false', 'support_webp': 'false', 'xiguavideopcwebid': '7134967546256016900', 'xiguavideopcwebid.sig': 'xxRww5R1VEMJN_dQepHorEu_eAc', } urllib3.disable_warnings() response = requests.get(url=url, headers=headers, params=params, cookies=cookies, verify=False) if 'data' not in response.json() or response.json()['data'] == '': Common.logger(log_type, crawler).warning('get_video_info: response: {}', response) else: video_info = response.json()['data']['gidInformation']['packerData']['video'] video_url_dict = {} # video_url if 'videoResource' not in video_info: video_url_dict["video_url"] = '' video_url_dict["audio_url"] = '' video_url_dict["video_width"] = 0 video_url_dict["video_height"] = 0 elif 'dash_120fps' in video_info['videoResource']: if "video_list" in video_info['videoResource']['dash_120fps'] and len(video_info['videoResource']['dash_120fps']['video_list']) != 0: video_url = video_info['videoResource']['dash_120fps']['video_list'][-1]['backup_url_1'] audio_url = video_info['videoResource']['dash_120fps']['video_list'][-1]['backup_url_1'] if len(video_url) % 3 == 1: video_url += '==' elif len(video_url) % 3 == 2: video_url += '=' elif len(audio_url) % 3 == 1: audio_url += '==' elif len(audio_url) % 3 == 2: audio_url += '=' video_url = base64.b64decode(video_url).decode('utf8') audio_url = base64.b64decode(audio_url).decode('utf8') video_width = video_info['videoResource']['dash_120fps']['video_list'][-1]['vwidth'] video_height = video_info['videoResource']['dash_120fps']['video_list'][-1]['vheight'] video_url_dict["video_url"] = video_url video_url_dict["audio_url"] = audio_url video_url_dict["video_width"] = video_width video_url_dict["video_height"] = video_height elif 'dynamic_video' in video_info['videoResource']['dash_120fps'] \ and 'dynamic_video' in video_info['videoResource']['dash_120fps'] \ and 'dynamic_video_list' in video_info['videoResource']['dash_120fps']['dynamic_video'] \ and 'dynamic_audio_list' in video_info['videoResource']['dash_120fps']['dynamic_video'] \ and len(video_info['videoResource']['dash_120fps']['dynamic_video']['dynamic_video_list']) != 0 \ and len(video_info['videoResource']['dash_120fps']['dynamic_video']['dynamic_audio_list']) != 0: video_url = video_info['videoResource']['dash_120fps']['dynamic_video']['dynamic_video_list'][-1]['backup_url_1'] audio_url = video_info['videoResource']['dash_120fps']['dynamic_video']['dynamic_audio_list'][-1]['backup_url_1'] if len(video_url) % 3 == 1: video_url += '==' elif len(video_url) % 3 == 2: video_url += '=' elif len(audio_url) % 3 == 1: audio_url += '==' elif len(audio_url) % 3 == 2: audio_url += '=' video_url = base64.b64decode(video_url).decode('utf8') audio_url = base64.b64decode(audio_url).decode('utf8') video_width = video_info['videoResource']['dash_120fps']['dynamic_video']['dynamic_video_list'][-1]['vwidth'] video_height = video_info['videoResource']['dash_120fps']['dynamic_video']['dynamic_video_list'][-1]['vheight'] video_url_dict["video_url"] = video_url video_url_dict["audio_url"] = audio_url video_url_dict["video_width"] = video_width video_url_dict["video_height"] = video_height elif 'dash' in video_info['videoResource'] \ and 'dynamic_video' in video_info['videoResource']['dash'] \ and 'dynamic_video_list' in video_info['videoResource']['dash']['dynamic_video']: video_url = video_info['videoResource']['dash']['dynamic_video']['dynamic_video_list'][-1]['backup_url_1'] audio_url = video_info['videoResource']['dash']['dynamic_video']['dynamic_audio_list'][-1]['backup_url_1'] if len(video_url) % 3 == 1: video_url += '==' elif len(video_url) % 3 == 2: video_url += '=' elif len(audio_url) % 3 == 1: audio_url += '==' elif len(audio_url) % 3 == 2: audio_url += '=' video_url = base64.b64decode(video_url).decode('utf8') audio_url = base64.b64decode(audio_url).decode('utf8') video_width = video_info['videoResource']['dash']['dynamic_video']['dynamic_video_list'][-1]['vwidth'] video_height = video_info['videoResource']['dash']['dynamic_video']['dynamic_video_list'][-1]['vheight'] elif 'normal' in video_info['videoResource']: video_url = video_info['videoResource']['normal']['video_list'][-1]['backup_url_1'] audio_url = video_info['videoResource']['normal']['video_list'][-1]['backup_url_1'] if len(video_url) % 3 == 1: video_url += '==' elif len(video_url) % 3 == 2: video_url += '=' elif len(audio_url) % 3 == 1: audio_url += '==' elif len(audio_url) % 3 == 2: audio_url += '=' video_url = base64.b64decode(video_url).decode('utf8') audio_url = base64.b64decode(audio_url).decode('utf8') video_width = video_info['videoResource']['normal']['video_list'][-1]['vwidth'] video_height = video_info['videoResource']['normal']['video_list'][-1]['vheight'] else: video_url = 0 audio_url = 0 video_width = 0 video_height = 0 return video_url_dict # except Exception as e: # Common.logger(log_type).error(f'get_video_info异常:{e}\n') @classmethod def get_videolist(cls, log_type, crawler, out_uid, machine): signature = cls.get_signature(log_type, crawler, out_uid, machine) url = "https://www.ixigua.com/api/videov2/author/new_video_list?" params = { 'to_user_id': str(out_uid), 'offset': str(cls.offset), 'limit': '30', 'maxBehotTime': '0', 'order': 'new', 'isHome': '0', 'msToken': 'G0eRzNkw189a8TLaXjc6nTHVMQwh9XcxVAqTbGKi7iPJdQcLwS3-XRrJ3MZ7QBfqErpxp3EX1WtvWOIcZ3NIgr41hgcd-v64so_RRj3YCRw1UsKW8mIssNLlIMspsg==', 'X-Bogus': 'DFSzswVuEkUANjW9ShFTgR/F6qHt', '_signature': signature, } headers = { 'authority': 'www.ixigua.com', 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6', 'cache-control': 'no-cache', 'cookie': f'MONITOR_WEB_ID=7168304743566296612; __ac_signature={signature}; ixigua-a-s=1; support_webp=true; support_avif=false; csrf_session_id=a5355d954d3c63ed1ba35faada452b4d; msToken=G0eRzNkw189a8TLaXjc6nTHVMQwh9XcxVAqTbGKi7iPJdQcLwS3-XRrJ3MZ7QBfqErpxp3EX1WtvWOIcZ3NIgr41hgcd-v64so_RRj3YCRw1UsKW8mIssNLlIMspsg==; tt_scid=o4agqz7u9SKPwfBoPt6S82Cw0q.9KDtqmNe0JHxMqmpxNHQWq1BmrQdgVU6jEoX7ed99; ttwid=1%7CHHtv2QqpSGuSu8r-zXF1QoWsvjmNi1SJrqOrZzg-UCY%7C1676618894%7Cee5ad95378275f282f230a7ffa9947ae7eff40d0829c5a2568672a6dc90a1c96; ixigua-a-s=1', 'pragma': 'no-cache', 'referer': f'https://www.ixigua.com/home/{out_uid}/video/?preActiveKey=hotsoon&list_entrance=userdetail', 'sec-ch-ua': '"Chromium";v="110", "Not A(Brand";v="24", "Microsoft Edge";v="110"', '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/110.0.0.0 Safari/537.36 Edg/110.0.1587.41', 'x-secsdk-csrf-token': '00010000000119e3f9454d1dcbb288704cda1960f241e2d19bd21f2fd283520c3615a990ac5a17448bfbb902a249' } urllib3.disable_warnings() response = requests.get(url=url, headers=headers, params=params, proxies=proxies, verify=False) cls.offset += 30 if response.status_code != 200: Common.logger(log_type, crawler).warning(f"get_videolist_response:{response.text}\n") elif 'data' not in response.text: Common.logger(log_type, crawler).warning(f"get_videolist_response:{response.text}\n") elif 'videoList' not in response.json()["data"]: Common.logger(log_type, crawler).warning(f"get_videolist_response:{response.json()}\n") else: videoList = response.json()['data']['videoList'] for i in range(len(videoList)): # video_title if 'title' not in videoList[i]: video_title = 0 else: video_title = videoList[i]['title'].strip().replace('手游', '') \ .replace('/', '').replace('\/', '').replace('\n', '') # video_id if 'video_id' not in videoList[i]: video_id = 0 else: video_id = videoList[i]['video_id'] # gid if 'gid' not in videoList[i]: gid = 0 else: gid = videoList[i]['gid'] # play_cnt if 'video_detail_info' not in videoList[i]: play_cnt = 0 elif 'video_watch_count' not in videoList[i]['video_detail_info']: play_cnt = 0 else: play_cnt = videoList[i]['video_detail_info']['video_watch_count'] # comment_cnt if 'comment_count' not in videoList[i]: comment_cnt = 0 else: comment_cnt = videoList[i]['comment_count'] # like_cnt if 'digg_count' not in videoList[i]: like_cnt = 0 else: like_cnt = videoList[i]['digg_count'] # share_cnt share_cnt = 0 # video_duration if 'video_duration' not in videoList[i]: video_duration = 0 else: video_duration = videoList[i]['video_duration'] # send_time if 'publish_time' not in videoList[i]: publish_time = 0 else: publish_time = videoList[i]['publish_time'] # is_top if 'is_top' not in videoList[i]: is_top = 0 else: is_top = videoList[i]['is_top'] # user_name if 'user_info' not in videoList[i]: user_name = 0 elif 'name' not in videoList[i]['user_info']: user_name = 0 else: user_name = videoList[i]['user_info']['name'] # user_id if 'user_info' not in videoList[i]: user_id = 0 elif 'user_id' not in videoList[i]['user_info']: user_id = 0 else: user_id = videoList[i]['user_info']['user_id'] # avatar_url if 'user_info' not in videoList[i]: avatar_url = 0 elif 'avatar_url' not in videoList[i]['user_info']: avatar_url = 0 else: avatar_url = videoList[i]['user_info']['avatar_url'] # cover_url if 'video_detail_info' not in videoList[i]: cover_url = 0 elif 'detail_video_large_image' not in videoList[i]['video_detail_info']: cover_url = 0 elif 'url' in videoList[i]['video_detail_info']['detail_video_large_image']: cover_url = videoList[i]['video_detail_info']['detail_video_large_image']['url'] else: cover_url = videoList[i]['video_detail_info']['detail_video_large_image']['url_list'][0]['url'] Common.logger(log_type, crawler).info( f'send_time:{time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(publish_time))}') video_url_dict = cls.get_video_url(log_type, crawler, gid) video_url = video_url_dict["video_url"] audio_url = video_url_dict["audio_url"] video_width = video_url_dict["video_width"] video_height = video_url_dict["video_height"] video_dict = {'video_title': video_title, 'video_id': video_id, 'gid': gid, 'play_cnt': play_cnt, 'comment_cnt': comment_cnt, 'like_cnt': like_cnt, 'share_cnt': share_cnt, 'video_width': video_width, 'video_height': video_height, 'video_duration': video_duration, 'publish_time': publish_time, 'is_top': is_top, 'user_name': user_name, 'user_id': user_id, 'avatar_url': avatar_url, 'cover_url': cover_url, 'audio_url': audio_url, 'video_url': video_url} for k, v in video_dict.items(): print(f"{k}:{v}") print("\n") if __name__ == '__main__': # print(Follow.get_signature("follow", "xigua", "95420624045", "local")) Follow.get_videolist("follow", "xigua", "95420624045", "local") pass