text: #! usr/bin/python
  # encoding=utf-8
  
  import socket
  import codecs
  import time
  from threading import Thread
  from collections import deque
  from multiprocessing import Process, cpu_count
  
  import bencoder
  
  from .utils import get_logger, get_nodes_info, get_rand_id, get_neighbor
  from .database import RedisClient
  
  # 服务器 tracker
  BOOTSTRAP_NODES = [
      "udp://tracker.open-internet.nl:6969/announce",
      "udp://tracker.coppersurfer.tk:6969/announce",
      "udp://exodus.desync.com:6969/announce",
      "udp://tracker.opentrackr.org:1337/announce",
      "udp://tracker.internetwarriors.net:1337/announce",
      "udp://9.rarbg.to:2710/announce",
      "udp://public.popcorn-tracker.org:6969/announce",
      "udp://tracker.vanitycore.co:6969/announce",
      "https://1.track.ga:443/announce",
      "udp://tracker.tiny-vps.com:6969/announce",
      "udp://tracker.cypherpunks.ru:6969/announce",
      "udp://thetracker.org:80/announce",
      "udp://tracker.torrent.eu.org:451/announce",
      "udp://retracker.lanta-net.ru:2710/announce",
      "udp://bt.xxx-tracker.com:2710/announce",
      "http://retracker.telecom.by:80/announce",
      "http://retracker.mgts.by:80/announce",
      "http://0d.kebhana.mx:443/announce",
      "udp://torr.ws:2710/announce",
      "udp://open.stealth.si:80/announce",
      ("router.bittorrent.com", 6881),
      ("dht.transmissionbt.com", 6881),
      ("router.utorrent.com", 6881),
  ]
  
  # 双端队列容量
  MAX_NODE_QSIZE = 10000
  # UDP 报文 buffsize
  UDP_RECV_BUFFSIZE = 65535
  # 服务 host
  SERVER_HOST = "0.0.0.0"
  # 服务端口
  SERVER_PORT = 9090
  # 磁力链接前缀
  MAGNET_PER = "magnet:?xt=urn:btih:{}"
  # while 循环休眠时间
  SLEEP_TIME = 1e-5
  # 节点 id 长度
  PER_NID_LEN = 20
  # 执行 bs 定时器间隔（秒）
  PER_SEC_BS_TIMER = 8
  # 是否使用全部进程
  MAX_PROCESSES = cpu_count() // 2 or cpu_count()
  
  
  class HNode:
      def __init__(self, nid, ip=None, port=None):
          self.nid = nid
          self.ip = ip
          self.port = port
  
  
  class DHTServer:
      def __init__(self, bind_ip, bind_port, process_id):
          self.bind_ip = bind_ip
          self.bind_port = bind_port
          self.process_id = process_id
          self.nid = get_rand_id()
          # nodes 节点是一个双端队列
          self.nodes = deque(maxlen=MAX_NODE_QSIZE)
          # KRPC 协议是由 bencode 编码组成的一个简单的 RPC 结构，使用 UDP 报文发送。
          self.udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
          # UDP 地址绑定
          self.udp.bind((self.bind_ip, self.bind_port))
          # redis 客户端
          self.rc = RedisClient()
          self.logger = get_logger("logger_{}".format(bind_port))
  
      def bootstrap(self):
          """
          利用 tracker 服务器，伪装成 DHT 节点，加入 DHT 网络
          """
          for address in BOOTSTRAP_NODES:
              self.send_find_node(address)
  
      def bs_timer(self):
          """
          定时执行 bootstrap()
          """
          t = 1
          while True:
              if t % PER_SEC_BS_TIMER == 0:
                  t = 1
                  self.bootstrap()
              t += 1
              time.sleep(1)
  
      def send_krpc(self, msg, address):
          """
          发送 krpc 协议
  
          :param msg: 发送 UDP 报文信息
          :param address: 发送地址，(ip, port) 元组
          """
          try:
              # msg 要经过 bencode 编码
              self.udp.sendto(bencoder.bencode(msg), address)
          except:
              pass
  
      def send_error(self, tid, address):
          """
          发送错误回复
          """
          msg = dict(t=tid, y="e", e=[202, "Server Error"])
          self.send_krpc(msg, address)
  
      def send_find_node(self, address, nid=None):
          """
          发送 find_node 请求。
  
          `find_node 请求`
          find_node 被用来查找给定 ID 的节点的联系信息。这时 KPRC 协议中的
          "q" == "find_node"。find_node 请求包含 2 个参数，第一个参数是 id，
          包含了请求节点的 ID。第二个参数是 target，包含了请求者正在查找的
          节点的 ID。当一个节点接收到了 find_node 的请求，他应该给出对应的
          回复，回复中包含 2 个关键字 id 和 nodes，nodes 是字符串类型，
          包含了被请求节点的路由表中最接近目标节点的 K(8) 个最接近的节点的联系信息。
  
          `示例`
          参数: {"id" : "<querying nodes id>", "target" : "<id of target node>"}
          回复: {"id" : "<queried nodes id>", "nodes" : "<compact node info>"}
  
          :param address: 地址元组（ip, port)
          :param nid: 节点 id
          """
          nid = get_neighbor(nid) if nid else self.nid
          tid = get_rand_id()
          msg = dict(
              t=tid,
              y="q",
              q="find_node",  # 指定请求为 find_node
              a=dict(id=nid, target=get_rand_id()),
          )
          self.send_krpc(msg, address)
  
      def send_find_node_forever(self):
          """
          循环发送 find_node 请求
          """
          self.logger.info("send find node forever...")
          while True:
              try:
                  # 弹出一个节点
                  node = self.nodes.popleft()
                  self.send_find_node((node.ip, node.port), node.nid)
                  time.sleep(SLEEP_TIME)
              except IndexError:
                  # 一旦节点队列为空，则重新加入 DHT 网络
                  self.bootstrap()
  
      def save_magnet(self, info_hash):
          """
          将磁力链接保存到数据库
  
          :param info_hash:  磁力链接的 info_hash
          """
          # 使用 codecs 解码 info_hash
          hex_info_hash = codecs.getencoder("hex")(info_hash)[0].decode()
          magnet = MAGNET_PER.format(hex_info_hash)
          self.rc.add_magnet(magnet)
          # self.logger.info("pid " + str(self.process_id) + " - " + magnet)
          self.logger.info("pid_{0} - {1}".format(self.process_id, magnet))
  
      def on_message(self, msg, address):
          """
          负责返回信息的处理
  
          :param msg: 报文信息
          :param address: 报文地址
          """
          try:
              # `回复`
              # 对应于 KPRC 消息字典中的 y 关键字的值是 r，包含了一个附加的关键字 r。
              # 关键字 r 是字典类型，包含了返回的值。发送回复消息是在正确解析了请求消息的
              # 基础上完成的。
              if msg[b"y"] == b"r":
                  # nodes 是字符串类型，包含了被请求节点的路由表中最接近目标节点
                  # 的 K个最接近的节点的联系信息。
                  if msg[b"r"].get(b"nodes", None):
                      self.on_find_node_response(msg)
              # `请求`
              # 对应于 KPRC 消息字典中的 y 关键字的值是 q，它包含 2 个附加的关键字
              # q 和 a。关键字 q 是字符串类型，包含了请求的方法名字。关键字 a 一个字典
              # 类型包含了请求所附加的参数。
              # 而实际上我们只需要获取这两者中的 info hash，用于构造磁力链接进而获取种子。
              elif msg[b"y"] == b"q":
                  # get_peers 与 torrent 文件的 info_hash 有关。这时 KPRC 协议中的
                  # "q" = "get_peers"。get_peers 请求包含 2 个参数。第一个参数是 id，
                  # 包含了请求节点的 ID。第二个参数是 info_hash，它代表 torrent 文件的 info_hash
                  if msg[b"q"] == b"get_peers":
                      self.on_get_peers_request(msg, address)
                  # announce_peer 表明请求的节点正在某个端口下载 torrent
                  # 文件。announce_peer 包含 4 个参数。第一个参数是 id，包含了请求节点的 ID；
                  # 第二个参数是 info_hash，包含了 torrent 文件的 info_hash；第三个参数是 port
                  # 包含了整型的端口号，表明 peer 在哪个端口下载；第四个参数数是 token，
                  # 这是在之前的 get_peers 请求中收到的回复中包含的。
                  elif msg[b"q"] == b"announce_peer":
                      self.on_announce_peer_request(msg, address)
          except KeyError:
              pass
  
      def on_find_node_response(self, msg):
          """
          解码 nodes 节点信息，并存储在双端队列
  
          :param msg: 节点报文信息
          """
          nodes = get_nodes_info(msg[b"r"][b"nodes"])
          for node in nodes:
              nid, ip, port = node
              # 进行节点有效性判断
              if len(nid) != PER_NID_LEN or ip == self.bind_ip:
                  continue
              # 将节点加入双端队列
              self.nodes.append(HNode(nid, ip, port))
  
      def on_get_peers_request(self, msg, address):
          """
          处理 get_peers 请求，获取 info hash
  
          :param msg: 节点报文信息
          :param address: 节点地址
          """
          tid = msg[b"t"]
          try:
              info_hash = msg[b"a"][b"info_hash"]
              self.save_magnet(info_hash)
          except KeyError:
              # 没有对应的 info hash，发送错误回复
              self.send_error(tid, address)
  
      def on_announce_peer_request(self, msg, address):
          """
          处理 get_announce 请求，获取 info hash，address, port
          本爬虫目的暂时只是爬取磁链，所以忽略 address, port 有需要的
          开发者可自行完善这部分内容
  
          :param msg: 节点报文信息
          :param address: 节点地址
          """
          tid = msg[b"t"]
          try:
              info_hash = msg[b"a"][b"info_hash"]
              self.save_magnet(info_hash)
          except KeyError:
              # 没有对应的 info hash，发送错误回复
              self.send_error(tid, address)
  
      def receive_response_forever(self):
          """
          循环接受 udp 数据
          """
          self.logger.info(
              "receive response forever {}:{}".format(self.bind_ip, self.bind_port)
          )
          # 首先加入到 DHT 网络
          self.bootstrap()
          while True:
              try:
                  # 接受返回报文
                  data, address = self.udp.recvfrom(UDP_RECV_BUFFSIZE)
                  # 使用 bdecode 解码返回数据
                  msg = bencoder.bdecode(data)
                  # 处理返回信息
                  self.on_message(msg, address)
                  time.sleep(SLEEP_TIME)
              except Exception as e:
                  self.logger.warning(e)
  
  
  def _start_thread(offset):
      """
      启动线程
  
      :param offset: 端口偏移值
      """
      dht = DHTServer(SERVER_HOST, SERVER_PORT + offset, offset)
      threads = [
          Thread(target=dht.send_find_node_forever),
          Thread(target=dht.receive_response_forever),
          Thread(target=dht.bs_timer),
      ]
  
      for t in threads:
          t.start()
  
      for t in threads:
          t.join()
  
  
  def start_server():
      """
      多线程启动服务
      """
      processes = []
      for i in range(MAX_PROCESSES):
          processes.append(Process(target=_start_thread, args=(i,)))
  
      for p in processes:
          p.start()
  
      for p in processes:
          p.join()
  