import json import sys import threading import time import warnings from libs.proxy import Proxy from log.print_log import PrintLog from rds_model.db_redis import DbRedis from rds_model.rds_user_video_list import RdsUserVideoList from web_cookie import Cookie from web_dy import * start_time = time.time() # -- coding: utf-8 --** def set_score(flag): rds = RdsUserVideoList() if flag == 'success': data_score = rds.get_score() if data_score is None: data_score = '1@@@1@@@0' else: data_score = data_score.split('@@@') total, success, fail = int(data_score[0]), int(data_score[1]), int(data_score[2]) success = success + 1 data_score = str(total) + '@@@' + str(success) + '@@@' + str(fail) rds.record_score(data_score) elif flag == 'fail': data_score = rds.get_score() if data_score is None: data_score = '1@@@0@@@1' else: data_score = data_score.split('@@@') total, success, fail = int(data_score[0]), int(data_score[1]), int(data_score[2]) fail = fail + 1 data_score = str(total) + '@@@' + str(success) + '@@@' + str(fail) rds.record_score(data_score) elif flag == 'all': data_score = rds.get_score() if data_score is None: data_score = '1@@@0@@@0' else: data_score = data_score.split('@@@') total, success, fail = int(data_score[0]), int(data_score[1]), int(data_score[2]) total = total + 1 data_score = str(total) + '@@@' + str(success) + '@@@' + str(fail) rds.record_score(data_score) return None def get_signature(url=None,method='_signature'): with open('/mnt/shop_live_scraper/signature.js', 'r', encoding='utf-8') as f: b = f.read() c = execjs.compile(b) # url=url.replace('%28','(').replace('%29',')').replace('%2C',',') d = c.call(method, url.replace('\n','')) return d def get_ua_ck(): ua_list=[ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.25 Safari/537.36 Core/1.70.3870.400 QQBrowser/10.8.4405.400" ] ck_list=[ 'ttwid=1%7CTVzdM0P0u-8dtsmh6c-EaQEtBoTSOs_MG85FAg07AbA%7C1631502013%7C66442d8594de8e93ad18b73f3dfe0c94ed864c3d932824bcde9918b5be172321; passport_csrf_token=866923f1a32045fd82e47053158402a2', 'ttwid=1%7CGPDDu9-w3RGs2Pcd0wRlvLYoktpDt-v8LP5ZMyb1NBM%7C1630319594%7Cffb8de47e6da87dcfd76349b5ad34aa1f9b9d4332261a3a8436b932a893366c1; passport_csrf_token=79284b8777a7a54f3066cefef9af539e', 'ttwid=1%7CGsfqc7NpdOg4N-U-VX7Q77KsWjVTZ7gxLNifsisj8YE%7C1631618570%7Cafbb13a27fd1c2d7a8245454b1e0d7cd654d80848a320933a25d9ef77638c18c; passport_csrf_token=84911c9af94040a99cc10416bd27533d', 'ttwid=1%7C82FGr05YUOReYUB301ao_erqOQ3ilbXZdEy0tkMsdXY%7C1631863641%7C1dcebe643a96f00841a3b490db60de886bfe07ff3d276e509717abc4e1681ba6; passport_csrf_token=494ae3fffe00328101fd40e050ce49db', 'ttwid=1%7CwfnX3T9LY4_60iGoQNzyqYe5ahILFeRxfMuZ1pdgXf8%7C1632724192%7Cb613fddc0b533d5578dad4d5f9290705fdc6432aa854d492f4761d164dd3fdd5; passport_csrf_token=4a8afba333103b033e537003b72ee91b' ] return random.choice(ua_list),random.choice(ck_list) def get_user_videos(sec_user_id, max_cursor=0, count=20): ua,ck=get_ua_ck() # ua="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" url='https://www.douyin.com/aweme/v1/web/aweme/post/?' param={ 'device_platform': 'webapp', 'aid': '6383', 'channel': 'channel_pc_web', 'sec_user_id': sec_user_id, 'max_cursor': str(max_cursor), 'count': str(count), 'publish_video_strategy_type': '2', 'version_code': '170400', 'version_name': '17.4.0', 'cookie_enabled': 'true', 'screen_width': '1920', 'screen_height': '1080', 'browser_language': 'zh-CN', 'browser_platform': 'Win32', 'browser_name': 'Mozilla', 'browser_version':ua.replace('Mozilla/',''), 'browser_online': 'true', "source" : "channel_pc_web" } url = url + parse.urlencode(param) _signature = get_signature(url) url+='&_signature='+quote(_signature) ck = Cookie.get() if ck is None: print('获取cookie失败') return None headers = { "authority": "www.douyin.com", "method": "GET", "path": str(url).replace('https://www.douyin.com',''), "scheme": "https", "accept": "application/json, text/plain, */*", # "accept-encoding": "gzip, deflate, br", "accept-language": "zh-CN,zh;q=0.9", "cookie": ck, "referer": "https://www.douyin.com/user/{sec_user_id}?enter_method=search_result&enter_from=search_result".format(sec_user_id=sec_user_id), "user-agent":ua, "withcredentials": "true", "sec-ch-ua" : '"Google Chrome";v="93", " Not;A Brand";v="99", "Chromium";v="93"', "sec-ch-ua-mobile" : "?0", "sec-ch-ua-platform" : "Windows", "sec-fetch-dest" : "empty", "sec-fetch-mode" : "cors", "sec-fetch-site" : "same-origin" } if ck: headers['cookie']=ck retry = 0 response_json = None while True: if retry > 20: Cookie.del_cookie(ck) break retry += 1 proxy = Proxy.get() proxies = { "http": "http://" + proxy, "https": "http://" + proxy } try: response = requests.get( url, headers=headers, proxies=proxies, timeout=8 ) if (response.status_code == 200) and (response.text is not None) and (response.text != ''): # print(response) response_json = response.json() if (response_json.get('aweme_list') is not None): print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' 数据获取成功！' + '\n' + str(sec_user_id) ) break else: print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' 数据获取失败！' + '\n' + str(sec_user_id) + '\n' + response.text + Proxy.proxy_info ) else: print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' 爬取http连接失败！' + str(response.status_code) + '\n' + Proxy.proxy_info + '\n' + str(sec_user_id) + '\n' + '爬取结果：' + str(response) + '\n' ) time.sleep(1) except requests.exceptions.ProxyError as e: print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' 代理过期！' + str(e) + '\n' + str(sec_user_id) + '\n' + Proxy.proxy_info ) Proxy.del_proxy(proxy) pass except requests.exceptions.ConnectTimeout as e: print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' ConnectTimeout！' + str(e) + '\n' + str(sec_user_id) + '\n' + Proxy.proxy_info ) Proxy.del_proxy(proxy) pass except Exception as e: print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' 请求抛出异常！' + str(e) + '\n' + str(e.__traceback__.tb_lineno) + '\n' + str(sec_user_id) + '\n' + Proxy.proxy_info ) pass return response_json def scrape(): rds = RdsUserVideoList() user_info = rds.get_request_param() if user_info is None: return None sec_user_id = str(user_info) print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' ' + str(sec_user_id) ) try: videos = get_user_videos(sec_user_id=sec_user_id,max_cursor=0,count=50) if videos is None: # rds.push_request_id(sec_user_id) print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' 数据获取失败！响应数据为空！' + '\n' + str(sec_user_id) + '\n' ) sys.exit(0) if isinstance(videos, dict): awemes = videos.get('aweme_list') else: # print(videos) awemes = None if awemes: set_score('success') data = str(sec_user_id) + '@@@' + json.dumps(videos) rds.push_data_list(data) else: set_score('fail') except Exception as e: set_score('fail') # rds.push_request_id(sec_user_id) print( time.strftime("%H:%M:%S", time.localtime()) + ' ' + str(sec_user_id) + '数据异常：' + str(e) ) sys.exit(0) if __name__ == '__main__': print("主方法开始执行") # 并行线程数 threading_count = int(sys.argv[1]) rds = RdsUserVideoList() warnings.filterwarnings("ignore") print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' ' + ' 开始执行，用户队列长度：' + str(rds.get_len()) ) while True: sys.stdout.flush() # 减去主线程 active_count = threading.active_count() - 1 increment = threading_count - active_count while increment > 0: sys.stdout.flush() # scrape() task = threading.Thread(target=scrape, args=()) task.start() # 准备就绪, 等待cpu执行 increment = increment - 1 current_time = time.time() if current_time - start_time > 3600: print( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' 主方法执行终止' ) sys.exit(0) time.sleep(1)