用Python+Requests+SQLite3，我给自己做了个抖音直播间数据监控小工具

用PythonRequestsSQLite3构建抖音直播间数据监控系统最近发现身边不少朋友沉迷于抖音直播经常讨论某个主播的直播间人气变化。作为技术爱好者我萌生了一个想法能不能用Python写个小工具自动记录我关注的主播数据经过一周的摸索终于完成了一个轻量级的监控系统。这个工具不仅能定时抓取直播间状态还能把数据存入本地数据库生成可视化报表。下面就把我的实现思路和踩坑经验分享给大家。1. 系统架构设计这个小工具的核心功能很简单定期获取直播间数据并存储。但要让整个流程自动化运行需要考虑以下几个模块数据采集层通过Requests库模拟浏览器请求获取直播间JSON数据数据存储层使用SQLite3建立本地数据库结构化存储历史记录任务调度层利用Python内置的sched模块实现定时任务数据分析层通过Pandas进行基础统计Matplotlib生成趋势图表关键技术选型对比技术选项优势适用场景Requests简单易用无需浏览器环境基础HTTP请求SQLite3零配置单文件存储本地小型数据管理APScheduler功能丰富支持多种触发器复杂定时任务Basic SchedPython内置无需安装简单定时需求提示如果只是个人使用建议优先选择Python标准库组件减少依赖问题。2. 环境配置与基础搭建首先创建项目目录结构/douyin-monitor ├── config.py # 配置文件 ├── database.py # 数据库操作 ├── collector.py # 数据采集 ├── scheduler.py # 任务调度 └── analysis.py # 数据分析安装必要依赖推荐使用虚拟环境pip install requests pandas matplotlib核心配置文件示例config.py# 直播间ID列表 ROOM_IDS [123456, 654321] # 采集间隔秒 INTERVAL 300 # 数据库路径 DB_PATH data/live_monitor.db3. 数据采集实现抖音的直播间数据主要通过其内部API获取。经过抓包分析发现关键接口形如https://live.douyin.com/webcast/room/web/enter/?room_id{room_id}实现数据采集的核心函数import requests from config import ROOM_IDS def fetch_room_data(room_id): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Cookie: 你的登录Cookie } url fhttps://live.douyin.com/webcast/room/web/enter/?room_id{room_id} try: response requests.get(url, headersheaders, timeout10) if response.status_code 200: return parse_data(response.json(), room_id) except Exception as e: print(f采集失败: {str(e)}) return None def parse_data(raw_data, room_id): room_info raw_data.get(data, {}).get(data, {}) return { room_id: room_id, timestamp: int(time.time()), viewers: room_info.get(user_count, 0), likes: room_info.get(like_count, 0), gifts: room_info.get(gift_value, 0), status: room_info.get(status, 0) }注意实际使用时需要替换有效的Cookie信息否则可能获取不到完整数据。4. 数据存储方案使用SQLite3创建数据表import sqlite3 from config import DB_PATH def init_db(): conn sqlite3.connect(DB_PATH) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS room_stats ( id INTEGER PRIMARY KEY AUTOINCREMENT, room_id TEXT NOT NULL, timestamp INTEGER NOT NULL, viewers INTEGER DEFAULT 0, likes INTEGER DEFAULT 0, gifts INTEGER DEFAULT 0, status INTEGER DEFAULT 0 ) ) conn.commit() conn.close()数据插入函数def save_to_db(data): conn sqlite3.connect(DB_PATH) cursor conn.cursor() cursor.execute( INSERT INTO room_stats (room_id, timestamp, viewers, likes, gifts, status) VALUES (?, ?, ?, ?, ?, ?) , ( data[room_id], data[timestamp], data[viewers], data[likes], data[gifts], data[status] )) conn.commit() conn.close()5. 定时任务调度使用Python标准库sched实现基础定时功能import sched import time from collector import fetch_room_data from database import save_to_db scheduler sched.scheduler(time.time, time.sleep) def job(): print(f开始执行采集任务: {time.strftime(%Y-%m-%d %H:%M:%S)}) for room_id in ROOM_IDS: data fetch_room_data(room_id) if data: save_to_db(data) scheduler.enter(INTERVAL, 1, job) if __name__ __main__: scheduler.enter(0, 1, job) scheduler.run()对于更复杂的调度需求可以改用APSchedulerfrom apscheduler.schedulers.blocking import BlockingScheduler sched BlockingScheduler() sched.scheduled_job(interval, secondsINTERVAL) def timed_job(): # 采集逻辑... sched.start()6. 数据分析与可视化基础统计查询示例import sqlite3 import pandas as pd def get_daily_stats(room_id, date): conn sqlite3.connect(DB_PATH) df pd.read_sql(f SELECT strftime(%H, datetime(timestamp, unixepoch)) as hour, AVG(viewers) as avg_viewers, MAX(viewers) as peak_viewers FROM room_stats WHERE room_id {room_id} AND date(timestamp, unixepoch) {date} GROUP BY hour ORDER BY hour , conn) conn.close() return df生成观看人数趋势图import matplotlib.pyplot as plt def plot_viewer_trend(room_id, days7): conn sqlite3.connect(DB_PATH) df pd.read_sql(f SELECT datetime(timestamp, unixepoch) as time, viewers FROM room_stats WHERE room_id {room_id} ORDER BY timestamp DESC LIMIT {24*days} , conn) plt.figure(figsize(12, 6)) plt.plot(df[time], df[viewers]) plt.title(f直播间 {room_id} 近{days}天观看人数趋势) plt.xlabel(时间) plt.ylabel(观看人数) plt.xticks(rotation45) plt.grid() plt.tight_layout() plt.savefig(freport_{room_id}.png) plt.close()7. 系统优化与扩展在实际使用过程中我发现几个可以改进的地方反爬策略应对使用代理IP轮询推荐快代理等服务随机化请求间隔30-90秒之间模拟真实用户行为轨迹数据异常处理def validate_data(data): if not isinstance(data, dict): return False required_fields [room_id, viewers, status] return all(field in data for field in required_fields)消息通知功能import smtplib from email.mime.text import MIMEText def send_alert(subject, content): msg MIMEText(content) msg[Subject] subject msg[From] your_emailexample.com msg[To] target_emailexample.com with smtplib.SMTP(smtp.example.com, 587) as server: server.login(username, password) server.send_message(msg)Web可视化界面可选 使用FlaskECharts构建简单的数据看板from flask import Flask, render_template import sqlite3 app Flask(__name__) app.route(/dashboard/room_id) def dashboard(room_id): conn sqlite3.connect(DB_PATH) # 获取数据逻辑... return render_template(dashboard.html, datachart_data)这个项目最让我惊喜的是SQLite的性能表现——即使存储了上万条记录查询速度依然很快。后来我又添加了自动生成日报的功能每天早上都会收到邮件报告关注主播的昨日表现。