之前我在写百度网盘爬虫，百度图片爬虫的时候答应网友说，抽时间要把ok搜搜的的源码公开，如今是时候兑现诺言了，下面就是爬虫的所有代码，完全，彻底的公开，你会不会写程序都可以使用，不过请先装个linux系统，具备公网条件，然后运行：
python startCrawler.py
有必要提醒你，数据库字段代码中都有，请你自己建张表格，这个太简单了，就不多说了。同时我也提供一下下载地址，源码都在：下载地址1 下载地址2
#!/usr/bin/env python # encoding: utf-8 """ author:haoning create time:2015.8.1 """ import hashlib import os import time import datetime import traceback import sys import random import json import socket import threading from hashlib import sha1 #进行hash加密 from random import randint from struct import unpack from socket import inet_ntoa from threading import Timer, Thread from time import sleep from collections import deque from Queue import Queue import MySQLdb as mdb #数据库连接器 import metautils import downloadTorrent from bencode import bencode, bdecode import pygeoip DB_HOST = '127.0.0.1' DB_USER = 'root' DB_PASS = 'root' BOOTSTRAP_NODES = ( ("67.215.246.10", 6881), ("82.221.103.244", 6881), ("23.21.224.150", 6881) ) RATE = 1 #调控速率 TID_LENGTH = 2 RE_JOIN_DHT_INTERVAL = 3 TOKEN_LENGTH = 2 INFO_HASH_LEN = 500000 #50w数据很小，限制内存不至于消耗太大 CACHE_LEN = 100 #更新数据库缓存 WAIT_DOWNLOAD = 80 geoip = pygeoip.GeoIP('GeoIP.dat') def is_ip_allowed(ip): country = geoip.country_code_by_addr(ip) if country in ('CN','TW','JP','HK', 'KR'): return True return False def entropy(length): return "".join(chr(randint(0, 255)) for _ in xrange(length)) def random_id(): h = sha1() h.update(entropy(20)) return h.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 timer(t, f): Timer(t, f).start() def get_neighbor(target, nid, end=10): return target[:end]+nid[end:] class KNode(object): def __init__(self, nid, ip, port): self.nid = nid self.ip = ip self.port = port class DHTClient(Thread): def __init__(self, max_node_qsize): Thread.__init__(self) self.setDaemon(True) self.max_node_qsize = max_node_qsize self.nid = random_id() self.nodes = deque(maxlen=max_node_qsize) def send_krpc(self, msg, address): try: self.ufd.sendto(bencode(msg), address) except Exception: pass def send_find_node(self, address, nid=None): nid = get_neighbor(nid, self.nid) if nid else self.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 join_DHT(self): for address in BOOTSTRAP_NODES: self.send_find_node(address) def re_join_DHT(self): if len(self.nodes) == 0: self.join_DHT() timer(RE_JOIN_DHT_INTERVAL, self.re_join_DHT) def auto_send_find_node(self): wait = 1.0 / self.max_node_qsize while True: try: node = self.nodes.popleft() self.send_find_node((node.ip, node.port), node.nid) except IndexError: pass try: sleep(wait) except KeyboardInterrupt: os._exit(0) def process_find_node_response(self, msg, address): nodes = decode_nodes(msg["r"]["nodes"]) for node in nodes: (nid, ip, port) = node if len(nid) != 20: continue if ip == self.bind_ip: continue n = KNode(nid, ip, port) self.nodes.append(n) class DHTServer(DHTClient): #获得info_hash def __init__(self, master, bind_ip, bind_port, max_node_qsize): DHTClient.__init__(self, max_node_qsize) self.master = master self.bind_ip = bind_ip self.bind_port = bind_port self.speed=0 self.process_request_actions = { "get_peers": self.on_get_peers_request, "announce_peer": self.on_announce_peer_request, } self.ufd = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) self.ufd.bind((self.bind_ip, self.bind_port)) timer(RE_JOIN_DHT_INTERVAL, self.re_join_DHT) def run(self): self.re_join_DHT() while True: try: (data, address) = self.ufd.recvfrom(65536) msg = bdecode(data) self.on_message(msg, address) except Exception: pass def on_message(self, msg, address): global RATE #设为全局量 try: if msg["y"] == "r": if msg["r"].has_key("nodes"): self.process_find_node_response(msg, address) #发现节点 elif msg["y"] == "q": try: self.speed+=1 if self.speed % 10000 ==0: RATE=random.randint(1,3) if RATE==2: RATE=1 if RATE==3: RATE=10 if self.speed>100000: self.speed=0 if self.speed % RATE==0: #数据过多，占用cpu太多，划分限速,1,1,10 self.process_request_actions[msg["q"]](msg, address) #处理其他节点的请求，这个过程获取info_hash #self.process_request_actions[msg["q"]](msg, address) #处理其他节点的请求，这个过程获取info_hash except KeyError: self.play_dead(msg, address) except KeyError: pass def on_get_peers_request(self, msg, address): try: infohash = msg["a"]["info_hash"] tid = msg["t"] nid = msg["a"]["id"] token = infohash[:TOKEN_LENGTH] msg = { "t": tid, "y": "r", "r": { "id": get_neighbor(infohash, self.nid), "nodes": "", "token": token } } self.master.log(infohash, address) self.send_krpc(msg, address) except KeyError: pass def on_announce_peer_request(self, msg, address): try: infohash = msg["a"]["info_hash"] token = msg["a"]["token"] nid = msg["a"]["id"] tid = msg["t"] if infohash[:TOKEN_LENGTH] == token: if msg["a"].has_key("implied_port ") and msg["a"]["implied_port "] != 0: port = address[1] else: port = msg["a"]["port"] self.master.log_announce(infohash, (address[0], port)) except Exception: print 'error' pass finally: self.ok(msg, address) def play_dead(self, msg, address): try: tid = msg["t"] msg = { "t": tid, "y": "e", "e": [202, "Server Error"] } self.send_krpc(msg, address) except KeyError: pass def ok(self, msg, address): try: tid = msg["t"] nid = msg["a"]["id"] msg = { "t": tid, "y": "r", "r": { "id": get_neighbor(nid, self.nid) } } self.send_krpc(msg, address) except KeyError: pass class Master(Thread): #解析info_hash def __init__(self): Thread.__init__(self) self.setDaemon(True) self.queue = Queue() self.cache = Queue() self.count=0 self.mutex = threading.RLock() #可重入锁，使单线程可以再次获得已经获得的? self.waitDownload = Queue() self.metadata_queue = Queue() self.dbconn = mdb.connect(DB_HOST, DB_USER, DB_PASS, 'oksousou', charset='utf8') self.dbconn.autocommit(False) self.dbcurr = self.dbconn.cursor() self.dbcurr.execute('SET NAMES utf8') self.visited = set() def lock(self): #加锁 self.mutex.acquire() def unlock(self): #解锁 self.mutex.release() def work(self,item): print "start thread",item while True: self.prepare_download_metadata() self.lock() self.download_metadata() self.unlock() self.lock() self.got_torrent() self.unlock() def start_work(self,max): for item in xrange(max): t = threading.Thread(target=self.work, args=(item,)) t.setDaemon(True) t.start() #入队的种子效率更高 def log_announce(self, binhash, address=None): if self.queue.qsize() < INFO_HASH_LEN : #大于INFO_HASH_LEN就不要入队，否则后面来不及处理 if is_ip_allowed(address[0]): self.queue.put([address, binhash]) #获得info_hash def log(self, infohash, address=None): if self.queue.qsize() < INFO_HASH_LEN: #大于INFO_HASH_LEN/2就不要入队，否则后面来不及处理 if is_ip_allowed(address[0]): self.queue.put([address, infohash]) def prepare_download_metadata(self): if self.queue.qsize() == 0: sleep(2) #从queue中获得info_hash用来下载 address, binhash= self.queue.get() if binhash in self.visited: return if len(self.visited) > 100000: #大于100000重置队列,认为已经访问过了 self.visited = set() self.visited.add(binhash) #跟新已经访问过的info_hash info_hash = binhash.encode('hex') utcnow = datetime.datetime.utcnow() self.cache.put((address,binhash,utcnow)) #装入缓存队列 def download_metadata(self): if self.cache.qsize() > CACHE_LEN/2: #出队更新下载 while self.cache.qsize() > 0: #排空队列 address,binhash,utcnow = self.cache.get() info_hash = binhash.encode('hex') self.dbcurr.execute('SELECT id FROM search_hash WHERE info_hash=%s', (info_hash,)) y = self.dbcurr.fetchone() if y: # 更新最近发现时间，请求数 self.dbcurr.execute('UPDATE search_hash SET last_seen=%s, requests=requests+1 WHERE info_hash=%s', (utcnow, info_hash)) else: self.waitDownload.put((address, binhash)) self.dbconn.commit() if self.waitDownload.qsize() > WAIT_DOWNLOAD: while self.waitDownload.qsize() > 0: address,binhash = self.waitDownload.get() t = threading.Thread(target=downloadTorrent.download_metadata, args=(address, binhash, self.metadata_queue)) t.setDaemon(True) t.start() def decode(self, s): if type(s) is list: s = ';'.join(s) u = s for x in (self.encoding, 'utf8', 'gbk', 'big5'): try: u = s.decode(x) return u except: pass return s.decode(self.encoding, 'ignore') def decode_utf8(self, d, i): if i+'.utf-8' in d: return d[i+'.utf-8'].decode('utf8') return self.decode(d[i]) def parse_metadata(self, data): #解析种子 info = {} self.encoding = 'utf8' try: torrent = bdecode(data) #编码后解析 if not torrent.get('name'): return None except: return None detail = torrent info['name'] = self.decode_utf8(detail, 'name') if 'files' in detail: info['files'] = [] for x in detail['files']: if 'path.utf-8' in x: v = {'path': self.decode('/'.join(x['path.utf-8'])), 'length': x['length']} else: v = {'path': self.decode('/'.join(x['path'])), 'length': x['length']} if 'filehash' in x: v['filehash'] = x['filehash'].encode('hex') info['files'].append(v) info['length'] = sum([x['length'] for x in info['files']]) else: info['length'] = detail['length'] info['data_hash'] = hashlib.md5(detail['pieces']).hexdigest() return info def got_torrent(self): if self.metadata_queue.qsize() == 0: return binhash, address, data,start_time = self.metadata_queue.get() if not data: return try: info = self.parse_metadata(data) if not info: return except: traceback.print_exc() return temp = time.time() x = time.localtime(float(temp)) utcnow = time.strftime("%Y-%m-%d %H:%M:%S",x) # get time now info_hash = binhash.encode('hex') #磁力 info['info_hash'] = info_hash # need to build tags info['tagged'] = False info['classified'] = False info['requests'] = 1 info['last_seen'] = utcnow info['create_time'] = utcnow info['source_ip'] = address[0] if info.get('files'): files = [z for z in info['files'] if not z['path'].startswith('_')] if not files: files = info['files'] else: files = [{'path': info['name'], 'length': info['length']}] files.sort(key=lambda z:z['length'], reverse=True) bigfname = files[0]['path'] info['extension'] = metautils.get_extension(bigfname).lower() info['category'] = metautils.get_category(info['extension']) try: try: print ' ', 'Saved', info['info_hash'], info['name'], (time.time()-start_time), 's', address[0] except: print ' ', 'Saved', info['info_hash'] ret = self.dbcurr.execute('INSERT INTO search_hash(info_hash,category,data_hash,name,extension,classified,source_ip,tagged,' + 'length,create_time,last_seen,requests) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)', (info['info_hash'], info['category'], info['data_hash'], info['name'], info['extension'], info['classified'], info['source_ip'], info['tagged'], info['length'], info['create_time'], info['last_seen'], info['requests'])) if self.count %50 ==0: self.dbconn.commit() if self.count>100000: self.count=0 except: print self.name, 'save error', self.name, info traceback.print_exc() return if __name__ == "__main__": #启动客户端 master = Master() master.start_work(150) #启动服务器 dht = DHTServer(master, "0.0.0.0", 6881, max_node_qsize=200) dht.start() dht.auto_send_find_node()
注意，上面的代码有一段代码需要下载种子，所以下面的这段代码十分重要：
#!/usr/bin/env python # encoding: utf-8 """ author:haoning create time:2015.8.1 """ from hashlib import sha1 import math from socket import inet_ntoa import socket from struct import pack, unpack from threading import Timer, Thread from time import sleep, time from bencode import bencode, bdecode from startCrawler import entropy BT_PROTOCOL = "BitTorrent protocol" BT_MSG_ID = 20 EXT_HANDSHAKE_ID = 0 def random_id(): hash = sha1() hash.update(entropy(20)) return hash.digest() def send_packet(the_socket, msg): the_socket.send(msg) def send_message(the_socket, msg): msg_len = pack(">I", len(msg)) send_packet(the_socket, msg_len + msg) def send_handshake(the_socket, infohash): bt_header = chr(len(BT_PROTOCOL)) + BT_PROTOCOL ext_bytes = "x00x00x00x00x00x10x00x00" peer_id = random_id() packet = bt_header + ext_bytes + infohash + peer_id send_packet(the_socket, packet) def check_handshake(packet, self_infohash): try: bt_header_len, packet = ord(packet[:1]), packet[1:] if bt_header_len != len(BT_PROTOCOL): return False except TypeError: return False bt_header, packet = packet[:bt_header_len], packet[bt_header_len:] if bt_header != BT_PROTOCOL: return False packet = packet[8:] infohash = packet[:20] if infohash != self_infohash: return False return True def send_ext_handshake(the_socket): msg = chr(BT_MSG_ID) + chr(EXT_HANDSHAKE_ID) + bencode({"m":{"ut_metadata": 1}}) send_message(the_socket, msg) def request_metadata(the_socket, ut_metadata, piece): """bep_0009""" msg = chr(BT_MSG_ID) + chr(ut_metadata) + bencode({"msg_type": 0, "piece": piece}) send_message(the_socket, msg) def get_ut_metadata(data): ut_metadata = "_metadata" index = data.index(ut_metadata)+len(ut_metadata) + 1 return int(data[index]) def get_metadata_size(data): metadata_size = "metadata_size" start = data.index(metadata_size) + len(metadata_size) + 1 data = data[start:] return int(data[:data.index("e")]) def recvall(the_socket, timeout=5): the_socket.setblocking(0) total_data = [] data = "" begin = time() while True: sleep(0.05) if total_data and time()-begin > timeout: break elif time()-begin > timeout*2: break try: data = the_socket.recv(1024) if data: total_data.append(data) begin = time() except Exception: pass return "".join(total_data) def download_metadata(address, infohash, metadata_queue, timeout=5): metadata = None start_time = time() the_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: the_socket.settimeout(timeout) the_socket.connect(address) # handshake send_handshake(the_socket, infohash) packet = the_socket.recv(4096) # handshake error if not check_handshake(packet, infohash): return # ext handshake send_ext_handshake(the_socket) packet = the_socket.recv(4096) # get ut_metadata and metadata_size ut_metadata, metadata_size = get_ut_metadata(packet), get_metadata_size(packet) # request each piece of metadata metadata = [] for piece in range(int(math.ceil(metadata_size/(16.0*1024)))): #piece是个控制块，根据控制块下载数据 request_metadata(the_socket, ut_metadata, piece) packet = recvall(the_socket, timeout) #the_socket.recv(1024*17) metadata.append(packet[packet.index("ee")+2:]) metadata = "".join(metadata) except socket.timeout: pass except Exception, e: pass finally: #print "metadata= %s" %(metadata) the_socket.close() #确保没回都关闭socket if metadata != None: #只让不空的种子入? metadata_queue.put((infohash, address, metadata,start_time))
其实下载种子还有一种方式就是借助libtorrent，但这个太耗费cpu了，所以我一般不用他，如下：
#coding: utf8 import threading import traceback import random import time import os import socket import libtorrent as lt threading.stack_size(200*1024) socket.setdefaulttimeout(30) def fetch_torrent(session, ih, timeout): name = ih.upper() url = 'magnet:?xt=urn:btih:%s' % (name,) data = '' params = { 'save_path': '/tmp/downloads/', 'storage_mode': lt.storage_mode_t(2), 'paused': False, 'auto_managed': False, 'duplicate_is_error': True} try: handle = lt.add_magnet_uri(session, url, params) except: return None status = session.status() handle.set_sequential_download(1) meta = None down_time = time.time() down_path = None for i in xrange(0, timeout): if handle.has_metadata(): info = handle.get_torrent_info() down_path = '/tmp/downloads/%s' % info.name() #print 'status', 'p', status.num_peers, 'g', status.dht_global_nodes, 'ts', status.dht_torrents, 'u', status.total_upload, 'd', status.total_download meta = info.metadata() break time.sleep(1) if down_path and os.path.exists(down_path): os.system('rm -rf "%s"' % down_path) session.remove_torrent(handle) return meta def download_metadata(address, binhash, metadata_queue, timeout=20): metadata = None start_time = time.time() try: session = lt.session() r = random.randrange(10000, 50000) session.listen_on(r, r+10) session.add_dht_router('router.bittorrent.com',6881) session.add_dht_router('router.utorrent.com',6881) session.add_dht_router('dht.transmission.com',6881) session.add_dht_router('127.0.0.1',6881) session.start_dht() metadata = fetch_torrent(session, binhash.encode('hex'), timeout) session = None except: traceback.print_exc() finally: metadata_queue.put((binhash, address, metadata,start_time))
这个爬虫还是耗费了本人和其他网上高手的很多时间的，请看到这篇博客的朋友保持钻研精神，开源精神，多多交流，秉承分享。本人建立个qq群作为去转盘网的官方群，人数现在也不多，如果有兴趣的话来逛逛吧，多个粉丝去转盘多一份热闹，qq群号：512245829