别再只会下载了!用Python手把手教你解析.torrent文件,提取磁力链接和Tracker信息
深入解析.torrent文件:用Python提取磁力链接与Tracker信息的完整指南
你是否曾经好奇过.torrent文件内部究竟藏着什么秘密?这些小巧的文件如何指挥着下载软件完成复杂的任务?今天我们将抛开图形界面工具,直接深入.torrent文件的二进制世界,用Python代码揭开它的神秘面纱。
1. 理解.torrent文件的基础结构
.torrent文件本质上是一个经过Bencoding编码的字典结构,包含了下载所需的全部元数据。与常见的JSON或XML不同,Bencoding是一种专为BitTorrent协议设计的紧凑编码格式,它使用简单的标记规则来表示四种基本数据类型:
- 字符串:格式为
长度:内容,例如4:test表示字符串"test" - 整数:格式为
i数字e,例如i42e表示数字42 - 列表:格式为
l内容e,例如li42e4:teste表示列表[42, "test"] - 字典:格式为
d键值对e,键必须是字符串,值可以是任意类型
典型的.torrent文件包含以下关键部分:
{ "announce": "http://tracker.example.com/announce", "announce-list": [ ["http://tracker1.example.com/announce"], ["http://tracker2.example.com/announce"] ], "info": { "name": "example_file.txt", "piece length": 262144, "pieces": "<二进制哈希数据>", "length": 1234567 # 单文件时 # 或多文件时的结构 "files": [ {"path": ["dir", "file1.txt"], "length": 12345}, {"path": ["file2.txt"], "length": 67890} ] } }注意:
info字典是整个.torrent文件的核心,它的SHA-1哈希值就是生成磁力链接的关键信息。
2. Python解析Bencoding编码
要解析.torrent文件,我们需要先实现Bencoding解码器。下面是一个完整的Python实现:
import re from collections import OrderedDict def decode_bencode(data): if isinstance(data, bytes): data = data.decode('utf-8') def _decode(s): if s.startswith('i'): # 整数解码 i数字e match = re.match(r'i(-?\d+)e', s) return int(match.group(1)), s[match.end():] elif s.startswith('l'): # 列表解码 l内容e s = s[1:] lst = [] while not s.startswith('e'): item, s = _decode(s) lst.append(item) return lst, s[1:] elif s.startswith('d'): # 字典解码 d键值对e s = s[1:] dic = OrderedDict() while not s.startswith('e'): key, s = _decode(s) value, s = _decode(s) dic[key] = value return dic, s[1:] else: # 字符串解码 长度:内容 match = re.match(r'(\d+):', s) length = int(match.group(1)) start = match.end() end = start + length return s[start:end], s[end:] result, _ = _decode(data) return result这个解码器可以处理所有Bencoding类型,并保持字典的有序性(这在处理.torrent文件时很重要)。让我们测试一下:
# 测试解码器 test_data = b'd4:name8:example12:piece lengthi262144e6:pieces20:\x00\x01\x02\x03...e' decoded = decode_bencode(test_data) print(decoded)3. 读取和解析.torrent文件
现在我们可以编写代码来读取.torrent文件并解析其内容:
import hashlib import urllib.parse def parse_torrent_file(file_path): with open(file_path, 'rb') as f: content = f.read() torrent_data = decode_bencode(content) # 提取基本信息 announce = torrent_data.get('announce', '') announce_list = torrent_data.get('announce-list', []) comment = torrent_data.get('comment', '') created_by = torrent_data.get('created by', '') creation_date = torrent_data.get('creation date', 0) # 处理info字典 info = torrent_data['info'] name = info['name'] piece_length = info['piece length'] pieces = info['pieces'] # 判断是单文件还是多文件 if 'length' in info: # 单文件 file_size = info['length'] files = [{'path': [name], 'length': file_size}] else: # 多文件 files = info['files'] return { 'announce': announce, 'announce_list': announce_list, 'comment': comment, 'created_by': created_by, 'creation_date': creation_date, 'info': { 'name': name, 'piece_length': piece_length, 'pieces': pieces, 'files': files } }使用这个函数,我们可以轻松获取.torrent文件中的所有信息:
torrent_info = parse_torrent_file('example.torrent') print("Tracker服务器:", torrent_info['announce']) print("文件名:", torrent_info['info']['name']) print("文件列表:") for file in torrent_info['info']['files']: print(f"- {'/'.join(file['path'])} ({file['length']} bytes)")4. 生成磁力链接
磁力链接(Magnet URI)的核心是info字典的SHA-1哈希值。下面是生成磁力链接的完整代码:
def generate_magnet_link(torrent_info): # 计算info哈希 info_dict = torrent_info['info'] info_bencoded = bencode(info_dict) # 需要实现bencode函数 info_hash = hashlib.sha1(info_bencoded).hexdigest() # 构建磁力链接参数 params = { 'xt': f'urn:btih:{info_hash}', 'dn': urllib.parse.quote(info_dict['name']), } # 添加主Tracker if torrent_info['announce']: params['tr'] = urllib.parse.quote(torrent_info['announce']) # 添加备用Trackers announce_list = [] for tier in torrent_info.get('announce_list', []): for tracker in tier: if tracker not in announce_list and tracker != torrent_info['announce']: announce_list.append(tracker) # 将参数编码为查询字符串 query = '&'.join(f'{k}={v}' for k, v in params.items()) for tracker in announce_list: query += f'&tr={urllib.parse.quote(tracker)}' return f'magnet:?{query}' # 辅助函数:将Python对象编码为Bencoding格式 def bencode(data): if isinstance(data, int): return f'i{data}e'.encode() elif isinstance(data, str): return f'{len(data)}:{data}'.encode() elif isinstance(data, bytes): return f'{len(data)}:'.encode() + data elif isinstance(data, list): return b'l' + b''.join(bencode(item) for item in data) + b'e' elif isinstance(data, dict): return b'd' + b''.join( bencode(k) + bencode(v) for k, v in sorted(data.items()) ) + b'e' else: raise TypeError(f"Unsupported type for bencoding: {type(data)}")现在我们可以生成完整的磁力链接了:
magnet_link = generate_magnet_link(torrent_info) print("生成的磁力链接:", magnet_link)5. 处理常见问题与错误
在实际应用中,你可能会遇到各种.torrent文件的变体和特殊情况。以下是几个常见问题及其解决方案:
5.1 损坏的.torrent文件
有些.torrent文件可能因为各种原因损坏。我们可以添加一些验证逻辑:
def validate_torrent(torrent_data): required_keys = ['announce', 'info'] for key in required_keys: if key not in torrent_data: raise ValueError(f"Missing required key: {key}") info = torrent_data['info'] info_required = ['name', 'piece length', 'pieces'] for key in info_required: if key not in info: raise ValueError(f"Missing required info key: {key}") if 'length' not in info and 'files' not in info: raise ValueError("Info must contain either 'length' or 'files'") return True5.2 非标准Bencoding实现
有些.torrent生成工具可能不完全遵循Bencoding规范。我们可以增强解码器的容错能力:
def robust_decode_bencode(data): try: return decode_bencode(data) except Exception as e: # 尝试修复常见问题 if isinstance(data, str): data = data.encode('utf-8') # 尝试去除可能的头部垃圾数据 start = data.find(b'd') if start > 0: return decode_bencode(data[start:]) raise ValueError(f"Failed to decode bencoded data: {str(e)}")5.3 大文件处理
对于非常大的.torrent文件,我们可以使用流式处理来避免内存问题:
def stream_parse_torrent(file_path): with open(file_path, 'rb') as f: # 读取足够的数据来获取info字典的位置 chunk = f.read(1024) info_start = chunk.find(b'4:info') # 定位info字典的起始和结束位置 f.seek(info_start) stack = 0 start_pos = f.tell() while True: byte = f.read(1) if not byte: break if byte == b'd': stack += 1 elif byte == b'e': stack -= 1 if stack == 0: end_pos = f.tell() break # 读取info字典部分 f.seek(start_pos) info_data = f.read(end_pos - start_pos) info_dict = decode_bencode(info_data) # 计算info哈希 info_hash = hashlib.sha1(info_data).hexdigest() return { 'info': info_dict, 'info_hash': info_hash }6. 完整示例与进阶应用
让我们把这些代码整合成一个完整的脚本,并添加一些实用功能:
#!/usr/bin/env python3 import argparse import hashlib import json import re from collections import OrderedDict import urllib.parse # 这里插入之前定义的decode_bencode、bencode等函数 class TorrentParser: def __init__(self, file_path): self.file_path = file_path self.raw_data = None self.parsed_data = None self.info_hash = None def parse(self): with open(self.file_path, 'rb') as f: self.raw_data = f.read() self.parsed_data = decode_bencode(self.raw_data) # 计算info哈希 info_start = self.raw_data.find(b'4:info') + 6 info_end = self._find_info_end(info_start) info_bencoded = self.raw_data[info_start-6:info_end] self.info_hash = hashlib.sha1(info_bencoded).hexdigest() return self.parsed_data def _find_info_end(self, start_pos): stack = 0 data = self.raw_data[start_pos:] for i, byte in enumerate(data): if byte == ord('d'): stack += 1 elif byte == ord('e'): stack -= 1 if stack == -1: return start_pos + i + 1 raise ValueError("Invalid info dictionary: missing closing 'e'") def get_magnet_link(self): if not self.parsed_data: self.parse() params = { 'xt': f'urn:btih:{self.info_hash}', 'dn': urllib.parse.quote(self.parsed_data['info']['name']), } # 添加Trackers trackers = set() if 'announce' in self.parsed_data and self.parsed_data['announce']: trackers.add(self.parsed_data['announce']) if 'announce-list' in self.parsed_data: for tier in self.parsed_data['announce-list']: for tracker in tier: if tracker: trackers.add(tracker) query = '&'.join(f'{k}={v}' for k, v in params.items()) for tracker in trackers: query += f'&tr={urllib.parse.quote(tracker)}' return f'magnet:?{query}' def get_file_list(self): if not self.parsed_data: self.parse() info = self.parsed_data['info'] if 'files' in info: return [{'path': '/'.join(f['path']), 'length': f['length']} for f in info['files']] else: return [{'path': info['name'], 'length': info['length']}] def to_json(self, indent=None): if not self.parsed_data: self.parse() def default_encoder(obj): if isinstance(obj, bytes): try: return obj.decode('utf-8') except UnicodeDecodeError: return list(obj) raise TypeError(f"Object of type {type(obj)} is not JSON serializable") return json.dumps(self.parsed_data, indent=indent, default=default_encoder) def main(): parser = argparse.ArgumentParser( description='Parse .torrent files and extract information') parser.add_argument('torrent_file', help='Path to the .torrent file') parser.add_argument('--magnet', action='store_true', help='Generate magnet link') parser.add_argument('--files', action='store_true', help='List files in the torrent') parser.add_argument('--json', action='store_true', help='Output full torrent info as JSON') parser.add_argument('--pretty', action='store_true', help='Pretty-print JSON output') args = parser.parse_args() tp = TorrentParser(args.torrent_file) tp.parse() if args.magnet: print(tp.get_magnet_link()) if args.files: for file in tp.get_file_list(): print(f"{file['path']} ({file['length']} bytes)") if args.json: print(tp.to_json(indent=4 if args.pretty else None)) if __name__ == '__main__': main()这个脚本提供了命令行界面,可以方便地提取.torrent文件中的各种信息:
# 生成磁力链接 python torrent_parser.py example.torrent --magnet # 列出文件 python torrent_parser.py example.torrent --files # 输出完整JSON信息 python torrent_parser.py example.torrent --json --pretty7. 实际应用场景与扩展思路
掌握了.torrent文件解析技术后,你可以开发许多实用工具:
- 批量磁力链接生成器:扫描目录下的所有.torrent文件,生成对应的磁力链接列表
- Torrent文件校验工具:验证.torrent文件的完整性和有效性
- Tracker服务器分析工具:统计.torrent文件中使用的Tracker服务器
- Torrent文件编辑器:修改.torrent文件中的Tracker列表或其他元数据
- 资源搜索工具:通过info哈希值在多个平台搜索资源
下面是一个批量处理的示例:
import os from concurrent.futures import ThreadPoolExecutor def process_torrent_file(file_path): try: tp = TorrentParser(file_path) tp.parse() return { 'file': os.path.basename(file_path), 'name': tp.parsed_data['info']['name'], 'magnet': tp.get_magnet_link(), 'size': sum(f['length'] for f in tp.get_file_list()) } except Exception as e: return { 'file': os.path.basename(file_path), 'error': str(e) } def batch_process(directory): torrent_files = [ os.path.join(directory, f) for f in os.listdir(directory) if f.endswith('.torrent') ] with ThreadPoolExecutor() as executor: results = list(executor.map(process_torrent_file, torrent_files)) # 输出结果 for result in results: if 'error' in result: print(f"{result['file']}: ERROR - {result['error']}") else: print(f"{result['file']}: {result['name']}") print(f" Size: {result['size']} bytes") print(f" Magnet: {result['magnet']}")这个批量处理工具可以高效地处理大量.torrent文件,提取关键信息并生成报告。
