使用Python libtorrent创建守护进程以获取100k +种子的元数据

我正在尝试使用python libtorrent获取每天大约10k + torrents的元数据。

这是目前的代码流

  • 启动libtorrent会话。
  • 获取最近一天内我们需要的元数据上传的种子总数。
  • 从数据库中获取洪流散列
  • 使用这些散列创建磁链,并通过为每个磁URI创建句柄来在会话中添加这些磁URI。
  • 在获取元数据时睡一会儿,并持续检查是否找到元数据。
  • 如果接收到元数据,请将其添加到数据库中,否则检查我们是否已经查找了大约10分钟的元数据,如果是,则删除句柄,即不要再查找元数据。
  • 无限期地做。 并为未来保存会话状态。
  • 到目前为止,我已经尝试过。

    #!/usr/bin/env python
    # this file will run as client or daemon and fetch torrent meta data i.e. torrent files from magnet uri
    
    import libtorrent as lt # libtorrent library
    import tempfile # for settings parameters while fetching metadata as temp dir
    import sys #getting arguiments from shell or exit script
    from time import sleep #sleep
    import shutil # removing directory tree from temp directory 
    import os.path # for getting pwd and other things
    from pprint import pprint # for debugging, showing object data
    import MySQLdb # DB connectivity 
    import os
    from datetime import date, timedelta
    
    session = lt.session(lt.fingerprint("UT", 3, 4, 5, 0), flags=0)
    session.listen_on(6881, 6891)
    session.add_extension('ut_metadata')
    session.add_extension('ut_pex')
    session.add_extension('smart_ban')
    session.add_extension('metadata_transfer')
    
    session_save_filename = "/magnet2torrent/magnet_to_torrent_daemon.save_state"
    
    if(os.path.isfile(session_save_filename)):
    
        fileread = open(session_save_filename, 'rb')
        session.load_state(lt.bdecode(fileread.read()))
        fileread.close()
        print('session loaded from file')
    else:
        print('new session started')
    
    session.add_dht_router("router.utorrent.com", 6881)
    session.add_dht_router("router.bittorrent.com", 6881)
    session.add_dht_router("dht.transmissionbt.com", 6881)
    session.add_dht_router("dht.aelitis.com", 6881)
    
    session.start_dht()
    session.start_lsd()
    session.start_upnp()
    session.start_natpmp()
    
    alive = True
    while alive:
    
        db_conn = MySQLdb.connect(  host = '',  user = '',  passwd = '',    db = '',    unix_socket='/mysql/mysql.sock') # Open database connection
        #print('reconnecting')
        #get all records where enabled = 0 and uploaded within yesterday 
        subset_count = 100 ;
    
        yesterday = date.today() - timedelta(1)
        yesterday = yesterday.strftime('%Y-%m-%d %H:%M:%S')
        #print(yesterday)
    
        total_count_query = ("SELECT COUNT(*) as total_count FROM content WHERE upload_date > '"+ yesterday +"' AND enabled = '0' ")
        #print(total_count_query)
        try:
            total_count_cursor = db_conn.cursor()# prepare a cursor object using cursor() method
            total_count_cursor.execute(total_count_query) # Execute the SQL command
            total_count_results = total_count_cursor.fetchone() # Fetch all the rows in a list of lists.
            total_count = total_count_results[0]
            print(total_count)
        except:
                print "Error: unable to select data"
    
        total_pages = total_count/subset_count
        #print(total_pages)
    
        current_page = 1
        while(current_page <= total_pages):
            from_count = (current_page * subset_count) - subset_count
    
            #print(current_page)
            #print(from_count)
    
            hashes = []
    
            get_mysql_data_query = ("SELECT hash FROM content WHERE upload_date > '" + yesterday +"' AND enabled = '0' ORDER BY record_num DESC LIMIT "+ str(from_count) +" , " + str(subset_count) +" ")
            #print(get_mysql_data_query)
            try:
                get_mysql_data_cursor = db_conn.cursor()# prepare a cursor object using cursor() method
                get_mysql_data_cursor.execute(get_mysql_data_query) # Execute the SQL command
                get_mysql_data_results = get_mysql_data_cursor.fetchall() # Fetch all the rows in a list of lists.
                for row in get_mysql_data_results:
                    hashes.append(row[0].upper())
            except:
                print "Error: unable to select data"
    
            #print(hashes)
    
            handles = []
    
            for hash in hashes:
                tempdir = tempfile.mkdtemp()
                add_magnet_uri_params = {
                    'save_path': tempdir,
                    'duplicate_is_error': True,
                    'storage_mode': lt.storage_mode_t(2),
                    'paused': False,
                    'auto_managed': True,
                    'duplicate_is_error': True
                }
                magnet_uri = "magnet:?xt=urn:btih:" + hash.upper() + "&tr=udp%3A%2F%2Ftracker.openbittorrent.com%3A80&tr=udp%3A%2F%2Ftracker.publicbt.com%3A80&tr=udp%3A%2F%2Ftracker.ccc.de%3A80"
                #print(magnet_uri)
                handle = lt.add_magnet_uri(session, magnet_uri, add_magnet_uri_params)
                handles.append(handle) #push handle in handles list
    
            #print("handles length is :")
            #print(len(handles))
    
            while(len(handles) != 0):
                for h in handles:
                    #print("inside handles for each loop")
                    if h.has_metadata():
                        torinfo = h.get_torrent_info()
                        final_info_hash = str(torinfo.info_hash())
                        final_info_hash = final_info_hash.upper()
                        torfile = lt.create_torrent(torinfo)
                        torcontent = lt.bencode(torfile.generate())
                        tfile_size = len(torcontent)
                        try:
                            insert_cursor = db_conn.cursor()# prepare a cursor object using cursor() method
                            insert_cursor.execute("""INSERT INTO dht_tfiles (hash, tdata) VALUES (%s, %s)""",  [final_info_hash , torcontent] )
                            db_conn.commit()
                            #print "data inserted in DB"
                        except MySQLdb.Error, e:
                            try:
                                print "MySQL Error [%d]: %s" % (e.args[0], e.args[1])
                            except IndexError:
                                print "MySQL Error: %s" % str(e)    
    
    
                        shutil.rmtree(h.save_path())    #   remove temp data directory
                        session.remove_torrent(h) # remove torrnt handle from session   
                        handles.remove(h) #remove handle from list
    
                    else:
                        if(h.status().active_time > 600):   # check if handle is more than 10 minutes old i.e. 600 seconds
                            #print('remove_torrent')
                            shutil.rmtree(h.save_path())    #   remove temp data directory
                            session.remove_torrent(h) # remove torrnt handle from session   
                            handles.remove(h) #remove handle from list
                    sleep(1)        
                    #print('sleep1')
    
            #print('sleep10')
            #sleep(10)
            current_page = current_page + 1
    
            #save session state
            filewrite = open(session_save_filename, "wb")
            filewrite.write(lt.bencode(session.save_state()))
            filewrite.close()
    
    
        print('sleep60')
        sleep(60)
    
        #save session state
        filewrite = open(session_save_filename, "wb")
        filewrite.write(lt.bencode(session.save_state()))
        filewrite.close()
    

    我试图在上面的脚本上运行一夜,发现在隔夜会话中只发现大约1200个洪流的元数据。 所以我正在寻找改进脚本的性能。

    我甚至尝试解码save_state文件,并注意到有700多个DHT nodes连接到。 所以它不像DHT没有运行,

    我打算做的是,无限期地keep the handles active ,而不提取元数据。 如果10分钟内没有获取元数据,10分钟后就不会去除句柄,就像我目前正在做的那样。

    我有几个关于lib-torrent python绑定的问题。

  • 我可以继续运行多少把手? 运行手柄有没有限制?
  • 将运行10k +或100k手柄减慢我的系统? 或吃掉资源? 如果是,那么哪些资源? 我的意思是内存,网络?
  • 我在防火墙后面,可能是阻塞的传入端口,导致元数据获取速度慢?
  • DHT服务器可以像router.bittorrent.com或任何其他BAN我的IP地址发送太多的请求?
  • 其他同行可以禁止我的IP地址,如果他们发现我只提取元数据太多请求?
  • 我可以运行这个脚本的多个实例吗? 或者可能是多线程? 它会提供更好的性能?
  • 如果使用同一脚本的多个实例,则每个脚本都将获得唯一的节点ID,具体取决于我使用的IP和端口,这是否可行?
  • 有没有更好的方法? 为了实现我所尝试的?


    我无法回答特定于libtorrent API的问题,但您的一些问题一般适用于bittorrent。

    将运行10k +或100k手柄减慢我的系统? 或吃掉资源? 如果是,那么哪些资源? 我的意思是内存,网络?

    元数据下载不应该占用太多资源,因为它们还没有完整的torrent下载,即它们不能分配实际的文件或类似的东西。 但是一旦他们抓住了第一部分元数据,他们将需要一些内存/磁盘空间用于元数据本身。

    我在防火墙后面,可能是阻塞的传入端口,导致元数据获取速度慢?

    是的,通过减少可建立连接的对等点的数量,在具有较低对等点数的群集上获取元数据(或根本建立任何连接)变得更加困难。

    NAT可能会导致相同的问题。

    DHT服务器可以像router.bittorrent.com或任何其他BAN我的IP地址发送太多的请求?

    router.bittorrent.com是一个引导节点,而不是服务器本身。 查找不查询单个节点,他们查询许多不同(数以百万计)。 但是,是的,个别节点可以禁止,或更可能限制您的速度。

    这可以通过查找随机分布的ID来分散DHT密钥空间中的负载来缓解。

    我可以运行这个脚本的多个实例吗? 或者可能是多线程? 它会提供更好的性能?

    AIUI libtorrent是非阻塞或多线程的,您可以一次安排多个种子。

    我不知道libtorrent是否有DHT请求传出的速率限制。

    如果使用同一个脚本的多个实例,每个脚本将根据我使用的ip和端口获得唯一的节点ID,这种可行的解决方案是什么?

    如果您的意思是DHT节点ID,那么它们是从IP(按照BEP 42)派生的,而不是端口。 虽然包含了一些随机元素,但是每个IP可以获得有限数量的ID。

    其中一些可能也适用于您的情况:http://blog.libtorrent.org/2012/01/seeding-a-million-torrents/

    另一种选择是我自己的DHT实施,其中包括一个批量获取种子的CLI。

    链接地址: http://www.djcxy.com/p/29443.html

    上一篇: creating daemon using Python libtorrent for fetching meta data of 100k+ torrents

    下一篇: Setting up Mesos with Ansible on Ubuntu 14.04 on Digital Ocean