无限获取虚拟和 cookie 并保存到 SQLite

.gitignore

@@ -1,3 +1,4 @@
 output
 __pycache__
-env
+env
+.vscode/

conf/config.py

@@ -1,6 +1,7 @@
 import sys
 import os
 from DrissionPage import ChromiumOptions
+import dataset
 from loguru import logger
 HOST='localhost'
 PORT=9226
@@ -26,7 +27,8 @@ logger.add(os.path.join(OUTPUT, "all.log"), level="DEBUG", format='<green>{time:
 logger.debug(f"WORK_DIR {WORK_DIR}")
 logger.debug(f"INI_PATH {INI_PATH}")
 
-  
+db = dataset.connect(f'sqlite:///{OUTPUT}/douyin.db')
+
 def find_edge_path_in_registry():  
     import winreg as reg  
     path = None  
@@ -49,7 +51,6 @@ if 'win' in sys.platform:
         # 生成默认配置文件
         chrome_options = ChromiumOptions(False, None)
         chrome_options.set_address(f"{HOST}:{PORT}")
-        USER_DATA += str(PORT)
         chrome_options.set_browser_path(path)
         chrome_options.set_user_data_path(USER_DATA)
         chrome_options.save(INI_PATH)

conf/dp_configs.ini

@@ -16,7 +16,11 @@ system_user_path = False
 existing_only = False
 
 [session_options]
-headers = {'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/603.3.8 (KHTML, like Gecko) Version/10.1.2 Safari/603.3.8', 'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'connection': 'keep-alive', 'accept-charset': 'GB2312,utf-8;q=0.7,*;q=0.7'}
+headers = {
+    'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/603.3.8 (KHTML, like Gecko) Version/10.1.2 Safari/603.3.8', 'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'connection': 'keep-alive', 'accept-charset': 'GB2312,utf-8;q=0.7,*;q=0.7',
+    'Sec-Ch-Ua-Platform:':'macOS',
+    "Sec-Ch-Ua": ""
+    }
 
 [timeouts]
 base = 10

dp/cookies.py

@@ -0,0 +1,155 @@
+import asyncio
+import os
+import random
+import re
+import time
+import os
+import sys
+sys.path.append(os.path.dirname(os.path.dirname(__file__)))
+
+from DrissionPage import ChromiumPage
+from DrissionPage import ChromiumOptions
+from conf.config import logger,PAGE_OUTPUT,INI_PATH,chrome_options,db,USER_DATA,OUTPUT
+from DrissionPage import ChromiumOptions
+from DrissionPage.common import Settings
+from faker import Faker
+import DrissionPage
+from fake_useragent import UserAgent
+from dataset import Table
+table:Table = db['cookie']
+
+def get_browser_fake_info():
+    # https://github.com/joke2k/faker/wiki
+    fake = Faker()
+    while True:
+        user_agent = fake.user_agent()
+        mac_platform_token = fake.mac_platform_token()
+        if "Windows" in user_agent:
+            break
+    return user_agent
+page = None
+# https://dataset.readthedocs.io/en/latest/api.html
+
+def create_ua_header():
+    ua = UserAgent(os=["windows", "macos"])
+    res = ua.getRandom
+    logger.info(f"{ua}")
+    version = res['version']
+    vnum =  int(version)
+    header = {
+        "Sec-Ch-Ua": f'";Not A Brand";v="99", "Chromium";v="{vnum}"',
+        "Sec-Ch-Ua-Platform": "Windows" if 'win' in res['os'] else "macOS",
+        "User-Agent": res['useragent'],
+
+    }
+    # logger.info(f"{res}")
+    # logger.info(f"{header}")
+    return header
+
+
+def gen_douyin_cookies():
+    # header = {'Sec-Ch-Ua': '";Not A Brand";v="99", "Chromium";v="122"', 'Sec-Ch-Ua-Platform': 'Windows', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36 Config/92.2.2788.20'}
+    header = create_ua_header()
+    logger.info(f"header {header}")
+    chrome_options.set_user_data_path(USER_DATA + '1')
+    page = ChromiumPage(chrome_options)
+    try:
+        page.set.cookies.clear()
+    except Exception as e:
+        logger.exception(f"{e}")
+        page.quit()
+        return
+    page.set.load_mode.normal()
+    page.set.headers(headers=header)
+    logger.info(f"address {chrome_options.address}")
+    logger.info(f"user_data_path {chrome_options.user_data_path}")
+    logger.info(f"start '{page._chromium_options._browser_path}'")
+    logger.info(f"process_id {page.process_id}")
+    # page.get("edge://version/")
+    url="https://www.douyin.com/"
+    # tab = page.new_tab()
+    tab = page
+    retry = 3
+    while retry:
+        try:
+            logger.info(f"start listen")
+            tab.listen.start("www.douyin.com", method="GET")
+            tab.get(url)
+            i = 0
+            for packet in tab.listen.steps(timeout=5):
+                if not packet.url.startswith('https'):
+                    continue
+                # logger.info(f"{packet.url} {packet.request.headers}")
+                cookie = packet.request.extra_info.headers["cookie"]
+                logger.debug(f"cookie {cookie}")
+                # page.quit()
+                if len(cookie) > 1600:
+                    ret = {
+                        "cookies": cookie,
+                        "user_agent": header['User-Agent'],
+                        "update_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())),
+                        "create_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())),
+                    }
+                    logger.debug(f"ret {ret}")
+                    page.quit()
+                    return ret
+                i += 1
+                if i == 5:
+                    break
+            # res = tab.listen.wait(1, timeout=5, raise_err=True)
+            # logger.debug(f"res {res.request.headers}")
+            # logger.debug(f"res.request.extra_info.headers {res.request.extra_info.headers}")
+        except DrissionPage.errors.WaitTimeoutError as e:
+            logger.error(f"{e}")
+        except Exception as e:
+            logger.exception(e)
+        tab.listen.clear()
+        retry -= 1
+        logger.info(f"retry {retry}")
+    page.quit()
+
+def save_to_db(data):
+    table = db['cookie']
+    table.insert(data)
+
+def try_gen_douyin_cookies():
+    while True:
+        try:
+            ret = gen_douyin_cookies()
+            # logger.info(f"ret {ret}")
+            save_to_db(ret)
+            break
+        # Ctrl+c cancel_error
+        except KeyboardInterrupt:
+            logger.info("KeyboardInterrupt")
+            page.quit()
+            sys.exit(0)
+        except DrissionPage.errors.WaitTimeoutError as e:
+            logger.error(f"{e}")
+        except Exception as e:
+            logger.exception(e)
+
+def get_from_table():
+    count = table.count()
+    # random choise
+    if count > 0:
+        index = random.randint(0, count-1)
+        row = table.find_one(id=index)
+        # logger.info(f"{row['id']}") 
+        # logger.info(f"row {row}")
+        return row
+    else:
+        logger.info("no cookie")
+        return None
+
+def main():
+    
+    # ua = create_ua_header()
+    # gen_douyin_cookies()
+    # try_gen_douyin_cookies()
+    get_from_table()
+    # table.create_column('key')
+    # table.delete(id=6)
+
+if __name__ == "__main__":
+    main()

dp/index.html

@@ -0,0 +1,72 @@
+<!DOCTYPE html>  
+<html lang="zh-CN">  
+<head>  
+    <meta charset="UTF-8">  
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">  
+    <title>直播间信息</title>  
+    <script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>  
+    <style>  
+        body, html {  
+            height: 100%;  
+            margin: 0;  
+            display: flex;  
+            align-items: center;  
+            justify-content: center;  
+            flex-direction: column;  
+            font-family: Arial, sans-serif;  
+        }  
+        .container {  
+            text-align: center;  
+        }  
+        .input-group {  
+            margin-bottom: 20px;  
+            /* height: 30px;   */
+
+        } 
+        #liveUrlInput {
+        width: 100%; /* 使输入框宽度为父元素的100% */
+        height: 100%;  
+        padding: 12px 20px; /* 增加上下填充使其更高，增加左右填充使其更宽 */
+        } 
+        .result {  
+            margin-top: 10px;  
+        }  
+        button {  
+            padding: 10px 20px;  
+        }  
+    </style>  
+</head>  
+<body>  
+    <div class="container">  
+        <div class="input-group">  
+            <input type="text" id="liveUrlInput" placeholder="请输入直播间地址">  
+        </div>  
+        <button onclick="getAudienceNumber()">获取人数</button>  
+        <div class="result" id="result"></div>  
+    </div>  
+    <script>    
+        function getAudienceNumber() {    
+            var liveUrl = document.getElementById('liveUrlInput').value;    
+            if (!liveUrl) {    
+                alert('请输入直播间地址！');    
+                return;    
+            }    
+            $.ajax({    
+                url: '/live_url', // 使用相对路径，不再指定域名和端口  
+                type: 'POST',    
+                data: { live_url: liveUrl },    
+                success: function(response) {    
+                    // 假设后端返回的是JSON格式的数据    
+                    // var data = JSON.parse(response);    
+                    // 这里简单展示返回的数据，实际应用中可能需要根据返回的数据结构进行相应处理    
+                    // document.getElementById('result').innerText = JSON.stringify(data);    
+                    document.getElementById('result').innerText = response
+                },    
+                error: function(xhr, status, error) {    
+                    document.getElementById('result').innerText = '请求失败：' + error;    
+                }    
+            });    
+        }    
+    </script>
+</body>  
+</html>

dp/page.py

@@ -13,15 +13,25 @@ from DrissionPage import ChromiumOptions
 from DrissionPage.common import Settings
 
 Settings.raise_when_ele_not_found=True
+chrome_options.set_user_agent("Mozilla/5.0 (Windows NT 5.2) AppleWebKit/536.1 (KHTML, like Gecko) Chrome/52.0.893.0 Safari/536.1")
 
-        
 page = ChromiumPage(chrome_options)
 logger.debug(f"address {chrome_options.address}")
+logger.debug(f"user_data_path {chrome_options.user_data_path}")
 logger.debug(f"start '{page._chromium_options._browser_path}'")
+logger.debug(f"process_id {page.process_id}")
 # 设置 none 的时候 page.get() 不会等待加载完成，而是直接返回，page.ele 会阻塞，不过一旦找到元素也会立即返回
 # 因此设置为 none 是最高效率、最迅速的，甚至不用 page.stop_loading() 因为停止过程中也要花费时间，而是直接请求空页面 about:blank 断开所有连接
 page.set.load_mode.none()
 # page.set.NoneElement_value('没找到')
-# page.get("edge://version/")
+page.get("edge://version/")
 # page.new_tab("http://www.baidu.com")
 
+def gen_cookies():
+    ret = {
+        "cookies": "",
+        "user-agent":"",
+        "update_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())),
+        "create_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())),
+    }
+    page

gpt.md

@@ -0,0 +1,81 @@
+## 构思
+假设要做一个直播信息统计的工具，产生的数据如下：
+- 每个主播开启一个独立的直播间，可能有上万个主播，也就是数万个直播间。
+- 每个主播可能每天开几场直播，或者每个月开几十场直播
+- 每个直播间有几百或上百万用户观看
+- 每个观众可能会发布弹幕或礼物信息，一场直播可能有几千到几百万条弹幕或礼物数据
+- 每条弹幕或礼物数据包含发表弹幕的用户头像、信息、主页链接、昵称等等。
+- 所有的内容都是 json 格式的数据，少则 几k多则 64k 的数据。
+- 每个观众和主播都可以用用户 id 来表示
+
+可能需要统计的信息如下：
+- 每个主播、每场直播开播时长，频率，收到的礼物总数
+- 每个主播每天的直播时长，直播间数，观众数，平均每小时的观众数，平均每小时的弹幕数
+- 记录直播间总弹幕、礼物数
+- 分析每一条弹幕频率，或分析观众在直播间的建议和需求
+
+根据以上需求，如果设计一个合理的软件数据架构，在 PostgreSQL 、本地 SQLite 、对象存储 等方案中，你会怎么设计？
+但是如果只创建一个弹幕表，或者只有一个直播间表，那么数据量会非常庞大，如果每天有几千个主播开直播，那么一个弹幕表和礼物表显然会非常庞大，如果每天有上万场直播，那么一张表的数据量会达到几亿甚至几十亿。
+一个可能的构思如下：
+- 主播表 anchors ，记录每个主播的基本信息，包括头像、昵称、简介等
+- 直播间表 live_rooms ，记录每场直播的开播时间、直播间 id、主播 id 等
+- 直播场次表 live_sessions ，记录每场直播的开播时间、结束时间，直播间 id、主播 id 等
+- 主播直播间弹幕表 danmus ，根据 live_sessions id 进行表分区，例如 CREATE TABLE danmus_{live_sessions_id} PARTITION OF danmus 
+- 主播直播间礼物表  gifts ，根据 live_sessions id 进行表分区，例如 CREATE TABLE gifts_{live_sessions_id} PARTITION OF gifts 
+
+上述是否合理，给出最佳合理的表结构，以及如何设计数据存储方案。
+
+## 数据库表设计
+anchors (主播表)
+
+anchor_id (主键)
+nickname
+avatar
+description
+... (其他主播相关信息)
+live_rooms (直播间表)
+
+room_id (主键)
+anchor_id (外键，关联anchors表)
+room_name
+... (其他直播间相关信息)
+live_sessions (直播场次表)
+
+session_id (主键)
+room_id (外键，关联live_rooms表)
+start_time
+end_time
+viewer_count
+... (其他直播场次相关信息)
+danmus (弹幕表) —— 使用表分区
+
+danmu_id (主键)
+session_id (外键，关联live_sessions表)
+content
+user_id (发送弹幕的用户ID)
+timestamp
+... (其他弹幕相关信息)
+可以根据session_id进行表分区，以提高查询效率。
+gifts (礼物表) —— 使用表分区
+
+gift_id (主键)
+session_id (外键，关联live_sessions表)
+user_id (发送礼物的用户ID)
+gift_name
+timestamp
+... (其他礼物相关信息)
+同样可以根据session_id进行表分区。
+users (用户表) —— 记录所有观众和主播的信息
+
+user_id (主键)
+nickname
+avatar
+... (其他用户相关信息)
+
+总结
+使用PostgreSQL作为主要的数据库解决方案，利用其强大的功能和表分区技术来管理大规模数据。
+根据直播场次（session_id）对弹幕和礼物表进行分区，以提高查询效率和数据管理。
+避免在单个表中存储过多的数据，通过合理的表设计和分区策略来优化性能。
+对于非结构化数据（如用户头像等），可以考虑使用对象存储进行保存，并在数据库中保存相应的链接或标识符。
+
+

main.py

@@ -1,40 +1,93 @@
-from conf.config import logger
+import json
+import os
+import signal
+import sys
+import time
+from conf.config import logger,OUTPUT
 from dp.page import page
+from DrissionPage import ChromiumPage
+from bottle import route, run, template,static_file,post,request
+import threading
 
-# url = input("输入直播间地址（如 https://live.douyin.com/1234568）：")
-url = "https://live.douyin.com/389150144924"
-page.listen.start('live.douyin.com/webcast/room/web/enter')
-page.get(url)
-page._wait_loaded(5)
-logger.info(f"页面加载成功")
-res = page.listen.wait(1)
-logger.info(f"res.request {res.request}")
-# self._data_packet = data_packet
-# self._request = raw_request
-# self._raw_post_data = post_data
-# self._postData = None
-# self._headers = None
-logger.info(f"res.request._data_packet {res.request.data_packet}")
-logger.info(f"res.request._raw_request {res.request.raw_request}")
-logger.info(f"res.request._raw_post_data {res.request.raw_post_data}")
-logger.info(f"res.request._postData {res.request.post_data}")
-logger.info(f"res.request._headers {res.request.headers}")
-logger.info(f"res.request.cookies {res.request.cookies}")
-logger.info(f"res.request.extra_info {res.request.extra_info.all_info}")
-
-
-logger.info(f"res.response {res.response}")
-# self._data_packet = data_packet
-# self._response = raw_response
-# self._raw_body = raw_body
-# self._is_base64_body = base64_body
-# self._body = None
-# self._headers = None
-logger.info(f"res.response._data_packet {res.response.data_packet}")
-logger.info(f"res.response._raw_response {res.response.raw_response}")
-logger.info(f"res.response._raw_body {res.response.raw_body}")
-logger.info(f"res.response._is_base64_body {res.response.is_base64_body}")
-logger.info(f"res.response._body {res.response.body}")
-logger.info(f"res.response._headers {res.response.headers}")
-page.listen.stop()
-# page.quit()
+def get_url_enterdata(tab:ChromiumPage, url):
+    # url = input("输入直播间地址（如 https://live.douyin.com/1234568）：")
+    tab.listen.start('live.douyin.com/webcast/room/web/enter')
+    tab.get(url)
+    tab._wait_loaded(5)
+    logger.info(f"页面加载成功")
+    res = tab.listen.wait(count=1,timeout=3.5, raise_err=True)
+    data = res.response.body.get("data")
+    if not data:
+        os._exit(1)
+
+    with open(os.path.join(OUTPUT, "data.json"), 'w') as f:
+        json.dump(data, f)
+
+    tab.listen.stop()
+    return data
+
+def get_anchor(data:dict):
+    room_view_stats = data.get("room_view_stats")
+    if not room_view_stats:
+        logger.error(f"room_view_stats not fount")
+        return None, None
+    display_value = room_view_stats.get("display_value")
+    # "display_long_anchor"
+    anchor = room_view_stats.get("display_long_anchor")
+    return display_value, anchor
+
+
+@route('/')
+def index():
+    return static_file('dp/index.html', root='./')
+
+@post('/live_url')
+def input_url():
+    live_url = request.forms.get('live_url')
+    if live_url:
+        # 这里可以添加您的逻辑，比如日志记录、数据处理等
+        logger.info(f"input_url {live_url}")
+        tab = page.new_tab()
+        try:
+            data = get_url_enterdata(tab, live_url)
+            display_value, anchor = get_anchor(data.get("data")[0])
+            name = data.get("user").get("nickname")
+            ret = f"主播： {name}  |  {anchor} （{display_value}）"
+        except Exception as e:
+            logger.error(e)
+            ret = f"获取数据失败 {e}"
+        tab.close()
+        logger.info(ret)
+        return ret
+    else:
+        return "live_url 参数未提供"
+
+def load_index():
+    import DrissionPage
+    page.get("http://localhost:9230")
+    page._wait_loaded(5)
+    page.tab
+    while True:
+        try:
+            page.tabs_count
+        except DrissionPage.errors.PageDisconnectedError as e:
+            logger.info(f"{e}")
+            break
+        except Exception as e:
+            logger.info(f"{e}")
+            break
+        time.sleep(1)
+    logger.info(f"退出程序")
+    os._exit(0)
+    
+
+def main():
+    page_index = threading.Thread(target=load_index)
+    page_index.start()
+    run(host='localhost', port=9230, debug=logger.debug)
+
+    logger.info(f"server is stop")
+
+if __name__ == "__main__":
+    main()
+    page.quit()