python开发的 dht网络爬虫

使用 libtorrent 的python绑定库实现一个dht网络爬虫，抓取dht网络中的磁力链接。

dht 网络简介

p2p网络

在P2P网络中，通过种子文件下载资源时，要知道资源在P2P网络中哪些计算机中，这些传输资源的计算机称作peer。在传统的P2P网络中，使用tracker服务器跟踪资源的peer。要下载资源，首先需要取得这些peer。

dht网络

tracker服务器面临一些版权和法律问题。于是出现了DHT，它把tracker上的资源peer信息分散到了整个网络中。dht网络是由分布 式节点构成，节点(node)是实现了DHT协议的p2p客户端。P2P客户端程序既是peer也是node。DHT网络有多种算法，常用的有 Kademlia。

dht网络下载

P2P客户端使用种子文件下载资源时，如果没有tracker服务器，它就向DHT网络查询资源的peer列表， 然后从peer下载资源。

Magnet是磁力链接

资源的标识在DHT网络中称为infohash，是一个通过sha1算法得到的20字节长的字符串。infohash是使用种子文件的文件描述信息 计算得到。磁力链接是把infohash编码成16进制字符串得到。P2P客户端使用磁力链接，下载资源的种子文件，然后根据种子文件下载资源。

Kademlia 算法

Kademlia是DHT网络的一种实现， 具体的算法参见：DHT协议

KRPC 协议

KRPC 是节点之间的交互协议，使用UDP来传送。

包括4种请求：ping，find_node，get_peer，announce_peer。其中get_peer和announce_peer是节点间查询资源的主要消息。

dht 爬虫原理

主要的思路就是伪装为p2p客户端，加入dht网络，收集dht网络中的get_peer和announce_peer消息，这些消息是其他node发送给伪装的p2p客户端的udp消息。

本文dht爬虫的实现

爬虫运行环境

1. linux 系统

2. python 2.7

3. libtorrent 库的python绑定

4. twisted 网络库

5. 防火墙开启固定的udp和tcp端口

libtorrent 库的介绍

libtorrent库是p2p下载的客户端库，有丰富的接口，可以用来开发下载p2p网络上的资源。它有python的绑定库，本爬虫就是使用它的python库开发的。

在libtorrent中有几个概念需要解释下。 session 相当于p2p客户端，session开启一个tcp和一个udp端口，用来与其他p2p客户端交换数据。可以在一个进程内定义多个session，也就是多个p2p客户端，来加快收集速度。

alert是libtorrent中用来收集各种消息的队列，每个session都有一个自己的alert消息队列。KRPC协议的get_peer和announce_peer消息也是从这个队列中获取，就是用这两个消息收集磁力链接的。

主要实现代码

爬虫实现的主要代码比较简单

# 事件通知处理函数
    def _handle_alerts(self, session, alerts):
        while len(alerts):
            alert = alerts.pop()
            # 获取dht_announce_alert和dht_get_peer_alert消息
            # 从这两消息收集磁力链接
            if isinstance(alert, lt.add_torrent_alert):
                alert.handle.set_upload_limit(self._torrent_upload_limit)
                alert.handle.set_download_limit(self._torrent_download_limit)
            elif isinstance(alert, lt.dht_announce_alert):
                info_hash = alert.info_hash.to_string().encode('hex')
                if info_hash in self._meta_list:
                    self._meta_list[info_hash] += 1
                else:
                    self._meta_list[info_hash] = 1
                    self._current_meta_count += 1
            elif isinstance(alert, lt.dht_get_peers_alert):
                info_hash = alert.info_hash.to_string().encode('hex')
                if info_hash in self._meta_list:
                    self._meta_list[info_hash] += 1
                else:
                    self._infohash_queue_from_getpeers.append(info_hash)
                    self._meta_list[info_hash] = 1
                    self._current_meta_count += 1

    def start_work(self):
        '''主工作循环，检查消息，显示状态'''
        # 清理屏幕
        begin_time = time.time()
        show_interval = self._delay_interval
        while True:
            for session in self._sessions:
                session.post_torrent_updates()
                # 从队列中获取信息
                self._handle_alerts(session, session.pop_alerts())
            time.sleep(self._sleep_time)
            if show_interval > 0:
                show_interval -= 1
                continue
            show_interval = self._delay_interval

            # 统计信息显示
            show_content = ['torrents:']
            interval = time.time() - begin_time
            show_content.append('  pid: %s' % os.getpid())
            show_content.append('  time: %s' %
                                time.strftime('%Y-%m-%d %H:%M:%S'))
            show_content.append('  run time: %s' % self._get_runtime(interval))
            show_content.append('  start port: %d' % self._start_port)
            show_content.append('  collect session num: %d' %
                                len(self._sessions))
            show_content.append('  info hash nums from get peers: %d' %
                                len(self._infohash_queue_from_getpeers))
            show_content.append('  torrent collection rate: %f /minute' %
                                (self._current_meta_count * 60 / interval))
            show_content.append('  current torrent count: %d' %
                                self._current_meta_count)
            show_content.append('  total torrent count: %d' %
                                len(self._meta_list))
            show_content.append('n')

            # 存储运行状态到文件
            try:
                with open(self._stat_file, 'wb') as f:
                    f.write('n'.join(show_content))
                with open(self._result_file, 'wb') as f:
                    json.dump(self._meta_list, f)
            except Exception as err:
                pass

            # 测试是否到达退出时间
            if interval >= self._exit_time:
                # stop
                break

            # 每天结束备份结果文件
            self._backup_result()

        # 销毁p2p客户端
        for session in self._sessions:
            torrents = session.get_torrents()
            for torrent in torrents:
                session.remove_torrent(torrent)

运行效率

在我的一台512M内存，单cpu机器上。爬虫刚开始运行稍慢，运行几分钟后收集速度稳定在 180个每分钟，1小时采集10000左右。

运行状态

run times: 12

torrents:
  pid: 11480  time: 2014-08-18 22:45:01
  run time: day: 0, hour: 0, minute: 12, second: 25
  start port: 32900
  collect session num: 20
  info hash nums from get peers: 2222
  torrent collection rate: 179.098480 /minute
  current torrent count: 2224
  total torrent count: 58037

爬虫完整代码

完整的代码参见：https://github.com/blueskyz/DHTCrawler

还包括一个基于twisted的监控进程，用来查看爬虫状态，在爬虫进程退出后重新启动。

原文链接：python开发的 dht网络爬虫