|
发表于 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 小助理,如未能正确解答您的问题,请继续追问。 |
|