视频脚本多线程

douyin_live_lottery_scraper.py

@@ -32,7 +32,6 @@ def get_random(i, random_type=1):
         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

douyin_video_scraper_web.py

@@ -1,159 +1,305 @@
 # -- coding: utf-8 --**
-
-import json
-import time
-import sys
+import json,time,sys,threading,warnings
 
 from rds_model.db_redis import DbRedis
 from log.print_log import PrintLog
 from libs.proxy import Proxy
 from web_dy import *
 from rds_model.rds_user_video_list import RdsUserVideoList
+from libs.web_cookie import Cookie
 
-if __name__ == '__main__':
-    import warnings
-    start_time = time.time()
+start_time = time.time()
+
+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 ;
+
+def get_signature(url=None,method='_signature'):
+    with open('/mnt/shop_live_scraper/signature.js', 'r', encoding='utf-8') as f:
+        b = f.read()
     
-    warnings.filterwarnings("ignore")
+    c = execjs.compile(b)
+    
+    # url=url.replace('%28','(').replace('%29',')').replace('%2C',',')
+    d = c.call(method, url.replace('\n',''))
+    
+    return d
+
+def get_user_videos(sec_user_id, max_cursor=0, count=20):
+
+    # ua,ck=get_ua_ck('get_user_videos')
+    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': '160100',
+        'version_name': '16.1.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',
+    }
+
+    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",
+    }
+
+    if ck:
+        headers['cookie']=ck
+
+    retry = 0
+    cookie_retry = 100
+    response_json = None
+
     while True:
+        if retry > 10:
+            Cookie.del_cookie(ck)
+            break
 
-        server_time = int(time.time())
-        current_time = time.time()
+        retry += 1
+        cookie_retry += 1
 
-        if current_time - start_time > 300:
-            print(
-                time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 
-                + ' 主方法执行终止'
-            )
-            sys.exit(0)
-		
         proxy = Proxy.get()
-        print(proxy)
+
         proxies = {
             "http": "http://" + proxy,
             "https": "http://" + proxy
         }
-        
-        errorn = 0
-        suc_f = 0
-        webdy=WebDouYin()
-        
-        rds = RdsUserVideoList()
 
-        ck = rds.get_cookie()
-        if ck is None:
-            ck = webdy.get_ck(proxy=proxies)
-            if ck:
-                rds.set_cookie(ck)
-            else:
-                print('获取cookie失败')
-                break
-
-        for index in range(1000):
-            try:
-                # ukey = 'BrandLiveData.DouyinUserVideo'
-                # users = DbRedis.douyin_connect().rpop(ukey)
-                users = rds.get_request_param()
-                
-                if users is None:
-                    time.sleep(1)
-                    continue
-                user = json.loads(users)
-                user_id = user.get('uid')
-                sec_user_id = user.get('sec_uid')
-                re_times = int(user.get('re_times')) + 1
-
-                user.update({
-                    're_times':re_times,
-                    'sec_uid' : sec_user_id
-                })
-                
-                users = json.dumps(user)
-                videos = webdy.get_user_videos(sec_user_id=sec_user_id,max_cursor=0,count=20,proxy=proxies,cookie=ck)
-                # skey = 'BrandLiveData.DouyinUserVideoScore'
-                # data_score = DbRedis.douyin_connect().rpop(skey)
-                
-                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)
-                # DbRedis.douyin_connect().lpush(skey, data_score)
-
-                if isinstance(videos, dict):
-                    awemes = videos.get('aweme_list')
+        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 != ''):
+                response_json = response.json()
+
+                if (response_json.get('aweme_list') is not None):
+                    break
                 else:
-                    print(videos)
-                    awemes = None
+                    print(
+                        time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 
+                        + ' 数据获取失败！' 
+                        + '\n'
+                        + str(sec_user_id) 
+                        + '\n'
+                        + response.text
+                        + Proxy.proxy_info
+                    )
                 
-                stime = time.strftime("%H:%M:%S", time.localtime())
-                if awemes:
-                    # PrintLog.print(stime+" 成功"+str(index)+" "+str(sec_user_id))
-                    print(stime+" 成功"+str(index))
-                    # dkey = 'BrandLiveData.DouyinUserVideoResponsePython'
-                    data = str(user_id) + '@@@' + json.dumps(videos)
-                    # DbRedis.douyin_connect().lpush(dkey, data)
-                    rds.push_data_list(data)
-
-                    # skey = 'BrandLiveData.DouyinUserVideoScore'
-                    # data_score = DbRedis.douyin_connect().rpop(skey)
-                    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)
-                    # DbRedis.douyin_connect().lpush(skey, data_score)
-                    rds.record_score(data_score)
-
-                else:  
-                    # skey = 'BrandLiveData.DouyinUserVideoScore'
-                    # data_score = DbRedis.douyin_connect().rpop(skey)
-                    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)
-                    # DbRedis.douyin_connect().lpush(skey, data_score) 
-                    rds.record_score(data_score)
-                    # PrintLog.print(stime+" 失败"+str(index)+' '+ str(sec_user_id))
-                    print(stime+" 失败"+str(index))
-                    errorn = errorn + 1
-                    if re_times<1:
-                        # ukey = 'BrandLiveData.DouyinUserVideo'
-                        # DbRedis.douyin_connect().lpush(ukey, users)
-                        rds.push_request_id(users)
-                    if errorn>2:
-                        break
-            except Exception as e:
-                # skey = 'BrandLiveData.DouyinUserVideoScore'
-                # data_score = DbRedis.douyin_connect().rpop(skey)
-                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)
-                # DbRedis.douyin_connect().lpush(skey, data_score) 
-                rds.record_score(data_score)
+            else:
                 print(
-                    time.strftime("%H:%M:%S", time.localtime())
-                    + ' 请求抛出异常！行号：' 
-                    + str(e.__traceback__.tb_lineno)
-                    + ' 错误:'
-                    + str(e) 
-                    + "\n"
+                    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'
                 )
-                break
+                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(sec_user_id) 
+                + '\n'
+                + Proxy.proxy_info
+            )
+            pass
+
+    return response_json
+
+def scrape(sec_user_id):
+        
+    rds = RdsUserVideoList()
+    sec_user_id = str(sec_user_id)
+
+    print(
+        time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 
+        + ' ' 
+        + str(sec_user_id)
+    )
+
+    time.sleep(0.1)
+
+    try:
+        videos = get_user_videos(sec_user_id=sec_user_id,max_cursor=0,count=20)
+       
+        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()
+            
+            user_info = rds.get_request_param()
+            if user_info is None:
+                time.sleep(0.1)
+                continue
+
+            user_info = json.loads(user_info)
+            sec_user_id = user_info.get('sec_uid')
+           
+            task = threading.Thread(target=scrape, args=(sec_user_id))
+            task.start()  # 准备就绪, 等待cpu执行
+            increment = increment - 1
+           
+        current_time = time.time()
+
+        if current_time - start_time > 300:
+            print(
+                time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 
+                + ' 主方法执行终止'
+            )
+            sys.exit(0)
+
+        time.sleep(0.01)
 

libs/web_cookie.py

@@ -0,0 +1,226 @@
+import requests,json,random,uuid,sys,time,threading
+from urllib import parse
+from urllib.parse import quote
+from libs.proxy import Proxy
+from rds_model.rds_user_video_list import RdsUserVideoList
+from rds_model.db_redis import DbRedis
+# -- coding: utf-8 --**
+
+start_time = time.time()
+
+class Cookie:
+    redis = DbRedis.douyin_connect()
+
+    cookie_info = ''
+
+    @staticmethod
+    def get():
+        while True:
+            cookie = Cookie.random_cookie()
+
+            if cookie is not None:
+                return cookie
+
+            print('未拿到有效的代理')
+            time.sleep(0.1)
+
+    @staticmethod
+    def random_cookie():
+        key = 'CookieHash'
+        cookie_dict = Cookie.redis.hgetall(key)
+
+        if (cookie_dict is None) or (len(cookie_dict) == 0):
+            return
+
+        cookie_list = list(cookie_dict)
+
+        now = int(time.time())
+        while True:
+            cookie = random.choice(cookie_list)#Cookie.choice_cookie()
+
+            if cookie is None:
+                return
+
+            cookie_info = cookie_dict.get(cookie)
+
+            if cookie_info is None:
+                continue
+
+            Cookie.cookie_info = cookie_info
+
+            cookie_info = json.loads(cookie_info)
+
+            expire_at = int(cookie_info.get('expired_at'))
+
+            # 删除过期的代理
+            if expire_at <= now:
+                Cookie.redis.hdel(key, cookie)
+                cookie_list.remove(cookie)
+                continue
+
+            return cookie
+
+    @staticmethod
+    def del_cookie(cookie):
+        cookie_info = Cookie.redis.hget('CookieHash', cookie)
+        if cookie_info is None:
+            return
+
+        cookie_info = json.loads(cookie_info)
+        min_expired_at = cookie_info.get('min_expired_at')
+
+        if min_expired_at is None:
+            min_expired_at = 0
+
+        min_expired_at = int(min_expired_at)
+        now = int(time.time())
+
+        if min_expired_at >= now:
+            return
+
+        # 删除失效的Cookie
+        print('删除失效Cookie：' + cookie)
+
+    def set(cookie):
+        Cookie.redis.hset('CookieHash', cookie)
+
+    def get_ck(self):
+        requests.packages.urllib3.disable_warnings()
+        headers = {
+            'authority': 'www.douyin.com',
+            'method': 'GET',
+            'path': '/',
+            'scheme': 'https',
+            'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
+            'accept-encoding': 'gzip, deflate, br',
+            'accept-language': 'zh-CN,zh;q=0.9',
+            'upgrade-insecure-requests': '1',
+            'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36',
+        }
+
+        retry = 0
+        ck = None
+        # 屏蔽验证https证书提示
+        requests.packages.urllib3.disable_warnings()
+
+        while True:
+            if retry > 10:
+                break
+
+            retry += 1
+
+            proxy = Proxy.get()
+
+            proxies = {
+                "http": "http://" + proxy,
+                "https": "http://" + proxy
+            }
+
+            rds = RdsUserVideoList()
+
+            try:
+                response = requests.get(
+                    'https://www.douyin.com/',
+                    headers=headers,
+                    proxies=proxies,
+                    timeout=8,
+                    verify=False
+                )
+
+                cookie_dict=response.cookies.get_dict()
+                ck=''
+
+                if cookie_dict:
+                    for k,v in cookie_dict.items():
+                        ck+='%s=%s; '%(k,v)
+                    ck=ck[:-2]
+                    ck+='; passport_csrf_token='+str(uuid.uuid4()).replace('-','')
+
+                    Cookie.set(ck)
+                    
+                    print(
+                        time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 
+                        + ' 数据获取成功！' 
+                        + '\n'
+                        + ck
+                        + '\n'
+                        + Proxy.proxy_info
+                    )
+                else:
+                    print(
+                        time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 
+                        + ' 爬取http连接失败！' 
+                        + str(response.status_code) 
+                        + '\n'
+                        + Proxy.proxy_info 
+                    )
+                    time.sleep(1)
+            except requests.exceptions.ProxyError as e:
+                print(
+                    time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 
+                    + ' 代理过期！' 
+                    + str(e) 
+                    + 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'
+                    + 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'
+                    + Proxy.proxy_info
+                )
+                pass
+
+        return ck
+
+if __name__ == "__main__":
+    print("主方法开始执行")
+    # 并行线程数
+    threading_count = int(sys.argv[1])
+
+    print(
+        time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 
+        + ' ' 
+        + ' 开始执行'
+    )
+
+    while True:
+        sys.stdout.flush()
+        
+        # 减去主线程
+        active_count = threading.active_count() - 1
+       
+        increment = threading_count - active_count
+
+        cookie = Cookie()
+        
+        while increment > 0:
+            sys.stdout.flush()
+           
+            task = threading.Thread(target=cookie.get_ck,)
+            task.start()  # 准备就绪, 等待cpu执行
+            increment = increment - 1
+           
+        current_time = time.time()
+
+        if current_time - start_time > 300:
+            print(
+                time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 
+                + ' 主方法执行终止'
+            )
+            sys.exit(0)
+
+        time.sleep(0.01)

rds_model/rds_user_video_list.py

@@ -37,9 +37,11 @@ class RdsUserVideoList:
     
     def get_cookie(self):
         key = 'BrandLiveData.DouyinUserVideoWebCookie'
-        return self.redis.get(key)
+        return self.redis.rpop(key)
 
     def set_cookie(self, data):
         key = 'BrandLiveData.DouyinUserVideoWebCookie'
-        self.redis.set(key, data)
-        self.redis.expire(key, 7200)
+        self.redis.lpush(key, data)
+
+    def del_cookie(self):
+        

web_dy.py

@@ -1,6 +1,7 @@
 import requests,json,random,execjs,uuid
 from urllib import parse
 from urllib.parse import quote
+from libs.proxy import Proxy
 # -- coding: utf-8 --**
 
 class WebDouYin:
@@ -167,4 +168,12 @@ if __name__ == '__main__':
     # res = webdy.get_user_videos('MS4wLjABAAAAqLPgx-hHf27EqGEtRQ6YyuQQTmikB5CBO1jXy61yhWKujGd8KO5G8V2vdcLQJAym')
     # print(res)
     # webdy.get_user_info('MS4wLjABAAAAC2euvL-0qMZyd80aNwZa-wX5KXuz_r7YVNHSBOogfVg')
-    webdy.get_ck()
+
+    proxy = Proxy.get()
+    print(proxy)
+    proxies = {
+        "http": "http://" + proxy,
+        "https": "http://" + proxy
+    }
+
+    ck = webdy.get_ck()