理解到位！那我们只需要调整代码的位置，让 `finally` 块在所有截图完成之后执行，而不是每次截图后执 ...
帮我分析一下这个和你写的
import requests
import re
import json
import logging
import subprocess
from datetime import datetime
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 初始的多个room_url
room_urls = ["https://live.douyin.com/705102825087",
"https://live.douyin.com/395156372226",
"https://live.douyin.com/569866064557"]
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")
screenshot_path = f"{date_str}_{room_id}.jpg"
# 使用ffmpeg截取截图
command = [
"ffmpeg",
"-i", stream_url,
"-vframes", "1", # 只截取一帧
"-q:v", "2", # 设置截图质量
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}")
Linker0662 发表于 2024-07-18 14:21
帮我分析一下这个和你写的
import requests
import...
好的，让我们一起看看这个代码并分析一下它的工作原理。这个代码的主要任务是从多个抖音直播间中抓取截图。我们会逐步分析各个部分，并给出需要的改进建议。
### 代码分析
#### 导入模块和日志配置
import requests
import re
import json
import logging
import subprocess
from datetime import datetime
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
这个部分导入了需要的模块，并配置了日志记录，便于在程序运行时记录重要信息。
#### 初始的多个直播间URL列表
room_urls = []# 在这里填入你的多个直播间URL
这里定义了一个空列表 `room_urls`，你需要在这里填入你要抓取截图的直播间URL。
#### 获取直播流URL函数
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
这个函数 `get_stream_url` 通过一系列 HTTP 请求和正则表达式解析来获取直播流的 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}")
这个函数 `capture_screenshot` 使用 `ffmpeg` 命令来截取直播流的截图，并将截图保存为文件。
#### 主逻辑
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}")
这个部分是代码的主逻辑，它遍历每个直播间 URL，提取 room_id，获取直播流 URL，然后截取截图。
### 修改建议
为了保存多个不同直播间的截图并放在一个以当天日期命名的文件夹中，我们可以做以下调整：
1. 创建一个以当前日期命名的文件夹。
2. 将截图保存到这个文件夹中。
修改后的代码如下：
```python
import os
import requests
import re
import json
import logging
import subprocess
from datetime import datetime
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 初始的多个room_url
room_urls = []# 在这里填入你的多个直播间URL
# 创建一个以当天日期命名的文件夹
date_str = datetime.now().strftime("%Y%m%d")
if not os.path.exists(date_str):
os.makedirs(date_str)
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(
以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。 FishC 发表于 2024-7-18 14:22
好的，让我们一起看看这个代码并分析一下它的工作原理。这个代码的主要任务是从多个抖音直播间中抓取截 ...
你怎么在
# 解析JSON响应
data = json.loads(
就结束了呀 Linker0662 发表于 2024-07-18 14:36
你怎么在
# 解析JSON响应
data = json. ...
啊哈，抱歉让你有点困惑了！我们继续分析完这个代码。让我们从解析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 小助理，如未能正确解答您的问题，请继续追问。 2024-07-18 14:38:44,269 - INFO - 处理直播间URL: https://live.douyin.com/1234567890
Traceback (most recent call last):
File "E:\PyCharm\pythonProject\MaoZhua2.py", line 109, in <module>
stream_url = get_stream_url(room_id)
^^^^^^^^^^^^^^^^^^^^^^^
File "E:\PyCharm\pythonProject\MaoZhua2.py", line 66, in get_stream_url
status = data.get("data", {}).get("data", {}).get("status", 0)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'list' object has no attribute 'get'
为什么报错了呀 Linker0662 发表于 2024-07-18 14:39
2024-07-18 14:38:44,269 - INFO - 处理直播间URL: ht ...
这是一个典型的JSON解析错误，表明 `data.get("data", {})` 返回的是一个列表而不是字典，因此不能调用 `get` 方法。我们需要确认 `data` 的结构，以便正确地提取我们需要的信息。为了调试这个问题，我们可以先打印出 `data` 看看它的结构，然后相应地调整我们的代码。
### 调试和修正代码
1. **打印 JSON 响应**：先打印出 `data`，看看到底是什么结构。
print(json.dumps(data, indent=4, ensure_ascii=False))
把这个打印语句放在 `data = json.loads(ress.text)` 之后，重新运行程序，看下打印出来的结果。
2. **检查数据结构**：根据打印出来的 `data` 结构，调整提取数据的方式。假设 `data` 的结构如下：
{
"data": {
"data": [
{
"status": 2,
"stream_url": {
"live_core_sdk_data": {
"pull_data": {
"stream_data": "{\"data\": {\"origin\": {\"main\": {\"flv\": \"http://example.com/stream.flv\"}}}}"
}
}
}
}
]
}
}
在这种情况下，`data["data"]["data"]` 返回的是一个列表。
### 更新代码
基于上面的假设结构，下面是更新后的代码：
import requests
import re
import json
import logging
import subprocess
from datetime import datetime
# 配置日志
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)
print(json.dumps(data, indent=4, ensure_ascii=False))# 打印JSON响应结构
data_list = data.get("data", {}).get("data", [])
if data_list:
status = data_list.get("status", 0)
if status == 2:
stream_data = data_list.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("未能获取到有效的数据")
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")
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}")
### 说明
1. **打印 JSON 响应**：在解析 JSON 之前添加了 `print(json.dumps(data, indent=4, ensure_ascii=False))`，这样可以看到实际返回的数据结构。
2. **处理 JSON 数据**：假设 `data["data"]["data"]` 是一个列表，所以我们首先获取这个列表 `data_list`，然后检查列表是否为空，再根据列表的第一个元素来提取所需的信息。
通过这些调整，我们可以更好地适应抖音返回的数据结构，避免因为数据结构不匹配导致的错误。如果你看到打印出来的 JSON 结构和假设的不一样，请根据实际的结构进一步调整代码。
以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。 FishC 发表于 2024-7-18 14:40
这是一个典型的JSON解析错误，表明 `data.get("data", {})` 返回的是一个列表而不是字典，因此不能调用 ...
E:\Python\python.exe E:\PyCharm\pythonProject\MaoZhua2.py
2024-07-19 19:15:44,998 - INFO - 处理直播间URL: https://live.douyin.com/1234567890
2024-07-19 19:15:46,338 - WARNING - 直播未开始或不可用
2024-07-19 19:15:46,338 - WARNING - 未能获取直播流URL: 1234567890
2024-07-19 19:15:46,338 - INFO - 处理直播间URL: https://live.douyin.com/0987654321
{
"data": {
"data": [
{
"id_str": "7306889246596483852",
"status": 4,
"status_str": "4",
"title": "順心如意",
"user_count_str": "0",
"mosaic_status": 0,
"mosaic_status_str": "",
"admin_user_ids": [],
"admin_user_ids_str": [],
"live_room_mode": 0,
"has_commerce_goods": false,
"linker_map": {},
"AnchorABMap": {},
"like_count": 0,
"owner_user_id_str": "",
"admin_user_open_ids": [],
"admin_user_open_ids_str": [],
"owner_open_id_str": ""
}
],
"enter_room_id": "7306889246596483852",
"user": {
"id_str": "78774060474",
"sec_uid": "MS4wLjABAAAA3rw3vYyLpKUFQ7tLRrzjzDGgxfAHBvt_3l4HYXrEpk4",
"nickname": "顺心如意88",
"avatar_thumb": {
"url_list": [
"https://p26.douyinpic.com/aweme/100x100/aweme-avatar/tos-cn-avt-0015_f7357d5fadf5df5f6b110526dfa3821d.jpeg?from=3067671334",
"https://p11.douyinpic.com/aweme/100x100/aweme-avatar/tos-cn-avt-0015_f7357d5fadf5df5f6b110526dfa3821d.jpeg?from=3067671334",
"https://p3.douyinpic.com/aweme/100x100/aweme-avatar/tos-cn-avt-0015_f7357d5fadf5df5f6b110526dfa3821d.jpeg?from=3067671334"
]
},
"follow_info": {
"follow_status": 0,
"follow_status_str": "0"
},
"open_id_str": ""
},
"qrcode_url": "",
"enter_mode": 0,
"room_status": 2,
"partition_road_map": {},
"similar_rooms": [],
"shark_decision_conf": ""
},
"extra": {
"now": 1721387746335
},
"status_code": 0
}
2024-07-19 19:15:46,926 - WARNING - 未能获取到有效的数据
2024-07-19 19:15:46,926 - WARNING - 未能获取直播流URL: 0987654321
{
"data": {
"prompts": "该内容暂时无法无法查看"
},
"extra": {
"now": 1721387746918
},
"status_code": 4001038
}
进程已结束，退出代码为 0
为什么还是不行呀 文件夹也没有创建 Linker0662 发表于 2024-07-19 19:17
E:\Python\python.exe E:\PyCharm\pythonProject\MaoZ ...
看起来你已经通过某种方式获取到了一些数据，但它仍然没有成功获取直播流的URL。这里有几种可能的原因：
1. **API响应格式改变**：从你的输出来看，JSON响应结构和你预期的可能不太一样。
2. **直播状态问题**：有一些直播间可能还未开始或者不可用，这会影响你的抓取。
我们结合你的错误信息和当前情况，调整一下代码逻辑来处理这些细节。
### 调整后的代码示例
假设 `get_stream_url` 函数是一个从API获取直播流URL的函数，我们需要确保正确解析API的响应，并处理不同的返回情况。
import json
import logging
def get_stream_url(room_id):
# 模拟API调用
response = simulate_api_call(room_id)
try:
data = json.loads(response)
# 确保数据格式正确
if isinstance(data, dict) and isinstance(data.get('data', {}), dict):
# 获取直播状态列表（假设是data["data"]["data"]）
stream_data = data['data'].get('data')
if isinstance(stream_data, list) and len(stream_data) > 0:
live_info = stream_data
status = int(live_info.get("status", 0))# 强制转换为整数
if status == 2:# 假设2表示直播中
stream_url = live_info.get("stream_url")# 替换为正确的键名
return stream_url
else:
logging.warning("直播未开始或不可用")
else:
logging.warning("未能获取到有效的数据")
else:
logging.warning("API响应数据格式不正确")
except json.JSONDecodeError as e:
logging.error(f"JSON解析错误: {e}")
return None
def simulate_api_call(room_id):
# 模拟返回数据，你可以根据实际情况替换此部分
if room_id == "1234567890":
return '''
{
"data": {
"data": [
{
"id_str": "7306889246596483852",
"status": 4,
"status_str": "4",
"title": "心如意",
"user_count_str": "0",
"mosaic_status": 0,
"mosaic_status_str": ""
}
],
"enter_room_id": "7306889246596483852",
"room_status": 2
},
"extra": {
"now": 1721387746335
},
"status_code": 0
}'''
else:
return '''
{
"data": {
"prompts": "该内容暂时无法查看"
},
"extra": {
"now": 1721387746918
},
"status_code": 4001038
}'''
# 示例：处理多个直播间URL
room_ids = ["1234567890", "0987654321"]
for room_id in room_ids:
logging.info(f"处理直播间URL: https://live.douyin.com/{room_id}")
stream_url = get_stream_url(room_id)
if stream_url:
logging.info(f"获取直播流URL成功: {stream_url}")
# 继续处理，比如截图等操作
else:
logging.warning(f"未能获取直播流URL: {room_id}")
### 解释
1. **模拟API调用**：为了测试方便，我创建了一个 `simulate_api_call` 函数来模拟API的响应。你可以将其替换为实际的API请求代码。
2. **解析JSON响应**：尝试解析JSON响应并校验数据结构，确保其符合预期。如果不符合，则记录警告信息。
3. **判断直播状态**：根据解析到的状态码判断直播间是否在直播中，然后获取直播流的URL。
4. **日志记录**：通过记录日志来监控每一步的状态和潜在的问题，便于调试。
希望这次能够帮助你顺利获取到直播流的URL！如果还有问题，请继续提供详细信息，我会尽力帮你解决！
以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。 我觉得我们偏离了问题解决方向，请看下面的代码，可以实现对不同直播间的视频流实现截图，我还需要实现自动新建文件夹和对应直播间截图的命名功能
import requests
import re
import json
import logging
import subprocess
from datetime import datetime
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 初始的多个room_url
room_urls = ["https://live.douyin.com/705102825087",
"https://live.douyin.com/395156372226",
"https://live.douyin.com/569866064557"]
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")
screenshot_path = f"{date_str}_{room_id}.jpg"
# 使用ffmpeg截取截图
command = [
"ffmpeg",
"-i", stream_url,
"-vframes", "1", # 只截取一帧
"-q:v", "2", # 设置截图质量
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}")
Linker0662 发表于 2024-07-19 19:22
我觉得我们偏离了问题解决方向，请看下面的代码，可以 ...
明白了，看来你已经在实现抓取抖音直播间视频流并进行截图的功能。那么，我们来添加一些代码，实现自动创建文件夹及为截图命名的功能。
主要的步骤包括：
1. 在脚本开始时创建一个目录来保存截图。
2. 生成保存截图的路径，并包含日期和直播间ID。
3. 使用 `os` 模块来确保目录存在，然后使用 `subprocess` 执行 `ffmpeg` 命令进行截图。
我们将这些步骤集成到你的现有代码中。下面是修改后的代码：
import requests
import re
import json
import logging
import subprocess
import os
from datetime import datetime
# 配置日志
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"
# 添加更多的直播间URL
]
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")
# 创建截图文件夹，如果不存在
folder_path = os.path.join(os.getcwd(), "screenshots")
if not os.path.exists(folder_path):
os.makedirs(folder_path)
screenshot_path = os.path.join(folder_path, f"{date_str}_{room_id}.jpg")
# 使用ffmpeg截取截图
command = [
"ffmpeg", "-i", stream_url, "-frames:v", "1", "-q:v", "2", 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. **创建截图文件夹**: 在 `capture_screenshot` 函数中，使用 `os.makedirs` 来创建一个名为 `screenshots` 的文件夹。如果该文件夹不存在则创建它。
2. **生成截图路径**: 将文件名设为包含日期和直播间ID的字符串，并存储在 `screenshots` 文件夹中。
3. **执行 `ffmpeg` 命令**: 通过 `subprocess.run` 调用 `ffmpeg` 截取直播流的截图并保存到指定路径。
这样你就可以根据不同的直播间ID和当前日期保存截图了。希望这能帮助你解决问题！
以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。
页:
1
[2]