Last active
February 28, 2019 16:26
-
-
Save maliubiao/f4d7448eb2b48a9575cd to your computer and use it in GitHub Desktop.
access torrent data
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import torrent | |
import sys | |
import os.path | |
import time | |
def get_meta(path): | |
f = open(path, "r") | |
s = f.read() | |
f.close() | |
meta = torrent.loads(s) | |
meta["torrent"] = os.path.basename(path) | |
if "creation date" in meta: | |
meta["creation date"] = time.ctime(meta["creation date"]) | |
return meta | |
def print_overview(meta): | |
print "torrent:", meta["torrent"] | |
keys = ("encoding", | |
"creation date", | |
"created by", | |
"announce", | |
"comment") | |
for key in keys: | |
if key in meta: | |
print "============\n%s: %s" % (key, meta[key]) | |
def print_info_overview(meta): | |
print "torrent:", meta["torrent"] | |
info = meta["info"] | |
keys = ("publisher", "name", "publisher-url") | |
for key in keys: | |
if key in info: | |
print "============\n%s: %s" % (key, info[key]) | |
def print_info_files(meta): | |
print "torrent:", meta["torrent"] | |
files = meta["info"]["files"] | |
keys = ("path", "length", "filehash", "ed2k") | |
for file in files: | |
print "*************************" | |
for key in keys: | |
if key not in file: | |
continue | |
if key == "path": | |
print "============\n%s: %s" % (key, ", ".join(file[key])) | |
elif key == "filehash": | |
print "============\n%s: %s" % (key, | |
file[key].encode("HEX").upper()) | |
elif key == "ed2k": | |
print "============\n%s: %s" % (key, | |
file[key].encode("HEX").upper()) | |
def print_usage(): | |
print """readtorrrent.py option torrentfile" | |
""" | |
def main(): | |
if len(sys.argv) < 3: | |
print_usage() | |
arg1 = sys.argv[1] | |
path = sys.argv[2] | |
if arg1 == "-o": | |
print_overview(get_meta(path)) | |
elif arg1 == "-i": | |
print_info_overview(get_meta(path)) | |
elif arg1 == "-f": | |
print_info_files(get_meta(path)) | |
elif arg1 == "-r": | |
replace_files(path) | |
if __name__ == "__main__": | |
main() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import torrent | |
import os.path | |
import sys | |
import uuid | |
def get_meta(path): | |
f = open(path, "r") | |
s = f.read() | |
f.close() | |
return torrent.loads(s) | |
def generate_name_insert(encoding, name): | |
try: | |
name = name.decode(encoding) | |
except: | |
name = name.decode("utf8") | |
ext = name.rfind(".") | |
if ext < 0: | |
final = "=".join([x for x in name]) | |
else: | |
final = "=".join([x for x in name[:ext]]) + name[ext:] | |
return final.encode("utf-8") | |
def generate_name_random(encoding, name): | |
try: | |
name = name.decode(encoding) | |
except: | |
name = name.decode("utf8") | |
ext = name.rfind(".") | |
if ext < 0: | |
final = str(uuid.uuid4()) | |
else: | |
final = (str(uuid.uuid4()) + name[ext:]) | |
return final.encode("utf-8") | |
def main(): | |
option = sys.argv[1] | |
name = sys.argv[2] | |
if option == "-r": | |
generate_name = generate_name_random | |
else: | |
generate_name = generate_name_insert | |
info = get_meta(name) | |
#torrent_name | |
encoding = info["encoding"].lower() | |
if encoding == "utf-8": | |
encoding = "utf8" | |
newname = generate_name(encoding, info["info"]["name"]) | |
info["info"]["name"] = newname | |
if "name.utf-8" in info["info"]: | |
unewname = generate_name(encoding, info["info"]["name.utf-8"]) | |
info["info"]["name.utf-8"] = unewname | |
#override encoding | |
info["encoding"] = "utf-8" | |
#torrent files | |
files = info["info"]["files"] | |
for f in files: | |
if "path" in f: | |
for i, p in enumerate(f["path"]): | |
newpath = generate_name(encoding, p) | |
f["path"][i] = newpath | |
if "path.utf-8" in f: | |
for i, p in enumerate(f["path.utf-8"]): | |
newpath = generate_name(encoding, p) | |
f["path.utf-8"][i] = newpath | |
f = open(os.path.basename(name), "w+") | |
f.write(torrent.dumps(info)) | |
f.close() | |
if __name__ == "__main__": | |
main() | |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#coding=utf8 | |
""" | |
针对在windows下运行 | |
python torrent-generate-ed2k.py dir | |
dir是包含种子的目录 | |
脚本把种子里的文件分别转换成ed2k单文件链接 | |
这样可以避开网盘对torrent hash的封杀 | |
""" | |
import torrent | |
import pdb | |
import sys | |
import os | |
def convert_torrent_to_ed2k(torrent_fn): | |
meta = torrent.loads(open(torrent_fn, "rb").read()) | |
text = [] | |
for file in meta["info"]["files"]: | |
if "ed2k" not in file: continue | |
template = "ed2k://|file|%s|%s|%s|/" | |
fn = "".join(file["path"]).decode("utf8") | |
fn_utf8 = fn.encode("utf8") | |
goout = False | |
for word in ("注册", "美女", "直播", "文宣"): | |
if word in fn_utf8: | |
goout = True | |
if goout: | |
continue | |
ext = fn.split(".")[-1].lower() | |
if ext in ("txt", "html", "htm", "mht", "url", "jpg", "chm", "rar", "gif", "png"): | |
continue | |
fn = fn.encode("gb18030", "ignore") | |
link = template %(fn, file["length"], file["ed2k"].encode("hex")) | |
text.append("%s\n" %( link)) | |
print "".join(text) | |
if __name__ == "__main__": | |
d = sys.argv[1] | |
for fn in os.listdir(d): | |
if fn.endswith(".torrent"): | |
try: | |
convert_torrent_to_ed2k(os.path.join(d, fn)) | |
except: | |
print "bad torrent", fn | |
continue |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#coding=utf8 | |
""" | |
解析种子文件, 像json.loads一样使用 | |
""" | |
import hashlib | |
import pdb | |
def loads(bencoded_string): | |
be = bencoded_string.strip("\n") | |
stack = [[]] | |
lastkey = None | |
i = 0 | |
info_start = 0 | |
ulen = len(bencoded_string) | |
counter = 0 | |
while i < ulen and counter < ulen: | |
counter += 1 | |
char = be[i] | |
#dict d3:key8:items...e | |
if char == "d": | |
last = stack[-1] | |
keyoffset = be.find(":", i) | |
if keyoffset < 0: | |
raise ValueError("torrent.loads: Error") | |
else: | |
stack.append({}) | |
if isinstance(last, list): | |
last.append(stack[-1]) | |
elif isinstance(last, dict): | |
last[lastkey] = stack[-1] | |
lastkey = None | |
keylen = int(be[i+1:keyoffset]) | |
lastkey = be[keyoffset+1:keyoffset+1+keylen] | |
stack[-1][lastkey] = None | |
i = keylen + keyoffset + 1 | |
#list lc....e | |
elif char == "l": | |
last = stack[-1] | |
stack.append([]) | |
if isinstance(last, list): | |
last.append(stack[-1]) | |
elif isinstance(last, dict): | |
last[lastkey] = stack[-1] | |
lastkey = None | |
i += 1 | |
elif char == "e": | |
del stack[-1] | |
i += 1 | |
elif char == "i": | |
last = stack[-1] | |
integer = None | |
intoffset = be.find("e", i) | |
if intoffset < 0: | |
raise ValueError("torrent.loads int error") | |
else: | |
integer = int(be[i+1:intoffset]) | |
i = intoffset+1 | |
if isinstance(last, list): | |
last.append(integer) | |
elif isinstance(last, dict): | |
if lastkey: | |
last[lastkey] = integer | |
lastkey = None | |
else: | |
last[string] = None | |
lastkey = string | |
elif char.isdigit(): | |
last = stack[-1] | |
string = None | |
stroffset = be.find(":", i) | |
if stroffset < 0: | |
raise ValueError("torrent.loads: str Error") | |
else: | |
strlen = int(be[i:stroffset]) | |
string = be[stroffset+1:stroffset+1+strlen] | |
i = stroffset+1+strlen | |
if isinstance(last, list): | |
last.append(string) | |
elif isinstance(last, dict): | |
if lastkey: | |
last[lastkey] = string | |
lastkey = None | |
else: | |
last[string] = None | |
lastkey = string | |
if len(stack[0]) == 1: | |
return stack[0][0] | |
else: | |
return stack[0] | |
def dumps(obj): | |
#buffer | |
string = [] | |
#stack | |
s = [obj] | |
#siblings | |
cs = [[obj]] | |
#loop | |
ls = [id(obj)] | |
lastkey = [] | |
while s: | |
obj = s[-1] | |
siblings = cs[-1] | |
if isinstance(obj, dict): | |
string.append("d") | |
items = obj.items() | |
#empty dict | |
if not items: | |
raise ValueError("empty dict") | |
s.append(items[0]) | |
cs.append(items) | |
for k, v in items: | |
if (id(k) in ls) or (id(v) in ls): | |
raise ValueError("loop detected") | |
ls.append(id(items[0])) | |
elif isinstance(obj, list): | |
string.append("l") | |
#emtpy list | |
if not obj: | |
raise ValueError("empty list") | |
s.append(obj[0]) | |
cs.append(obj) | |
for i in obj: | |
if id(i) in ls: | |
raise ValueError("loop detected") | |
ls.append(id(obj[0])) | |
elif isinstance(obj, tuple): | |
#child of a dict | |
if len(obj) != 2 or (not isinstance(s[-2], dict)): | |
raise ValueError("don't use tuple: %s" % obj) | |
if not isinstance(obj[0], str): | |
raise ValueError("illegal key type") | |
#write key | |
string.append("%d:%s" % (len(obj[0]), obj[0])) | |
#write value | |
s[-1] = obj[1] | |
ls[-1] = id(obj[1]) | |
lastkey.append(obj[0]) | |
elif isinstance(obj, (str, int)): | |
if isinstance(obj, str): | |
if not str: | |
raise ValueError("empty string") | |
string.append("%d:%s" % (len(obj), obj)) | |
else: | |
string.append("i%de" % obj) | |
while len(s) != 1: | |
p = s[-1] | |
ps = cs[-1] | |
if isinstance(s[-2], dict): | |
pindex = ps.index((lastkey.pop(), p)) | |
else: | |
#find me | |
i = 0 | |
while True: | |
pindex = ps.index(p, i) | |
if id(ps[pindex]) == id(p): | |
break | |
i = pindex + 1 | |
assert pindex >= 0 | |
#go to next sibling | |
pindex += 1 | |
if pindex < len(ps): | |
s[-1] = ps[pindex] | |
break | |
#or go to parent | |
s.pop() | |
ls.pop() | |
cs.pop() | |
string.append("e") | |
#done | |
if len(s) == 1: | |
break | |
else: | |
raise ValueError("illegal type: %s" % type(obj)) | |
return "".join(string) | |
def get_info_hash(torrent): | |
if isinstance(torrent, file): | |
k = torrent.read() | |
else: | |
k = torrent | |
if not (k[0] == "d" and k[-1] == "e"): | |
raise ValueError("broken torrent") | |
info_offset = k.find("infod") | |
if info_offset < 0: | |
raise ValueError("broken torrent") | |
if not k[info_offset+5].isdigit(): | |
raise ValueError("broken torrent") | |
end_offset = k.find("5:nodes") | |
if end_offset < 0: | |
end_offset = -1 | |
bencoded = k[info_offset+4:end_offset] | |
return hashlib.sha1(bencoded).hexdigest().upper() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment