開博第一篇：DHT 爬蟲的學習記錄

經過一段時間的研究和學習，大致了解了DHT網絡的一些信息，大部分還是參會別人的相關代碼，一方面主要對DHT爬蟲原理感興趣，最主要的是為了學習python，大部分是別人的東西原理還是引用別人的吧

DHT網絡爬蟲的實現 | 學步園   http://www.xuebuyuan.com/1287052.html

DHT協議原理以及一些重點分析：

   要做DHT的爬蟲，首先得透徹理解DHT，這樣才能知道在什么地方究竟該應用什么算法去解決問題。關於DHT協議的細節以及重要的參考文章，請參考文末1

   DHT協議作為BT協議的一個輔助，是非常好玩的。它主要是為了在BT正式下載時得到種子或者BT資源。傳統的網絡，需要一台中央服務器存放種子或者BT資源，不僅浪費服務器資源，還容易出現單點的各種問題，而DHT網絡則是為了去中心化，也就是說任意時刻，這個網絡總有節點是亮的，你可以去詢問問這些亮的節點，從而將自己加入DHT網絡。

   要實現DHT協議的網絡爬蟲，主要分3步，第一步是得到資源信息(infohash，160bit，20字節，可以編碼為40字節的十六進制字符串)，第二步是確認這些infohash是有效的，第三步是通過有效的infohash下載到BT的種子文件，從而得到對這個資源的完整描述。

   其中第一步是其他節點用DHT協議中的get_peers方法向爬蟲發送請求得到的，第二步是其他節點用DHT協議中的announce_peer向爬蟲發送請求得到的，第三步可以有幾種方式得到，比如可以去一些保存種子的網站根據infohash直接下載到，或者通過announce_peer的節點來下載到，具體如何實現，可以取決於你自己的爬蟲。

   DHT協議中的主要幾個操作：

 

   主要負責通過UDP與外部節點交互，封裝4種基本操作的請求以及相應。

   ping：檢查一個節點是否“存活”

   在一個爬蟲里主要有兩個地方用到ping，第一是初始路由表時，第二是驗證節點是否存活時

   find_node：向一個節點發送查找節點的請求

   在一個爬蟲中主要也是兩個地方用到find_node，第一是初始路由表時，第二是驗證桶是否存活時

   get_peers：向一個節點發送查找資源的請求

   在爬蟲中有節點向自己請求時不僅像個正常節點一樣做出回應，還需要以此資源的info_hash為機會盡可能多的去認識更多的節點。如圖，get_peers實際上最后一步是announce_peer，但是因為爬蟲不能announce_peer，所以實際上get_peers退化成了find_node操作。

   announce_peer：向一個節點發送自己已經開始下載某個資源的通知

   爬蟲中不能用announce_peer，因為這就相當於通報虛假資源，對方很容易從上下文中判斷你是否通報了虛假資源從而把你禁掉

   DHT協議中有幾個重點的需要澄清的地方：

   1. node與infohash同樣使用160bit的表示方式，160bit意味着整個節點空間有2^160 = 730750818665451459101842416358141509827966271488，是48位10進制，也就是說有百億億億億億個節點空間，這么大的節點空間，是足夠存放你的主機節點以及任意的資源信息的。

   2. 每個節點有張路由表。每張路由表由一堆K桶組成，所謂K桶，就是桶中最多只能放K個節點，默認是8個。而桶的保存則是類似一顆前綴樹的方式。相當於一張8桶的路由表中最多有160-4個K桶。

   3. 根據DHT協議的規定，每個infohash都是有位置的，因此，兩個infohash之間就有距離一說，而兩個infohash的距離就可以用異或來表示，即infohash1 xor infohash2，也就是說，高位一樣的話，他們的距離就近，反之則遠，這樣可以快速的計算兩個節點的距離。計算這個距離有什么用呢，在DHT網絡中，如果一個資源的infohash與一個節點的infohash越近則該節點越有可能擁有該資源的信息，為什么呢？可以想象，因為人人都用同樣的距離算法去遞歸的詢問離資源接近的節點，並且只要該節點做出了回應，那么就會得到一個announce信息，也就是說跟資源infohash接近的節點就有更大的概率拿到該資源的infohash

   4. 根據上述算法，DHT中的查詢是跳躍式查詢，可以迅速的跨越的的節點桶而接近目標節點桶。之所以在遠處能夠大幅度跳躍，而在近處只能小幅度跳躍，原因是每個節點的路由表中離自身越接近的節點保存得越多，如下圖

   5. 在一個DHT網絡中當爬蟲並不容易，不像普通爬蟲一樣，看到資源就可以主動爬下來，相反，因為得到資源的方式(get_peers, announce_peer)都是被動的，所以爬蟲的方式就有些變化了，爬蟲所要做的事就是像個正常節點一樣去響應其他節點的查詢，並且得到其他節點的回應，把其中的數據收集下來就算是完成工作了。而爬蟲唯一能做的，是盡可能的去多認識其他節點，這樣，才能有更多其他節點來向你詢問。

   6. 有人說，那么我把DHT爬蟲的K桶中的容量K增大是不是就能增加得到資源的機會，其實不然，之前也分析過了，DHT爬蟲最重要的信息來源全是被動的，因為你不能增大別人的K，所以距離遠的節點保存你自身的概率就越小，當然距離遠的節點去請求你的概率相對也比較小。

   一些主要的組件（實際實現更加復雜一些，有其他的模塊，這里僅列舉主要幾個）：

   DHT crawler：

   這個就是DHT爬蟲的主邏輯，為了簡化多線程問題，跟server用了生產者消費者模型，負責消費，並且復用server的端口。

   主要任務就是負責初始化，包括路由表的初始化，以及初始的請求。另外負責處理所有進來的消息事件，由於生產者消費者模型的使用，里面的操作都基本上是單線程的，簡化了不少問題，而且相信也比上鎖要提升速度（當然了，加鎖這步按理是放到了queue這里了，不過對於這種生產者源源不斷生產的類型，可以用ring-buffer大幅提升性能）。

   DHT server：

   這里是DHT爬蟲的服務器端，DHT網絡中的節點不單是client，也是server，所以要有server擔當生產者的角色，最初也是每個消費者對應一個生產者，但實際上發現可以利用IO多路復用來達到消息事件的目的，這樣一來大大簡化了系統中線程的數量，如果client可以的話，也應該用同樣的方式來組織，這樣系統的速度應該會快很多。（尚未驗證）

   DHT route table：

   主要負責路由表的操作。

   路由表有如下操作：

   init：剛創建路由表時的操作。分兩種情況：

   1. 如果之前已經初始化過，並且將上次路由表的數據保存下來，則只需要讀入保存數據。

   2. 如果之前沒有初始化過，則首先應當初始化。

   首先，應當有一個接入點，也就是說，你要想加進這個網絡，必須認識這個網絡中某個節點i並將i加入路由表，接下來對i用find_node詢問自己的hash_info，這里巧妙的地方就在於，理論上通過一定數量的詢問就會找到離自己距離很近的節點(也就是經過一定步驟就會收斂)。find_node目的在於盡可能早的讓自己有數據，並且讓網絡上別的節點知道自己，如果別人不認識你，就不會發送消息過來，意味着你也不能獲取到想要的信息。

   search：比較重要的方法，主要使用它來定位當前infohash所在的桶的位置。會被其他各種代理方法調用到。

   findNodes：找到路由表中與傳入的infohash最近的k個節點

   getPeer：找到待查資源是否有peer（即是否有人在下載，也就是是否有人announce過）

   announcePeer：通知該資源正在被下載

   DHT bucket:

   acitiveNode：邏輯比較多，分如下幾點。

 

        1. 查找所要添加的節點對應路由表的桶是否已經滿，如果未滿，添加節點

        2. 如果已經滿，檢查該桶中是否包含爬蟲節點自己，如果不包含，拋棄待添加節點

        3. 如果該桶中包含本節點，則平均分裂該桶

   其他的諸如locateNode, 
replaceNode, updateNode, 
removeNode，就不一一說明了

   DHT torrent parser：  

   主要從bt種子文件中解析出以下幾個重要的信息：name，size，file list(sub file name, sub file size)，比較簡單，用bencode方向解碼就行了

   Utils：

   distance：計算兩個資源之間的距離。在kad中用a xor b表示

 

   為了增加難度，選用了不太熟悉的語言python，結果步步為營，但是也感慨python的簡潔強大。在實現中，也碰到很多有意思的問題。比如如何保存一張路由表中的所有桶，之前想出來幾個辦法，甚至為了節省資源，打算用bit數組+dict直接保存，但是因為估計最終的幾個操作不是很方便直觀容易出錯而放棄，選用的結構就是前綴樹，操作起來果然是沒有障礙；

   在超時問題上，比如桶超時和節點超時，一直在思考一個高效但是比較優雅的做法，可以用一個同步調用然后等待它的超時，但是顯然很低效，尤其我沒有用更多線程的情況，一旦阻塞了就等於該端口所有事件都被阻塞了。所以必須用異步操作，但是異步操作很難去控制它的精確事件，當然，我可以在每個事件來的時候檢查一遍是否超時，但是顯然也是浪費和低效。那么，剩下的只有采用跟tomcat類似的方式了，增加一個線程來監控，當然，這個監控線程最好是全局的，能監控所有crawler中所有事務的超時。另外，超時如果控制不當，容易導致內存沒有回收以至於內存泄露，也值得注意。超時線程是否會與其他線程互相影響也應當仔細檢查。

   最初超時的控制沒處理好，出現了ping storm，運行一定時間后大多數桶已經滿了，如果按照協議中的方式去跑的話會發現大量的事件都是在ping以確認這個節點是否ok以至於大量的cpu用於處理ping和ping響應。深入理解后發現，檢查節點狀態是不需要的，因為節點狀態只是為了提供給詢問的人一些好的節點，既然如此，可以將每次過來的節點替換當前桶中最老的節點，如此一來，我們將總是保存着最新的節點。

   搜索算法也是比較讓我困惑的地方，簡而言之，搜索的目的並不是真正去找資源，而是去認識那些能夠保存你的節點。為什么說是能夠保存你，因為離你越遠，桶的數量越少，這樣一來，要想進他們的桶中去相對來說就比較困難，所以搜索的目標按理應該是附近的節點最好，但是不能排除遠方節點也可能保存你的情況，這種情況會發生在遠方節點初始化時或者遠方節點的桶中節點超時的時候，但總而言之，概率要小些。所以搜索算法也不應該不做判斷就胡亂搜索，但是也不應該將搜索的距離嚴格限制在附近，所以這是一個權衡問題，暫時沒有想到好的方式，覺得暫時讓距離遠的以一定概率發生，而距離近的必然發生

   還有一點，就是搜索速度問題，因為DHT網絡的這種結構，決定了一個節點所認識的其他節點必然是有限的附近節點，於是每個節點在一定時間段內能拿到的資源數必然是有限的，所以應當分配多個節點同時去抓取，而抓取資源的數量很大程度上就跟分配節點的多少有關了。

   最后一個值得優化的地方是findnodes方法，之前的方式是把一個桶中所有數據拿出來排序，然后取其中前K個返回回去，但是實際上我們做了很多額外的工作，這是經典的topN問題，使用排序明顯是浪費時間的，因為這個操作非常頻繁，所以即便所有保存的節點加起來很少（(160 - 4) * 8），也會一定程度上增加時間。而采用的算法是在一篇論文《可擴展的DHT網絡爬蟲設計和優化》中找到的，基本公式是IDi = IDj xor 2 ^(160 - i)，這樣，已知IDi和i就能知道IDj，若已知IDi和IDj就能知道i，通過這種方式，可以快速的查找該桶A附近的其他桶（顯然是離桶A層次最近的桶中的節點距離A次近），比起全部遍歷再查找效率要高不少。

  

    dht協議http://www.bittorrent.org/beps/bep_0005.html 及其翻譯http://gobismoon.blog.163.com/blog/static/5244280220100893055533/

 

爬蟲源碼參考別人的，非原創，只為學習

 

 

#encoding: utf-8

from hashlib import sha1
from random import randint
from struct import unpack, pack
from socket import inet_aton, inet_ntoa
from bisect import bisect_left
from threading import Timer
from time import sleep

from bencode import bencode, bdecode

BOOTSTRAP_NODES = [
    ("router.bittorrent.com", 6881),
    ("dht.transmissionbt.com", 6881),
    ("router.utorrent.com", 6881)
] 
TID_LENGTH = 4
KRPC_TIMEOUT = 10
REBORN_TIME = 5 * 60
K = 8

def entropy(bytes):
    s = ""
    for i in range(bytes):
        s += chr(randint(0, 255))
    return s

    # """把爬蟲"偽裝"成正常node, 一個正常的node有ip, port, node ID三個屬性, 因為是基於UDP協議,   
    # 所以向對方發送信息時, 即使沒"明確"說明自己的ip和port時, 對方自然會知道你的ip和port,   
    # 反之亦然. 那么我們自身node就只需要生成一個node ID就行, 協議里說到node ID用sha1算法生成,   
    # sha1算法生成的值是長度是20 byte, 也就是20 * 8 = 160 bit, 正好如DHT協議里說的那范圍: 0 至 2的160次方,   
    # 也就是總共能生成1461501637330902918203684832716283019655932542976個獨一無二的node.   
    # ok, 由於sha1總是生成20 byte的值, 所以哪怕你寫SHA1(20)或SHA1(19)或SHA1("I am a 2B")都可以,   
    # 只要保證大大降低與別人重復幾率就行. 注意, node ID非十六進制,   
    # 也就是說非FF5C85FE1FDB933503999F9EB2EF59E4B0F51ECA這個樣子, 即非hash.hexdigest(). """
def random_id():
    hash = sha1()
    hash.update( entropy(20) )
    return hash.digest()

def decode_nodes(nodes):
    n = []
    length = len(nodes)
    if (length % 26) != 0: 
        return n
    for i in range(0, length, 26):
        nid = nodes[i:i+20]
        ip = inet_ntoa(nodes[i+20:i+24])
        port = unpack("!H", nodes[i+24:i+26])[0]
        n.append( (nid, ip, port) )
    return n

def encode_nodes(nodes):
    strings = []
    for node in nodes:
        s = "%s%s%s" % (node.nid, inet_aton(node.ip), pack("!H", node.port))
        strings.append(s)

    return "".join(strings)

def intify(hstr):
    #"""這是一個小工具, 把一個node ID轉換為數字. 后面會頻繁用到.""" 
    return long(hstr.encode('hex'), 16)    #先轉換成16進制, 再變成數字

def timer(t, f):
    Timer(t, f).start()


class BucketFull(Exception):
    pass


class KRPC(object):
    def __init__(self):
        self.types = {
            "r": self.response_received,
            "q": self.query_received
        }
        self.actions = {
            "ping": self.ping_received,
            "find_node": self.find_node_received,
            "get_peers": self.get_peers_received,
            "announce_peer": self.announce_peer_received,
        }

        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.socket.bind(("0.0.0.0", self.port))

    def response_received(self, msg, address):
        self.find_node_handler(msg)

    def query_received(self, msg, address):
        try:
            self.actions[msg["q"]](msg, address)
        except KeyError:
            pass

    def send_krpc(self, msg, address):
        try:
            self.socket.sendto(bencode(msg), address)
        except:
            pass


class Client(KRPC):
    def __init__(self, table):
        self.table = table

        timer(KRPC_TIMEOUT, self.timeout)
        timer(REBORN_TIME, self.reborn)
        KRPC.__init__(self)

    def find_node(self, address, nid=None):
        nid = self.get_neighbor(nid) if nid else self.table.nid
        tid = entropy(TID_LENGTH)
        
        msg = {
            "t": tid,
            "y": "q",
            "q": "find_node",
            "a": {"id": nid, "target": random_id()}
        }
        self.send_krpc(msg, address)

    def find_node_handler(self, msg):
        try:
            nodes = decode_nodes(msg["r"]["nodes"])
            for node in nodes:
                (nid, ip, port) = node
                if len(nid) != 20: continue
                if nid == self.table.nid: continue
                self.find_node( (ip, port), nid )
        except KeyError:
            pass

    def joinDHT(self):
        for address in BOOTSTRAP_NODES: 
            self.find_node(address)

    def timeout(self):
        if len( self.table.buckets ) < 2:
            self.joinDHT()
        timer(KRPC_TIMEOUT, self.timeout)

    def reborn(self):
        self.table.nid = random_id()
        self.table.buckets = [ KBucket(0, 2**160) ]
        timer(REBORN_TIME, self.reborn)

    def start(self):
        self.joinDHT()

        while True:
            try:
                (data, address) = self.socket.recvfrom(65536)
                msg = bdecode(data)
                self.types[msg["y"]](msg, address)
            except Exception:
                pass

    def get_neighbor(self, target):
        return target[:10]+random_id()[10:]


class Server(Client):
    def __init__(self, master, table, port):
        self.table = table
        self.master = master
        self.port = port
        Client.__init__(self, table)

    def ping_received(self, msg, address):
        try:
            nid = msg["a"]["id"]
            msg = {
                "t": msg["t"],
                "y": "r",
                "r": {"id": self.get_neighbor(nid)}
            }
            self.send_krpc(msg, address)
            self.find_node(address, nid)
        except KeyError:
            pass

    def find_node_received(self, msg, address):
        try:
            target = msg["a"]["target"]
            neighbors = self.table.get_neighbors(target)
            
            nid = msg["a"]["id"]
            msg = {
                "t": msg["t"],
                "y": "r",
                "r": {
                    "id": self.get_neighbor(target), 
                    "nodes": encode_nodes(neighbors)
                }
            }
            self.table.append(KNode(nid, *address))
            self.send_krpc(msg, address)
            self.find_node(address, nid)
        except KeyError:
            pass

    def get_peers_received(self, msg, address):
        try:
            infohash = msg["a"]["info_hash"]

            neighbors = self.table.get_neighbors(infohash)

            nid = msg["a"]["id"]
            msg = {
                "t": msg["t"],
                "y": "r",
                "r": {
                    "id": self.get_neighbor(infohash), 
                    "nodes": encode_nodes(neighbors)
                }
            }
            self.table.append(KNode(nid, *address))
            self.send_krpc(msg, address)
            self.master.log(infohash)
            self.find_node(address, nid)
        except KeyError:
            pass

    def announce_peer_received(self, msg, address):
        try:
            infohash = msg["a"]["info_hash"]
            nid = msg["a"]["id"]

            msg = { 
                "t": msg["t"],
                "y": "r",
                "r": {"id": self.get_neighbor(infohash)}
            }

            self.table.append(KNode(nid, *address))
            self.send_krpc(msg, address)
            self.master.log(infohash)
            self.find_node(address, nid)
        except KeyError:
            pass
# 該類只實例化一次. 
class KTable(object):
    # 這里的nid就是通過node_id()函數生成的自身node ID. 協議里說道, 每個路由表至少有一個bucket,   
 #    還規定第一個bucket的min=0, max=2^160次方, 所以這里就給予了一個buckets屬性來存儲bucket, 這個是列表.
    def __init__(self, nid):
        self.nid = nid
        self.buckets = [ KBucket(0, 2**160) ]

    def append(self, node):
        index = self.bucket_index(node.nid)
        try:
            bucket = self.buckets[index]
            bucket.append(node)
        except IndexError:
            return
        except BucketFull:
            if not bucket.in_range(self.nid): 
                return
            self.split_bucket(index)
            self.append(node)


        # 返回與目標node ID或infohash的最近K個node.  
 
        # 定位出與目標node ID或infohash所在的bucket, 如果該bucuck有K個節點, 返回.   
        # 如果不夠到K個節點的話, 把該bucket前面的bucket和該bucket后面的bucket加起來, 只返回前K個節點.  
        # 還是不到K個話, 再重復這個動作. 要注意不要超出最小和最大索引范圍.  
        # 總之, 不管你用什么算法, 想盡辦法找出最近的K個節點.  
    def get_neighbors(self, target):
        nodes = []
        if len(self.buckets) == 0: return nodes
        if len(target) != 20 : return nodes

        index = self.bucket_index(target)
        try:
            nodes = self.buckets[index].nodes
            min = index - 1
            max = index + 1

            while len(nodes) < K and ((min >= 0) or (max < len(self.buckets))):
                if min >= 0:
                    nodes.extend(self.buckets[min].nodes)

                if max < len(self.buckets):
                    nodes.extend(self.buckets[max].nodes)

                min -= 1
                max += 1

            num = intify(target)
            nodes.sort(lambda a, b, num=num: cmp(num^intify(a.nid), num^intify(b.nid)))
            return nodes[:K] #K是個常量, K=8 
        except IndexError:
            return nodes

    def bucket_index(self, target):
        return bisect_left(self.buckets, intify(target))


        # 拆表  
 
        # index是待拆分的bucket(old bucket)的所在索引值.   
        # 假設這個old bucket的min:0, max:16. 拆分該old bucket的話, 分界點是8, 然后把old bucket的max改為8, min還是0.   
        # 創建一個新的bucket, new bucket的min=8, max=16.  
        # 然后根據的old bucket中的各個node的nid, 看看是屬於哪個bucket的范圍里, 就裝到對應的bucket里.   
        # 各回各家,各找各媽.  
        # new bucket的所在索引值就在old bucket后面, 即index+1, 把新的bucket插入到路由表里. 
    def split_bucket(self, index):
        old = self.buckets[index]
        point = old.max - (old.max - old.min)/2
        new = KBucket(point, old.max)
        old.max = point
        self.buckets.insert(index + 1, new)
        for node in old.nodes[:]:
            if new.in_range(node.nid):
                new.append(node)
                old.remove(node)

    def __iter__(self):
        for bucket in self.buckets:
            yield bucket


class KBucket(object):
    __slots__ = ("min", "max", "nodes")

    # min和max就是該bucket負責的范圍, 比如該bucket的min:0, max:16的話,   
    # 那么存儲的node的intify(nid)值均為: 0到15, 那16就不負責, 這16將會是該bucket后面的bucket的min值.   
    # nodes屬性就是個列表, 存儲node. last_accessed代表最后訪問時間, 因為協議里說到,   
    # 當該bucket負責的node有請求, 回應操作; 刪除node; 添加node; 更新node; 等這些操作時,   
    # 那么就要更新該bucket, 所以設置個last_accessed屬性, 該屬性標志着這個bucket的"新鮮程度". 用linux話來說, touch一下.  
    # 這個用來便於后面說的定時刷新路由表.  

    def __init__(self, min, max):
        self.min = min
        self.max = max
        self.nodes = []


    # 添加node, 參數node是KNode實例.  

    # 如果新插入的node的nid屬性長度不等於20, 終止.  
    # 如果滿了, 拋出bucket已滿的錯誤, 終止. 通知上層代碼進行拆表.  
    # 如果未滿, 先看看新插入的node是否已存在, 如果存在, 就替換掉, 不存在, 就添加,  
    # 添加/替換時, 更新該bucket的"新鮮程度".  
    def append(self, node):
        if node in self:
            self.remove(node)
            self.nodes.append(node)
        else:
            if len(self) < K:
                self.nodes.append(node)
            else:
                raise BucketFull

    def remove(self, node):
        self.nodes.remove(node)

    def in_range(self, target):
        return self.min <= intify(target) < self.max

    def __len__(self):
        return len(self.nodes)

    def __contains__(self, node):
        return node in self.nodes

    def __iter__(self):
        for node in self.nodes:
            yield node

    def __lt__(self, target):
        return self.max <= target


class KNode(object):
     # """  
  #       nid就是node ID的簡寫, 就不取id這么模糊的變量名了. __init__方法相當於別的OOP語言中的構造方法,   
  #       在python嚴格來說不是構造方法, 它是初始化, 不過, 功能差不多就行.  
  #       """ 
    __slots__ = ("nid", "ip", "port")
    
    def __init__(self, nid, ip, port):
        self.nid = nid
        self.ip = ip
        self.port = port

    def __eq__(self, other):
        return self.nid == other.nid



#using example
class Master(object):
    def __init__(self, f):
        self.f = f

    def log(self, infohash):
        self.f.write(infohash.encode("hex")+"\n")
        self.f.flush()
try:
    f = open("infohash.log", "a")
    m = Master(f)
    s = Server(Master(f), KTable(random_id()), 8001)
    s.start()     
except KeyboardInterrupt:
    s.socket.close()
    f.close()

 

種子從迅雷下，初期為學習從http://torrage.com/sync/下的infohash，去重用了別人寫的Bloom Filter算法,數據庫用Mysql，建表語句如下，其中uinthash是根據infohash的頭四個字節和最后四個字節組成的一個int整數，先這樣設計，看后期查詢的時候用得到不，總覺得用infohash來查很慢

CREATE TABLE `torrentinfo` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `infohash` char(40) NOT NULL DEFAULT '',
  `filename` varchar(128) DEFAULT NULL,
  `filelength` bigint(11) DEFAULT NULL,
  `recvtime` datetime DEFAULT NULL,
  `filecontent` text,
  `uinthash` int(11) unsigned NOT NULL DEFAULT '0',
  PRIMARY KEY (`id`),
  KEY `uinthash_index` (`uinthash`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

 

　

Thunder.py

# _*_ coding: utf-8 _*_
import socket
import os,glob
import time as time_p
import requests
from bencode import bdecode, BTL
from torrent import *
import threading, signal
import MySQLdb
from BloomFilter import *

class Thunder(object):
    def __init__(self):
        self.connstr={'host':'127.0.0.1','user':'root','passwd':'123456','port':3306,'charset':"UTF8"}
    def download(self, infohash):
        try:
            tc = self._download(infohash)
            if(tc==-1):
                return
            tc = bdecode(tc)
            info = torrentInfo(tc)
            # print info['name']
            # print info['length']
            # print info['files']
            uint=int(infohash[:4]+infohash[-4:],16)
            time_now=time_p.strftime('%Y-%m-%d %H:%M:%S',time_p.localtime(time_p.time()))
            sql="insert into torrentinfo(infohash,filename,filelength,recvtime,filecontent,uinthash) values('%s','%s','%d','%s','%s','%d')"%(infohash,MySQLdb.escape_string(info['name']),info['length'],time_now,MySQLdb.escape_string(info['files']),uint)
            self.executeSQL(sql)
        except Exception,e:
            print  e
            pass

    def openConnection(self):
        try:
            self.conn=MySQLdb.connect(**self.connstr)
            self.cur=self.conn.cursor()
            self.conn.select_db('dht')
        except MySQLdb.Error,e:
            print 'mysql error %d:%s'%(e.args[0],e.args[1])


    def executeSQL(self,sql):
        try:
            self.cur.execute(sql)
            self.conn.commit()
        except MySQLdb.Error,e:
            print 'mysql error %d:%s'%(e.args[0],e.args[1])
    def closeConnection(self):
        try:
            self.cur.close()
            self.conn.close()
        except MySQLdb.Error,e:
            print 'mysql error %d:%s'%(e.args[0],e.args[1])    

    def _download(self, infohash):
        infohash = infohash.upper()
        start = infohash[0:2]
        end = infohash[-2:]
        url = "http://bt.box.n0808.com/%s/%s/%s.torrent" % (start, end, infohash)
        headers = {
            "Referer": "http://bt.box.n0808.com"
        }
        try:
            r = requests.get(url, headers=headers, timeout=10)
            if r.status_code == 200:
                # f=open("d:\\"+infohash+'.torrent','wb')
                # f.write(r.content)
                # f.close()
                return r.content
        except (socket.timeout, requests.exceptions.Timeout), e:
            pass
        return -1

class torrentBean(object):
    """docstring for torrentBean"""
    __slots__=('infohash','filename','recvtime','filecontent','uinthash')

    def __init__(self, infohash,filename,recvtime,filecontent,uinthash):
        super(torrentBean, self).__init__()
        self.infohash = infohash
        self.filename = filename
        self.recvtime = recvtime
        self.filecontent = filecontent
        self.uinthash = uinthash


bf = BloomFilter(0.001, 1000000)
a=Thunder()
a.openConnection()
# info_hash="a02d2735e6e1daa6f7d58f21bd7340a7b7c4b7a5"
# info_hash='cf3a6a4f07da0b90beddae838462ca0012bef285'
# a.download('cf3a6a4f07da0b90beddae838462ca0012bef285')


files=glob.glob('./*.txt')
for fl in files:
    print os.path.basename(fl)
    f=open(fl,'r')
    for line in f:
        infohash=line.strip('\n')
        if not bf.is_element_exist(infohash):
            bf.insert_element(infohash)
            a.download(infohash)
a.closeConnection()

 

　　torrent種子文件經過bencode解析，獲取key為info對應value值，種子大致的格式如下，有亂碼，不影響觀看

{
    'files': [{
        'path': ['PGD660.avi'],
        'length': 1367405512,
        'filehash': 'J\xef\xfe\xb3K\xd4g\x8d\x07m\x03\xbb\xb3\xadt\xa1\xa0\xf0\xec\xab',
        'ed2k': '/\xfb\xe55#n\xbd1\xb6\x1c\x0f\xf3\xe4\x9dP\xfb',
        'path.utf-8': ['PGD660.avi']
    }, {
        'path': ['PGD660B.jpg'],
        'length': 135899,
        'filehash': '*$O\x17w\xe9E\x95>O\x1f\xfb\x0e\x9b\x16\x15B\\Q\x9d',
        'ed2k': 'T/L*\xbb\x8e.\xe2d\xddu\nR\x07\xca\x19',
        'path.utf-8': ['PGD660B.jpg']
    }, {
        'path': ['yoy123@\xe8\x8d\x89\xe6\xa6\xb4\xe7\xa4\xbe\xe5\x8c\xba@\xe6\x9c\x80\xe6\x96\xb0\xe5\x9c\xb0\xe5\x9d\x80.mht'],
        'length': 472,
        'filehash': '&\xa92\xb7\xdd8\xeel3\xcc-S\x07\xb5e\xd35\xc0\xb7r',
        'ed2k': '\x13\xd2 a\x0cA\xb4\xf2X\x12\xea\xd4\xe8\xac`\x92',
        'path.utf-8': ['yoy123@\xe8\x8d\x89\xe6\xa6\xb4\xe7\xa4\xbe\xe5\x8c\xba@\xe6\x9c\x80\xe6\x96\xb0\xe5\x9c\xb0\xe5\x9d\x80.mht']
    }, {
        'path': ['yoy123@\xe8\x8d\x89\xe6\xa6\xb4\xe7\xa4\xbe\xe5\x8c\xba\xe5\xae\xa3\xe4\xbc\xa0.txt'],
        'length': 363,
        'filehash': '\x96nA*\xe2\xb6Y+[\xe3\xaf\xd4\x14A\x94\xf5@\xcd\xc1\x91',
        'ed2k': '8V\xa6X\xd9\x82l\xdbNO8\xe8D\xe9E\xed',
        'path.utf-8': ['yoy123@\xe8\x8d\x89\xe6\xa6\xb4\xe7\xa4\xbe\xe5\x8c\xba\xe5\xae\xa3\xe4\xbc\xa0.txt']
    }, {
        'path': ['\xe2\x98\x85\xe5\xb0\x91\xe5\xa6\x87 \xe8\xae\xba\xe5\x9d\x9b \xe9\x99\x90\xe9\x87\x8f\xe5\xbc\x80\xe6\x94\xbe\xe4\xb8\xad\xe3\x80\x82\xe3\x80\x82.mht'],
        'length': 475,
        'filehash': '\xec\xde\xeb-6\x86\x1avB\xdd\xd8q\x8b\x8f\xc06\xf0XX\x0e',
        'ed2k': '\xa7\x8dU\xfd\xfc=\x12\x15>yE\x8f&A\xc2u',
        'path.utf-8': ['\xe2\x98\x85\xe5\xb0\x91\xe5\xa6\x87 \xe8\xae\xba\xe5\x9d\x9b \xe9\x99\x90\xe9\x87\x8f\xe5\xbc\x80\xe6\x94\xbe\xe4\xb8\xad\xe3\x80\x82\xe3\x80\x82.mht']
    }, {
        'path': ['\xe6\x9f\x8f\xe6\x8b\x89\xe5\x9c\x96\xe7\xa7\x98\xe5\xaf\x86\xe8\x8a\xb1\xe5\x9c\x92.mht'],
        'length': 478,
        'filehash': "\xe4\xb5'Td\x0b=P\xc0\x9aG\xa2\xd7\xfapg\xc6.\x8e\xa7",
        'ed2k': '\xdd\x8d\xbb\x0b\x04\xcb\x03O\xb1\x18"\x03\xb1\x1d\xba\x08',
        'path.utf-8': ['\xe6\x9f\x8f\xe6\x8b\x89\xe5\x9c\x96\xe7\xa7\x98\xe5\xaf\x86\xe8\x8a\xb1\xe5\x9c\x92.mht']
    }, {
        'path': ['\xe7\xbe\x8e\xe5\xa5\xb3\xe4\xb8\x8a\xe9\x96\x80\xe6\x8f\xb4\xe4\xba\xa4\xe6\x9c\x8d\xe5\x8b\x99.mht'],
        'length': 478,
        'filehash': "\xe4\xb5'Td\x0b=P\xc0\x9aG\xa2\xd7\xfapg\xc6.\x8e\xa7",
        'ed2k': '\xdd\x8d\xbb\x0b\x04\xcb\x03O\xb1\x18"\x03\xb1\x1d\xba\x08',
        'path.utf-8': ['\xe7\xbe\x8e\xe5\xa5\xb3\xe4\xb8\x8a\xe9\x96\x80\xe6\x8f\xb4\xe4\xba\xa4\xe6\x9c\x8d\xe5\x8b\x99.mht']
    }],
    'publisher': 'yoy123',
    'piece length': 524288,
    'name': 'PGD660 \xe6\x83\xb3\xe8\xa9\xa6\xe8\x91\x97\xe5\x85\xa8\xe5\x8a\x9b\xe6\x93\x8d\xe6\x93\x8d\xe7\x9c\x8b\xe9\x80\x99\xe5\x80\x8b\xe6\xb7\xab\xe8\x95\xa9\xe7\xbe\x8e\xe5\xa5\xb3\xe5\x97\x8e \xe5\xb0\x8f\xe5\xb7\x9d\xe3\x81\x82\xe3\x81\x95\xe7\xbe\x8e',
    'publisher.utf-8': 'yoy123',
}

解析代碼torrent.py

# _*_ coding: utf-8 _*_
from time import time

def torrentInfo(torrentContent):
    metadata = torrentContent["info"]
    print metadata
    info = {
        "name": getName(metadata),
        "length": calcLength(metadata),
        "timestamp": getCreateDate(torrentContent),
        "files": extraFiles(metadata)
    }
    return info

def calcLength(metadata):
    length = 0
    try:
        length = metadata["length"]
    except KeyError:
        try:
            for file in metadata["files"]:
                length += file["length"]
        except KeyError:
            pass
    return length

def extraFiles(metadata):
    files = []
    try:
        for file in metadata["files"]:
            path = file["path.utf-8"]
            size=file['length']
            if len(path) > 1:
                main = path[0]
                for f in path[1:2]:
                    files.append("%s/%s    %d bytes" % (main, f,size))
            else:
                files.append("%s    %d bytes" % (path[0],size) )
        if files:
            return '\r\n'.join(files)
        else:
            return getName(metadata)
    except KeyError:
        return getName(metadata)

def getName(metadata):
    try:
        name = metadata["name.utf-8"]
        if name.strip()=="": 
                raise KeyError
    except KeyError:
        try:
            name = metadata["name"]
            if name.strip()=="": 
                raise KeyError
        except KeyError:
            name = getMaxFile(metadata)

    return name
def getMaxFile(metadata):
    try:
        maxFile = metadata["files"][0]
        for file in metadata["files"]:
            if file["length"] > maxFile["length"]:
                maxFile = file
        name = maxFile["path"][0]
        return name
    except KeyError:
        return ""

def getCreateDate(torrentContent):
    try:
        timestamp = torrentContent["creation date"]
    except KeyError:
        timestamp = int( time() )
    return timestamp

 

　　最后還有別人寫的BloomFilter代碼

#encoding: utf-8
'''
Created on 2012-11-7

@author: palydawn
'''
import cmath
from BitVector import BitVector

class BloomFilter(object):
    def __init__(self, error_rate, elementNum):
        #計算所需要的bit數
        self.bit_num = -1 * elementNum * cmath.log(error_rate) / (cmath.log(2.0) * cmath.log(2.0))
        
        #四字節對齊
        self.bit_num = self.align_4byte(self.bit_num.real)
        
        #分配內存
        self.bit_array = BitVector(size=self.bit_num)
        
        #計算hash函數個數
        self.hash_num = cmath.log(2) * self.bit_num / elementNum
        
        self.hash_num = self.hash_num.real
        
        #向上取整
        self.hash_num = int(self.hash_num) + 1
        
        #產生hash函數種子
        self.hash_seeds = self.generate_hashseeds(self.hash_num)
        
    def insert_element(self, element):
        for seed in self.hash_seeds:
            hash_val = self.hash_element(element, seed)
            #取絕對值
            hash_val = abs(hash_val)
            #取模，防越界
            hash_val = hash_val % self.bit_num
            #設置相應的比特位
            self.bit_array[hash_val] = 1
    
    #檢查元素是否存在，存在返回true，否則返回false 
    def is_element_exist(self, element):
        for seed in self.hash_seeds:
            hash_val = self.hash_element(element, seed)
            #取絕對值
            hash_val = abs(hash_val)
            #取模，防越界
            hash_val = hash_val % self.bit_num
            
            #查看值
            if self.bit_array[hash_val] == 0:
                return False
        return True
        
    #內存對齊    
    def align_4byte(self, bit_num):
        num = int(bit_num / 32)
        num = 32 * (num + 1)
        return num
    
    #產生hash函數種子,hash_num個素數
    def generate_hashseeds(self, hash_num):
        count = 0
        #連續兩個種子的最小差值
        gap = 50
        #初始化hash種子為0
        hash_seeds = []
        for index in xrange(hash_num):
            hash_seeds.append(0)
        for index in xrange(10, 10000):
            max_num = int(cmath.sqrt(1.0 * index).real)
            flag = 1
            for num in xrange(2, max_num):
                if index % num == 0:
                    flag = 0
                    break
            
            if flag == 1:
                #連續兩個hash種子的差值要大才行
                if count > 0 and (index - hash_seeds[count - 1]) < gap:
                    continue
                hash_seeds[count] = index
                count = count + 1
            
            if count == hash_num:
                break
        return hash_seeds
    
    def hash_element(self, element, seed):
        hash_val = 1
        for ch in str(element):
            chval = ord(ch)
            hash_val = hash_val * seed + chval
        return hash_val


    def SaveBitToFile(self,f):
        self.bit_array.write_bits_to_fileobject(f)
        pass

 

表內容見下圖