ppf/new/storage.py

   1 """
   2 Storage class for P2P logging information.
   3
   4 Built on previous work by Adriana Draghici and Razvan Deaconescu.
   5
   6 2011, Razvan Deaconescu, razvan.deaconescu@cs.pub.ro
   7 2011, Mariana Marasoiu, mariana.marasoiu@gmail.com
   8 """
   9
  10 import os
  11 import os.path
  12 import re
  13 import logging
  14 import sqlite3
  15 import MySQLdb
  16 import datetime
  17
  18 #
  19 # Logging code heavily inspired by Logging HOWTO documentation:
  20 #     http://docs.python.org/dev/howto/logging.html#configuring-logging
  21 #
  22
  23 # Create logger; default logging level is DEBUG.
  24 logger = logging.getLogger(__name__)
  25 logger.setLevel(logging.DEBUG)
  26
  27 # Create console handler and set level to ERROR.
  28 ch = logging.StreamHandler()
  29 ch.setLevel(logging.DEBUG)
  30
  31 # Create formatter.
  32 formatter = logging.Formatter('%(filename)s:%(lineno)s - %(levelname)s: %(message)s')
  33
  34 # Add formatter to console handler.
  35 ch.setFormatter(formatter)
  36
  37 # Add console handler to logger.
  38 logger.addHandler(ch)
  39
  40
  41 message_types = {
  42         'CHOKE': {'id': 1, 'parameters': None},
  43         'UNCHOKE': {'id': 2, 'parameters': None},
  44         'INTERESTED': {'id': 3, 'parameters': None},
  45         'NOT_INTERESTED': {'id': 4, 'parameters': None},
  46         'HAVE': {'id': 5, 'parameters': None},
  47         'BITFIELD': {'id': 6, 'parameters': None},
  48         'REQUEST': {'id': 7, 'parameters': None},
  49         'PIECE': {'id': 8, 'parameters': None},
  50         'CANCEL': {'id': 9, 'parameters': None},
  51         'DHT_PORT': {'id': 10, 'parameters': None}
  52 }
  53
  54 bittorrent_clients = {
  55         'Tribler': {
  56             'id': 1,
  57             'language': 'Python',
  58             'url': 'http://www.tribler.org/trac',
  59             'dht_support': True,
  60             'streaming_support': True,
  61             'pxe_support': None,
  62             'features': None
  63         },
  64         'NextShare': {
  65             'id': 2,
  66             'language': 'Python',
  67             'url': 'https://trac.p2p-next.org/',
  68             'dht_support': True,
  69             'streaming_support': True,
  70             'pxe_support': None,
  71             'features': None
  72         },
  73         'libtorrent-rasterbar': {
  74             'id': 3,
  75             'language': 'C++',
  76             'url': 'http://www.rasterbar.com/products/libtorrent/',
  77             'dht_support': True,
  78             'streaming_support': True,
  79             'pxe_support': None,
  80             'features': None
  81         },
  82         'Vuze': {
  83             'id': 4,
  84             'language': 'Java',
  85             'url': 'http://www.vuze.com/',
  86             'dht_support': True,
  87             'streaming_support': True,
  88             'pxe_support': None,
  89             'features': None
  90         },
  91         'Transmission': {
  92             'id': 5,
  93             'language': 'C',
  94             'url': 'http://www.transmissionbt.com/',
  95             'dht_support': True,
  96             'streaming_support': False,
  97             'pxe_support': None,
  98             'features': None
  99         },
 100         'Aria': {
 101             'id': 6,
 102             'language': 'C',
 103             'url': 'http://aria2.sourceforge.net/',
 104             'dht_support': True,
 105             'streaming_support': False,
 106             'pxe_support': None,
 107             'features': None
 108         },
 109         'Mainline': {
 110             'id': 7,
 111             'language': 'Python',
 112             'url': 'http://www.bittorrent.com/',
 113             'dht_support': True,
 114             'streaming_support': False,
 115             'pxe_support': None,
 116             'features': None
 117         }
 118 }
 119
 120 transfer_directions = {
 121         'receive': 1,
 122         'send': 2
 123 }
 124
 125 class Swarm(object):
 126     """ Class mimics a C structure. """
 127     def __init__(self, torrent_filename=None, data_size=None,
 128             description=None):
 129         self.torrent_filename = torrent_filename
 130         self.data_size = data_size
 131         self.description = description
 132
 133 class ClientSession(object):
 134     """ Class mimics a C structure. """
 135     # TODO: Add timezone.
 136     def __init__(self, swarm_id=None, btclient=None, system_os=None,
 137             system_os_version=None, system_ram=None, system_cpu=None,
 138             public_ip=None, public_port=None, ds_limit=None, us_limit=None,
 139             start_time=None, dht_enabled=None, pxe_enabled=None,
 140             streaming_enabled=None, features=None, description=None):
 141         self.swarm_id = swarm_id
 142         self.btclient = btclient
 143         self.system_os = system_os
 144         self.system_os_version = system_os_version
 145         self.system_ram = system_ram
 146         self.system_cpu = system_cpu
 147         self.public_ip = public_ip
 148         self.public_port = public_port
 149         self.ds_limit = ds_limit
 150         self.us_limit = us_limit
 151         self.start_time = start_time
 152         self.dht_enabled = dht_enabled
 153         self.pxe_enabled = pxe_enabled
 154         self.streaming_enabled = streaming_enabled
 155         self.features = features
 156         self.description = description
 157
 158 class PeerStatusMessage(object):
 159     """ Class mimics a C structure. """
 160     def __init__(self, swarm_id=None, client_session_id=None, timestamp=None,
 161             peer_ip=None, peer_port=None, download_speed=None,
 162             upload_speed=None):
 163         self.swarm_id = swarm_id
 164         self.client_session_id = client_session_id
 165         self.timestamp = timestamp
 166         self.peer_ip = peer_ip
 167         self.peer_port = peer_port
 168         self.download_speed = download_speed
 169         self.upload_speed = upload_speed
 170
 171 class StatusMessage(object):
 172     """ Class mimics a C structure. """
 173     def __init__(self, swarm_id=None, client_session_id=None, timestamp=None,
 174             time=None, num_peers=None, num_dht_peers=None,
 175             download_speed=None, upload_speed=None, download_size=None,
 176             upload_size=None, eta=None):
 177         self.swarm_id = swarm_id
 178         self.client_session_id = client_session_id
 179         self.timestamp = timestamp
 180         self.num_peers = num_peers
 181         self.num_dht_peers = num_dht_peers
 182         self.download_speed = download_speed
 183         self.upload_speed = upload_speed
 184         self.download_size = download_size
 185         self.upload_size = upload_size
 186         self.eta = eta
 187
 188 class VerboseMessage(object):
 189     """ Class mimics a C structure. """
 190     def __init__(self, swarm_id=None, client_session_id=None, timestamp=None,
 191             transfer_direction=None, peer_ip=None, peer_port=None,
 192             message_type=None, index=None, begin=None, length=None,
 193             listen_port=None):
 194         self.swarm_id = swarm_id
 195         self.client_session_id = client_session_id
 196         self.timestamp = timestamp
 197         self.transfer_direction = transfer_direction
 198         self.peer_ip = peer_ip
 199         self.peer_port = peer_port
 200         self.message_type = message_type
 201         self.index = index
 202         self.begin = begin
 203         self.length = length
 204         self.listen_port = listen_port
 205
 206 class SwarmDataAccess(object):
 207     def __init__(self):
 208         pass
 209
 210     def add_swarm(self, swarm):
 211         pass
 212
 213     def remove_swarm(self):
 214         pass
 215
 216     def get_swarm(self):
 217         pass
 218
 219     def update_swarm(self):
 220         pass
 221
 222     def add_client_session(self, session):
 223         pass
 224
 225     def remove_client_session(self):
 226         pass
 227
 228     def get_client_session(self):
 229         pass
 230
 231     def update_client_session(self):
 232         pass
 233
 234     def add_peer_status_message(self, msg):
 235         pass
 236
 237     def remove_peer_status_message(self):
 238         pass
 239
 240     def get_peer_status_message(self):
 241         pass
 242
 243     def update_peer_status_message(self):
 244         pass
 245
 246     def add_status_message(self, msg):
 247         pass
 248
 249     def remove_status_message(self):
 250         pass
 251
 252     def get_status_message(self):
 253         pass
 254
 255     def update_status_message(self):
 256         pass
 257
 258     def add_verbose_message(self, msg):
 259         pass
 260
 261     def remove_verbose_message(self):
 262         pass
 263
 264     def get_verbose_message(self):
 265         pass
 266
 267     def update_verbose_message(self):
 268         pass
 269
 270 class FileAccess(SwarmDataAccess):
 271     def __init__(self, path):
 272         self.base_path = path
 273
 274 def find_last_numeric_subfolder(path):
 275     """
 276     Find last numeric folder in base_path folder.
 277     The last numeric folder is the last swarm_id.
 278     """
 279     dir_list = []
 280     pattern = re.compile("[0-9]+")
 281
 282     # Browse entries in base_path folder.
 283     listing = os.listdir(path)
 284     for entry in listing:
 285         # If directory name is a number (id) add it to the list.
 286         if os.path.isdir(os.path.join(path, entry)):
 287             if pattern.match(entry):
 288                 dir_list.append(int(entry))
 289
 290     if not dir_list:
 291         return None
 292     else:
 293         dir_list.sort()
 294         return dir_list[len(dir_list)-1]
 295
 296 class TreeTextFileAccess(FileAccess):
 297     def __init__(self, path):
 298         super(TreeTextFileAccess, self).__init__(path)
 299
 300     def add_swarm(self, swarm):
 301         """
 302         Create a subfolder with an unique id. Add 1 to the last numeric
 303         subfolder id. In case none exists, use 1 as id.
 304         """
 305         id = find_last_numeric_subfolder(self.base_path)
 306         if id == None:
 307             id = 1
 308         else:
 309             id = id+1
 310
 311         swarm_path = os.path.join(self.base_path, str(id))
 312         os.mkdir(swarm_path)
 313
 314         swarm_config = os.path.join(swarm_path, "swarm.conf")
 315         f = open(swarm_config, 'w')
 316         f.write("""id = %s
 317         torrent_filename = %s
 318         data_size = %s
 319         description = %s
 320         """ %(id, swarm.torrent_filename, swarm.data_size, swarm.description))
 321         f.close()
 322
 323     def add_client_session(self, session):
 324         """
 325         Create session subfolder in swarm subfolder and add config file.
 326         TODO: Throw exception in case swarm subfolder doesn't exist.
 327         """
 328         swarm_path = os.path.join(self.base_path, str(session.swarm_id))
 329
 330         # Search first available folder in swarm_path.
 331         id = find_last_numeric_subfolder(swarm_path)
 332         if id == None:
 333             id = 1
 334         else:
 335             id = id+1
 336
 337         # Create session subfolder.
 338         session_path = os.path.join(swarm_path, str(id))
 339         os.mkdir(session_path)
 340
 341         # Create and populate configuration file.
 342         session_config = os.path.join(session_path, "client_session.conf")
 343         f = open(session_config, 'w')
 344         f.write("""id = %s
 345         swarm_id = %s
 346         btclient = %s
 347         system_os = %s
 348         system_os_version = %s
 349         system_ram = %s
 350         system_cpu = %s
 351         public_ip = %s
 352         public_port = %s
 353         ds_limit = %s
 354         us_limit = %s
 355         start_time = %s
 356         dht_enabled = %s
 357         pxe_enabled = %s
 358         streaming_enabled = %s
 359         features = %s
 360         description = %s
 361         """ %(id, session.swarm_id, session.btclient, session.system_os,
 362             session.system_os_version, session.system_ram, session.system_cpu,
 363             session.public_ip, session.public_port, session.ds_limit,
 364             session.us_limit, session.start_time, session.dht_enabled,
 365             session.pxe_enabled, session.streaming_enabled,
 366             session.features, session.description))
 367         f.close()
 368
 369     def add_peer_status_message(self, msg):
 370         # TODO: id is number of lines in file.
 371         swarm_path = os.path.join(self.base_path, str(msg.swarm_id))
 372         session_path = os.path.join(swarm_path, str(msg.client_session_id))
 373         message_file = os.path.join(session_path, "peer_status.txt")
 374
 375         f = open(message_file, 'a')
 376         f.write("""%s,%s,%s,%s,%s,%s,%s,%s\n"""
 377                 %(1, msg.swarm_id, msg.client_session_id, msg.timestamp,
 378                     msg.peer_ip, msg.peer_port, msg.download_speed,
 379                     msg.upload_speed))
 380         f.close()
 381
 382     def add_status_message(self, msg):
 383         # TODO: id is number of lines in file.
 384         swarm_path = os.path.join(self.base_path, str(msg.swarm_id))
 385         session_path = os.path.join(swarm_path, str(msg.client_session_id))
 386         message_file = os.path.join(session_path, "status.txt")
 387
 388         f = open(message_file, 'a')
 389         f.write("""%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n"""
 390                 %(1, msg.swarm_id, msg.client_session_id, msg.timestamp,
 391                     msg.num_peers, msg.num_dht_peers, msg.download_speed,
 392                     msg.upload_speed, msg.download_size, msg.upload_size,
 393                     msg.eta))
 394         f.close()
 395
 396     def add_verbose_message(self, msg):
 397         # TODO: id is number of lines in file.
 398         swarm_path = os.path.join(self.base_path, str(msg.swarm_id))
 399         session_path = os.path.join(swarm_path, str(msg.client_session_id))
 400         message_file = os.path.join(session_path, "verbose.txt")
 401
 402         f = open(message_file, 'a')
 403         f.write("""%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n"""
 404                 %(1, msg.swarm_id, msg.client_session_id, msg.timestamp,
 405                     msg.transfer_direction, msg.peer_ip, msg.peer_port,
 406                     msg.message_type, msg.index, msg.begin, msg.length,
 407                     msg.listen_port))
 408         f.close()
 409
 410 class DatabaseAccess(SwarmDataAccess):
 411     def __init__(self, database):
 412         self.database = database
 413
 414     def connect(self):
 415         self.conn = None
 416         self.cursor = None
 417
 418     def disconnect(self):
 419         self.cursor.close()
 420         self.conn.close()
 421
 422 class SQLiteDatabaseAccess(DatabaseAccess):
 423     def __init___(self, database):
 424         super(SQLiteDatabaseAccess, self).__init__(database)
 425
 426     def connect(self):
 427         self.conn = sqlite3.connect(self.database)
 428         self.cursor = self.conn.cursor()
 429         # Use foreign key support if available.
 430         self.cursor.execute("PRAGMA foreign_keys = ON")
 431         self.conn.commit()
 432
 433     def reset_query(self):
 434         self.columns = ""
 435         self.values = ""
 436
 437     def append_to_insert_query(self, data_name, data_value):
 438         if data_value is not None:
 439             self.columns = self.columns + data_name + ", "
 440             self.values = self.values + "'" + str(data_value) + "'" + ", "
 441
 442     def add_swarm(self, swarm):
 443         self.reset_query()
 444         self.append_to_insert_query("torrent_filename", swarm.torrent_filename)
 445         self.append_to_insert_query("data_size", swarm.data_size)
 446         self.append_to_insert_query("description", swarm.description)
 447
 448         self.columns = re.sub(',\s*$', '', self.columns)
 449         self.values = re.sub(',\s*$', '', self.values)
 450         insert_query = "INSERT INTO swarms(" + self.columns +")" + \
 451                 " VALUES(" + self.values + ")"
 452         self.cursor.execute(insert_query)
 453         self.conn.commit()
 454
 455     def add_client_session(self, session):
 456         self.reset_query()
 457         self.append_to_insert_query("swarm_id", session.swarm_id)
 458         # TODO: search database for client ID
 459         self.append_to_insert_query("btclient_id",
 460                 bittorrent_clients[session.btclient]['id'])
 461         self.append_to_insert_query("system_os", session.system_os)
 462         self.append_to_insert_query("system_os_version", session.system_os_version)
 463         self.append_to_insert_query("system_ram", session.system_ram)
 464         self.append_to_insert_query("system_cpu", session.system_cpu)
 465         self.append_to_insert_query("public_ip", session.public_ip)
 466         self.append_to_insert_query("public_port", session.public_port)
 467         self.append_to_insert_query("ds_limit", session.ds_limit)
 468         self.append_to_insert_query("us_limit", session.us_limit)
 469         self.append_to_insert_query("start_time", session.start_time)
 470         self.append_to_insert_query("dht_enabled", session.dht_enabled)
 471         self.append_to_insert_query("pxe_enabled", session.pxe_enabled)
 472         self.append_to_insert_query("streaming_enabled", session.streaming_enabled)
 473         self.append_to_insert_query("features", session.features)
 474         self.append_to_insert_query("description", session.description)
 475
 476         self.columns = re.sub(',\s*$', '', self.columns)
 477         self.values = re.sub(',\s*$', '', self.values)
 478         insert_query = "INSERT INTO client_sessions(" + self.columns +")" + \
 479                 " VALUES(" + self.values + ")"
 480         self.cursor.execute(insert_query)
 481         self.conn.commit()
 482
 483     def get_string_timestamp(self, ts):
 484         # Timestamp is Python Datatime. Convert it to string format and
 485         # pass it to internal SQLITE julianday() function.
 486         return "%s-%s-%s %s:%s:%s.%s" \
 487                 %(ts.year, ts.month, ts.day, ts.hour, ts.minute, ts.second,
 488                         1000 * ts.microsecond)
 489
 490     def add_peer_status_message(self, msg):
 491         self.reset_query()
 492         self.append_to_insert_query("client_session_id", msg.client_session_id)
 493
 494         # TODO: Check msg.timestamp is not None. Raise exception.
 495         timestamp_string = self.get_string_timestamp(msg.timestamp)
 496         value = "julianday(" + timestamp_string + ")"
 497         self.append_to_insert_query("timestamp", value)
 498
 499         self.append_to_insert_query("timestamp", msg.timestamp)
 500         self.append_to_insert_query("peer_ip", msg.peer_ip)
 501         self.append_to_insert_query("peer_port", msg.peer_port)
 502         self.append_to_insert_query("download_speed", msg.download_speed)
 503         self.append_to_insert_query("upload_speed", msg.upload_speed)
 504
 505         self.columns = re.sub(',\s*$', '', self.columns)
 506         self.values = re.sub(',\s*$', '', self.values)
 507         insert_query = "INSERT INTO peer_status_messages(" + \
 508                 self.columns +")" + " VALUES(" + self.values + ")"
 509         self.cursor.execute(insert_query)
 510         self.conn.commit()
 511
 512     def add_status_message(self, msg):
 513         self.reset_query()
 514         self.append_to_insert_query("client_session_id", msg.client_session_id)
 515
 516         # TODO: Check msg.timestamp is not None. Raise exception.
 517         timestamp_string = self.get_string_timestamp(msg.timestamp)
 518         value = "julianday(" + timestamp_string + ")"
 519         self.append_to_insert_query("timestamp", value)
 520
 521         self.append_to_insert_query("num_peers", msg.num_peers)
 522         self.append_to_insert_query("num_dht_peers", msg.num_dht_peers)
 523         self.append_to_insert_query("download_speed", msg.download_speed)
 524         self.append_to_insert_query("upload_speed", msg.upload_speed)
 525         self.append_to_insert_query("download_size", msg.download_size)
 526         self.append_to_insert_query("upload_size", msg.upload_size)
 527         self.append_to_insert_query("eta", msg.eta)
 528
 529         self.columns = re.sub(',\s*$', '', self.columns)
 530         self.values = re.sub(',\s*$', '', self.values)
 531         insert_query = "INSERT INTO status_messages(" + self.columns +")" + \
 532                 " VALUES(" + self.values + ")"
 533         self.cursor.execute(insert_query)
 534         self.conn.commit()
 535
 536     def add_verbose_message(self, msg):
 537         self.reset_query()
 538         self.append_to_insert_query("client_session_id", msg.client_session_id)
 539
 540         # TODO: Check msg.timestamp is not None. Raise exception.
 541         timestamp_string = self.get_string_timestamp(msg.timestamp)
 542         value = "julianday(" + timestamp_string + ")"
 543         self.append_to_insert_query("timestamp", value)
 544
 545         self.append_to_insert_query("transfer_direction_id",
 546                 transfer_directions[msg.transfer_direction])
 547         self.append_to_insert_query("peer_ip", msg.peer_ip)
 548         self.append_to_insert_query("peer_port", msg.peer_port)
 549         self.append_to_insert_query("message_type_id",
 550                 message_types[msg.message_type]['id'])
 551         self.append_to_insert_query("index", msg.index)
 552         self.append_to_insert_query("begin", msg.begin)
 553         self.append_to_insert_query("length", msg.length)
 554         self.append_to_insert_query("listen_port", msg.listen_port)
 555
 556         self.columns = re.sub(',\s*$', '', self.columns)
 557         self.values = re.sub(',\s*$', '', self.values)
 558         insert_query = "INSERT INTO verbose_messages(" + self.columns +")" + \
 559                 " VALUES(" + self.values + ")"
 560         self.cursor.execute(insert_query)
 561         self.conn.commit()
 562
 563 class MySQLDatabaseAccess(DatabaseAccess):
 564     def __init___(self, database):
 565         super(MySQLDatabaseAccess, self).__init__(database)
 566
 567     def connect(self):
 568         # TODO Add support for reading connection information from config_file
 569         self.conn = MySQLdb.Connection(db=self.database['database'],
 570                                        user=self.database['user'],
 571                                        passwd=self.database['password'],
 572                                        host=self.database['host'],
 573                                        port=self.database['port'])
 574         self.cursor = self.conn.cursor()
 575         self.conn.commit()
 576
 577     def reset_query(self):
 578         self.columns = ""
 579         self.values = ""
 580
 581     def append_to_insert_query(self, data_name, data_value):
 582         if data_value is not None:
 583             self.columns = self.columns + data_name + ", "
 584             self.values = self.values + "'" + str(data_value) + "'" + ", "
 585
 586     def add_swarm(self, swarm):
 587         self.reset_query()
 588         self.append_to_insert_query("torrent_filename", swarm.torrent_filename)
 589         self.append_to_insert_query("data_size", swarm.data_size)
 590         self.append_to_insert_query("description", swarm.description)
 591
 592         self.columns = re.sub(',\s*$', '', self.columns)
 593         self.values = re.sub(',\s*$', '', self.values)
 594         insert_query = "INSERT INTO swarms(" + self.columns +")" + \
 595                 " VALUES(" + self.values + ")"
 596         self.cursor.execute(insert_query)
 597         self.conn.commit()
 598
 599     def add_client_session(self, session):
 600         self.reset_query()
 601         self.append_to_insert_query("swarm_id", session.swarm_id)
 602         # TODO: search database for client ID
 603         self.append_to_insert_query("btclient_id",
 604                 bittorrent_clients[session.btclient]['id'])
 605         self.append_to_insert_query("system_os", session.system_os)
 606         self.append_to_insert_query("system_os_version", session.system_os_version)
 607         self.append_to_insert_query("system_ram", session.system_ram)
 608         self.append_to_insert_query("system_cpu", session.system_cpu)
 609         self.append_to_insert_query("public_ip", session.public_ip)
 610         self.append_to_insert_query("public_port", session.public_port)
 611         self.append_to_insert_query("ds_limit", session.ds_limit)
 612         self.append_to_insert_query("us_limit", session.us_limit)
 613         self.append_to_insert_query("start_time", session.start_time)
 614         self.append_to_insert_query("dht_enabled", session.dht_enabled)
 615         self.append_to_insert_query("pxe_enabled", session.pxe_enabled)
 616         self.append_to_insert_query("streaming_enabled", session.streaming_enabled)
 617         self.append_to_insert_query("features", session.features)
 618         self.append_to_insert_query("description", session.description)
 619
 620         self.columns = re.sub(',\s*$', '', self.columns)
 621         self.values = re.sub(',\s*$', '', self.values)
 622         insert_query = "INSERT INTO client_sessions(" + self.columns +")" + \
 623                 " VALUES(" + self.values + ")"
 624         self.cursor.execute(insert_query)
 625         self.conn.commit()
 626
 627     def get_string_timestamp(self, ts):
 628         # Timestamp is Python Datatime. Convert it to string format.
 629         return "%s-%s-%s %s:%s:%s" \
 630                 %(ts.year, ts.month, ts.day, ts.hour, ts.minute, ts.second)
 631
 632     def add_peer_status_message(self, msg):
 633         self.reset_query()
 634         self.append_to_insert_query("client_session_id", msg.client_session_id)
 635
 636         # TODO: Check msg.timestamp is not None. Raise exception.
 637         timestamp_string = self.get_string_timestamp(msg.timestamp)
 638
 639         self.append_to_insert_query("timestamp", timestamp_string)
 640         self.append_to_insert_query("peer_ip", msg.peer_ip)
 641         self.append_to_insert_query("peer_port", msg.peer_port)
 642         self.append_to_insert_query("download_speed", msg.download_speed)
 643         self.append_to_insert_query("upload_speed", msg.upload_speed)
 644
 645         self.columns = re.sub(',\s*$', '', self.columns)
 646         self.values = re.sub(',\s*$', '', self.values)
 647         insert_query = "INSERT INTO peer_status_messages(" + \
 648                 self.columns +")" + " VALUES(" + self.values + ")"
 649         self.cursor.execute(insert_query)
 650         self.conn.commit()
 651
 652     def add_status_message(self, msg):
 653         self.reset_query()
 654         self.append_to_insert_query("client_session_id", msg.client_session_id)
 655
 656         # TODO: Check msg.timestamp is not None. Raise exception.
 657         timestamp_string = self.get_string_timestamp(msg.timestamp)
 658
 659         self.append_to_insert_query("timestamp", timestamp_string)
 660         self.append_to_insert_query("num_peers", msg.num_peers)
 661         self.append_to_insert_query("num_dht_peers", msg.num_dht_peers)
 662         self.append_to_insert_query("download_speed", msg.download_speed)
 663         self.append_to_insert_query("upload_speed", msg.upload_speed)
 664         self.append_to_insert_query("download_size", msg.download_size)
 665         self.append_to_insert_query("upload_size", msg.upload_size)
 666         self.append_to_insert_query("eta", msg.eta)
 667
 668         self.columns = re.sub(',\s*$', '', self.columns)
 669         self.values = re.sub(',\s*$', '', self.values)
 670         insert_query = "INSERT INTO status_messages(" + self.columns +")" + \
 671                 " VALUES(" + self.values + ")"
 672         self.cursor.execute(insert_query)
 673         self.conn.commit()
 674
 675     def add_verbose_message(self, msg):
 676         self.reset_query()
 677         self.append_to_insert_query("client_session_id", msg.client_session_id)
 678
 679         # TODO: Check msg.timestamp is not None. Raise exception.
 680         timestamp_string = self.get_string_timestamp(msg.timestamp)
 681
 682         self.append_to_insert_query("timestamp", timestamp_string)
 683         self.append_to_insert_query("transfer_direction_id",
 684                 transfer_directions[msg.transfer_direction])
 685         self.append_to_insert_query("peer_ip", msg.peer_ip)
 686         self.append_to_insert_query("peer_port", msg.peer_port)
 687         self.append_to_insert_query("message_type_id",
 688                 message_types[msg.message_type]['id'])
 689         self.append_to_insert_query("index", msg.index)
 690         self.append_to_insert_query("begin", msg.begin)
 691         self.append_to_insert_query("length", msg.length)
 692         self.append_to_insert_query("listen_port", msg.listen_port)
 693
 694         self.columns = re.sub(',\s*$', '', self.columns)
 695         self.values = re.sub(',\s*$', '', self.values)
 696         insert_query = "INSERT INTO verbose_messages(" + self.columns +")" + \
 697                 " VALUES(" + self.values + ")"
 698         self.cursor.execute(insert_query)
 699         self.conn.commit()
 700
 701 class SwarmWriter(object):
 702     """
 703     Wrapper class for swarm storage write actions. Multiple *Access
 704     objects may be added to the clas resulting in multiple storage types.
 705     For example, adding a swarm could result in
 706        * adding a table entry in a MySQL database (MySQLDatabaseAccess)
 707        * adding a table entry in an SQLite database (SQLiteDatabaseAccess)
 708        * adding a new folder in a tree structure (TreeTextFileAccess)
 709     """
 710
 711     def __init__(self):
 712         self.handlers = []
 713
 714     def add_access_handle(self, handle):
 715         self.handlers.append(handle)
 716
 717     def remove_access_handle(self, handle):
 718         self.handlers.remove(handle)
 719
 720     def add_swarm(self, swarm):
 721         for h in self.handlers:
 722             h.add_swarm(swarm)
 723
 724     def add_client_session(self, session):
 725         for h in self.handlers:
 726             h.add_client_session(session)
 727
 728     def add_peer_status_message(self, msg):
 729         for h in self.handlers:
 730             h.add_peer_status_message(msg)
 731
 732     def add_status_message(self, msg):
 733         for h in self.handlers:
 734             h.add_status_message(msg)
 735
 736     def add_verbose_message(self, msg):
 737         for h in self.handlers:
 738             h.add_verbose_message(msg)