上传文件至 /

检测直播在线_roomId.py

@@ -0,0 +1,102 @@
+import os
+import re
+import time
+import pickle
+import requests
+from datetime import datetime
+
+
+session = requests.session()
+def load_cookies_from_file(filename):
+    # 以二进制写入模式打开文件
+
+    with open(filename, 'rb') as f:
+        return pickle.load(f)
+
+# if os.path.isfile(f'{wjm}.pkl'):
+#     print(f'{wjm}.pkl文件存在')
+
+user_name = ''
+with open(r'D:\job_demo\demo\测试\cookie_name.txt','r',encoding='utf-8') as f:
+        
+        user_name = f.read().split('|')[1].strip()
+Cookie = load_cookies_from_file(r'D:\job_demo\demo\测试\MS4wLjABAAAAc9m4vmspNsSiYP348XRC-OLVqSXRn8Rdtl618fbzYag.pkl')
+# print(Cookie)
+# print(type(Cookie))
+cookies_dict = {c.name: c.value for c in Cookie}
+cookie_str = "; ".join(f"{k}={v}" for k, v in cookies_dict.items())
+
+def detect_live():
+    # user_name = ''
+    # with open(r'D:\job_demo\demo\测试\cookie_name.txt','r',encoding='utf-8') as f:
+        
+    #     user_name = f.read().split('|')[1].strip()
+    # Cookie = load_cookies_from_file(r'D:\job_demo\demo\测试\MS4wLjABAAAAc9m4vmspNsSiYP348XRC-OLVqSXRn8Rdtl618fbzYag.pkl')
+    # print(Cookie)
+    # print(type(Cookie))
+    # cookies_dict = {c.name: c.value for c in Cookie}
+    # cookie_str = "; ".join(f"{k}={v}" for k, v in cookies_dict.items())
+    # print(cookie_str)
+    # print(type(cookie_str))
+    url = 'https://www.douyin.com/user/self?from_tab_name=live'
+
+    headers = {
+            'Connection': 'keep-alive',
+            'Pragma': 'no-cache',
+            'Cache-Control': 'no-cache',
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36',
+            '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.7',
+            'Accept-Encoding': 'gzip, deflate',
+            'Accept-Language': 'zh-CN,zh;q=0.9',
+        }
+    res = session.get(url, headers=headers,cookies=Cookie)
+
+
+    while True:
+        if not res.status_code == 200:
+            print(datetime.now(),'请求失败，重试一次')
+            # app_edit(f'{datetime.now()}----请求失败，重试一次')
+            
+        #循环查询是否有开播的信息
+        res = session.get(url,headers=headers,cookies=Cookie)
+        if 'const verify_data' in res.text:
+            print(datetime.now(), f'{user_name}--Cookie失效')
+        if '直播中</span>' in res.text:
+            with open('index.html','w',encoding='utf-8') as f:
+                f.write(res.text)
+            # zburl = re.findall('<a href="(https://live.douyin.com/.*?)"',res.text)
+            zburl = re.findall(r'<a href="(?:https://live.douyin.com/\d+)"',res.text)
+
+            if zburl:
+                print(datetime.now(),f'{user_name}--检查到已开播')
+                # app_edit(f'{datetime.now()}----检查到已开播')
+                return zburl[0]
+        time.sleep(2)
+        print(datetime.now(),f'{user_name}--未直播等待2秒后再次查询')
+
+
+def get_roomId(url):
+    print(datetime.now(),'正在请求直播房间')
+    headers = {
+        'Connection': 'keep-alive',
+        'Pragma': 'no-cache',
+        'Cache-Control': 'no-cache',
+        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36',
+        '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.7',
+        'Accept-Encoding': 'gzip, deflate',
+        'Accept-Language': 'zh-CN,zh;q=0.9',
+    }
+    for _ in range(3):
+        res = session.get(url, headers=headers,cookies=Cookie)
+        roomId = re.findall(r'\\"roomId\\":\\"(\d+?)\\"', res.text)
+        if roomId:
+            return roomId
+        time.sleep(2)
+
+
+
+if __name__ == '__main__':
+    # zburl = detect_live()
+    zburl = 'https://live.douyin.com/80017709309'
+    roomId = get_roomId(zburl)
+    print(roomId)