抖音酒类直播间监控系统 - 实时数据采集与可视化

1
douyin-live-monitor/
2
├── backend/
3
│   └── app/
4
│       ├── main.py           # FastAPI 应用入口
5
│       ├── config.py         # 配置管理
6
│       ├── database.py       # 数据库连接
7
│       ├── models.py         # SQLAlchemy 模型
8
│       ├── schemas.py        # Pydantic 数据模型
9
│       ├── crud.py           # 数据库 CRUD 操作
10
│       └── crawler/
11
│           ├── douyin_client.py  # 抖音直播间爬虫客户端
12
│           └── tasks.py          # 后台采集任务
13
├── database/                 # 数据库相关脚本
14
├── shot/                     # 项目截图
15
├── requirements.txt          # Python 依赖
16
└── README.md

1
class DouyinLiveClient:
2
    """抖音直播间爬虫客户端"""
3

4
    async def init(self) -> None:
5
        """初始化浏览器实例"""
6
        self._playwright = await async_playwright().start()
7
        self._browser = await self._playwright.chromium.launch(
8
            headless=self._headless,
9
        )
10
        # 模拟真实用户环境
11
        self._context = await self._browser.new_context(
12
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...",
13
            locale="zh-CN",
14
        )
15

16
    async def fetch_room_info(self, room_name: str, room_url: str) -> LiveRoomInfo:
17
        """获取直播间实时数据"""
18
        page = await self._open_room_page(room_url)
19
        online_count, like_count = await self._extract_room_metrics(page, ...)
20
        return LiveRoomInfo(
21
            room_name=room_name,
22
            is_live=online_count > 0 or like_count > 0,
23
            online_count=online_count,
24
            like_count=like_count,
25
        )

1
async def _simulate_human_actions(self, page: Page) -> None:
2
    """模拟人工操作，提高采集稳定性"""
3
    # 1. 随机等待
4
    await page.wait_for_timeout(random.randint(800, 2500))
5

6
    # 2. 鼠标移动到页面中间区域
7
    x = random.randint(int(width * 0.3), int(width * 0.7))
8
    y = random.randint(int(height * 0.3), int(height * 0.7))
9
    await page.mouse.move(x, y, steps=random.randint(5, 20))
10
    await page.mouse.click(x, y, delay=random.randint(50, 200))
11

12
    # 3. 模拟滚动
13
    await page.mouse.wheel(0, random.randint(200, 800))

1
async def fetch_danmu_snapshot(self, room_name: str, room_url: str) -> List[DanmuItem]:
2
    """抓取当前页面弹幕快照"""
3
    # 定位弹幕容器
4
    container_locator = page.locator(
5
        "div.webcast-chatroom___list "
6
        "div:has(span.v8LY0gZF)"  # 昵称
7
        ":has(span.webcast-chatroom___content-with-emoji-text)"  # 内容
8
    )
9

10
    # 去重处理
11
    key = f"{room_name}|{sender_nickname}|{content}"
12
    if key not in seen_set:
13
        seen_set.add(key)
14
        danmu_items.append(DanmuItem(...))
15

16
    return danmu_items

1
@staticmethod
2
def _parse_number_from_text(text: str) -> int:
3
    """解析中文数字文本"""
4
    m = re.search(r"([\d.]+)\s*([wW万]?)", raw)
5
    value = float(m.group(1))
6
    unit = m.group(2)
7

8
    if unit in ("w", "W", "万"):
9
        value *= 10000
10

11
    return int(value)

1
CREATE DATABASE jj_douyin CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;