哎呀，爱播总是在晚上才开播，不知道什么时候开播，怎么办？
操作流程
选择执行服务的设备：电脑、手机、云服务器、NAS
我这边做演示的是NAS搭建的虚拟化服务器，也算云服务器
准备
一台设备、Docker、青龙面板、脚本代码
安装Docker并部署青龙面板
Windows系统
- 访问Docker官网下载安装包
- 安装后启动Docker服务
docker pull whyour/qinglong:latest
docker run -dit \
-v /path/to/your/qinglong:/ql/data \
-p 5700:5700 \
--name qinglong \
--hostname qinglong \
--restart unless-stopped \
whyour/qinglong:latest将/path/to/your/qinglong替换为你本地的实际路径
打开浏览器访问 http://localhost:5700
macOS系统
- 安装Docker Desktop for Mac
- 从Docker官网下载
- 拉取青龙镜像并运行
docker pull whyour/qinglong:latest
docker run -dit \
-v ~/qinglong:/ql/data \
-p 5700:5700 \
--name qinglong \
--hostname qinglong \
--restart unless-stopped \
whyour/qinglong:latestLinux系统
安装Docker
curl -fsSL https://get.docker.com | sh
sudo systemctl start docker
sudo systemctl enable docker运行青龙容器
docker pull whyour/qinglong:latest
docker run -dit \
-v ~/qinglong:/ql/data \
-p 5700:5700 \
--name qinglong \
--hostname qinglong \
--restart unless-stopped \
whyour/qinglong:latest安卓手机
- 安装Termux
- 从F-Droid或Google Play安装
- 安装必要组件
pkg update
pkg install wget git docker运行青龙容器（需要root权限）
docker run -dit \
-v /data/ql:/ql/data \
-p 5700:5700 \
--name qinglong \
whyour/qinglong:latestiOS系统
- 从App Store安装iSH Shell
- 安装Alpine Linux包
apk update
apk add docker运行青龙容器
docker pull whyour/qinglong:latest
docker run -dit \
-v ~/qinglong:/ql/data \
-p 5700:5700 \
--name qinglong \
--hostname qinglong \
--restart unless-stopped \
whyour/qinglong:latest青龙面板不管是用什么系统部署，默认端口都是5700或者15700
这里有一点需要注意，因为青龙面板更新，数据挂载目录由/ql到/ql/data
部署脚本
进入青龙面板 脚本管理 右上角+号
类型：空文件 文件名：douyin_monitor.py 点击确定
点击右上角编辑 置入代码
#!/usr/bin/env python3
# 抖音直播监控(完整功能版) - 支持开播/关播通知开关
import requests
import time
import json
import random
import os
from urllib.parse import quote
# 配置区域 ==================================
DOUYIN_IDS = ["66871771615"] # 监控的抖音直播间ID 主播抖音号
WXPUSHER_APP_TOKEN = "AT_XXXXXXXXXXXXXXXXXX" # WXPUSHER应用的TOKEN
# 从 https://wxpusher.zjiecode.com/admin/ 获取
WXPUSHER_UIDS = [
"UID_XXXXXXXXXXXXXXXXXXXXXXXXXX ", # 第一个用户的UID
#"UID_第二个用户的UID", # 第二个用户的UID
#"UID_第三个用户的UID" # 第三个用户的UID
# 可以继续添加更多UID
] # 替换为你的WxPusher用户UID列表
CHECK_INTERVAL = 300 # 与青龙定时任务保持一致(秒)
DEBUG_MODE = False # 是否输出调试信息 True False
NOTIFY_LIVE_START = True # 是否发送开播通知
NOTIFY_LIVE_END = True # 是否发送关播通知
# ==========================================
# 状态存储文件路径(青龙面板环境适配)
STATUS_FILE = "/ql/data/scripts/douyin_live_status.json" if os.path.exists("/ql/data") else "douyin_live_status.json"
def init_browser_meta():
"""动态生成浏览器指纹"""
chrome_versions = [
("124.0.6367.91", '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"'),
("123.0.6312.106", '"Google Chrome";v="123", "Not:A-Brand";v="8", "Microsoft Edge";v="123"'),
("122.0.6261.95", '"Chromium";v="122", "Not.A/Brand";v="24", "Microsoft Edge";v="122"')
]
version, sec_ua = random.choice(chrome_versions)
return {
"ua": f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{version} Safari/537.36",
"sec_ua": sec_ua,
"platform": random.choice(["Windows", "Win32"]),
"width": random.randint(1280, 1920),
"height": random.randint(720, 1080),
"last_reset": time.time()
}
def load_status():
"""加载状态记录"""
try:
if os.path.exists(STATUS_FILE):
with open(STATUS_FILE, 'r') as f:
return json.load(f)
except Exception as e:
print(f"⚠️ 加载状态文件失败: {str(e)}")
return {}
def save_status(status):
"""保存状态记录"""
try:
with open(STATUS_FILE, 'w') as f:
json.dump(status, f, indent=2)
except Exception as e:
print(f"⚠️ 保存状态文件失败: {str(e)}")
def get_real_ttwid(retry=3):
"""获取有效ttwid(带重试机制)"""
headers = {
"User-Agent": "Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Mobile Safari/537.36",
"Accept": "application/json, text/plain, */*",
"Origin": "https://www.douyin.com",
"Sec-Fetch-Site": "same-site",
"Sec-Fetch-Mode": "cors",
}
for _ in range(retry):
try:
s = requests.Session()
s.get("https://www.douyin.com/", headers=headers, timeout=10)
r = s.post(
"https://ttwid.bytedance.com/ttwid/union/register/",
json={
"region": "cn",
"aid": 1768,
"service": "www.ixigua.com",
"migrate_info": {"ticket": "", "source": "node"}
},
headers=headers,
timeout=15
)
if r.status_code == 200 and 'ttwid' in s.cookies:
if DEBUG_MODE:
print("✅ 成功获取ttwid:", s.cookies['ttwid'][:30]+"...")
return s.cookies['ttwid']
print(f"❌ 获取ttwid失败，状态码：{r.status_code} 响应：{r.text[:200]}")
time.sleep(2)
except Exception as e:
print(f"⚠️ ttwid获取异常: {str(e)}")
time.sleep(3)
print("❌ 连续获取ttwid失败，请检查网络或稍后重试")
return None
def call_live_api(douyin_id, ttwid, browser_meta):
"""调用直播API(带随机参数)"""
params = {
"aid": 6383,
"device_platform": "web",
"web_rid": douyin_id,
"enter_from": "web_search",
"cookie_enabled": "true",
"screen_width": browser_meta["width"],
"screen_height": browser_meta["height"],
"browser_language": "zh-CN",
"browser_platform": browser_meta["platform"],
"browser_name": "Chrome",
"browser_version": browser_meta["ua"].split("Chrome/")[1].split(" ")[0],
"browser_online": "true",
"os_name": "Windows",
"os_version": "10",
"msToken": "".join(random.choices("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", k=107)),
"_t": int(time.time() * 1000),
"random": random.randint(100000, 999999)
}
headers = {
"User-Agent": browser_meta["ua"],
"Sec-Ch-Ua": browser_meta["sec_ua"],
"Sec-Ch-Ua-Mobile": "?0",
"Sec-Ch-Ua-Platform": f'"{browser_meta["platform"]}"',
"Accept": "*/*",
"Origin": "https://live.douyin.com",
"Referer": f"https://live.douyin.com/{douyin_id}",
"Cookie": f"ttwid={ttwid}; s_v_web_id=verify_{''.join(random.choices('1234567890abcdef', k=16))};"
}
try:
response = requests.get(
"https://live.douyin.com/webcast/room/web/enter/",
headers=headers,
params=params,
timeout=20
)
return response
except Exception as e:
print(f"🚨 API请求异常: {str(e)}")
return None
def wxpusher_send(nickname, message_type, room_title="", live_url="", cover_url=""):
"""发送微信通知"""
# 检查通知开关
if message_type == "live_start" and not NOTIFY_LIVE_START:
print("⏳ 开播通知已关闭，跳过发送")
return
if message_type == "live_end" and not NOTIFY_LIVE_END:
print("⏳ 关播通知已关闭，跳过发送")
return
status = load_status()
notification_key = f"{nickname}_{message_type}"
# 6小时内不重复通知
if notification_key in status:
last_time = status[notification_key].get("time", 0)
if time.time() - last_time < 21600:
print(f"⏳ 已跳过重复通知（6小时内已发送过）: {nickname} {message_type}")
return
url = "https://wxpusher.zjiecode.com/api/send/message"
headers = {"Content-Type": "application/json"}
if message_type == "live_start":
content = f"## 🎬 {nickname} 开播啦！\n\n**直播间标题**: {room_title}\n\n[点击进入直播间]({live_url})"
summary = f"{nickname}开播了"
elif message_type == "live_end":
content = f"## 🏁 {nickname} 直播结束\n\n**最后标题**: {room_title}\n\n[直播间历史链接]({live_url})"
summary = f"{nickname}直播已结束"
else:
return
data = {
"appToken": WXPUSHER_APP_TOKEN,
"content": content,
"summary": summary,
"contentType": 2,
"uids": WXPUSHER_UIDS,
"url": live_url
}
try:
r = requests.post(url, headers=headers, json=data, timeout=10)
resp = r.json()
if isinstance(resp, dict) and resp.get("code") == 1000:
print(f"📢 {message_type}通知发送成功 | 主播: {nickname}")
status[notification_key] = {
"time": int(time.time()),
"title": room_title,
"url": live_url
}
save_status(status)
else:
error_msg = resp.get("msg", str(resp))
print(f"🚨 通知失败 | 错误信息: {error_msg}")
except Exception as e:
print(f"🚨 通知异常: {str(e)}")
def check_live(douyin_id, browser_meta):
"""执行直播状态检查"""
ttwid = get_real_ttwid()
if not ttwid:
return
response = call_live_api(douyin_id, ttwid, browser_meta)
if not response:
return
try:
if response.status_code != 200:
print(f"🚨 API响应异常 | 状态码：{response.status_code}")
return
data = response.json()
if DEBUG_MODE:
print("🔧 完整响应：", json.dumps(data, indent=2, ensure_ascii=False))
if not data.get("data") or not isinstance(data["data"], dict):
print("🚨 异常响应结构，可能被风控")
return
room_info = data["data"].get("data", [{}])[0]
user_info = data["data"].get("user", {})
nickname = user_info.get("nickname", "未知主播")
room_title = room_info.get("title", "无标题")
live_url = f"https://live.douyin.com/{douyin_id}"
cover_url = room_info.get("cover", {}).get("url_list", [""])[0]
is_living = room_info.get("status") == 2
status = load_status()
current_status = status.get(douyin_id, {"is_living": False})
if douyin_id not in status or current_status["is_living"] != is_living:
if is_living:
print(f"🔴 检测到开播：{nickname} | 标题：{room_title}")
wxpusher_send(
nickname=nickname,
message_type="live_start",
room_title=room_title,
live_url=live_url,
cover_url=cover_url
)
else:
print(f"⚪ 检测到关播：{nickname}")
wxpusher_send(
nickname=nickname,
message_type="live_end",
room_title=current_status.get("last_title", "无记录"),
live_url=live_url
)
status[douyin_id] = {
"is_living": is_living,
"last_title": room_title,
"last_check": int(time.time())
}
save_status(status)
else:
print(f"🔄 状态未变化：{nickname} | {'直播中' if is_living else '未开播'}")
except Exception as e:
print(f"🚨 检测异常: {str(e)}")
if __name__ == "__main__":
print("====== 🚀 抖音直播监控已启动 ======")
#print(f"定时规则: */5 * * * *")
print(f"开播通知: {'开启' if NOTIFY_LIVE_START else '关闭'} | 关播通知: {'开启' if NOTIFY_LIVE_END else '关闭'}")
browser_meta = init_browser_meta()
try:
for did in DOUYIN_IDS:
check_live(did, browser_meta)
print("✅ 本轮检测完成")
except Exception as e:
print(f"🚨 主程序异常: {str(e)}")
创建好脚本之后创建定时任务
定时任务 创建任务
名称：心碎007开播通知
python3 /ql/data/scripts/douyin_monitor.py定时规则：*/5 * * * *
这样就创建好了，可以查看日志里的执行状态
调试输出的数据代表
🔧 完整响应： {
"data": {
"data": [
{
"id_str": "7490288179593906981",
"status": 2, # 直播状态(2表示正在直播4表示直播结束)
"status_str": "2",
"title": "无美颜 灯牌连麦", # 直播间标题
"user_count_str": "100+",
"cover": { # 直播间封面图
"url_list": [
"https://p3-webcast-sign.douyinpic.com/image-cut-tos-priv/af95bc493fc73baab08c87d1e7ee15a4~tplv-qz53dukwul-common-resize:0:0.image?..."
]
},
"owner": { # 主播信息
"nickname": "心碎007", # 主播昵称
"avatar_thumb": { # 主播头像
"url_list": [
"https://p26.douyinpic.com/aweme/100x100/aweme-avatar/tos-cn-avt-0015_1a1b00b7317d53a716aea35c094f643d.jpeg?..."
]
}
},
"stats": { # 直播间统计数据
"total_user_str": "2万+", # 累计观众
"user_count_str": "152" # 当前在线观众
}
}
]
}
}历史更新
2025.04.07
增加关播通知
增加开播关播开关配置
增加API风控检测
增加通知多个用户UID
删除通知卡片显示直播间标题
修复重复通知
修复脚本状态一直是运行中
结语
可见实现一个小功能也并不是很省事，但是重点在学习
关注我爱播：心碎007
本文代码产权归本站所有
© 版权声明
THE END
喜欢就支持一下吧
相关推荐
