import requests import uuid import random import time import json import hashlib from libs.aesgzip import tt_encrypt from urllib import parse from libs.proxy import Proxy from rds_model.db_redis import DbRedis def get_mc(): def a(): seed = "1234567890ABCDEF" sa = [] for i in range(2): sa.append(random.choice(seed)) salt = ''.join(sa) return salt k = '' for i in range(6): k += a() + ':' return k[:-1] def get_random(i, random_type=1): if random_type == 1: return str(random.randint(1 * 10 ** (i - 1), 1 * 10 ** i - 1)) elif random_type == 8: seed = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" sa = [] for i in range(i): sa.append(random.choice(seed)) salt = ''.join(sa) return salt else: seed = "1234567890abcde" sa = [] for i in range(i): sa.append(random.choice(seed)) salt = ''.join(sa) return salt def get_random_brand_type(): brand_type = get_random(3, random_type=8) + '-' + get_random(2, random_type=8) + '00' return brand_type V1 = str(random.randint(5,8)) V2 = str(random.randint(0,9)) channel = 'oppo' device_type_list = ['VTR-AL00', 'KNT-AL10', 'PRA-AL00', 'DUA-TL00', 'PLK-AL10', 'KIW-AL10', 'LON-AL00', 'ANA-AN00', 'JSC-AN00', 'TAS-AN00', 'TAS-AL00'] # device_type = 'GRL-CL00' device_type = get_random_brand_type() device_brand = 'HUAWEI' # Y79， Y66i， V7， X20， X20Plus， V7+， X9s L，V5s，Y66，Y30 # device_type = random.choice(device_type_list) # channel_list = ['huawei', 'googleplay', 'douyin-huidu-guanwang-test', 'douyin_tengxun_wzl', # 'wandoujia_zhiwei', 'update'] # channel = random.choice(channel_list) # print(channel, device_type) cookie = '' class DouYinApi: #USER_AGENT = 'okhttp/3.10.0.1' USER_AGENT = f'com.ss.android.ugc.aweme/{V1}{V2}0 (Linux; U; Android 5.1.1; zh_CN; {device_type}; Build/LMY47V; Cronet/58.0.2991.0)' COMMON_DEVICE_PARAMS = { 'retry_type': 'no_retry', 'ac': '4g', 'channel': channel, 'aid': '2329', 'app_name': 'aweme', 'version_code': f'{V1}{V2}0', 'version_name': f'{V1}.{V2}.0', 'device_platform': 'android', 'ssmix': 'a', 'device_type': device_type, 'device_brand': device_brand, 'language': 'zh', 'os_api': '25', 'os_version': '7.1.2', 'manifest_version_code': f'{V1}{V2}0', 'resolution': '720*1280', 'dpi': '192', 'update_version_code': f'{V1}{V2}02', 'mcc_mnc': '46000' } PROXY = {} def __init__(self, sessionid, proxies): """ :param cid: client id """ self.proxies = proxies self.__cid = '' self.__device_id = '' self.__iid = '' self.__uuid = '' self.__openudid = '' self.__device_params = {} self.__cookie = { } def init_device_ids(self, device_id, iid, udid, openudid, cc=None): """初始化设备id参数 :param device_id: device id :param iid: install id :param udid: imei :param openudid: open udid :param serial_number: serial no :param clientudid: client udid :param sim_serial_number: sim serial number :param mc: mac address :return: none """ self.__device_id = device_id self.__iid = iid self.__uuid = udid self.__openudid = openudid device_ids = { 'uuid': udid, 'openudid': openudid } if device_id and iid: device_ids.update({ 'device_id': device_id, 'iid': iid, }) self.__device_params = self.COMMON_DEVICE_PARAMS.copy() self.__device_params.update(device_ids) if cc: self.__cookie.update(cc) def __get_encrypted_device_info(self, device_id, openudid, udid, clientudid, serial_number, mac, iid): register_info = { "magic_tag": "ss_app_log", "header": { "display_name": "抖音短视频", "update_version_code": int(self.COMMON_DEVICE_PARAMS['update_version_code']), "manifest_version_code": int(self.COMMON_DEVICE_PARAMS['manifest_version_code']), "aid": '2329', "channel": self.COMMON_DEVICE_PARAMS['channel'], "appkey": "57bfa27c67e58e7d920028d3", "package": "com.ss.android.ugc.aweme", "app_version": self.COMMON_DEVICE_PARAMS['version_name'], "version_code": int(self.COMMON_DEVICE_PARAMS['version_code']), "sdk_version": "2.5.5.8", "os": "Android", "os_version": self.COMMON_DEVICE_PARAMS['os_version'], "os_api": self.COMMON_DEVICE_PARAMS['os_api'], "device_model": self.COMMON_DEVICE_PARAMS['device_type'], "device_brand": self.COMMON_DEVICE_PARAMS['device_brand'], "device_manufacturer": self.COMMON_DEVICE_PARAMS['device_brand'], "cpu_abi": "armeabi-v7a", "build_serial": serial_number, "release_build": "fbfabbc_20190522", "density_dpi": self.COMMON_DEVICE_PARAMS['dpi'], "display_density": "mdpi", "resolution": "1280x720", "language": "zh", "mc": mac, "timezone": 8, "access": "4G", "not_request_sender": 0, "rom": "EMUI-rel.se.infra.20191210.155925", "rom_version": "HUAWEI-user 7.1.2 20171130.276299 release-keys", "sig_hash": "aea615ab910015038f73c47e45d21466", # "device_id": str(device_id), "openudid": str(openudid), "udid": str(udid), "clientudid": str(clientudid), "serial_number": str(serial_number), "sim_serial_number": [ ], "region": "CN", "tz_name": "Asia/Shanghai", "tz_offset": 28800 }, "_gen_time": str(round(time.time() * 1000)) } if device_id: register_info['header']['device_id'] = str(device_id) if iid: register_info['header']['iid'] = str(iid) register_info['header']['push_sdk'] = '[1, 2, 6, 7, 8, 9]' return tt_encrypt((json.dumps(register_info))) def log_settings(self, ids): device_id = ids['device_id'] openudid = ids['openudid'] udid = ids['uuid'] clientudid = ids['clientudid'] serial_number = ids['serial_number'] iid = ids['iid'] mc = ids['mc'] d = self.__get_encrypted_device_info(device_id, openudid, udid, clientudid, serial_number, mc, iid) url = 'https://log.snssdk.com/service/2/log_settings/?' url = self.__add_other_params(url, {}) sign = self.__get_sign(url) if self.proxies: resp = requests.post(url, headers=self.__get_headers(sign), cookies=self.__cookie, proxies=self.proxies, data=d, verify=False, timeout=20) else: resp = requests.get(url, headers=self.__get_headers(sign), cookies=self.__cookie, data=d, verify=False, timeout=20) # print(resp.text) def xlog(self, device_id, iid, udid, openudid, serial_number): xlog = { "dpod": { "pod": "" }, "p1": f"{device_id}", "p2": f"{iid}", "ut": 72, "ait": int(time.time() - 72), "pkg": "com.ss.android.ugc.aweme", "fp": "HUAWEI/HUAWEI/VOG-AL00:5.1.1/20171130.376229:user/release-keys", "vc": 0, "vpn": 0, "wifisid": "", "wifimac": "", "wifip": "", "aplist": [ ], "route": { "iip": "0", "gip": "0", "ghw": "0", "type": "0" }, "location": "", "apps": ["com.microvirt.guide", "com.android.browser", "com.ss.android.ugc.aweme"], "mdi_if": { }, "extra": "install", "hw": { "brand": self.COMMON_DEVICE_PARAMS['device_brand'], "model": self.COMMON_DEVICE_PARAMS['device_type'], "board": self.COMMON_DEVICE_PARAMS['device_brand'], "device": self.COMMON_DEVICE_PARAMS['device_type'], "product": self.COMMON_DEVICE_PARAMS['device_type'], "display": "720*1280", "dpi": 0, "bat": 1000, "cpu": { "core": 2, "hw": "placeholder", "max": "2400000", "min": "1000000", "ft": "neon vfp swp half thumb fastmult edsp vfpv3" }, "mem": { "ram": "2120286208", "rom": "8320901120" } }, "id": { "i": 22, "r": "5.1.1", "imei": f"{udid}", "imsi": '46000' + str(self.__get_random(10)), "adid": f"{openudid}", "adid_ex": f"{openudid}", "mac": '', "serial": serial_number }, "emulator": { }, "env": { "ver": "0.6.02.17", "tag": "default", "pkg": "com.ss.android.ugc.aweme", "tz": "GMT+08:00", "ml": "zh_CN", "uid": 10045, "rebuild": 0, "jd": 0, "dbg": -1, "tid": 0, "xposed": 0, "hk": [ "__system_property_get", "fopen", "fgets", "fread", "getpid", "gettid", "dlopen", "dlsym", "dladdr"], "su": 0, "sp": "/system/xbin/su", "ro.secure_s": "1", "ro.debuggable_s": "0", "click": "", "hph": "", "hpp": "", "mc": 0, "fc": 946933748, "jexp": 0, "frida": 0, "cydia": 0, "vapp": "" }, "extension": { "bt": -1, "notify": 2212858197, "sign": "AEA615AB910015038F73C47E45D21466", "bytes64": "Aa1IrwSQLRXfmUvJokUpTkHJ9Yh1G4lpVgZSOaZeri+QVgSy80D5ID5UbH6x6xcSHsdjXqljImqg/LxoSbAeyqYZ67ZUTE23KmeyzHjTOeTtGogSEQY2/D8UWmY7Xt815Ch96RiLEgmzqnzCGZ0/N2AAaS5FynrMMfNv9jPlZgcZBjlWv5Ept44wrLJ+G1rJ7CzjL567O2ClTXKGTbHDTZsh91lWXmcNKzsh9PReifo2kf9f2MvJnJ9mN80soe7hw+/96olR5tTH6yVpsbuJt0NjxXI+y5yKp+GLueKCeYbG4QIqqI4buMEsjnPQCOm1bZtYxqeevC1aJVlNMdoQSZjZ7xqsWk8cNBuXqe/XXQua4QIqqI4buMEsjnPQCOm1bWat8qmH4OhJGLtb/aaNzHlXbo7P3DIJUsaOqMvCZvAN5apd9qaCQuvbISV7KJveRYfCRQGc0wiPPEaBEF7iO6qDRjlTDdNocBDAc6VdJS27EXEPHFPIH40PlLnx7vTDN6kUtGcG6G07pnlwwdAp6VZJskS5EmTY0MWyrPBOzBug+bRJCza4pcoC5A8ALNOUc39lxIAX7NuMJcf5rmTmkg8asiKpeeQHm2lSK5YNJkldp4+JLVggF/F+Dr6JjhyhERFxDxxTyB+ND5S58e70wzcNUOIjqUHt95eTwSOIFJvTZ+ZxXM4kDOGCeZXOBoqAfUbYgYfGrRZtcMdw1Lra0HN713ei+bcfoILrHoJO4fS3ZEq1jSU2d8XDEIid38oMwYrPyzXQOAQ/eRZlCiMSW83LMCj/4N0KpL/qlOzmB0uiBqKuF73Qhd0IfQTPpVCrwO6P/upwe1Q8yj+kXQBntfARcQ8cU8gfjQ+UufHu9MM3QpsZRTCVwNPhHp7m8lPb3/6YAZdknQzHMOVZp/LM1kf9MvhPN3heC2rp/QwWY1p+IPPe4wuRVVSO2vyJ8PhyDQxltLZjJW23706uyu+wGLFBiE7+4WBKMHWO3JmR6TWuHglclf+zh1CT21Wy5n/3Ex732O0WkYD01c/U27BcQ2Gb5RZdXYfhXTJe6VIzQnf02PCccLK2HnTdZ6tZ4qquKhkB0gKbMhtx/V5sBYe38cgU+5ct4Zy3fRwT/i5HkB2Wh6UlY2Ty1vIfaakVcR9dC7TKHDAWLIO0CPgBDeZFdb1zfIGyz50R3D9FNBoXkK6o5MRSDrM+pK5mlxZPJ0n56WiqDTEjAaxwEDWLejQFgyczb8DTeJnn5+nvyP/5jmfdv9sb7e8tpPlu02HsOtNzRLZRdpI3mgFYLPF0AJz2DHf90TVVUkETnF3ts+CqH34sIKT0zmoYjvpKGzdznMgRkJMPB3/Q4n/SZqJYVx9Bae0yYyFhtXt6XCdLgBPreyCy57dyB/svbMR4tvQ6K6W28yo9zGrt6gp20h6UzBfquQkOfB1SPLAqdyCyEcbja+ew+LgOejtotPR6lPJ7gUDbIoBURxkdLpINVc0RirdDBYeDaH3sZPli2YneQthTQCW91O6bRzlAEbljd7NTi3OomYuf+aUu8Y5rTjnyJAig+jSORIsyBHYeyrnBuAbFZUsvC5yRvAfEf6Uf1KWgGjcplhfxnBb4ocrYARhO5NCkCRg66GqLCXUDUpm2Gh2mN17vPXAF2ZH+W/hUtmsTs5zkJwks+vnWRDGs9QZWzkUUd2A/eHRxxHqx51yqZKT6v6uuF716rYsthonl01CqAJe7q0jq//hm7ZwCwWyY+GxkxON9PRgo36h0MOgFSURMf/C8pOUz21p3kXrPxmoPxXrisOD5U3oCiOe01fWZoxtAHCzgQpWtnDwqPVxohTwXyShv", }, "rl": { }, "ssp": { }, "grilock": "eyJvcyI6IkFuZHJvaWQiLCJ2ZXJzaW9uIjoiMS4wLjAiLCJ0b2tlbl9pZCI6IiIsImNvZGUiOjUwNH0=" } s = (json.dumps(xlog)) url = f"https://xlog.snssdk.com/v2/r?os=0&ver=0.6.02.17&m=1&app_ver=5.5.0®ion=CN&aid=2329&did={device_id}" data = {"post_d": s, "url": url} r = requests.post("http://192.168.77.186:18400/xlogr", data=str(data), timeout=20).text # print(r) def register_device(self): udid = '8678' + get_random(11) serial_number = "71623411" openudid = 'c2a5' + str(uuid.uuid4())[-12:] clientudid = str(uuid.uuid4()) mc = get_mc() """获取设备信息，传参数注册老设备，不传参数注册新设备 :return: """ # serial_number = str(uuid.uuid4())[-12:] # if openudid is None: # openudid = 'fa23' + str(uuid.uuid4())[-12:] # clientudid = str(uuid.uuid4()) # if udid is None: # udid = '861' + self.__get_random(12) # mc = self.__get_random_mac() # if device_id is None: # device_id = '825739' + self.__get_random(9) params = { 'uuid': udid, 'openudid': openudid, '_rticket': str(int(round(time.time() * 1000))) } requests.packages.urllib3.disable_warnings() params.update(self.COMMON_DEVICE_PARAMS) device_register_url = 'https://log.snssdk.com/service/2/device_register/?' + parse.urlencode(params) headers = { 'User-Agent': DouYinApi.USER_AGENT } d = self.__get_encrypted_device_info(None, openudid, udid, clientudid, serial_number, mc, iid=None) if self.proxies: resp = requests.post(device_register_url, data=d, proxies=self.proxies, headers=headers, verify=False, timeout=20) else: resp = requests.post(device_register_url, data=d, headers=headers, verify=False, timeout=20) cookie = resp.cookies.get_dict() if len(cookie) != 0: self.__cookie.update(cookie) # print(self.__cookie) resp = resp.json() ids = { 'new_user': resp['new_user'], 'device_id': str(resp['device_id']), 'iid': str(resp['install_id']), 'uuid': udid, 'openudid': openudid, 'serial_number': serial_number, 'clientudid': clientudid, 'mc': mc, 'cookie': parse.urlencode(self.__cookie) } return ids def app_log(self): params = { 'uuid': udid, 'openudid': openudid, '_rticket': str(int(round(time.time() * 1000))) } params.update(self.COMMON_DEVICE_PARAMS) device_register_url = 'https://log.snssdk.com/service/2/app_log/?' + parse.urlencode(params) headers = { 'User-Agent': DouYinApi.USER_AGENT } data = {"device_info": device_info, 'url': device_register_url, 'ck': ''} gorgon = requests.post("http://192.168.77.186:18400/app_log", data=str(data), timeout=20).json() d = tt_encrypt(device_info) if self.proxies: resp = requests.post(device_register_url, data=d, proxies=self.proxies, headers=headers, verify=False, timeout=20) else: resp = requests.post(device_register_url, data=d, headers=headers, verify=False, timeout=20) cookie = resp.cookies.get_dict() if len(cookie) != 0: self.__cookie.update(cookie) resp = resp.json() # print(resp) return def get_feed(self): """获取首页推荐列表 """ douyin_url = 'https://aweme-eagle.snssdk.com/aweme/v1/feed/?type=0&max_cursor=0&min_cursor=-1&count=6&volume=0.0&pull_type=2&need_relieve_aweme=0&filter_warn=0&req_from&is_cold_start=0' return self.__http_get(douyin_url) def get_nearby_feed(self, city_id): """获取对应城市的推荐列表 :param cityid: 城市代码, 从https://wenku.baidu.com/view/af4281bafd0a79563c1e7287.html获取 :return: """ params = { 'city': city_id, } douyin_url = 'https://api.amemv.com/aweme/v1/nearby/feed/?max_cursor=0&min_cursor=0&count=20&feed_style=1&filter_warn=0&poi_class_code=0' return self.__http_get(douyin_url, params) def get_user_info(self, user_id): """获取用户信息 :param user_id: 用户ID :return: """ params = { 'user_id': user_id } douyin_url = 'https://aweme-eagle.snssdk.com/aweme/v1/user/?' return self.__http_get(douyin_url, params) def get_room_flv(self, room_id): """获取用户信息 :param user_id: 用户ID :return: """ params = { 'room_id': room_id, 'live_id':'1' } douyin_url = 'https://webcast-hl.amemv.com/webcast/room/reflow/info/?' return self.__http_get(douyin_url, params) def get_user_post(self, user_id, max_cursor, count): """获取用户作品 :param user_id: 用户ID :param max_cursor: 用于分页，第1页是0，后1页是上1页请求的时候返回的max_cursor :param count: 返回视频的条数 :return: """ params = { 'user_id': user_id, 'max_cursor': str(max_cursor), 'count': str(count) } douyin_url = 'https://aweme.snssdk.com/aweme/v1/aweme/post/' return self.__http_get(douyin_url, params) def comm_get(self, douyin_url, params): """获取用户信息 :param user_id: 用户ID :return: """ return self.__http_get(douyin_url, params) def get_sec_user_favorite(self, sec_user_id, max_cursor, count): """获取用户作品 :param user_id: 用户ID :param max_cursor: 用于分页，第1页是0，后1页是上1页请求的时候返回的max_cursor :param count: 返回视频的条数 :return: """ params = { 'sec_user_id': sec_user_id, 'max_cursor': str(max_cursor), 'count': str(count) } douyin_url = 'https://aweme-hl.snssdk.com/aweme/v1/aweme/favorite/?invalid_item_count=0&is_hiding_invalid_item=0&' return self.__http_get(douyin_url, params) def get_user_favorite(self, user_id, max_cursor, count): """获取用户作品 :param user_id: 用户ID :param max_cursor: 用于分页，第1页是0，后1页是上1页请求的时候返回的max_cursor :param count: 返回视频的条数 :return: """ params = { 'user_id': user_id, 'max_cursor': str(max_cursor), 'count': str(count) } douyin_url = 'https://aweme-hl.snssdk.com/aweme/v1/aweme/favorite/?invalid_item_count=0&is_hiding_invalid_item=0&' return self.__http_get(douyin_url, params) def get_user_forward_list(self, user_id, max_cursor, count): """获取用户动态 :param user_id: 用户ID :param max_cursor: 用于分页，第1页是0，后1页是上1页请求的时候返回的max_cursor :param count: 每次返回的动态条数 :return: """ params = { 'user_id': user_id, 'max_cursor': str(max_cursor), 'count': str(count) } douyin_url = 'https://aweme.snssdk.com/aweme/v1/forward/list/' return self.__http_get(douyin_url, params) def get_user_following_list(self, user_id, max_time, count): """获取用户关注列表 注意：关注列表请求太频繁会导致不返回数据 :param user_id: 用户ID :param max_time: 用于分页，第1页是0，后1页是上1页请求时返回的min_time :param count: 每次返回的条数 :return: """ params = { 'user_id': user_id, 'max_time': str(int(time.time()) if max_time == 0 else max_time), 'count': str(count), 'source_type': '1' } douyin_url = 'https://aweme.snssdk.com/aweme/v1/user/following/list/' return self.__http_get(douyin_url, params) def get_user_follower_list(self, user_id, min_time, count): """获取用户粉丝列表 :param user_id: 用户ID :param min_time: 用于分页，第1页是0，后1页是上1页请求时返回的min_time :param count: 每次返回的条数 :return: """ params = { 'user_id': user_id, 'max_time': str(int(time.time()) if min_time == 0 else min_time), 'count': str(count) } douyin_url = 'https://aweme.snssdk.com/aweme/v1/user/follower/list/' return self.__http_get(douyin_url, params) def get_user_favorite_list(self, user_id, min_time, count): """获取用户粉丝列表 :param user_id: 用户ID :param min_time: 用于分页，第1页是0，后1页是上1页请求时返回的min_time :param count: 每次返回的条数 :return: """ params = { 'user_id': user_id, 'max_time': str(int(time.time()) if min_time == 0 else min_time), 'count': str(count) } douyin_url = 'https://aweme.snssdk.com/aweme/v1/user/follower/list/' return self.__http_get(douyin_url, params) def get_hot_search_list(self): """获取抖音热搜榜 :return: """ douyin_url = 'https://api.amemv.com/aweme/v1/hot/search/list/?detail_list=1' return self.__http_get(douyin_url) def get_hot_video_list(self): """获取抖音视频榜 :return: """ douyin_url = 'https://aweme.snssdk.com/aweme/v1/hotsearch/aweme/billboard/' return self.__http_get(douyin_url) def get_hot_music_list(self): """获取抖音音乐榜 :return: """ douyin_url = 'https://aweme.snssdk.com/aweme/v1/hotsearch/music/billboard/' return self.__http_get(douyin_url) def get_hot_positive_energy_list(self): """获取抖音正能量榜 :return: """ douyin_url = 'https://aweme.snssdk.com/aweme/v1/hotsearch/positive_energy/billboard/' return self.__http_get(douyin_url) def get_hot_category_list(self, cursor, count): """获取热门分类列表 :param cursor: 分页用，第1页是0，下一页是上1页请求返回的cursor :param count: 每次返回的条数 :return: """ params = { 'cursor': str(cursor), 'count': str(count) } douyin_url = 'https://aweme.snssdk.com/aweme/v1/category/list/' return self.__http_get(douyin_url, params) def general_search(self, keyword, offset, count): """综合搜索 :param keyword: 关键词 :param offset: 分页，第1页是0，下1页是上1页请求返回的cursor :param count: 每次返回的条数 :return: """ params = { 'keyword': keyword, 'offset': str(offset), 'count': str(count), 'is_pull_refresh': '0', 'hot_search': '0', 'latitude': '0.0', 'longitude': '0.0' } douyin_url = 'https://aweme-hl.snssdk.com/aweme/v1/general/search/single/?' return self.__http_post(douyin_url, params) def item_search(self, keyword, offset, count): """综合搜索 :param keyword: 关键词 :param offset: 分页，第1页是0，下1页是上1页请求返回的cursor :param count: 每次返回的条数 :return: """ params = { 'query': keyword, 'cursor': str(offset), 'count': str(count), 'request_type': '1', 'search_filter': '1', 'search_source': 'commodity_search', 'enter_from': 'homepage_hot' } douyin_url = 'https://api5-normal-c-hl.amemv.com/aweme/v2/shop/search/aggregate/shopping/?' return self.__http_post(douyin_url, params) def video_search(self, keyword, offset, count): """ 视频搜索 :param keyword: 关键词 :param offset: 分页，第1页是0，下1页是上1页请求返回的cursor :param count: 每次返回的条数 :return: """ params = { 'keyword': keyword, 'offset': str(offset), 'count': str(count), 'is_pull_refresh': '0', 'hot_search': '0', 'source': 'video_search' } douyin_url = 'https://aweme-hl.snssdk.com/aweme/v1/search/item/?' return self.__http_post(douyin_url, params) def user_search(self, keyword, offset, count): """ 用户搜索 :param keyword: 关键词 :param offset: 分页，第1页是0，下1页是上1页请求返回的cursor :param count: 每次返回的条数 :return: """ params = { 'keyword': keyword, 'cursor': str(offset), 'count': str(count), 'type': '1', 'is_pull_refresh': '0', 'hot_search': '0', 'source': '' } douyin_url = 'https://aweme-hl.snssdk.com/aweme/v1/discover/search/?' return self.__http_post(douyin_url, params) def get_video_comment_list(self, aweme_id, cursor, count): """获取视频评论列表 :param awemeId: 视频ID :param cursor: 分页, 第1页是0, 下1页是上1页请求返回的cursor :param count: 每次返回的条数 :return: """ params = { 'aweme_id': aweme_id, 'cursor': str(cursor), 'count': str(count) } douyin_url = 'https://aweme.snssdk.com/aweme/v2/comment/list/' return self.__http_get(douyin_url, params) def get_video_detail(self, aweme_id): """获取视频详情 :param aweme_id: 视频ID :return: """ params = { 'aweme_id': aweme_id } douyin_url = 'https://aweme.snssdk.com/aweme/v1/aweme/detail/' return self.__http_get(douyin_url, params) def get_music_detail(self, music_id): """获取音乐详情 :param music_id: 音乐id :return: """ params = { 'music_id': str(music_id), 'click_reason': '0' } douyin_url = 'https://aweme.snssdk.com/aweme/v1/music/detail/' return self.__http_get(douyin_url, params) def get_music_videos(self, music_id, cursor, count): """获取音乐对应的视频列表 :param music_id: 音乐id :param cursor: 分页，首页是0，下一页是上一页请求返回的cursor :param count: 每次返回的条数 :return: """ params = { 'music_id': str(music_id), 'cursor': str(cursor), 'count': str(count), 'type': '6' } douyin_url = 'https://aweme.snssdk.com/aweme/v1/music/aweme/' return self.__http_get(douyin_url, params) def get_topic_videos(self, hashtag_name, cursor, count): """获取话题相关视频 :param hashtag_name: 话题 :param cursor: 分页，首页是0，下一页是上一页请求返回的cursor :param count: 每次返回的条数 :return: """ params = { 'cursor': str(cursor), 'count': str(count), 'source': 'challenge_video', 'hashtag_name': hashtag_name, 'type': '5', 'query_type': '1' } douyin_url = 'https://aweme-hl.snssdk.com/aweme/v1/challenge/aweme/' return self.__http_get(douyin_url, params) def get_promotion_list(self, user_id,sec_user_id, cursor, count): """获取商品橱窗列表 :param user_id: user id :param cursor: 分页，首页是0，下一页是上一页请求返回的cursor :param count: 每次返回的条数 :return: """ params = { # 'sec_user_id':sec_user_id, 'count': str(count), 'cursor': str(cursor), 'user_id': str(user_id), 'retry_type': 'no_retry', 'ac': '4g', 'channel': channel, 'aid': '2329', 'app_name': 'aweme', 'version_code': f'{V1}{V2}0', 'version_name': f'{V1}.{V2}.0', 'device_platform': 'android', 'ssmix': 'a', 'device_type': device_type, 'device_brand': device_brand, 'language': 'zh', 'os_api': '25', 'os_version': '7.1.2', 'manifest_version_code': f'{V1}{V2}0', 'resolution': '720*1280', 'dpi': '192', 'update_version_code': f'{V1}{V2}02', 'mcc_mnc': '46000' } douyin_url = 'https://aweme.snssdk.com/aweme/v1/promotion/user/promotion/list/' return self.__http_get(douyin_url, params) def get_share_video_detail(self, share_url): """获取分享链接对应的视频信息 :param share_url: 分享链接 :return: """ headers = { 'Accept': '*/*', 'Connection': 'keep-alive', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.116 Safari/537.36' } resp = requests.get(share_url, headers=headers, timeout=20) url_path = parse.urlparse(resp.url).path video_id = url_path.split('/')[3] return self.get_video_detail(video_id) def get_webcast_room_info(self, room_id): """获取直播房间信息 :param room_id: :return: """ params = { 'pack_level': '4', 'room_id': str(room_id), 'webcast_sdk_version': '1710' } douyin_url = 'https://webcast.amemv.com/webcast/room/info/?' return self.__http_get(douyin_url, params) def get_webcast_user_info(self, room_id, user_id): """获取直播用户信息 :param user_id: :return: """ params = { 'request_from': 'admin', 'current_room_id': str(room_id), 'target_uid': str(user_id), 'anchor_id': str(user_id), 'packed_level': '2', 'webcast_sdk_version': '1150' } douyin_url = 'https://webcast.amemv.com/webcast/user/' return self.__http_get(douyin_url, params) def get_webcast_ranklist(self, room_id): """获取直播本场榜 :param room_id: :param anchor_id: :return: """ params = { 'room_id': str(room_id), 'rank_type': '17', 'webcast_sdk_version': '1150' } douyin_url = 'https://webcast.amemv.com/webcast/ranklist/room/' + str(room_id) + '/contributor/' return self.__http_get(douyin_url, params) def __get_random(self, len): return ''.join(str(random.choice(range(10))) for _ in range(len)) def __get_msg(self, resp): return json.loads(resp.text)['msg'] def __get_random_mac(self): mac = [0x10, 0x2a, 0xb3, random.randint(0x00, 0x7f), random.randint(0x00, 0xff), random.randint(0x00, 0xff)] return ':'.join(map(lambda x: "%02x" % x, mac)) def __add_other_params(self, douyin_url, params=None): if params is None: params = {} if not douyin_url.__contains__('?'): douyin_url = douyin_url + '?' common_params = parse.urlencode(self.__device_params) if douyin_url.endswith('?') or douyin_url.endswith('&'): douyin_url = douyin_url + common_params else: douyin_url = douyin_url + '&' + common_params if len(params) > 0: douyin_url = douyin_url + '&' + parse.urlencode(params) douyin_url = douyin_url + "&_rticket=" + str(int(round(time.time() * 1000))) + "&ts=" + str(int(time.time())) return douyin_url def __get_cookie(self): if len(self.__cookie) == 0: return '' return "; ".join([str(x) + "=" + str(y) for x, y in self.__cookie.items()]) def get_webcast_feed(self): params = { } douyin_url = 'https://webcast.amemv.com/webcast/feed/?' return self.__http_get(douyin_url, params) def __get_sign(self, url, form_params=None): stub = '' if form_params: a = parse.urlencode(form_params) stub = hashlib.md5(a.encode('utf-8')).hexdigest() ts = int(time.time()) from libs.Xg04 import X_Gorgon params = url[url.index('?') + 1:] result = X_Gorgon(params, "", cookie) gorgon = result['X-Gorgon'] # ppp = url[url.index('?') + 1:] # s = getXGon(ppp, stub, parse.urlencode(self.__cookie)) # gorgon = xGorgon(ts, strToByte(s)) # # data = {"url": url, "headers": {}} # gorgon = requests.post("http://121.36.217.134:18500/gorgon", data=str(data)).text sign = { 'X-Khronos': str(ts), 'X-Gorgon': gorgon, 'X-Pods': '' } # print(gorgon) if stub: sign.update({ 'X-SS-STUB': stub.upper() }) return sign def __get_headers(self, sign=None): if sign is None: sign = {} headers = { 'User-Agent': DouYinApi.USER_AGENT, 'X-SS-REQ-TICKET': str(round(time.time() * 1000)), # 'response-format':'protobuf' } headers.update(sign) return headers def __http_get(self, url, query_params=None): if query_params is None: query_params = {} url = self.__add_other_params(url, query_params) # print(url) sign = self.__get_sign(url) if self.proxies: resp = requests.get(url, headers=self.__get_headers(sign), cookies=self.__cookie, proxies=self.proxies, verify=False, timeout=20) else: resp = requests.get(url, headers=self.__get_headers(sign), cookies=self.__cookie, verify=False, timeout=20) cookie = resp.cookies.get_dict() if len(cookie) != 0: self.__cookie.update(cookie) # print(resp.text[:]) return resp.json() def __http_post(self, url, form_params=None): if form_params is None: form_params = {} url = self.__add_other_params(url) sign = self.__get_sign(url, form_params) if self.proxies: resp = requests.post(url, headers=self.__get_headers(sign), data=form_params, cookies=self.__cookie, proxies=self.proxies, verify=False, timeout=20) else: resp = requests.post(url, headers=self.__get_headers(sign), data=form_params, cookies=self.__cookie, verify=False, timeout=20) cookie = resp.cookies.get_dict() if len(cookie) != 0: self.__cookie.update(cookie) # print(resp.text) return resp.text def comm_get(self, douyin_url, params={}, need_comm_param=True): """获取用户信息 :param user_id: 用户ID :return: """ if need_comm_param: return self.__http_get(douyin_url, params) else: params = parse.urlencode(params) url = douyin_url + params sign = self.__get_sign(douyin_url) if self.proxies: resp = requests.get(url, headers=self.__get_headers(sign), cookies=self.__cookie, proxies=self.proxies, verify=False, timeout=20) else: resp = requests.get(url, headers=self.__get_headers(sign), cookies=self.__cookie, verify=False, timeout=20) cookie = resp.cookies.get_dict() if len(cookie) != 0: self.__cookie.update(cookie) # print(resp.text) return resp.text def get_room_product_list(self, room_id, anchor_id, sec_author_id): """ 获取直播间商品信息 :return: """ params = { 'room_id': str(room_id), 'anchor_id': str(anchor_id), 'sec_author_id': sec_author_id, } douyin_url = 'https://lianmengapi.snssdk.com/live/promotions/?' return self.__http_get(douyin_url, params) def get_product_info(self, promotion_id, product_id, author_id): """ 获取商品信息 :param promotion_id:商品id :return: """ params = { "promotion_id": promotion_id, "product_id": product_id, "author_id": author_id, } douyin_url = 'https://api3-normal-c-lf.amemv.com/aweme/v2/shop/promotion/' return self.__http_post(douyin_url, params) def get_room_product_campaign(self, promotion_id): """ 获取直播间商品信息 :return: """ params = { 'promotion_ids': str(promotion_id), } douyin_url = 'https://lianmengapi.snssdk.com/live/promotion/campaign/?' return self.__http_get(douyin_url, params) def getDevice(self): result = self.register_device() # print(result) device_id, iid, udid, openudid, cookie = result['device_id'], result['iid'], result['uuid'], result['openudid'], result['cookie'] if device_id == '0': return '2532904814256759', '3166223513429303' self.init_device_ids(device_id, iid, udid, openudid) params = { "has_market": "0", "is_activated": '0' } self.comm_get('https://aweme.snssdk.com/service/2/app_alert/?', params) return device_id, iid if __name__ == '__main__': import warnings warnings.filterwarnings("ignore") for i in range(30): proxy = Proxy.get_yuanrenyun_list() print(proxy) proxies = { "http": "http://" + proxy, "https": "http://" + proxy } douApi = DouYinApi('', proxies) for i in range(10): try: result = douApi.register_device() print(result) device_id, iid, udid, openudid, cookie, cdid = result['device_id'], result['iid'], result['uuid'], \ result['openudid'], result['cookie'], result['clientudid'] if device_id == '0': continue douApi.init_device_ids(device_id, iid, udid, openudid) params = { "has_market": "0", "is_activated": '0' } douApi.comm_get('https://aweme.snssdk.com/service/2/app_alert/?', params) server_time = str(int(time.time())) data = device_id + '@@@' + iid + '@@@' + openudid + '@@@' + udid + '@@@' + cookie + '@@@' + server_time + '@@@' + cdid + '@@@' + proxy + '@@@' + V1 + '@@@' + V2 + '@@@' + device_type + '@@@' + device_brand print(data) key = 'DOUYIN_SCRAPE_DID_IID_TTREQ_0414' redis = DbRedis.kwai_connect().lpush(key, data) key = 'DOUYIN_SCRAPE_DID_IID_TTREQ_1221' if i > 10: key = 'DOUYIN_SCRAPE_DID_IID_TTREQ_1221C' redis = DbRedis.kwai_connect().lpush(key, data) except Exception as e: print(e) continue