抖音搜索视频python协议源码
async def searchVideos(cls, db, keyword, cursor,count=20,proxy=None, tries=TRY): if tries <= 0: return cls.NoResult version = random.choice(VERSION) params = { 'manifest_version_code':f'{version}', 'update_version_code':f'{version}2', # 'os_api':'28', 'app_type':'normal', 'device_id':f'{cls.random_device_id(db) }', 'aid':'1128', 'ac':'wifi', 'mcc_mnc':'46001', 'app_name':'aweme', 'version_code':version, 'device_platform':'android', 'version_name':'8.0.0', 'js_sdk_version':'1.19.2.0', 'channel':random.choice(CHANNEL), 'device_type':f'{fake_device_type()}', 'device_brand':f'{fake_device_type()}', 'os_version':'9', 'ssmix':'a', } data = { "count": count, "offset": cursor, "keyword": keyword, 'is_pull_refresh': '1', 'hot_search': '0', 'source': 'video_search', 'search_id': '', 'query_correct_type': '1' } api = gen_API(DYapi.SearchVideo, params) jobj = get_post_xg(api, data) cb = lambda x: False if x and x.get('aweme_list') else True ret = await async_request('POST', jobj['url'], proxy=proxy, callback=cb, data=jobj['data'], headers=jobj['headers']) if not ret: ret = await tornadoAyncRequest(jobj['url'], 'post', proxy=proxy, data=jobj['data'], headers=jobj['headers']) if cb(ret): proxy = await Proxy.get() tries -= 1 ret = await cls.searchVideos(db, keyword, cursor, count, proxy, tries) return ret def get_post_xg(url,data,cookie=''): index = url.find('?') + 1 params = url[index:] data_encode = urllib.parse.urlencode(data) x = X_Gorgon(params, data_encode,cookie) ck = cookie if cookie else f"install_id={int(random.randrange(10 ** 11, 10 ** 12))}" headers = { "Cookie":ck , "X-Khronos": x['X-Khronos'], "sdk-version": "1", "X-SS-REQ-TICKET": f"{x['X-Khronos']}233", "X-Gorgon": x['X-Gorgon'], "User-Agent": "com.ss.android.ugc.aweme/750 (Linux; U; Android 8.0.0; zh_CN; MIX 2; Build/OPR1.170623.027; Cronet/58.0.2991.0)", # "User-Agent": "ttnet okhttp/3.10.0.2", "Connection": "Keep-Alive", } return { 'url':url, 'data':data, 'headers':headers }