|
|
发表于 2024-7-18 14:37:45
|
显示全部楼层
啊哈，抱歉让你有点困惑了！我们继续分析完这个代码。让我们从解析JSON响应的部分接着往下看，并完成整个代码的分析。
### 解析JSON响应
# 解析JSON响应
data = json.loads(ress.text)
status = data.get("data", {}).get("data", {}).get("status", 0)
if status == 2:
stream_data = data.get("data", {}).get("data", {}).get("stream_url", {}).get("live_core_sdk_data", {}).get("pull_data", {}).get("stream_data", {})
value = json.loads(stream_data)
real_url = value.get("data", {}).get("origin", {}).get("main", {}).get("flv", "")
return real_url
else:
logging.warning("直播未开始或不可用")
这部分代码是从JSON响应中提取直播流的实际URL。以下是一些关键点：
1. **解析响应**：使用`json.loads`将响应文本解析成字典。
2. **检查状态**：检查直播的状态，`status`为2表示直播正在进行。
3. **提取流数据**：从解析的数据中逐层提取出实际的直播流URL。
### 处理直播间URL和截图
def capture_screenshot(stream_url, room_id):
# 获取当前日期时间，作为文件名的一部分
date_str = datetime.now().strftime("%Y%m%d")
screenshot_path = f"{date_str}_{room_id}.jpg"
# 使用ffmpeg截取截图
command = ["ffmpeg", "-i", stream_url, "-frames:v", "1", screenshot_path]
# 执行ffmpeg命令
try:
subprocess.run(command, check=True)
logging.info(f"截图已保存: {screenshot_path}")
except subprocess.CalledProcessError as e:
logging.error(f"截图失败: {e}")
for room_url in room_urls:
logging.info(f"处理直播间URL: {room_url}")
# 提取room ID
match = re.search(r'https://live.douyin.com/(\d+)', room_url)
if not match:
logging.warning("无法提取直播间ID")
continue
room_id = match.group(1)
stream_url = get_stream_url(room_id)
if stream_url:
capture_screenshot(stream_url, room_id)
else:
logging.warning(f"未能获取直播流URL: {room_id}")
### 修改建议
#### 创建目录
我们可以在截图保存之前创建一个以当天日期命名的目录。
#### 完整代码
import requests
import re
import json
import logging
import subprocess
from datetime import datetime
import os
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 初始的多个room_url
room_urls = [
"https://live.douyin.com/1234567890",
"https://live.douyin.com/0987654321"
]
def get_stream_url(room_id):
live_url = f"https://live.douyin.com/{room_id}"
session = requests.Session()
# 发送初始请求获取 __ac_nonce
headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36",
"Upgrade-Insecure-Requests": "1",
}
oresp = session.get(live_url, headers=headers)
oresp.close()
# 从 Set-Cookie 头中提取 __ac_nonce
ac_nonce_match = re.search(r'(?i)__ac_nonce=(.*?);', oresp.headers.get("Set-Cookie", ""))
if ac_nonce_match:
ac_nonce = ac_nonce_match.group(1)
else:
ac_nonce = None
if ac_nonce:
# 设置 __ac_nonce cookie 并发送另一个请求
session.cookies.set("__ac_nonce", ac_nonce)
resp = session.get(live_url, headers=headers)
# 从 Set-Cookie 头中提取 ttwid
ttwid_match = re.search(r'(?i)ttwid=.*?;', resp.headers.get("Set-Cookie", ""))
if ttwid_match:
ttwid = ttwid_match.group(0)
else:
ttwid = None
if ttwid:
# 构建最终请求的 URL
url = f"https://live.douyin.com/webcast/room/web/enter/?aid=6383&app_name=douyin_web&live_id=1&device_platform=web&language=zh-CN&enter_from=web_live&cookie_enabled=true&screen_width=1728&screen_height=1117&browser_language=zh-CN&browser_platform=MacIntel&browser_name=Chrome&browser_version=116.0.0.0&web_rid={room_id}"
headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36",
"Cookie": ttwid,
"Accept": "*/*",
"Host": "live.douyin.com",
"Connection": "keep-alive",
}
# 发送最终请求
ress = session.get(url, headers=headers)
ress.close()
# 解析JSON响应
data = json.loads(ress.text)
status = data.get("data", {}).get("data", {}).get("status", 0)
if status == 2:
stream_data = data.get("data", {}).get("data", {}).get("stream_url", {}).get("live_core_sdk_data", {}).get("pull_data", {}).get("stream_data", {})
value = json.loads(stream_data)
real_url = value.get("data", {}).get("origin", {}).get("main", {}).get("flv", "")
return real_url
else:
logging.warning("直播未开始或不可用")
else:
logging.warning("未能提取到 ttwid")
else:
logging.warning("未能提取到 __ac_nonce")
return None
def capture_screenshot(stream_url, room_id):
# 获取当前日期时间，作为文件名的一部分
date_str = datetime.now().strftime("%Y%m%d")
# 创建以当前日期命名的目录
os.makedirs(date_str, exist_ok=True)
screenshot_path = os.path.join(date_str, f"{room_id}.jpg")
# 使用ffmpeg截取截图
command = ["ffmpeg", "-i", stream_url, "-frames:v", "1", screenshot_path]
# 执行ffmpeg命令
try:
subprocess.run(command, check=True)
logging.info(f"截图已保存: {screenshot_path}")
except subprocess.CalledProcessError as e:
logging.error(f"截图失败: {e}")
for room_url in room_urls:
logging.info(f"处理直播间URL: {room_url}")
# 提取room ID
match = re.search(r'https://live.douyin.com/(\d+)', room_url)
if not match:
logging.warning("无法提取直播间ID")
continue
room_id = match.group(1)
stream_url = get_stream_url(room_id)
if stream_url:
capture_screenshot(stream_url, room_id)
else:
logging.warning(f"未能获取直播流URL: {room_id}")
### 总结
1. **解析直播URL**：通过发送请求和解析响应获取直播流的实际URL。
2. **截图保存**：使用`ffmpeg`从直播流中截取一帧并保存为图片。
3. **目录管理**：创建以当前日期命名的目录，并将所有截图保存到该目录中。
希望这个分析和改进对你有帮助！如果有任何问题，欢迎随时提问。
以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。 |
|
( 粤ICP备18085999号-1 | 粤公网安备 44051102000585号)
狗仔卡
楼主
显身卡