instrumentation/next-share/BaseLib/Core/Statistics/SeedingStatsCrawler.py

   1 # Written by Boxun Zhang, Boudewijn Schoon
   2 # see LICENSE.txt for license information
   3
   4 import sys
   5 import cPickle
   6
   7 from BaseLib.Core.BitTornado.BT1.MessageID import CRAWLER_SEEDINGSTATS_QUERY
   8 from BaseLib.Core.CacheDB.SqliteSeedingStatsCacheDB import *
   9
  10 DEBUG = False
  11
  12 class SeedingStatsCrawler:
  13     __single = None
  14
  15     @classmethod
  16     def get_instance(cls, *args, **kargs):
  17         if not cls.__single:
  18             cls.__single = cls(*args, **kargs)
  19         return cls.__single
  20
  21     def __init__(self):
  22         self._sqlite_cache_db = SQLiteSeedingStatsCacheDB.getInstance()
  23
  24     def query_initiator(self, permid, selversion, request_callback):
  25         """
  26         Established a new connection. Send a CRAWLER_DATABASE_QUERY request.
  27         @param permid The Tribler peer permid
  28         @param selversion The oberlay protocol version
  29         @param request_callback Call this function one or more times to send the requests: request_callback(message_id, payload)
  30         """
  31         if DEBUG:
  32             print >>sys.stderr, "crawler: SeedingStatsDB_update_settings_initiator"
  33         read_query = "SELECT * FROM SeedingStats WHERE crawled = 0"
  34         write_query = "UPDATE SeedingStats SET crawled = 1 WHERE crawled = 0"
  35         return request_callback(CRAWLER_SEEDINGSTATS_QUERY, cPickle.dumps([("read", read_query), ("write", write_query)], 2))
  36
  37     def update_settings_initiator(self, permid, selversion, request_callback):
  38         """
  39         Established a new connection. Send a CRAWLER_DATABASE_QUERY request.
  40         @param permid The Tribler peer permid
  41         @param selversion The oberlay protocol version
  42         @param request_callback Call this function one or more times to send the requests: request_callback(message_id, payload)
  43         """
  44         if DEBUG:
  45             print >>sys.stderr, "crawler: SeedingStatsDB_update_settings_initiator"
  46
  47         try:
  48             sql_update = "UPDATE SeedingStatsSettings SET crawling_interval=%s WHERE crawling_enabled=%s"%(1800, 1)
  49         except:
  50             print_exc()
  51         else:
  52             return request_callback(CRAWLER_SEEDINGSTATS_QUERY, cPickle.dumps(sql_update, 2))
  53
  54
  55     def handle_crawler_request(self, permid, selversion, channel_id, message, reply_callback):
  56         """
  57         Received a CRAWLER_DATABASE_QUERY request.
  58         @param permid The Crawler permid
  59         @param selversion The overlay protocol version
  60         @param channel_id Identifies a CRAWLER_REQUEST/CRAWLER_REPLY pair
  61         @param message The message payload
  62         @param reply_callback Call this function once to send the reply: reply_callback(payload [, error=123])
  63
  64         MESSAGE contains a cPickled list. Each list element is a
  65         tuple. Each tuple consists of a string (either 'read' or
  66         'write') and a string (the query)
  67         """
  68         if DEBUG:
  69             print >> sys.stderr, "crawler: handle_crawler_request", len(message)
  70
  71         results = []
  72         try:
  73             items = cPickle.loads(message)
  74             if DEBUG:
  75                 print >> sys.stderr, "crawler: handle_crawler_request", items
  76
  77             for action, query in items:
  78                 if action == "read":
  79                     cursor = self._sqlite_cache_db.execute_read(query)
  80                 elif action == "write":
  81                     cursor = self._sqlite_cache_db.execute_write(query)
  82                 else:
  83                     raise Exception("invalid payload")
  84
  85                 if cursor:
  86                     results.append(list(cursor))
  87                 else:
  88                     results.append(None)
  89         except Exception, e:
  90             if DEBUG:
  91                 print >> sys.stderr, "crawler: handle_crawler_request", e
  92             results.append(str(e))
  93             reply_callback(cPickle.dumps(results, 2), 1)
  94         else:
  95             reply_callback(cPickle.dumps(results, 2))
  96
  97         return True
  98
  99
 100     def handle_crawler_reply(self, permid, selversion, channel_id, channel_data, error, message, reply_callback):
 101         """
 102         Received a CRAWLER_DATABASE_QUERY request.
 103         @param permid The Crawler permid
 104         @param selversion The overlay protocol version
 105         @param channel_id Identifies a CRAWLER_REQUEST/CRAWLER_REPLY pair
 106         @param error The error value. 0 indicates success.
 107         @param message The message payload
 108         @param request_callback Call this function one or more times to send the requests: request_callback(message_id, payload)
 109         """
 110         if error:
 111             if DEBUG:
 112                 print >> sys.stderr, "seedingstatscrawler: handle_crawler_reply"
 113                 print >> sys.stderr, "seedingstatscrawler: error", error
 114
 115         else:
 116             try:
 117                 results = cPickle.loads(message)
 118
 119                 if DEBUG:
 120                     print >> sys.stderr, "seedingstatscrawler: handle_crawler_reply"
 121                     print >> sys.stderr, "seedingstatscrawler:", results
 122
 123                 # the first item in the list contains the results from the select query
 124                 if results[0]:
 125                     values = map(tuple, results[0])
 126                     self._sqlite_cache_db.insertMany("SeedingStats", values)
 127             except Exception, e:
 128
 129                 # 04/11/08 boudewijn: cPickle.loads(...) sometimes
 130                 # results in EOFError. This may be caused by message
 131                 # being interpreted as non-binary.
 132                 f = open("seedingstats-EOFError.data", "ab")
 133                 f.write("--\n%s\n--\n" % message)
 134                 f.close()
 135
 136                 print_exc()
 137                 return False
 138
 139         return True
 140
 141
 142     def handle_crawler_update_settings_request(self, permid, selversion, channel_id, message, reply_callback):
 143         """
 144         Received a CRAWLER_DATABASE_QUERY request.
 145         @param permid The Crawler permid
 146         @param selversion The overlay protocol version
 147         @param channel_id Identifies a CRAWLER_REQUEST/CRAWLER_REPLY pair
 148         @param message The message payload
 149         @param reply_callback Call this function once to send the reply: reply_callback(payload [, error=123])
 150         """
 151         if DEBUG:
 152             print >> sys.stderr, "crawler: handle_crawler_SeedingStats_request", message
 153
 154         # execute the sql
 155         sql_update = cPickle.loads(message)
 156
 157         try:
 158             self._sqlite_cache_db.execute_write(sql_update)
 159         except Exception, e:
 160             reply_callback(str(e), 1)
 161         else:
 162             reply_callback(cPickle.dumps('Update succeeded.', 2))
 163
 164         return True
 165
 166     def handle_crawler_update_setings_reply(self, permid, selversion, channel_id, message, reply_callback):
 167         """
 168         Received a CRAWLER_DATABASE_QUERY request.
 169         @param permid The Crawler permid
 170         @param selversion The overlay protocol version
 171         @param channel_id Identifies a CRAWLER_REQUEST/CRAWLER_REPLY pair
 172         @param message The message payload
 173         @param request_callback Call this function one or more times to send the requests: request_callback(message_id, payload)
 174         """
 175         if DEBUG:
 176             print >> sys.stderr, "olapps: handle_crawler_SeedingStats_reply"
 177
 178         return True