import os import json import time import qrcode import requests from PIL import Image from config import BILIBILI_API, DOWNLOAD_CONFIG import subprocess class BilibiliDownloader: def __init__(self): self.session = requests.Session() self.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' } self.session.headers.update(self.headers) self.cookies = self._load_cookies() if self.cookies: # 将加载的cookies设置到session中 self.session.cookies.update(self.cookies) def _load_cookies(self): """加载保存的cookies""" if os.path.exists(DOWNLOAD_CONFIG['cookie_file']): with open(DOWNLOAD_CONFIG['cookie_file'], 'r') as f: return json.load(f) return None def _save_cookies(self): """保存cookies到文件""" with open(DOWNLOAD_CONFIG['cookie_file'], 'w') as f: json.dump(dict(self.session.cookies), f) def login(self): """使用二维码登录B站""" # 获取二维码URL response = self.session.get('https://passport.bilibili.com/x/passport-login/web/qrcode/generate') data = response.json() if data['code'] != 0: print(f"获取二维码失败：{data}") return False qr_url = data['data']['url'] qr_key = data['data']['qrcode_key'] # 生成二维码，增加边距和容错率 qr = qrcode.QRCode( version=1, error_correction=qrcode.constants.ERROR_CORRECT_H, box_size=10, border=4, ) qr.add_data(qr_url) qr.make(fit=True) img = qr.make_image(fill_color="black", back_color="white") img.save('login_qr.png') # 打开二维码 try: # 在Windows系统下使用默认图片查看器打开 os.system('explorer "login_qr.png"') except: try: Image.open('login_qr.png').show() except: print("二维码已保存到login_qr.png文件，请手动打开扫描") print("\n请使用B站手机客户端扫描二维码登录") print("二维码文件保存在：", os.path.abspath('login_qr.png')) # 等待扫描登录 while True: try: response = self.session.get( 'https://passport.bilibili.com/x/passport-login/web/qrcode/poll', params={'qrcode_key': qr_key} ) data = response.json() print(f"登录状态：{data}") # 添加调试信息 # 检查并保存cookies if 'Set-Cookie' in response.headers: print("检测到新的cookies") self._save_cookies() except Exception as e: print(f"查询登录状态时出错：{e}") time.sleep(2) continue if data['code'] != 0: print(f"查询状态失败：{data}") time.sleep(2) continue result_data = data['data'] status_code = result_data['code'] if status_code == 0: print("登录成功！") # 验证登录状态 try: verify_response = self.session.get('https://api.bilibili.com/x/web-interface/nav') verify_data = verify_response.json() if verify_data['code'] == 0: print(f"欢迎，{verify_data['data']['uname']}!") self._save_cookies() os.remove('login_qr.png') return True else: print("登录验证失败，请重试") return False except Exception as e: print(f"验证登录状态时出错：{e}") return False elif status_code == 86038: print("二维码已过期，请重新运行程序") os.remove('login_qr.png') return False elif status_code == 86101: print("等待扫描二维码...") time.sleep(2) elif status_code == 86039: print("二维码已扫描，请在手机上确认登录") time.sleep(2) elif status_code == 86090: print("等待确认中...") time.sleep(2) else: print(f"等待操作中... (状态码：{status_code})") time.sleep(2) def verify_login(self): """验证登录状态""" try: print("验证登录状态...") response = self.session.get('https://api.bilibili.com/x/web-interface/nav') data = response.json() if data['code'] == 0: print(f"当前登录用户：{data['data']['uname']}") return True else: print("登录状态已失效，需要重新登录") return False except Exception as e: print(f"验证登录状态时出错：{e}") return False def get_favorite_folders(self): """获取用户收藏夹列表""" if not self.verify_login(): if not self.login(): return [] print("正在获取收藏夹列表...") try: response = self.session.get(BILIBILI_API['favorite_list']) data = response.json() print(f"API响应: {data}") # 打印完整响应 if data['code'] == 0: if 'data' in data and 'list' in data['data']: return data['data']['list'] else: print("响应中没有找到收藏夹列表数据") else: print(f"获取收藏夹失败，错误码：{data['code']}，信息：{data.get('message', '未知错误')}") except Exception as e: print(f"获取收藏夹时发生错误：{e}") return [] def get_favorite_videos(self, folder_id, page=1, pagesize=20): """获取收藏夹中的视频""" params = { 'media_id': folder_id, 'pn': page, 'ps': pagesize, 'order': 'mtime', 'type': 0, 'tid': 0, 'platform': 'web' } response = self.session.get(BILIBILI_API['favorite_content'], params=params) data = response.json() if data['code'] == 0: return data['data'] return None def download_video(self, bvid): """使用you-get下载视频""" if not os.path.exists(DOWNLOAD_CONFIG['output_dir']): os.makedirs(DOWNLOAD_CONFIG['output_dir']) url = f'https://www.bilibili.com/video/{bvid}' command = [ 'you-get', '--cookies', DOWNLOAD_CONFIG['cookie_file'], '-o', DOWNLOAD_CONFIG['output_dir'], url ] try: subprocess.run(command, check=True) return True except subprocess.CalledProcessError as e: print(f"下载失败：{e}") return False def main(): downloader = BilibiliDownloader() # 如果没有cookie或cookie已过期，进行登录 if not downloader.cookies: if not downloader.login(): return # 获取收藏夹列表 folders = downloader.get_favorite_folders() if not folders: print("未找到收藏夹，请确保已登录并有收藏内容") return # 显示收藏夹列表 print("\n=== 你的收藏夹列表 ===") for i, folder in enumerate(folders): print(f"{i+1}. {folder['title']} (共{folder['media_count']}个视频)") # 选择要下载的收藏夹 while True: try: choice = int(input("\n请选择要下载的收藏夹序号（输入0退出）：")) - 1 if choice == -1: return if 0 <= choice < len(folders): break print("无效的选择，请重试") except ValueError: print("请输入有效的数字") selected_folder = folders[choice] print(f"\n开始下载收藏夹：{selected_folder['title']}") # 获取并下载视频 page = 1 while True: videos = downloader.get_favorite_videos(selected_folder['id'], page) if not videos or not videos['medias']: break for video in videos['medias']: print(f"\n正在下载：{video['title']}") if downloader.download_video(video['bvid']): print(f"下载完成：{video['title']}") else: print(f"下载失败：{video['title']}") if videos['has_more']: page += 1 else: break print("\n所有视频下载完成！") if __name__ == "__main__": main()