# -*- coding: utf-8 -*- # by @嗷呜 import json import sys import uuid import copy sys.path.append('..') from base.spider import Spider from concurrent.futures import ThreadPoolExecutor, as_completed class Spider(Spider): def init(self, extend=""): self.dbody = { "page_params": { "channel_id": "", "filter_params": "sort=75", "page_type": "channel_operation", "page_id": "channel_list_second_page" } } self.body = self.dbody pass def getName(self): pass def isVideoFormat(self, url): pass def manualVideoCheck(self): pass def destroy(self): pass host = 'https://v.qq.com' apihost = 'https://pbaccess.video.qq.com' headers = { 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.5410.0 Safari/537.36', 'origin': host, 'referer': f'{host}/' } def homeContent(self, filter): cdata = { "电视剧": "100113", "电影": "100173", "综艺": "100109", "纪录片": "100105", "动漫": "100119", "少儿": "100150", "短剧": "110755" } result = {} classes = [] filters = {} for k in cdata: classes.append({ 'type_name': k, 'type_id': cdata[k] }) with ThreadPoolExecutor(max_workers=len(classes)) as executor: futures = [executor.submit(self.get_filter_data, item['type_id']) for item in classes] for future in futures: cid, data = future.result() if not data.get('data', {}).get('module_list_datas'): continue filter_dict = {} try: items = data['data']['module_list_datas'][-1]['module_datas'][-1]['item_data_lists']['item_datas'] for item in items: if not item.get('item_params', {}).get('index_item_key'): continue params = item['item_params'] filter_key = params['index_item_key'] if filter_key not in filter_dict: filter_dict[filter_key] = { 'key': filter_key, 'name': params['index_name'], 'value': [] } filter_dict[filter_key]['value'].append({ 'n': params['option_name'], 'v': params['option_value'] }) except (IndexError, KeyError): continue filters[cid] = list(filter_dict.values()) result['class'] = classes result['filters'] = filters return result def homeVideoContent(self): json_data = {'page_context':None,'page_params':{'page_id':'100101','page_type':'channel','skip_privacy_types':'0','support_click_scan':'1','new_mark_label_enabled':'1','ams_cookies':'',},'page_bypass_params':{'params':{'caller_id':'','data_mode':'default','page_id':'','page_type':'channel','platform_id':'2','user_mode':'default',},'scene':'channel','abtest_bypass_id':'',}} data = self.post(f'{self.apihost}/trpc.vector_layout.page_view.PageService/getPage',headers=self.headers, json=json_data).json() vlist = [] for it in data['data']['CardList'][0]['children_list']['list']['cards']: if it.get('params'): p = it['params'] tag = json.loads(p.get('uni_imgtag', '{}') or p.get('imgtag', '{}') or '{}') id = it.get('id') or p.get('cid') name = p.get('mz_title') or p.get('title') if name and 'http' not in id: vlist.append({ 'vod_id': id, 'vod_name': name, 'vod_pic': p.get('image_url'), 'vod_year': tag.get('tag_2', {}).get('text'), 'vod_remarks': tag.get('tag_4', {}).get('text') }) return {'list': vlist} def categoryContent(self, tid, pg, filter, extend): result = {} params = { "sort": extend.get('sort', '75'), "attr": extend.get('attr', '-1'), "itype": extend.get('itype', '-1'), "ipay": extend.get('ipay', '-1'), "iarea": extend.get('iarea', '-1'), "iyear": extend.get('iyear', '-1'), "theater": extend.get('theater', '-1'), "award": extend.get('award', '-1'), "recommend": extend.get('recommend', '-1') } if pg == '1': self.body = self.dbody.copy() self.body['page_params']['channel_id'] = tid self.body['page_params']['filter_params'] = self.josn_to_params(params) data = self.post( f'{self.apihost}/trpc.universal_backend_service.page_server_rpc.PageServer/GetPageData?video_appid=1000005&vplatform=2&vversion_name=8.9.10&new_mark_label_enabled=1', json=self.body, headers=self.headers).json() ndata = data['data'] if ndata['has_next_page']: result['pagecount'] = 9999 self.body['page_context'] = ndata['next_page_context'] else: result['pagecount'] = int(pg) vlist = [] for its in ndata['module_list_datas'][-1]['module_datas'][-1]['item_data_lists']['item_datas']: id = its.get('item_params', {}).get('cid') if id: p = its['item_params'] tag = json.loads(p.get('uni_imgtag', '{}') or p.get('imgtag', '{}') or '{}') name = p.get('mz_title') or p.get('title') pic = p.get('new_pic_hz') or p.get('new_pic_vt') vlist.append({ 'vod_id': id, 'vod_name': name, 'vod_pic': pic, 'vod_year': tag.get('tag_2', {}).get('text'), 'vod_remarks': tag.get('tag_4', {}).get('text') }) result['list'] = vlist result['page'] = pg result['limit'] = 90 result['total'] = 999999 return result def detailContent(self, ids): vbody = {"page_params":{"req_from":"web","cid":ids[0],"vid":"","lid":"","page_type":"detail_operation","page_id":"detail_page_introduction"},"has_cache":1} body = {"page_params":{"req_from":"web_vsite","page_id":"vsite_episode_list","page_type":"detail_operation","id_type":"1","page_size":"","cid":ids[0],"vid":"","lid":"","page_num":"","page_context":"","detail_page_type":"1"},"has_cache":1} with ThreadPoolExecutor(max_workers=2) as executor: future_detail = executor.submit(self.get_vdata, vbody) future_episodes = executor.submit(self.get_vdata, body) vdata = future_detail.result() data = future_episodes.result() pdata = self.process_tabs(data, body, ids) if not pdata: return self.handle_exception(None, "No pdata available") try: star_list = vdata['data']['module_list_datas'][0]['module_datas'][0]['item_data_lists']['item_datas'][ 0].get('sub_items', {}).get('star_list', {}).get('item_datas', []) actors = [star['item_params']['name'] for star in star_list] names = ['腾讯视频', '预告片'] plist, ylist = self.process_pdata(pdata, ids) if not plist: del names[0] if not ylist: del names[1] vod = self.build_vod(vdata, actors, plist, ylist, names) return {'list': [vod]} except Exception as e: return self.handle_exception(e, "Error processing detail") def searchContent(self, key, quick, pg="1"): headers = self.headers.copy() headers.update({'Content-Type': 'application/json'}) body = {'version':'25021101','clientType':1,'filterValue':'','uuid':str(uuid.uuid4()),'retry':0,'query':key,'pagenum':int(pg)-1,'pagesize':30,'queryFrom':0,'searchDatakey':'','transInfo':'','isneedQc':True,'preQid':'','adClientInfo':'','extraInfo':{'isNewMarkLabel':'1','multi_terminal_pc':'1','themeType':'1',},} data = self.post(f'{self.apihost}/trpc.videosearch.mobile_search.MultiTerminalSearch/MbSearch?vplatform=2', json=body, headers=headers).json() vlist = [] vname=["电视剧", "电影", "综艺", "纪录片", "动漫", "少儿", "短剧"] v=data['data']['normalList']['itemList'] d=data['data']['areaBoxList'][0]['itemList'] q=v+d if v[0].get('doc') and v[0]['doc'].get('id') =='MainNeed':q=d+v for k in q: if k.get('doc') and k.get('videoInfo') and k['doc'].get('id') and '外站' not in k['videoInfo'].get('subTitle') and k['videoInfo'].get('title') and k['videoInfo'].get('typeName') in vname: img_tag = k.get('videoInfo', {}).get('imgTag') if img_tag is not None and isinstance(img_tag, str): try: tag = json.loads(img_tag) except json.JSONDecodeError as e: tag = {} else: tag = {} pic = k.get('videoInfo', {}).get('imgUrl') vlist.append({ 'vod_id': k['doc']['id'], 'vod_name': self.removeHtmlTags(k['videoInfo']['title']), 'vod_pic': pic, 'vod_year': k['videoInfo'].get('typeName') +' '+ tag.get('tag_2', {}).get('text', ''), 'vod_remarks': tag.get('tag_4', {}).get('text', '') }) return {'list': vlist, 'page': pg} def playerContent(self, flag, id, vipFlags): ids = id.split('@') url = f"{self.host}/x/cover/{ids[0]}/{ids[1]}.html" return {'parse': 1, 'url': url, 'header': ''} def localProxy(self, param): pass def get_filter_data(self, cid): hbody = self.dbody.copy() hbody['page_params']['channel_id'] = cid data = self.post( f'{self.apihost}/trpc.universal_backend_service.page_server_rpc.PageServer/GetPageData?video_appid=1000005&vplatform=2&vversion_name=8.9.10&new_mark_label_enabled=1', json=hbody, headers=self.headers).json() return cid, data def get_vdata(self, body): try: vdata = self.post( f'{self.apihost}/trpc.universal_backend_service.page_server_rpc.PageServer/GetPageData?video_appid=3000010&vplatform=2&vversion_name=8.2.96', json=body, headers=self.headers ).json() return vdata except Exception as e: print(f"Error in get_vdata: {str(e)}") return {'data': {'module_list_datas': []}} def process_pdata(self, pdata, ids): plist = [] ylist = [] for k in pdata: if k.get('item_id'): pid = f"{k['item_params']['union_title']}${ids[0]}@{k['item_id']}" if '预告' in k['item_params']['union_title']: ylist.append(pid) else: plist.append(pid) return plist, ylist def build_vod(self, vdata, actors, plist, ylist, names): d = vdata['data']['module_list_datas'][0]['module_datas'][0]['item_data_lists']['item_datas'][0]['item_params'] urls = [] if plist: urls.append('#'.join(plist)) if ylist: urls.append('#'.join(ylist)) vod = { 'type_name': d.get('sub_genre', ''), 'vod_name': d.get('title', ''), 'vod_year': d.get('year', ''), 'vod_area': d.get('area_name', ''), 'vod_remarks': d.get('holly_online_time', '') or d.get('hotval', ''), 'vod_actor': ','.join(actors), 'vod_content': d.get('cover_description', ''), 'vod_play_from': '$$$'.join(names), 'vod_play_url': '$$$'.join(urls) } return vod def handle_exception(self, e, message): print(f"{message}: {str(e)}") return {'list': [{'vod_play_from': '哎呀翻车啦', 'vod_play_url': '翻车啦#555'}]} def process_tabs(self, data, body, ids): try: pdata = data['data']['module_list_datas'][-1]['module_datas'][-1]['item_data_lists']['item_datas'] tabs = data['data']['module_list_datas'][-1]['module_datas'][-1]['module_params'].get('tabs') if tabs and len(json.loads(tabs)): tabs = json.loads(tabs) remaining_tabs = tabs[1:] task_queue = [] for tab in remaining_tabs: nbody = copy.deepcopy(body) nbody['page_params']['page_context'] = tab['page_context'] task_queue.append(nbody) with ThreadPoolExecutor(max_workers=10) as executor: future_map = {executor.submit(self.get_vdata, task): idx for idx, task in enumerate(task_queue)} results = [None] * len(task_queue) for future in as_completed(future_map.keys()): idx = future_map[future] results[idx] = future.result() for result in results: if result: page_data = result['data']['module_list_datas'][-1]['module_datas'][-1]['item_data_lists'][ 'item_datas'] pdata.extend(page_data) return pdata except Exception as e: print(f"Error processing episodes: {str(e)}") return [] def josn_to_params(self, params, skip_empty=False): query = [] for k, v in params.items(): if skip_empty and not v: continue query.append(f"{k}={v}") return "&".join(query)