【Python实战】批量解码Base64磁力链接并自动生成下载清单
1. 什么是Base64编码的磁力链接磁力链接Magnet URI是一种基于文件内容标识的资源定位方式它通过哈希值来唯一标识文件而不是像传统URL那样依赖文件的位置。典型的磁力链接格式如下magnet:?xturn:btih:E7FC73D9E20697C6C440203F5884EF52F9E4BD28在实际应用中我们经常会遇到经过Base64编码的磁力链接。Base64是一种用64个字符来表示二进制数据的方法它可以将任意二进制数据转换为可打印的ASCII字符序列。编码后的磁力链接看起来像这样bWFnbmV0Oj94dD11cm46YnRpaDo0NWE4MDVkYmQ3OGI4ZGVjNzk2YTBhMTI3YzRiNGQyNDY2ZGRiYjlh这种编码方式有几个实际用途避免特殊字符在传输过程中被错误处理方便在不支持二进制数据的系统中存储和传输使链接更适合嵌入到其他文本格式中2. Python解码Base64磁力链接2.1 使用base64模块解码Python的标准库中已经包含了base64模块可以轻松实现编码和解码操作。下面是一个完整的解码示例import base64 encoded_link bWFnbmV0Oj94dD11cm46YnRpaDo0NWE4MDVkYmQ3OGI4ZGVjNzk2YTBhMTI3YzRiNGQyNDY2ZGRiYjlh decoded_bytes base64.b64decode(encoded_link) decoded_link decoded_bytes.decode(utf-8) print(decoded_link) # 输出magnet:?xturn:btih:45a805dbd78b8dec796a0a127c4b4d2466ddbb9a这里有几个关键点需要注意b64decode()返回的是bytes对象需要使用decode(utf-8)转换为字符串如果编码字符串包含换行符等特殊字符需要先进行清理Base64编码的字符串长度应该是4的倍数如果不是会自动补全2.2 处理批量解码任务当需要处理大量Base64编码的磁力链接时我们可以从文本文件中读取内容并批量解码import base64 def decode_magnet_links(input_file, output_file): with open(input_file, r, encodingutf-8) as f: encoded_links f.readlines() decoded_links [] for link in encoded_links: link link.strip() # 去除换行符和空格 if not link: continue try: decoded base64.b64decode(link).decode(utf-8) decoded_links.append(decoded) except Exception as e: print(f解码失败: {link}, 错误: {str(e)}) with open(output_file, w, encodingutf-8) as f: for link in decoded_links: f.write(link \n) # 使用示例 decode_magnet_links(encoded_links.txt, decoded_links.txt)这个函数会从输入文件读取所有行逐行解码Base64内容将解码后的磁力链接写入输出文件处理过程中会跳过空行并捕获解码错误3. 自动生成下载清单3.1 使用xlsxwriter创建Excel文件将解码后的磁力链接整理成Excel文件可以方便管理和分享。Python的xlsxwriter库非常适合这个任务import xlsxwriter def create_download_list(links, output_file): # 创建一个Excel工作簿 workbook xlsxwriter.Workbook(output_file) worksheet workbook.add_worksheet(磁力链接) # 设置标题行格式 header_format workbook.add_format({ bold: True, align: center, valign: vcenter, fg_color: #D7E4BC, border: 1 }) # 写入标题行 worksheet.write(A1, 序号, header_format) worksheet.write(B1, 磁力链接, header_format) worksheet.set_column(B:B, 80) # 设置列宽 # 写入数据 for i, link in enumerate(links, start1): worksheet.write(i, 0, i) worksheet.write(i, 1, link) # 添加自动筛选 worksheet.autofilter(A1:B1) workbook.close() # 使用示例 links [ magnet:?xturn:btih:45a805dbd78b8dec796a0a127c4b4d2466ddbb9a, magnet:?xturn:btih:d3b068553c34234669408efd01c009a52adbfc06 ] create_download_list(links, magnet_links.xlsx)这个Excel文件会包含带格式的标题行自动调整的列宽自动筛选功能序号和磁力链接两列数据3.2 完整工作流整合将解码和Excel生成整合成一个完整的工作流import base64 import xlsxwriter def process_magnet_links(input_file, output_excel): # 1. 解码Base64链接 with open(input_file, r, encodingutf-8) as f: encoded_links [line.strip() for line in f if line.strip()] decoded_links [] for link in encoded_links: try: decoded base64.b64decode(link).decode(utf-8) decoded_links.append(decoded) except: continue # 2. 创建Excel文件 workbook xlsxwriter.Workbook(output_excel) worksheet workbook.add_worksheet(磁力链接) # 设置格式 header_format workbook.add_format({ bold: True, border: 1, align: center, valign: vcenter, fg_color: #D9E1F2 }) # 写入数据 worksheet.write_row(A1, [序号, 磁力链接], header_format) worksheet.set_column(B:B, 80) for idx, link in enumerate(decoded_links, start1): worksheet.write(idx, 0, idx) worksheet.write(idx, 1, link) workbook.close() print(f成功处理 {len(decoded_links)} 条磁力链接已保存到 {output_excel}) # 使用示例 process_magnet_links(encoded_links.txt, magnet_collection.xlsx)4. 高级应用与错误处理4.1 处理特殊Base64变种有时我们会遇到URL安全的Base64编码它用-和_替换了和/。Python的base64模块也支持这种变种import base64 # URL安全的Base64解码 encoded bWFnbmV0Oj94dD11cm46YnRpaDo0NWE4MDVkYmQ3OGI4ZGVjNzk2YTBhMTI3YzRiNGQyNDY2ZGRiYjlh decoded base64.urlsafe_b64decode(encoded).decode(utf-8)4.2 验证磁力链接格式解码后我们可以验证磁力链接是否符合标准格式import re def is_valid_magnet_link(link): pattern r^magnet:\?xturn:btih:[0-9a-fA-F]{40}.*$ return re.match(pattern, link) is not None # 使用示例 print(is_valid_magnet_link(magnet:?xturn:btih:45a805dbd78b8dec796a0a127c4b4d2466ddbb9a)) # True print(is_valid_magnet_link(invalid_link)) # False4.3 性能优化技巧当处理大量链接时可以考虑以下优化使用列表推导式简化代码decoded_links [ base64.b64decode(link.strip()).decode(utf-8) for link in encoded_links if link.strip() ]使用多线程加速处理适用于超大量数据from concurrent.futures import ThreadPoolExecutor def decode_link(link): try: return base64.b64decode(link).decode(utf-8) except: return None with ThreadPoolExecutor() as executor: decoded_links list(filter(None, executor.map(decode_link, encoded_links)))分批写入Excel文件减少内存占用# 在原有代码基础上修改写入逻辑 BATCH_SIZE 1000 for i in range(0, len(decoded_links), BATCH_SIZE): batch decoded_links[i:iBATCH_SIZE] for j, link in enumerate(batch, starti1): worksheet.write(j, 0, j) worksheet.write(j, 1, link)5. 实际应用案例5.1 资源整理工具我们可以将上述功能整合成一个完整的资源整理工具import os import base64 import xlsxwriter from datetime import datetime class MagnetLinkProcessor: def __init__(self): self.decoded_links [] def load_links_from_file(self, file_path): 从文件加载Base64编码的磁力链接 if not os.path.exists(file_path): raise FileNotFoundError(f文件不存在: {file_path}) with open(file_path, r, encodingutf-8) as f: return [line.strip() for line in f if line.strip()] def decode_links(self, encoded_links): 批量解码Base64磁力链接 self.decoded_links [] for link in encoded_links: try: decoded base64.b64decode(link).decode(utf-8) if decoded.startswith(magnet:): self.decoded_links.append(decoded) except: continue return self.decoded_links def export_to_excel(self, output_path): 导出到Excel文件 if not self.decoded_links: raise ValueError(没有可导出的磁力链接) timestamp datetime.now().strftime(%Y%m%d_%H%M%S) if not output_path: output_path fmagnet_links_{timestamp}.xlsx workbook xlsxwriter.Workbook(output_path) worksheet workbook.add_worksheet(磁力链接) # 设置样式 header_style workbook.add_format({ bold: True, border: 1, bg_color: #4472C4, font_color: white, align: center }) # 写入数据 worksheet.write_row(A1, [序号, 磁力链接], header_style) worksheet.set_column(B:B, 80) for idx, link in enumerate(self.decoded_links, start1): worksheet.write(idx, 0, idx) worksheet.write(idx, 1, link) workbook.close() return output_path # 使用示例 processor MagnetLinkProcessor() encoded_links processor.load_links_from_file(links.txt) decoded_links processor.decode_links(encoded_links) output_file processor.export_to_excel(my_magnets.xlsx) print(f成功导出 {len(decoded_links)} 条链接到 {output_file})5.2 集成到下载工具我们可以进一步将解码功能集成到下载工具中实现自动化下载import subprocess def download_with_aria2(magnet_links, max_connections5): 使用aria2下载磁力链接 if not isinstance(magnet_links, list): magnet_links [magnet_links] for link in magnet_links: try: cmd [ aria2c, --max-connection-per-server str(max_connections), --seed-time0, # 不做种 link ] subprocess.run(cmd, checkTrue) except subprocess.CalledProcessError as e: print(f下载失败: {link}, 错误: {e}) # 使用示例 download_with_aria2(decoded_links)