Add files via upload

This commit is contained in:
Kero64 authored and GitHub committed 2025-02-16 20:22:59 +08:00
1 parent 47f939e412
commit fbef2f3170
30 files changed
+29198

No files matched your search

+367
View File
@@ -0,0 +1,367 @@
import json
import os
import subprocess
from typing import Dict
from tqdm import tqdm
import re
def convert_browser_cookies(browser_cookies: list) -> Dict:
"""转换浏览器导出的cookies格式为简单的键值对"""
essential_cookies = {}
required_keys = ['SESSDATA', 'bili_jct', 'DedeUserID']
for cookie in browser_cookies:
if cookie['name'] in required_keys:
essential_cookies[cookie['name']] = cookie['value']
return essential_cookies
def load_cookies_from_file() -> Dict:
"""从文件加载cookies"""
cookie_files = ['cookies.txt', 'bilibili_cookies.json']
required_keys = ['SESSDATA', 'bili_jct', 'DedeUserID']
for cookie_file in cookie_files:
if os.path.exists(cookie_file):
try:
with open(cookie_file, 'r', encoding='utf-8') as f:
cookies = {}
# 逐行读取文件
for line in f:
line = line.strip()
# 跳过空行和注释行(但不跳过#HttpOnly_开头的行)
if not line or (line.startswith('#') and not line.startswith('#HttpOnly_')):
continue
try:
# 处理HttpOnly标记
if line.startswith('#HttpOnly_'):
line = line[9:] # 移除'#HttpOnly_'前缀
# 分割字段并移除可能的空字段
fields = [f for f in line.split('\t') if f]
if len(fields) >= 7:
name = fields[5]
value = fields[6]
if name in required_keys:
# 对于SESSDATA,需要处理URL编码
if name == 'SESSDATA' and '%2C' in value:
value = value.replace('%2C', ',')
cookies[name] = value
print(f"找到cookie: {name}") # 调试信息
except Exception as e:
print(f"警告: 解析行 '{line}' 时出错: {str(e)}")
continue
# 检查是否找到所有必需的cookies
if all(key in cookies for key in required_keys):
print(f"已从 {cookie_file} 加载cookies")
return cookies
else:
missing = [key for key in required_keys if key not in cookies]
print(f"警告: {cookie_file} 中缺少以下cookie字段: {', '.join(missing)}")
except Exception as e:
print(f"警告: 读取 {cookie_file} 时出错: {str(e)}")
return {}
def parse_cookie_string(cookie_str: str) -> Dict:
"""解析浏览器直接复制的cookie字符串"""
cookies = {}
required_keys = ['SESSDATA', 'bili_jct', 'DedeUserID']
try:
# 分割cookie字符串
cookie_pairs = cookie_str.split(';')
for pair in cookie_pairs:
if '=' in pair:
name, value = pair.strip().split('=', 1)
if name in required_keys:
cookies[name] = value
except Exception:
return {}
return cookies
def get_cookies_from_user() -> Dict:
"""获取用户输入的cookies"""
print("请输入B站cookies (支持JSON格式、浏览器导出格式或直接复制的cookie字符串):")
cookies_str = input().strip()
# 首先尝试解析为JSON
try:
cookies_data = json.loads(cookies_str)
if isinstance(cookies_data, list):
cookies = convert_browser_cookies(cookies_data)
else:
cookies = cookies_data
except json.JSONDecodeError:
# 如果不是JSON,尝试解析为cookie字符串
cookies = parse_cookie_string(cookies_str)
# 验证必要的cookie字段
if all(key in cookies for key in ['SESSDATA', 'bili_jct', 'DedeUserID']):
return cookies
else:
print("Cookie缺少必要字段!需要SESSDATA、bili_jct和DedeUserID")
return {}
def save_cookies(cookies: Dict):
"""保存cookies为Netscape格式"""
with open('cookies.txt', 'w', encoding='utf-8') as f:
# 写入Netscape cookies文件头
f.write("# Netscape HTTP Cookie File\n")
f.write("# https://curl.haxx.se/rfc/cookie_spec.html\n")
f.write("# This is a generated file! Do not edit.\n\n")
# 为每个cookie写入Netscape格式的行
for name, value in cookies.items():
# 特殊处理SESSDATA,添加HttpOnly和Secure标记
if name == 'SESSDATA':
f.write(f"#HttpOnly_.bilibili.com\tTRUE\t/\tTRUE\t1735689600\t{name}\t{value}\n")
else:
f.write(f".bilibili.com\tTRUE\t/\tFALSE\t1735689600\t{name}\t{value}\n")
print(f"cookies已保存到cookies.txt,包含以下字段:{', '.join(cookies.keys())}")
def find_ffmpeg() -> str:
"""查找ffmpeg可执行文件的路径"""
try:
# 直接检查系统是否能执行ffmpeg命令
result = subprocess.run(['ffmpeg', '-version'],
capture_output=True,
text=True)
if result.returncode == 0:
print("找到系统ffmpeg")
return 'ffmpeg' # 直接返回命令名称
except Exception:
pass
# 如果直接执行失败,尝试完整路径
ffmpeg_path = r'C:\Program Files\ffmpeg\bin\ffmpeg.exe'
if os.path.isfile(ffmpeg_path):
print(f"找到ffmpeg: {ffmpeg_path}")
return ffmpeg_path
print("警告:未找到ffmpeg,请确保ffmpeg已正确安装并添加到系统PATH中")
return ''
def convert_to_mp3(input_file: str) -> bool:
"""将音频文件转换为MP3格式"""
if not os.path.exists(input_file):
print(f"错误:找不到文件 {input_file}")
return False
output_file = os.path.splitext(input_file)[0] + '.mp3'
try:
# 构建ffmpeg命令
cmd = [
'ffmpeg',
'-i', input_file,
'-acodec', 'libmp3lame',
'-q:a', '0', # 最高质量
'-threads', '4', # 使用4个线程
'-progress', 'pipe:1', # 输出进度信息到stdout
'-nostats', # 不输出额外统计信息
output_file
]
print(f"正在转换为MP3: {os.path.basename(input_file)}")
# 创建进度条
with tqdm(total=100,
desc="转换进度",
ncols=80,
bar_format='{desc}: {percentage:3.1f}%|{bar}| [{elapsed}<{remaining}]') as pbar:
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True
)
# 读取并更新进度
last_percent = 0
while True:
line = process.stdout.readline()
if not line and process.poll() is not None:
break
# 从输出中提取时间信息
if line.startswith('out_time_ms='):
time_ms = int(line.split('=')[1])
# 假设每秒处理速度相同,使用已处理时间估算进度
percent = min(100, max(0, int(time_ms / 10000000))) # 假设平均时长100秒
# 更新进度条
if percent > last_percent:
pbar.update(percent - last_percent)
last_percent = percent
# 确保进度条到达100%
if last_percent < 100:
pbar.update(100 - last_percent)
# 检查转换结果
if process.returncode == 0:
print(f"\n转换完成:{os.path.basename(output_file)}")
# 询问是否删除原始文件
if input("是否删除原始音频文件?(y/n): ").lower() == 'y':
os.remove(input_file)
print("原始文件已删除")
return True
else:
error = process.stderr.read()
print(f"转换失败!错误信息:\n{error}")
return False
except Exception as e:
print(f"转换过程中发生错误: {str(e)}")
return False
def download_audio(url: str, cookies: Dict):
"""下载音频"""
save_cookies(cookies)
try:
cmd = [
'yt-dlp',
'--cookies', 'cookies.txt',
'-f', 'ba',
'--no-playlist',
'--no-check-certificates',
'--progress',
'--newline',
]
cmd.append(url)
print("开始下载音频...")
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding='utf-8',
errors='replace',
bufsize=1,
universal_newlines=True
)
downloaded_file = None
pbar = None
total_size = None
# 用于解析下载进度的正则表达式
progress_pattern = re.compile(r'\[download\]\s+(\d+\.\d+)%\s+of\s+(\d+\.\d+)([KMG])iB\s+at\s+.*?ETA\s+(.*)')
while True:
output = process.stdout.readline()
if output == '' and process.poll() is not None:
break
if output:
output = output.strip()
# 获取文件名
if '[download] Destination:' in output:
downloaded_file = output.split('[download] Destination:', 1)[1].strip()
elif 'has already been downloaded' in output:
downloaded_file = output.split('[download] ', 1)[1].split(' has already', 1)[0].strip()
# 解析进度信息
match = progress_pattern.search(output)
if match:
percentage, size, unit, eta = match.groups()
# 计算总大小(转换为字节)
if total_size is None:
multiplier = {'K': 1024, 'M': 1024**2, 'G': 1024**3}
total_size = int(float(size) * multiplier[unit])
# 创建进度条
pbar = tqdm(
total=total_size,
unit='B',
unit_scale=True,
desc="下载进度",
ncols=80,
bar_format='{desc}: {percentage:3.1f}%|{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]'
)
# 更新进度条
if pbar:
current = int(total_size * float(percentage) / 100)
pbar.n = current
pbar.refresh()
# 其他输出信息
elif not output.startswith('[download]'):
print(output)
# 关闭进度条
if pbar:
pbar.close()
# 检查下载结果
if process.returncode == 0:
print("\n下载完成!")
if downloaded_file and os.path.exists(downloaded_file):
print(f"找到音频文件: {downloaded_file}")
if input("是否转换为MP3格式?(y/n): ").lower() == 'y':
if find_ffmpeg():
convert_to_mp3(downloaded_file)
else:
print("未找到ffmpeg,无法转换为MP3格式")
else:
print(f"警告:无法找到下载的文件")
m4a_files = [f for f in os.listdir('.') if f.endswith('.m4a')]
if m4a_files:
print(f"在当前目录找到以下.m4a文件:")
for i, file in enumerate(m4a_files, 1):
print(f"{i}. {file}")
choice = input("请选择要转换的文件编号(输入q取消):")
if choice.isdigit() and 1 <= int(choice) <= len(m4a_files):
if find_ffmpeg():
convert_to_mp3(m4a_files[int(choice)-1])
else:
print("未找到ffmpeg,无法转换为MP3格式")
else:
error = process.stderr.read()
print(f"下载失败!错误信息:\n{error}")
except FileNotFoundError:
print("错误:请先安装 yt-dlp")
print("可以使用以下命令安装:")
print("pip install yt-dlp")
except Exception as e:
print(f"发生错误: {str(e)}")
def main():
# 先尝试从文件加载cookies
cookies = load_cookies_from_file()
# 如果没有找到有效的cookies文件,则请求用户输入
if not cookies:
cookies = get_cookies_from_user()
if not cookies:
return
# 保存用户输入的cookies
save_cookies(cookies)
while True:
print("\n请输入B站视频URL (输入q退出):")
url = input().strip()
if url.lower() == 'q':
break
if not url.startswith(('https://www.bilibili.com', 'https://b23.tv')):
print("请输入有效的B站视频链接!")
continue
download_audio(url, cookies)
if __name__ == "__main__":
main()
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
+367
View File
@@ -0,0 +1,367 @@
import json
import os
import subprocess
from typing import Dict
from tqdm import tqdm
import re
def convert_browser_cookies(browser_cookies: list) -> Dict:
"""转换浏览器导出的cookies格式为简单的键值对"""
essential_cookies = {}
required_keys = ['SESSDATA', 'bili_jct', 'DedeUserID']
for cookie in browser_cookies:
if cookie['name'] in required_keys:
essential_cookies[cookie['name']] = cookie['value']
return essential_cookies
def load_cookies_from_file() -> Dict:
"""从文件加载cookies"""
cookie_files = ['cookies.txt', 'bilibili_cookies.json']
required_keys = ['SESSDATA', 'bili_jct', 'DedeUserID']
for cookie_file in cookie_files:
if os.path.exists(cookie_file):
try:
with open(cookie_file, 'r', encoding='utf-8') as f:
cookies = {}
# 逐行读取文件
for line in f:
line = line.strip()
# 跳过空行和注释行(但不跳过#HttpOnly_开头的行)
if not line or (line.startswith('#') and not line.startswith('#HttpOnly_')):
continue
try:
# 处理HttpOnly标记
if line.startswith('#HttpOnly_'):
line = line[9:] # 移除'#HttpOnly_'前缀
# 分割字段并移除可能的空字段
fields = [f for f in line.split('\t') if f]
if len(fields) >= 7:
name = fields[5]
value = fields[6]
if name in required_keys:
# 对于SESSDATA,需要处理URL编码
if name == 'SESSDATA' and '%2C' in value:
value = value.replace('%2C', ',')
cookies[name] = value
print(f"找到cookie: {name}") # 调试信息
except Exception as e:
print(f"警告: 解析行 '{line}' 时出错: {str(e)}")
continue
# 检查是否找到所有必需的cookies
if all(key in cookies for key in required_keys):
print(f"已从 {cookie_file} 加载cookies")
return cookies
else:
missing = [key for key in required_keys if key not in cookies]
print(f"警告: {cookie_file} 中缺少以下cookie字段: {', '.join(missing)}")
except Exception as e:
print(f"警告: 读取 {cookie_file} 时出错: {str(e)}")
return {}
def parse_cookie_string(cookie_str: str) -> Dict:
"""解析浏览器直接复制的cookie字符串"""
cookies = {}
required_keys = ['SESSDATA', 'bili_jct', 'DedeUserID']
try:
# 分割cookie字符串
cookie_pairs = cookie_str.split(';')
for pair in cookie_pairs:
if '=' in pair:
name, value = pair.strip().split('=', 1)
if name in required_keys:
cookies[name] = value
except Exception:
return {}
return cookies
def get_cookies_from_user() -> Dict:
"""获取用户输入的cookies"""
print("请输入B站cookies (支持JSON格式、浏览器导出格式或直接复制的cookie字符串):")
cookies_str = input().strip()
# 首先尝试解析为JSON
try:
cookies_data = json.loads(cookies_str)
if isinstance(cookies_data, list):
cookies = convert_browser_cookies(cookies_data)
else:
cookies = cookies_data
except json.JSONDecodeError:
# 如果不是JSON,尝试解析为cookie字符串
cookies = parse_cookie_string(cookies_str)
# 验证必要的cookie字段
if all(key in cookies for key in ['SESSDATA', 'bili_jct', 'DedeUserID']):
return cookies
else:
print("Cookie缺少必要字段!需要SESSDATA、bili_jct和DedeUserID")
return {}
def save_cookies(cookies: Dict):
"""保存cookies为Netscape格式"""
with open('cookies.txt', 'w', encoding='utf-8') as f:
# 写入Netscape cookies文件头
f.write("# Netscape HTTP Cookie File\n")
f.write("# https://curl.haxx.se/rfc/cookie_spec.html\n")
f.write("# This is a generated file! Do not edit.\n\n")
# 为每个cookie写入Netscape格式的行
for name, value in cookies.items():
# 特殊处理SESSDATA,添加HttpOnly和Secure标记
if name == 'SESSDATA':
f.write(f"#HttpOnly_.bilibili.com\tTRUE\t/\tTRUE\t1735689600\t{name}\t{value}\n")
else:
f.write(f".bilibili.com\tTRUE\t/\tFALSE\t1735689600\t{name}\t{value}\n")
print(f"cookies已保存到cookies.txt,包含以下字段:{', '.join(cookies.keys())}")
def find_ffmpeg() -> str:
"""查找ffmpeg可执行文件的路径"""
try:
# 直接检查系统是否能执行ffmpeg命令
result = subprocess.run(['ffmpeg', '-version'],
capture_output=True,
text=True)
if result.returncode == 0:
print("找到系统ffmpeg")
return 'ffmpeg' # 直接返回命令名称
except Exception:
pass
# 如果直接执行失败,尝试完整路径
ffmpeg_path = r'C:\Program Files\ffmpeg\bin\ffmpeg.exe'
if os.path.isfile(ffmpeg_path):
print(f"找到ffmpeg: {ffmpeg_path}")
return ffmpeg_path
print("警告:未找到ffmpeg,请确保ffmpeg已正确安装并添加到系统PATH中")
return ''
def convert_to_mp3(input_file: str) -> bool:
"""将音频文件转换为MP3格式"""
if not os.path.exists(input_file):
print(f"错误:找不到文件 {input_file}")
return False
output_file = os.path.splitext(input_file)[0] + '.mp3'
try:
# 构建ffmpeg命令
cmd = [
'ffmpeg',
'-i', input_file,
'-acodec', 'libmp3lame',
'-q:a', '0', # 最高质量
'-threads', '4', # 使用4个线程
'-progress', 'pipe:1', # 输出进度信息到stdout
'-nostats', # 不输出额外统计信息
output_file
]
print(f"正在转换为MP3: {os.path.basename(input_file)}")
# 创建进度条
with tqdm(total=100,
desc="转换进度",
ncols=80,
bar_format='{desc}: {percentage:3.1f}%|{bar}| [{elapsed}<{remaining}]') as pbar:
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True
)
# 读取并更新进度
last_percent = 0
while True:
line = process.stdout.readline()
if not line and process.poll() is not None:
break
# 从输出中提取时间信息
if line.startswith('out_time_ms='):
time_ms = int(line.split('=')[1])
# 假设每秒处理速度相同,使用已处理时间估算进度
percent = min(100, max(0, int(time_ms / 10000000))) # 假设平均时长100秒
# 更新进度条
if percent > last_percent:
pbar.update(percent - last_percent)
last_percent = percent
# 确保进度条到达100%
if last_percent < 100:
pbar.update(100 - last_percent)
# 检查转换结果
if process.returncode == 0:
print(f"\n转换完成:{os.path.basename(output_file)}")
# 询问是否删除原始文件
if input("是否删除原始音频文件?(y/n): ").lower() == 'y':
os.remove(input_file)
print("原始文件已删除")
return True
else:
error = process.stderr.read()
print(f"转换失败!错误信息:\n{error}")
return False
except Exception as e:
print(f"转换过程中发生错误: {str(e)}")
return False
def download_audio(url: str, cookies: Dict):
"""下载音频"""
save_cookies(cookies)
try:
cmd = [
'yt-dlp',
'--cookies', 'cookies.txt',
'-f', 'ba',
'--no-playlist',
'--no-check-certificates',
'--progress',
'--newline',
]
cmd.append(url)
print("开始下载音频...")
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding='utf-8',
errors='replace',
bufsize=1,
universal_newlines=True
)
downloaded_file = None
pbar = None
total_size = None
# 用于解析下载进度的正则表达式
progress_pattern = re.compile(r'\[download\]\s+(\d+\.\d+)%\s+of\s+(\d+\.\d+)([KMG])iB\s+at\s+.*?ETA\s+(.*)')
while True:
output = process.stdout.readline()
if output == '' and process.poll() is not None:
break
if output:
output = output.strip()
# 获取文件名
if '[download] Destination:' in output:
downloaded_file = output.split('[download] Destination:', 1)[1].strip()
elif 'has already been downloaded' in output:
downloaded_file = output.split('[download] ', 1)[1].split(' has already', 1)[0].strip()
# 解析进度信息
match = progress_pattern.search(output)
if match:
percentage, size, unit, eta = match.groups()
# 计算总大小(转换为字节)
if total_size is None:
multiplier = {'K': 1024, 'M': 1024**2, 'G': 1024**3}
total_size = int(float(size) * multiplier[unit])
# 创建进度条
pbar = tqdm(
total=total_size,
unit='B',
unit_scale=True,
desc="下载进度",
ncols=80,
bar_format='{desc}: {percentage:3.1f}%|{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]'
)
# 更新进度条
if pbar:
current = int(total_size * float(percentage) / 100)
pbar.n = current
pbar.refresh()
# 其他输出信息
elif not output.startswith('[download]'):
print(output)
# 关闭进度条
if pbar:
pbar.close()
# 检查下载结果
if process.returncode == 0:
print("\n下载完成!")
if downloaded_file and os.path.exists(downloaded_file):
print(f"找到音频文件: {downloaded_file}")
if input("是否转换为MP3格式?(y/n): ").lower() == 'y':
if find_ffmpeg():
convert_to_mp3(downloaded_file)
else:
print("未找到ffmpeg,无法转换为MP3格式")
else:
print(f"警告:无法找到下载的文件")
m4a_files = [f for f in os.listdir('.') if f.endswith('.m4a')]
if m4a_files:
print(f"在当前目录找到以下.m4a文件:")
for i, file in enumerate(m4a_files, 1):
print(f"{i}. {file}")
choice = input("请选择要转换的文件编号(输入q取消):")
if choice.isdigit() and 1 <= int(choice) <= len(m4a_files):
if find_ffmpeg():
convert_to_mp3(m4a_files[int(choice)-1])
else:
print("未找到ffmpeg,无法转换为MP3格式")
else:
error = process.stderr.read()
print(f"下载失败!错误信息:\n{error}")
except FileNotFoundError:
print("错误:请先安装 yt-dlp")
print("可以使用以下命令安装:")
print("pip install yt-dlp")
except Exception as e:
print(f"发生错误: {str(e)}")
def main():
# 先尝试从文件加载cookies
cookies = load_cookies_from_file()
# 如果没有找到有效的cookies文件,则请求用户输入
if not cookies:
cookies = get_cookies_from_user()
if not cookies:
return
# 保存用户输入的cookies
save_cookies(cookies)
while True:
print("\n请输入B站视频URL (输入q退出):")
url = input().strip()
if url.lower() == 'q':
break
if not url.startswith(('https://www.bilibili.com', 'https://b23.tv')):
print("请输入有效的B站视频链接!")
continue
download_audio(url, cookies)
if __name__ == "__main__":
main()
+760
View File
@@ -0,0 +1,760 @@
import json
import os
import subprocess
from typing import Dict
from tqdm import tqdm
import re
import requests
import time
def convert_browser_cookies(browser_cookies: list) -> Dict:
"""转换浏览器导出的cookies格式为简单的键值对"""
cookies = {}
for cookie in browser_cookies:
if isinstance(cookie, dict) and 'name' in cookie and 'value' in cookie:
cookies[cookie['name']] = cookie['value']
return cookies
def load_cookies_from_file() -> Dict:
"""从文件加载cookies"""
cookie_files = ['cookies.txt', 'bilibili_cookies.json']
for cookie_file in cookie_files:
if os.path.exists(cookie_file):
try:
with open(cookie_file, 'r', encoding='utf-8') as f:
content = f.read().strip()
cookies = {}
# 尝试作为JSON解析
try:
json_data = json.loads(content)
if isinstance(json_data, list):
cookies = convert_browser_cookies(json_data)
elif isinstance(json_data, dict):
cookies = json_data
except json.JSONDecodeError:
# 如果不是JSON,尝试作为Netscape格式或普通cookie字符串解析
if content.startswith('# Netscape HTTP Cookie File'):
# Netscape格式
for line in content.split('\n'):
line = line.strip()
if not line or line.startswith('#'):
continue
try:
fields = line.split('\t')
if len(fields) >= 7:
name, value = fields[5:7]
cookies[name] = value
except:
continue
else:
# 尝试作为普通cookie字符串解析
cookies = parse_cookie_string(content)
if cookies:
print(f"已从 {cookie_file} 加载cookies:")
for name, value in cookies.items():
print(f"找到cookie: {name}")
return cookies
else:
print(f"警告: 无法从 {cookie_file} 加载有效的cookies")
except Exception as e:
print(f"警告: 读取 {cookie_file} 时出错: {str(e)}")
return {}
def parse_cookie_string(cookie_str: str) -> Dict:
"""解析浏览器直接复制的cookie字符串"""
cookies = {}
try:
# 处理URL编码
def decode_value(value):
# 处理特殊的URL编码字符
replacements = {
'%2C': ',',
'%2F': '/',
'%3A': ':',
'%2B': '+',
'%3D': '=',
'%3B': ';'
}
for encoded, decoded in replacements.items():
value = value.replace(encoded, decoded)
return value
# 分割cookie字符串
cookie_pairs = cookie_str.split(';')
for pair in cookie_pairs:
if '=' in pair:
name, value = pair.strip().split('=', 1)
name = name.strip()
value = decode_value(value.strip())
cookies[name] = value
# 如果cookies非空,打印找到的cookie数量
if cookies:
print(f"字符串中解析出 {len(cookies)} 个cookies")
except Exception as e:
print(f"解析cookie字符串时出错: {str(e)}")
return {}
return cookies
def get_cookies_from_user() -> Dict:
"""获取用户输入的cookies"""
print("请输入B站cookies (支持JSON格式、浏览器导出格式或直接复制的cookie字符串):")
cookies_str = input().strip()
# 首先尝试解析JSON
try:
cookies_data = json.loads(cookies_str)
if isinstance(cookies_data, list):
cookies = convert_browser_cookies(cookies_data)
else:
cookies = cookies_data
except json.JSONDecodeError:
# 如果不是JSON,尝试解析为cookie字符串
cookies = parse_cookie_string(cookies_str)
# 检查是否获取到cookies
if cookies:
print("成功获取到以下cookies:")
for name in cookies.keys():
print(f"- {name}")
return cookies
else:
print("未能获取到有效的cookies")
return {}
def save_cookies(cookies: Dict):
"""保存cookies为Netscape格式"""
with open('cookies.txt', 'w', encoding='utf-8') as f:
f.write("# Netscape HTTP Cookie File\n")
f.write("# https://curl.haxx.se/rfc/cookie_spec.html\n")
f.write("# This is a generated file! Do not edit.\n\n")
# 为所有cookie写入Netscape格式的行
for name, value in cookies.items():
# 某些特殊cookie需要特殊处理
if name in ['SESSDATA']:
f.write(f"#HttpOnly_.bilibili.com\tTRUE\t/\tTRUE\t1735689600\t{name}\t{value}\n")
else:
f.write(f".bilibili.com\tTRUE\t/\tFALSE\t1735689600\t{name}\t{value}\n")
print(f"cookies已保存到cookies.txt,包含以下字段:{', '.join(cookies.keys())}")
def find_ffmpeg() -> str:
"""查找ffmpeg可执行文件的路径"""
try:
# 直接检查系统是否能执行ffmpeg命令
result = subprocess.run(['ffmpeg', '-version'],
capture_output=True,
text=True)
if result.returncode == 0:
print("找到系统ffmpeg")
return 'ffmpeg' # 直接返回命令名称
except Exception:
pass
# 如果直接执行失败,尝试完整路径
ffmpeg_path = r'C:\Program Files\ffmpeg\bin\ffmpeg.exe'
if os.path.isfile(ffmpeg_path):
print(f"找到ffmpeg: {ffmpeg_path}")
return ffmpeg_path
print("警告:未找到ffmpeg,请确保ffmpeg已正确安装并添加到系统PATH中")
return ''
def convert_to_mp3(input_file: str) -> bool:
"""将音频文件转换为MP3格式"""
if not os.path.exists(input_file):
print(f"错误:找不到文件 {input_file}")
return False
output_file = os.path.splitext(input_file)[0] + '.mp3'
try:
# 构建ffmpeg命令
cmd = [
'ffmpeg',
'-i', input_file,
'-acodec', 'libmp3lame',
'-q:a', '0', # 最高质量
'-threads', '4', # 使用4个线程
'-progress', 'pipe:1', # 输出进度信息到stdout
'-nostats', # 不输出额外统计信息
output_file
]
print(f"正在转换为MP3: {os.path.basename(input_file)}")
# 创建进度条
with tqdm(total=100,
desc="转换进度",
ncols=80,
bar_format='{desc}: {percentage:3.1f}%|{bar}| [{elapsed}<{remaining}]') as pbar:
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True
)
# 读取并更新进
last_percent = 0
while True:
line = process.stdout.readline()
if not line and process.poll() is not None:
break
# 从输出中提取时间信息
if line.startswith('out_time_ms='):
time_ms = int(line.split('=')[1])
# 假设每秒处理速度相同,使用已处理时间估算进度
percent = min(100, max(0, int(time_ms / 10000000))) # 假设平均时长100秒
# 更新进度条
if percent > last_percent:
pbar.update(percent - last_percent)
last_percent = percent
# 确保进度条到达100%
if last_percent < 100:
pbar.update(100 - last_percent)
# 检查转换结果
if process.returncode == 0:
print(f"\n转换完成:{os.path.basename(output_file)}")
# 询问是否删除原始文件
choice = input("是否删除原始音频文件?(y/n) [y]: ").lower()
if not choice or choice == 'y': # 直接回车或输入y都删除
os.remove(input_file)
print("原始文件已删除")
return True
else:
error = process.stderr.read()
print(f"转换失败!错误信息:\n{error}")
return False
except Exception as e:
print(f"转换过程中发生错误: {str(e)}")
return False
def download_audio(url: str, cookies: Dict):
"""下载音频"""
save_cookies(cookies)
try:
cmd = [
'yt-dlp',
'--cookies', 'cookies.txt',
'-f', 'ba',
'--no-playlist',
'--no-check-certificates',
'--progress',
'--newline',
]
cmd.append(url)
print("开始下载音频...")
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding='utf-8',
errors='replace',
bufsize=1,
universal_newlines=True
)
downloaded_file = None
pbar = None
total_size = None
# 用于解析下载进度的正则表达式
progress_pattern = re.compile(r'\[download\]\s+(\d+\.\d+)%\s+of\s+(\d+\.\d+)([KMG])iB\s+at\s+.*?ETA\s+(.*)')
while True:
output = process.stdout.readline()
if output == '' and process.poll() is not None:
break
if output:
output = output.strip()
# 获取文件名
if '[download] Destination:' in output:
downloaded_file = output.split('[download] Destination:', 1)[1].strip()
elif 'has already been downloaded' in output:
downloaded_file = output.split('[download] ', 1)[1].split(' has already', 1)[0].strip()
# 解析进度信息
match = progress_pattern.search(output)
if match:
percentage, size, unit, eta = match.groups()
# 计算总大小(转换为字节)
if total_size is None:
multiplier = {'K': 1024, 'M': 1024**2, 'G': 1024**3}
total_size = int(float(size) * multiplier[unit])
# 创建进度条
pbar = tqdm(
total=total_size,
unit='B',
unit_scale=True,
desc="下载进度",
ncols=80,
bar_format='{desc}: {percentage:3.1f}%|{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]'
)
# 更新进度条
if pbar:
current = int(total_size * float(percentage) / 100)
pbar.n = current
pbar.refresh()
# 其他输出信息
elif not output.startswith('[download]'):
print(output)
# 关闭进度条
if pbar:
pbar.close()
# 检查下载结果
if process.returncode == 0:
print("\n下载完成!")
if downloaded_file and os.path.exists(downloaded_file):
print(f"找到音频文件: {downloaded_file}")
if input("是否转换为MP3格式?(y/n): ").lower() == 'y':
if find_ffmpeg():
convert_to_mp3(downloaded_file)
else:
print("未找到ffmpeg,无法转换为MP3格式")
else:
print(f"警告:无法找到下载的文件")
m4a_files = [f for f in os.listdir('.') if f.endswith('.m4a')]
if m4a_files:
print(f"在当前目录找到以下.m4a文件:")
for i, file in enumerate(m4a_files, 1):
print(f"{i}. {file}")
choice = input("请选择要转换的文件编号(输入q取消):")
if choice.isdigit() and 1 <= int(choice) <= len(m4a_files):
if find_ffmpeg():
convert_to_mp3(m4a_files[int(choice)-1])
else:
print("未找到ffmpeg,无法转换为MP3格式")
else:
error = process.stderr.read()
print(f"下载失败!错误信息:\n{error}")
except FileNotFoundError:
print("错误:请先安装 yt-dlp")
print("可以使用以下命令安装:")
print("pip install yt-dlp")
except Exception as e:
print(f"发生错误: {str(e)}")
def get_user_info(cookies: Dict) -> None:
"""获取并显示用户信息"""
print("\n" + "="*50)
print("# 用户信息")
print("="*50)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
try:
# 获取用户基本信息
myinfo_url = 'http://api.bilibili.com/x/space/myinfo'
myinfo_resp = requests.get(myinfo_url, cookies=cookies, headers=headers)
myinfo_data = myinfo_resp.json()
user_id = None
if myinfo_data['code'] == 0:
user_data = myinfo_data['data']
user_id = user_data['mid'] # 获取用户ID
print(f"用户名: {user_data['name']}")
print(f"用户ID: {user_id}")
print(f"等级: LV{user_data['level']}")
print(f"会员类型: {'大会员' if user_data['vip']['status'] == 1 else '普通用户'}")
print(f"性别: {user_data['sex']}")
print(f"签名: {user_data['sign']}")
# 获取关注和粉丝数
if user_id:
stat_url = f'https://api.bilibili.com/x/relation/stat?vmid={user_id}'
stat_resp = requests.get(stat_url, headers=headers)
stat_data = stat_resp.json()
if stat_data['code'] == 0:
stat_info = stat_data['data']
print(f"关注数: {stat_info['following']}")
print(f"粉丝数: {stat_info['follower']}")
# 获取硬币数
coin_url = 'http://account.bilibili.com/site/getCoin'
coin_resp = requests.get(coin_url, cookies=cookies, headers=headers)
coin_data = coin_resp.json()
if coin_data['code'] == 0:
print(f"硬币数: {coin_data['data']['money']}")
# 获取今日投币经验
coin_exp_url = 'https://api.bilibili.com/x/web-interface/coin/today/exp'
coin_exp_resp = requests.get(coin_exp_url, cookies=cookies, headers=headers)
coin_exp_data = coin_exp_resp.json()
if coin_exp_data['code'] == 0:
today_exp = coin_exp_data['data']
max_coin_exp = 50 # B站每日投币多获得50经验
print(f"今日投币经验: {today_exp}/{max_coin_exp}")
remaining_coins = (max_coin_exp - today_exp) // 10 # 每个硬币获得10经验
if remaining_coins > 0:
print(f"今日还可投币: {remaining_coins}个")
# 获取登录信息
nav_url = 'https://api.bilibili.com/x/web-interface/nav'
nav_resp = requests.get(nav_url, cookies=cookies, headers=headers)
nav_data = nav_resp.json()
if nav_data['code'] == 0:
nav_info = nav_data['data']
print(f"当前登录状态: {'已登录' if nav_info['isLogin'] else '未登录'}")
if nav_info['isLogin']:
print(f"邮箱: {nav_info['email']}")
print(f"硬币余额: {nav_info['money']}")
print(f"B币余额: {nav_info['wallet']['bcoin_balance']}")
except Exception as e:
print(f"获取用户信息时出错: {str(e)}")
print("="*50 + "\n")
def extract_bvid_from_file(filename: str) -> list:
"""从文件中取URL并提取BV号"""
bvids = []
try:
with open(filename, 'r', encoding='utf-8') as f:
for line in f:
url = line.strip()
if url:
# 使用正则表达式匹配BV号
match = re.search(r'BV[a-zA-Z0-9]{10}', url)
if match:
bvid = match.group()
bvids.append(bvid)
except Exception as e:
print(f"读取文件出错: {str(e)}")
return bvids
def batch_coin(cookies: Dict) -> None:
"""批量投币功能"""
print("\n" + "="*50)
print("# 批量投币")
print("="*50)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
try:
# 先检查今日投币情况
coin_exp_url = 'https://api.bilibili.com/x/web-interface/coin/today/exp'
coin_exp_resp = requests.get(coin_exp_url, cookies=cookies, headers=headers)
coin_exp_data = coin_exp_resp.json()
if coin_exp_data['code'] == 0:
today_exp = coin_exp_data['data']
max_coin_exp = 50
remaining_coins = (max_coin_exp - today_exp) // 10
if remaining_coins <= 0:
print("今日已完成投币任务!")
return
print(f"今日还可投币: {remaining_coins}个")
# 检查bvid.txt文件是否存在
if not os.path.exists('bvid.txt'):
print("错误:找不到bvid.txt文件!")
print("请创建bvid.txt文件并在其中每行写入一个视频链接")
return
# 从文件中读取并提取BV号
bvids = extract_bvid_from_file('bvid.txt')
if not bvids:
print("错误:无法从文件中提取有效的BV号!")
return
print(f"从文件中提取到 {len(bvids)} 个BV号:")
for bvid in bvids:
print(f"- {bvid}")
# 询问是否继续投币
choice = input("\n是否开始批量投币?(y/n) [y]: ").lower()
if not choice or choice == 'y': # 直接回车或输入y都继续
# 获取csrf令牌
csrf = cookies.get('bili_jct')
if not csrf:
print("错误:无法获取csrf令牌!")
return
# 投币API
coin_url = 'https://api.bilibili.com/x/web-interface/coin/add'
# 创建进度条
with tqdm(total=min(remaining_coins, len(bvids)), desc="投币进度") as pbar:
for bvid in bvids:
if remaining_coins <= 0:
print("\n今日投币任务已完成!")
break
# 构造请求数据
data = {
'bvid': bvid,
'multiply': 1, # 投1个币
'select_like': 1, # 同时点赞
'csrf': csrf
}
try:
# 发送投币请求
response = requests.post(coin_url, data=data, cookies=cookies, headers=headers)
result = response.json()
if result['code'] == 0:
print(f"\n成功给视频 {bvid} 投币")
remaining_coins -= 1
pbar.update(1)
else:
print(f"\n给视频 {bvid} 投币失败:{result['message']}")
print("停止投币操作")
break # 投币失败就停止
# 添加延时,避免请求过快
time.sleep(2)
except Exception as e:
print(f"\n投币过程出错: {str(e)}")
print("停止投币操作")
break # 出错也停止
print("\n批量投币完成!")
# 获取当前硬币数
nav_url = 'https://api.bilibili.com/x/web-interface/nav'
nav_resp = requests.get(nav_url, cookies=cookies, headers=headers)
nav_data = nav_resp.json()
if nav_data['code'] == 0:
current_coins = nav_data['data']['money']
print(f"当前剩余硬币数: {current_coins}")
except Exception as e:
print(f"批量投币时出错: {str(e)}")
print("="*50 + "\n")
def show_menu() -> str:
"""显示功能菜单"""
print("\n" + "="*50)
print("B站工具箱")
print("="*50)
print("1. 批量点赞 (默认)")
print("2. 批量投币")
print("3. 下载视频音频")
print("4. 获取视频BV号")
print("5. 查看用户信息")
print("6. 退出程序")
print("="*50)
choice = input("请选择功能 (1-6) [1]: ").strip()
if not choice: # 如果直接回车,返回默认值'1'
return '1'
if choice in ['1', '2', '3', '4', '5', '6']:
return choice
print("无效的选择,使用默认选项1")
return '1'
def extract_bvid(url: str) -> str:
"""从URL中提取BV号"""
# 匹配BV号的正则表达式
bv_pattern = r'BV[a-zA-Z0-9]{10}'
match = re.search(bv_pattern, url)
if match:
return match.group()
return ''
def get_video_bvids() -> None:
"""获取视频BV号并保存到文件"""
bvids = []
print("请输入B站视频URL,每行一个,输入q结束:")
while True:
url = input().strip()
if url.lower() == 'q':
break
if not url.startswith(('https://www.bilibili.com', 'https://b23.tv')):
print("请输入有效的B站视频链接!")
continue
bvid = extract_bvid(url)
if bvid:
bvids.append(bvid)
print(f"已添加: {bvid}")
else:
print("无法从URL中提取BV号!")
if bvids:
with open('bvid.txt', 'w', encoding='utf-8') as f:
f.write('\n'.join(bvids))
print(f"已保存 {len(bvids)} 个BV号到 bvid.txt")
def batch_like(cookies: Dict) -> None:
"""批量点赞功能"""
print("\n" + "="*50)
print("# 批量点赞")
print("="*50)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
try:
# 检查bvid.txt文件是否存在
if not os.path.exists('bvid.txt'):
print("错误:找不到bvid.txt文件!")
print("请创建bvid.txt文件并在其中每行写入一个视频链接")
return
# 从文件中读取并提取BV号
bvids = extract_bvid_from_file('bvid.txt')
if not bvids:
print("错误:无法从文件中提取有效的BV号!")
return
print(f"从文件中提取到 {len(bvids)} 个BV号:")
for bvid in bvids:
print(f"- {bvid}")
# 询问是否继续点赞
choice = input("\n是否开始批量点赞?(y/n) [y]: ").lower()
if not choice or choice == 'y': # 直接回车或输入y都继续
# 获取csrf令牌
csrf = cookies.get('bili_jct')
if not csrf:
print("错误:无法获取csrf令牌!")
return
# 点赞API
like_url = 'https://api.bilibili.com/x/web-interface/archive/like'
# 创建进度条
with tqdm(total=len(bvids), desc="点赞进度") as pbar:
for bvid in bvids:
# 构造请求数据
data = {
'bvid': bvid,
'like': 1, # 1表示点赞,2表示取消点赞
'csrf': csrf
}
try:
# 发送点赞请求
response = requests.post(like_url, data=data, cookies=cookies, headers=headers)
result = response.json()
if result['code'] == 0:
print(f"\n成功给视频 {bvid} 点赞")
pbar.update(1)
else:
print(f"\n给视频 {bvid} 点赞失败:{result['message']}")
print("停止点赞操作")
break # 点赞失败就停止
# 添加延时,避免请求过快
time.sleep(2)
except Exception as e:
print(f"\n点赞过程出错: {str(e)}")
print("停止点赞操作")
break # 出错也停止
print("\n批量点赞完成!")
except Exception as e:
print(f"批量点赞时出错: {str(e)}")
print("="*50 + "\n")
def main():
# 先尝试从文件加载cookies
cookies = load_cookies_from_file()
# 如果没有找到有效的cookies文件,则请求用户输入
if not cookies:
cookies = get_cookies_from_user()
if not cookies:
return
# 保存用户输入的cookies
save_cookies(cookies)
# 获取并显示用户信息
get_user_info(cookies)
while True:
choice = show_menu()
if choice == '1':
# 批量点赞
batch_like(cookies)
elif choice == '2':
# 批量投币
batch_coin(cookies)
elif choice == '3':
# 下载视频音频
while True:
print("\n请输入B站视频URL (输入q返回主菜单):")
url = input().strip()
if url.lower() == 'q':
break
if not url.startswith(('https://www.bilibili.com', 'https://b23.tv')):
print("请输入有效的B站视频链接!")
continue
download_audio(url, cookies)
elif choice == '4':
# 获取视频BV号
get_video_bvids()
elif choice == '5':
# 查看用户信息
get_user_info(cookies)
elif choice == '6':
print("\n感谢使用!再见!")
break
if __name__ == "__main__":
main()
+998
View File
@@ -0,0 +1,998 @@
import json
import os
import subprocess
from typing import Dict
from tqdm import tqdm
import re
import requests
import time
def convert_browser_cookies(browser_cookies: list) -> Dict:
"""转换浏览器导出的cookies格式为简单的键值对"""
cookies = {}
for cookie in browser_cookies:
if isinstance(cookie, dict) and 'name' in cookie and 'value' in cookie:
cookies[cookie['name']] = cookie['value']
return cookies
def load_cookies_from_file() -> Dict:
"""从文件加载cookies"""
cookie_files = ['cookies.txt', 'bilibili_cookies.json']
for cookie_file in cookie_files:
if os.path.exists(cookie_file):
try:
with open(cookie_file, 'r', encoding='utf-8') as f:
content = f.read().strip()
cookies = {}
# 尝试作为JSON解析
try:
json_data = json.loads(content)
if isinstance(json_data, list):
cookies = convert_browser_cookies(json_data)
elif isinstance(json_data, dict):
cookies = json_data
except json.JSONDecodeError:
# 如果不是JSON,尝试作为Netscape格式或普通cookie字符串解析
if content.startswith('# Netscape HTTP Cookie File'):
# Netscape格式
for line in content.split('\n'):
line = line.strip()
if not line or line.startswith('#'):
continue
try:
fields = line.split('\t')
if len(fields) >= 7:
name, value = fields[5:7]
cookies[name] = value
except:
continue
else:
# 尝试作为普通cookie字符串解析
cookies = parse_cookie_string(content)
if cookies:
print(f"已从 {cookie_file} 加载cookies:")
for name, value in cookies.items():
print(f"找到cookie: {name}")
return cookies
else:
print(f"警告: 无法从 {cookie_file} 加载有效的cookies")
except Exception as e:
print(f"警告: 读取 {cookie_file} 时出错: {str(e)}")
return {}
def parse_cookie_string(cookie_str: str) -> Dict:
"""解析浏览器直接复制的cookie字符串"""
cookies = {}
try:
# 处理URL编码
def decode_value(value):
# 处理特殊的URL编码字符
replacements = {
'%2C': ',',
'%2F': '/',
'%3A': ':',
'%2B': '+',
'%3D': '=',
'%3B': ';'
}
for encoded, decoded in replacements.items():
value = value.replace(encoded, decoded)
return value
# 分割cookie字符串
cookie_pairs = cookie_str.split(';')
for pair in cookie_pairs:
if '=' in pair:
name, value = pair.strip().split('=', 1)
name = name.strip()
value = decode_value(value.strip())
cookies[name] = value
# 如果cookies非空,打印找到的cookie数量
if cookies:
print(f"字符串中解析出 {len(cookies)} 个cookies")
except Exception as e:
print(f"解析cookie字符串时出错: {str(e)}")
return {}
return cookies
def get_cookies_from_user() -> Dict:
"""获取用户输入的cookies"""
print("请输入B站cookies (支持JSON格式、浏览器导出格式或直接复制的cookie字符串):")
cookies_str = input().strip()
# 首先尝试解析JSON
try:
cookies_data = json.loads(cookies_str)
if isinstance(cookies_data, list):
cookies = convert_browser_cookies(cookies_data)
else:
cookies = cookies_data
except json.JSONDecodeError:
# 如果不是JSON,尝试解析为cookie字符串
cookies = parse_cookie_string(cookies_str)
# 检查是否获取到cookies
if cookies:
print("成功获取到以下cookies:")
for name in cookies.keys():
print(f"- {name}")
return cookies
else:
print("未能获取到有效的cookies")
return {}
def save_cookies(cookies: Dict):
"""保存cookies为Netscape格式"""
with open('cookies.txt', 'w', encoding='utf-8') as f:
f.write("# Netscape HTTP Cookie File\n")
f.write("# https://curl.haxx.se/rfc/cookie_spec.html\n")
f.write("# This is a generated file! Do not edit.\n\n")
# 为所有cookie写入Netscape格式的行
for name, value in cookies.items():
# 某些特殊cookie需要特殊处理
if name in ['SESSDATA']:
f.write(f"#HttpOnly_.bilibili.com\tTRUE\t/\tTRUE\t1735689600\t{name}\t{value}\n")
else:
f.write(f".bilibili.com\tTRUE\t/\tFALSE\t1735689600\t{name}\t{value}\n")
print(f"cookies已保存到cookies.txt,包含以下字段:{', '.join(cookies.keys())}")
def find_ffmpeg() -> str:
"""查找ffmpeg可执行文件的路径"""
try:
# 直接检查系统是否能执行ffmpeg命令
result = subprocess.run(['ffmpeg', '-version'],
capture_output=True,
text=True)
if result.returncode == 0:
print("找到系统ffmpeg")
return 'ffmpeg' # 直接返回命令名称
except Exception:
pass
# 如果直接执行失败,尝试完整路径
ffmpeg_path = r'C:\Program Files\ffmpeg\bin\ffmpeg.exe'
if os.path.isfile(ffmpeg_path):
print(f"找到ffmpeg: {ffmpeg_path}")
return ffmpeg_path
print("警告:未找到ffmpeg,请确保ffmpeg已正确安装添加到系统PATH中")
return ''
def convert_to_mp3(input_file: str) -> bool:
"""将音频文件转换为MP3格式"""
if not os.path.exists(input_file):
print(f"错误:找不到文件 {input_file}")
return False
output_file = os.path.splitext(input_file)[0] + '.mp3'
try:
# 构建ffmpeg命令
cmd = [
'ffmpeg',
'-i', input_file,
'-acodec', 'libmp3lame',
'-q:a', '0', # 最高质量
'-threads', '4', # 使用4个线程
'-progress', 'pipe:1', # 输出进度信息到stdout
'-nostats', # 不输出额外统计信息
output_file
]
print(f"正在转换为MP3: {os.path.basename(input_file)}")
# 创建进度条
with tqdm(total=100,
desc="转换进度",
ncols=80,
bar_format='{desc}: {percentage:3.1f}%|{bar}| [{elapsed}<{remaining}]') as pbar:
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True
)
# 读取并更新进
last_percent = 0
while True:
line = process.stdout.readline()
if not line and process.poll() is not None:
break
# 从输出中提取时间信息
if line.startswith('out_time_ms='):
time_ms = int(line.split('=')[1])
# 假设每秒处理速度相同,使用已处理时间估算进度
percent = min(100, max(0, int(time_ms / 10000000))) # 假设平均时长100秒
# 更新进度条
if percent > last_percent:
pbar.update(percent - last_percent)
last_percent = percent
# 确保进度条到达100%
if last_percent < 100:
pbar.update(100 - last_percent)
# 检查转换结果
if process.returncode == 0:
print(f"\n转换完成:{os.path.basename(output_file)}")
# 询问是否删除原始文件
choice = input("是否删除原始音频文件?(y/n) [y]: ").lower()
if not choice or choice == 'y': # 直接回车或输入y都删除
os.remove(input_file)
print("原始文件已删除")
return True
else:
error = process.stderr.read()
print(f"转换失败!错误信息:\n{error}")
return False
except Exception as e:
print(f"转换过程中发生错误: {str(e)}")
return False
def download_audio(url: str, cookies: Dict):
"""下载音频"""
save_cookies(cookies)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
try:
# 获取视频BV号并显示信息
bvid = extract_bvid(url)
if bvid:
pages = get_video_info(bvid, headers, show_info=False) # 添加参数控制是否显示信息
if len(pages) > 1:
choice = input("\n是否显示分P列表?(y/n) [n]: ").lower()
if choice == 'y':
print("\n分P列表:")
# 显示视频基本信息
get_video_info(bvid, headers, show_info=True) # 此时显示完整信息
# 显示分P信息
for page in pages:
duration_min = page['duration'] // 60
duration_sec = page['duration'] % 60
print(f"{page['page']}. {page['part']} ({duration_min}分{duration_sec}秒)")
print("\n请输入要下载的分P编号(用空格分隔,例如:1 3 5)")
print("支持范围选择(例如:1-5 7 9-11)")
choice = input("请选择 [1]: ").strip()
# 如果直接回车,默认下载第一P
if not choice:
choice = "1"
# 解析选择的分P
selected_pages = set()
for part in choice.split():
if '-' in part:
start, end = map(int, part.split('-'))
selected_pages.update(range(start, end + 1))
else:
selected_pages.add(int(part))
# 过滤无效的分P编号
valid_pages = sorted([p for p in selected_pages if 1 <= p <= len(pages)])
if valid_pages:
print(f"\n将下载以下分P: {', '.join(map(str, valid_pages))}")
for page_num in valid_pages:
page = pages[page_num - 1]
page_url = f"https://www.bilibili.com/video/{bvid}?p={page_num}"
print(f"\n开始下载P{page_num}: {page['part']}")
download_single_audio(page_url, cookies)
else:
print("未选择有效的分P编号,将下载P1")
download_single_audio(url, cookies)
else:
# 单P视频直接下载
download_single_audio(url, cookies)
except Exception as e:
print(f"下载过程出错: {str(e)}")
def get_user_info(cookies: Dict) -> None:
"""获取并显示用户信息"""
print("\n" + "="*50)
print("# 用户信息")
print("="*50)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
try:
# 获取用户基本信息
myinfo_url = 'http://api.bilibili.com/x/space/myinfo'
myinfo_resp = requests.get(myinfo_url, cookies=cookies, headers=headers)
myinfo_data = myinfo_resp.json()
user_id = None
if myinfo_data['code'] == 0:
user_data = myinfo_data['data']
user_id = user_data['mid'] # 获取用户ID
print(f"用户名: {user_data['name']}")
print(f"用户ID: {user_id}")
print(f"等级: LV{user_data['level']}")
print(f"会员类型: {'大会员' if user_data['vip']['status'] == 1 else '普通用户'}")
print(f"性别: {user_data['sex']}")
print(f"签名: {user_data['sign']}")
# 获取关注和粉丝数
if user_id:
stat_url = f'https://api.bilibili.com/x/relation/stat?vmid={user_id}'
stat_resp = requests.get(stat_url, headers=headers)
stat_data = stat_resp.json()
if stat_data['code'] == 0:
stat_info = stat_data['data']
print(f"关注数: {stat_info['following']}")
print(f"粉丝数: {stat_info['follower']}")
# 获取硬币数
coin_url = 'http://account.bilibili.com/site/getCoin'
coin_resp = requests.get(coin_url, cookies=cookies, headers=headers)
coin_data = coin_resp.json()
if coin_data['code'] == 0:
print(f"硬币数: {coin_data['data']['money']}")
# 获取今日投币经验
coin_exp_url = 'https://api.bilibili.com/x/web-interface/coin/today/exp'
coin_exp_resp = requests.get(coin_exp_url, cookies=cookies, headers=headers)
coin_exp_data = coin_exp_resp.json()
if coin_exp_data['code'] == 0:
today_exp = coin_exp_data['data']
max_coin_exp = 50 # B站每日投币多获得50经验
print(f"今日投币经验: {today_exp}/{max_coin_exp}")
remaining_coins = (max_coin_exp - today_exp) // 10 # 每个硬币获得10经验
if remaining_coins > 0:
print(f"今日还可投币: {remaining_coins}个")
# 获取登录信息
nav_url = 'https://api.bilibili.com/x/web-interface/nav'
nav_resp = requests.get(nav_url, cookies=cookies, headers=headers)
nav_data = nav_resp.json()
if nav_data['code'] == 0:
nav_info = nav_data['data']
print(f"当前登录状态: {'已登录' if nav_info['isLogin'] else '未登录'}")
if nav_info['isLogin']:
print(f"邮箱: {nav_info['email']}")
print(f"硬币余额: {nav_info['money']}")
print(f"B币余额: {nav_info['wallet']['bcoin_balance']}")
except Exception as e:
print(f"获取用户信息出错: {str(e)}")
print("="*50 + "\n")
def extract_bvid_from_file(filename: str) -> list:
"""从文件中取URL并提取BV号"""
bvids = []
try:
with open(filename, 'r', encoding='utf-8') as f:
for line in f:
url = line.strip()
if url:
# 使用正则表达式匹配BV号
match = re.search(r'BV[a-zA-Z0-9]{10}', url)
if match:
bvid = match.group()
bvids.append(bvid)
except Exception as e:
print(f"读取文件出错: {str(e)}")
return bvids
def batch_coin(cookies: Dict) -> None:
"""批量投币功能"""
print("\n" + "="*50)
print("# 批量投币")
print("="*50)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
try:
# 先检查今日投币情况
coin_exp_url = 'https://api.bilibili.com/x/web-interface/coin/today/exp'
coin_exp_resp = requests.get(coin_exp_url, cookies=cookies, headers=headers)
coin_exp_data = coin_exp_resp.json()
if coin_exp_data['code'] == 0:
today_exp = coin_exp_data['data']
max_coin_exp = 50
remaining_coins = (max_coin_exp - today_exp) // 10
if remaining_coins <= 0:
print("今日已完成投币任务!")
return
print(f"今日还可投币: {remaining_coins}个")
# 检查bvid.txt文件是否存在
if not os.path.exists('bvid.txt'):
print("错误:找不到bvid.txt文件!")
print("请创建bvid.txt文件并在其中每行写入一个视频链接")
return
# 从文件中读取并提取BV号
bvids = extract_bvid_from_file('bvid.txt')
if not bvids:
print("错误:无法从文件中提取效的BV号!")
return
print(f"从文件中提取到 {len(bvids)} 个BV号:")
# 获取每个视频的信息
for bvid in bvids:
get_video_info(bvid, headers)
# 询问是否继续投币
choice = input("\n是否开始批量投币?(y/n) [y]: ").lower()
if not choice or choice == 'y': # 直接回车或输入y都继续
# 获取csrf令牌
csrf = cookies.get('bili_jct')
if not csrf:
print("错误:无法获取csrf令牌!")
return
# 投API
coin_url = 'https://api.bilibili.com/x/web-interface/coin/add'
# 创建进度条
with tqdm(total=min(remaining_coins, len(bvids)), desc="投币进度") as pbar:
for bvid in bvids:
if remaining_coins <= 0:
print("\n今日投币任务已完成!")
break
# 构造请求数据
data = {
'bvid': bvid,
'multiply': 1, # 投1个币
'select_like': 1, # 同时点赞
'csrf': csrf
}
try:
# 发送投币请求
response = requests.post(coin_url, data=data, cookies=cookies, headers=headers)
result = response.json()
if result['code'] == 0:
print(f"\n成功给视频 {bvid} 投币")
remaining_coins -= 1
pbar.update(1)
else:
print(f"\n给视频 {bvid} 投币失败:{result['message']}")
print("停止投币操作")
break # 投币失败就停止
# 添加延时,避免请求过快
time.sleep(2)
except Exception as e:
print(f"\n投币过程出错: {str(e)}")
print("停止投币操作")
break # 出错也停止
print("\n批量投币完成!")
# 获取当前硬币数
nav_url = 'https://api.bilibili.com/x/web-interface/nav'
nav_resp = requests.get(nav_url, cookies=cookies, headers=headers)
nav_data = nav_resp.json()
if nav_data['code'] == 0:
current_coins = nav_data['data']['money']
print(f"当前剩余硬币数: {current_coins}")
except Exception as e:
print(f"批量投币时出错: {str(e)}")
print("="*50 + "\n")
def show_menu() -> str:
"""显示功能菜单"""
print("\n" + "="*50)
print("B站工具箱")
print("="*50)
print("1. 批量点赞 (默认)")
print("2. 批量投币")
print("3. 下载视频音频")
print("4. 获取视频BV号")
print("5. 查看用户信息")
print("6. bv转av")
print("7. 获取视频弹幕")
print("8. 退出程序")
print("="*50)
choice = input("请选择功能 (1-8) [1]: ").strip()
if not choice:
return '1'
if choice in ['1', '2', '3', '4', '5', '6', '7', '8']:
return choice
print("无效的选择,使用默认选项1")
return '1'
def extract_bvid(url: str) -> str:
"""从URL中提取BV号"""
# 匹配BV号的正则表达式
bv_pattern = r'BV[a-zA-Z0-9]{10}'
match = re.search(bv_pattern, url)
if match:
return match.group()
return ''
def get_video_bvids() -> None:
"""获取视频BV号并保存到文件"""
bvids = []
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
print("请输入B站视频URL,每行一个,输入q结束:")
while True:
url = input().strip()
if url.lower() == 'q':
break
if not url.startswith(('https://www.bilibili.com', 'https://b23.tv')):
print("请输入有效的B站视频链接!")
continue
bvid = extract_bvid(url)
if bvid:
bvids.append(bvid)
# 显示视频信息
get_video_info(bvid, headers)
else:
print("无法从URL中提取BV号!")
if bvids:
with open('bvid.txt', 'w', encoding='utf-8') as f:
f.write('\n'.join(bvids))
print(f"已保存 {len(bvids)} 个BV号到 bvid.txt")
def get_video_info(bvid: str, headers: dict, show_info: bool = True) -> list:
"""获取并显示视频信息"""
try:
video_url = f'https://api.bilibili.com/x/web-interface/view?bvid={bvid}'
response = requests.get(video_url, headers=headers)
data = response.json()
if data['code'] == 0:
info = data['data']
if show_info: # 只在需要时显示信息
print(f"\n视频信息 - {bvid}:")
print(f"标题: {info['title']}")
print(f"UP主: {info['owner']['name']}")
print(f"播放量: {info['stat']['view']}")
print(f"点赞数: {info['stat']['like']}")
print(f"投币数: {info['stat']['coin']}")
print(f"收藏数: {info['stat']['favorite']}")
print(f"分享数: {info['stat']['share']}")
print(f"发布时间: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(info['pubdate']))}")
print(f"视频简介: {info['desc'][:100]}..." if len(info['desc']) > 100 else f"视频简介: {info['desc']}")
print("-" * 50)
# 返回分P信息,供下载功能使用
return info.get('pages', [])
except Exception as e:
print(f"获取视频 {bvid} 信息失败: {str(e)}")
return []
def batch_like(cookies: Dict) -> None:
"""批量点赞功能"""
print("\n" + "="*50)
print("# 批量点赞")
print("="*50)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
try:
# 检查bvid.txt文件是否存在
if not os.path.exists('bvid.txt'):
print("错误:找不到bvid.txt文件!")
print("请创建bvid.txt文件并在其中每行写入一个视频链接")
return
# 从文件中读取并提取BV号
bvids = extract_bvid_from_file('bvid.txt')
if not bvids:
print("错误:无法从文件中提取有效的BV号!")
return
print(f"从文件中提取到 {len(bvids)} 个BV号:")
# 获取每个视频的信息
for bvid in bvids:
get_video_info(bvid, headers)
# 询问是否继续点赞
choice = input("\n是否开始批量点赞?(y/n) [y]: ").lower()
if not choice or choice == 'y':
# 获取csrf令牌
csrf = cookies.get('bili_jct')
if not csrf:
print("错误:无法获取csrf令牌!")
return
# 点赞API
like_url = 'https://api.bilibili.com/x/web-interface/archive/like'
# 创建进度条
with tqdm(total=len(bvids), desc="点赞进度") as pbar:
for bvid in bvids:
# 构造请求数据
data = {
'bvid': bvid,
'like': 1, # 1表示点赞,2表示取消点赞
'csrf': csrf
}
try:
# 发送点赞请求
response = requests.post(like_url, data=data, cookies=cookies, headers=headers)
result = response.json()
if result['code'] == 0:
print(f"\n成功给视频 {bvid} 点赞")
pbar.update(1)
else:
print(f"\n给视频 {bvid} 点赞失败:{result['message']}")
print("停止点赞操作")
break # 点赞失败就停止
# 添加延时,避免请求过快
time.sleep(2)
except Exception as e:
print(f"\n点赞过程出错: {str(e)}")
print("停止点赞操作")
break # 出错也停止
print("\n批量点赞完成!")
except Exception as e:
print(f"批量点赞时出错: {str(e)}")
print("="*50 + "\n")
def download_single_audio(url: str, cookies: Dict):
"""下载单个音频"""
try:
cmd = [
'yt-dlp',
'--cookies', 'cookies.txt',
'-f', 'ba',
'--no-playlist',
'--no-check-certificates',
'--progress',
'--newline',
]
cmd.append(url)
print("开始下载音频...")
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
encoding='utf-8',
errors='replace',
bufsize=1,
universal_newlines=True
)
downloaded_file = None
pbar = None
total_size = None
# 用于解析下载进度的正则表达式
progress_pattern = re.compile(r'\[download\]\s+(\d+\.\d+)%\s+of\s+(\d+\.\d+)([KMG])iB\s+at\s+.*?ETA\s+(.*)')
while True:
output = process.stdout.readline()
if output == '' and process.poll() is not None:
break
if output:
output = output.strip()
# 获取文件名
if '[download] Destination:' in output:
downloaded_file = output.split('[download] Destination:', 1)[1].strip()
elif 'has already been downloaded' in output:
downloaded_file = output.split('[download] ', 1)[1].split(' has already', 1)[0].strip()
# 解析进度信息
match = progress_pattern.search(output)
if match:
percentage, size, unit, eta = match.groups()
# 计算总大小(转换为字节)
if total_size is None:
multiplier = {'K': 1024, 'M': 1024**2, 'G': 1024**3}
total_size = int(float(size) * multiplier[unit])
# 创建进度条
pbar = tqdm(
total=total_size,
unit='B',
unit_scale=True,
desc="下载进度",
ncols=80,
bar_format='{desc}: {percentage:3.1f}%|{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]'
)
# 更新进度条
if pbar:
current = int(total_size * float(percentage) / 100)
pbar.n = current
pbar.refresh()
# 其他输出信息
elif not output.startswith('[download]'):
print(output)
# 关闭进度条
if pbar:
pbar.close()
# 查下载结果
if process.returncode == 0:
print("\n下载完成!")
if downloaded_file and os.path.exists(downloaded_file):
print(f"找到音频文件: {downloaded_file}")
if input("是否转换为MP3格式?(y/n) [y]: ").lower() in ['', 'y']:
if find_ffmpeg():
convert_to_mp3(downloaded_file)
else:
print("未找到ffmpeg,无法转换为MP3格式")
else:
print(f"警告:无法找到下载的文件")
m4a_files = [f for f in os.listdir('.') if f.endswith('.m4a')]
if m4a_files:
print(f"在当前目录找到以下.m4a文件:")
for i, file in enumerate(m4a_files, 1):
print(f"{i}. {file}")
choice = input("请选择要转换的文件编号(输入q取消)[1]: ")
if not choice:
choice = '1'
if choice.isdigit() and 1 <= int(choice) <= len(m4a_files):
if find_ffmpeg():
convert_to_mp3(m4a_files[int(choice)-1])
else:
print("未找到ffmpeg,无法转换为MP3格式")
else:
error = process.stderr.read()
print(f"下载失败!错误信息:\n{error}")
except FileNotFoundError:
print("错误:请先安装 yt-dlp")
print("可以使用以下命令安装:")
print("pip install yt-dlp")
except Exception as e:
print(f"发生��误: {str(e)}")
def main():
# 先尝试从文件加载cookies
cookies = load_cookies_from_file()
# 如果没有找到有效的cookies文件,则请求用户输入
if not cookies:
cookies = get_cookies_from_user()
if not cookies:
return
# 保存用户输入的cookies
save_cookies(cookies)
# 获取并显示用户信息
get_user_info(cookies)
while True:
choice = show_menu()
if choice == '1':
# 批量点赞
batch_like(cookies)
elif choice == '2':
# 批量投币
batch_coin(cookies)
elif choice == '3':
# 下载视频音频
while True:
print("\n请输入B站视频URL (输入q返回主菜单):")
url = input().strip()
if url.lower() == 'q':
break
if not url.startswith(('https://www.bilibili.com', 'https://b23.tv')):
print("请输入有效的B站视频链接!")
continue
download_audio(url, cookies)
elif choice == '4':
# 获取视频BV号
get_video_bvids()
elif choice == '5':
# 查看用户信息
get_user_info(cookies)
elif choice == '6':
# 新增的功能:bv转av
bvid = input("请输入BV号: ").strip()
if bvid.startswith('BV'):
av_number = bv_to_av(bvid)
print(f"对应的AV号是: {av_number}")
else:
print("请输入有效的BV号!")
elif choice == '7':
# 获取视频弹幕
url = input("请输入视频URL: ").strip()
if url.startswith(('https://www.bilibili.com', 'https://b23.tv')):
try:
# 获取视频cid
bvid = extract_bvid(url)
if bvid:
# 添加headers定义
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
video_url = f'https://api.bilibili.com/x/web-interface/view?bvid={bvid}'
response = requests.get(video_url, headers=headers)
data = response.json()
if data['code'] == 0:
cid = data['data']['cid']
# 获取弹幕分段
segment = input("请输入要获取第几个6分钟段的弹幕 (1-n) [1]: ").strip()
if not segment:
segment = "1"
if segment.isdigit():
get_danmaku(cid, int(segment))
else:
print("请输入有效的数字!")
else:
print(f"获取视频信息失败:{data['message']}")
else:
print("无法从URL中提取BV号!")
except Exception as e:
print(f"获取弹幕时出错: {str(e)}")
else:
print("请输入有效的B站视频链接!")
elif choice == '8':
print("\n感谢使用!再见!")
break
def bv_to_av(bvid: str) -> str:
"""将BV号转换为AV号"""
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com',
}
try:
# 使用B站API获取视频信息
video_url = f'https://api.bilibili.com/x/web-interface/view?bvid={bvid}'
response = requests.get(video_url, headers=headers)
data = response.json()
if data['code'] == 0:
# 从返回数据中获取aid(AV号)
av_number = data['data']['aid']
return str(av_number) # 直接返回数字部分
else:
print(f"获取AV号失败:{data['message']}")
return ""
except Exception as e:
print(f"转换过程出错: {str(e)}")
return ""
def get_danmaku(cid: str, segment_index: int = 1) -> None:
"""获取视频弹幕"""
try:
import google.protobuf.text_format as text_format
from bilibili.community.service.dm.v1 import dm_pb2 as Danmaku
except ImportError:
print("错误:缺少必要的库,请先安装:")
print("pip install protobuf")
print("同时需要编译proto文件生成dm_pb2.py")
return
url = 'https://api.bilibili.com/x/v2/dm/web/seg.so'
params = {
'type': 1, # 弹幕类型
'oid': cid, # 视频cid
'segment_index': segment_index # 弹幕分段
}
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Referer': 'https://www.bilibili.com'
}
try:
resp = requests.get(url, params=params, headers=headers)
if resp.status_code == 200:
danmaku_seg = Danmaku.DmSegMobileReply()
danmaku_seg.ParseFromString(resp.content)
print(f"\n获取到 {len(danmaku_seg.elems)} 条弹幕:")
print("-" * 50)
for elem in danmaku_seg.elems:
# 计算时间
seconds = elem.progress / 1000
minutes = int(seconds / 60)
seconds = int(seconds % 60)
# 获取弹幕类型
mode_types = {
1: "普通弹幕",
2: "普通弹幕",
3: "普通弹幕",
4: "底部弹幕",
5: "顶部弹幕",
6: "逆向弹幕",
7: "高级弹幕",
8: "代码弹幕",
9: "BAS弹幕"
}
mode_type = mode_types.get(elem.mode, "未知类型")
# 转换颜色为十六进制
color = f"#{elem.color:06x}"
print(f"时间: {minutes:02d}:{seconds:02d}")
print(f"内容: {elem.content}")
print(f"类型: {mode_type}")
print(f"颜色: {color}")
print(f"字号: {elem.fontsize}")
print(f"发送时间: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(elem.ctime))}")
print("-" * 50)
else:
print(f"获取弹幕失败: HTTP {resp.status_code}")
except Exception as e:
print(f"获取弹幕时出错: {str(e)}")
if __name__ == "__main__":
main()
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
+1
View File
@@ -0,0 +1 @@
# 空文件
@@ -0,0 +1 @@
# 空文件
@@ -0,0 +1 @@
# 空文件
@@ -0,0 +1 @@
# 空文件
@@ -0,0 +1 @@
# 空文件
@@ -0,0 +1,38 @@
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: dm.proto
# Protobuf Python Version: 5.28.1
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
_runtime_version.Domain.PUBLIC,
5,
28,
1,
'',
'dm.proto'
)
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x08\x64m.proto\x12 bilibili.community.service.dm.v1\"P\n\x10\x44mSegMobileReply\x12<\n\x05\x65lems\x18\x01 \x03(\x0b\x32-.bilibili.community.service.dm.v1.DanmakuElem\"\xe9\x01\n\x0b\x44\x61nmakuElem\x12\n\n\x02id\x18\x01 \x01(\x03\x12\x10\n\x08progress\x18\x02 \x01(\x05\x12\x0c\n\x04mode\x18\x03 \x01(\x05\x12\x10\n\x08\x66ontsize\x18\x04 \x01(\x05\x12\r\n\x05\x63olor\x18\x05 \x01(\r\x12\x0f\n\x07midHash\x18\x06 \x01(\t\x12\x0f\n\x07\x63ontent\x18\x07 \x01(\t\x12\r\n\x05\x63time\x18\x08 \x01(\x03\x12\x0e\n\x06weight\x18\t \x01(\x05\x12\x0e\n\x06\x61\x63tion\x18\n \x01(\t\x12\x0c\n\x04pool\x18\x0b \x01(\x05\x12\r\n\x05idStr\x18\x0c \x01(\t\x12\x0c\n\x04\x61ttr\x18\r \x01(\x05\x12\x11\n\tanimation\x18\x0e \x01(\tb\x06proto3')
_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'dm_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
DESCRIPTOR._loaded_options = None
_globals['_DMSEGMOBILEREPLY']._serialized_start=46
_globals['_DMSEGMOBILEREPLY']._serialized_end=126
_globals['_DANMAKUELEM']._serialized_start=129
_globals['_DANMAKUELEM']._serialized_end=362
# @@protoc_insertion_point(module_scope)
View File
Whitespace-only changes.
View File
Whitespace-only changes.
+26
View File
@@ -0,0 +1,26 @@
syntax = "proto3";
package bilibili.community.service.dm.v1;
// 弹幕数据包
message DmSegMobileReply {
repeated DanmakuElem elems = 1;
}
// 弹幕条目
message DanmakuElem {
int64 id = 1; // 弹幕dmid
int32 progress = 2; // 弹幕出现时间(毫秒)
int32 mode = 3; // 弹幕类型
int32 fontsize = 4; // 字号
uint32 color = 5; // 颜色
string midHash = 6; // 发送者hash
string content = 7; // 弹幕内容
int64 ctime = 8; // 发送时间
int32 weight = 9; // 权重
string action = 10; // 动作
int32 pool = 11; // 弹幕池
string idStr = 12; // 弹幕dmid字符串
int32 attr = 13; // 属性
string animation = 14; // 动画
}
+38
View File
@@ -0,0 +1,38 @@
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: dm.proto
# Protobuf Python Version: 5.28.1
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
_runtime_version.Domain.PUBLIC,
5,
28,
1,
'',
'dm.proto'
)
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x08\x64m.proto\x12 bilibili.community.service.dm.v1\"P\n\x10\x44mSegMobileReply\x12<\n\x05\x65lems\x18\x01 \x03(\x0b\x32-.bilibili.community.service.dm.v1.DanmakuElem\"\xe9\x01\n\x0b\x44\x61nmakuElem\x12\n\n\x02id\x18\x01 \x01(\x03\x12\x10\n\x08progress\x18\x02 \x01(\x05\x12\x0c\n\x04mode\x18\x03 \x01(\x05\x12\x10\n\x08\x66ontsize\x18\x04 \x01(\x05\x12\r\n\x05\x63olor\x18\x05 \x01(\r\x12\x0f\n\x07midHash\x18\x06 \x01(\t\x12\x0f\n\x07\x63ontent\x18\x07 \x01(\t\x12\r\n\x05\x63time\x18\x08 \x01(\x03\x12\x0e\n\x06weight\x18\t \x01(\x05\x12\x0e\n\x06\x61\x63tion\x18\n \x01(\t\x12\x0c\n\x04pool\x18\x0b \x01(\x05\x12\r\n\x05idStr\x18\x0c \x01(\t\x12\x0c\n\x04\x61ttr\x18\r \x01(\x05\x12\x11\n\tanimation\x18\x0e \x01(\tb\x06proto3')
_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'dm_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
DESCRIPTOR._loaded_options = None
_globals['_DMSEGMOBILEREPLY']._serialized_start=46
_globals['_DMSEGMOBILEREPLY']._serialized_end=126
_globals['_DANMAKUELEM']._serialized_start=129
_globals['_DANMAKUELEM']._serialized_end=362
# @@protoc_insertion_point(module_scope)