src/gpodder/dbsqlite.py

   1 # -*- coding: utf-8 -*-
   2 #
   3 # gPodder - A media aggregator and podcast client
   4 # Copyright (c) 2005-2008 Thomas Perl and the gPodder Team
   5 #
   6 # gPodder is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 3 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # gPodder is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 # GNU General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 #
  19 # dbsqlite.py -- SQLite interface
  20 # Justin Forest <justin.forest@gmail.com> 2008-06-13
  21
  22 have_sqlite = True
  23
  24 try:
  25     from sqlite3 import dbapi2 as sqlite
  26 except ImportError:
  27     try:
  28         from pysqlite2 import dbapi2 as sqlite
  29     except ImportError:
  30         have_sqlite = False
  31
  32 # TODO: show a message box
  33 if not have_sqlite:
  34     print "Please install pysqlite2 or upgrade to Python 2.5."
  35     import sys
  36     sys.exit()
  37
  38 from gpodder.liblogger import log
  39 from email.Utils import mktime_tz
  40 from email.Utils import parsedate_tz
  41 from email.Utils import formatdate
  42 from threading import RLock
  43 import string
  44
  45 class Storage(object):
  46     (STATE_NORMAL, STATE_DOWNLOADED, STATE_DELETED) = range(3)
  47
  48     lock = None
  49
  50     def __init__(self):
  51         self.settings = {}
  52         self.channel_map = {}
  53         self._db = None
  54         self.lock = RLock()
  55
  56     def setup(self, settings):
  57         self.settings = settings
  58         self.__check_schema()
  59
  60     def log(self, message, *args, **kwargs):
  61         if self.settings['gl'].config.log_sqlite:
  62             try:
  63                 message = message % args
  64                 log('%s', message, sender=self)
  65             except TypeError, e:
  66                 log('Exception in log(): %s: %s', e, message, sender=self)
  67
  68     def purge(self, max_episodes, channel_id):
  69         """
  70         Deletes old episodes.  Should be called
  71         before adding new episodes to a channel.
  72         """
  73         cur = self.cursor(lock=True)
  74
  75         self.log("purge(%s)", channel_id)
  76         sql = """
  77             DELETE FROM episodes
  78             WHERE channel_id = ?
  79             AND state <> ?
  80             AND id NOT IN
  81             (SELECT id FROM episodes WHERE channel_id = ?
  82             ORDER BY pubDate DESC LIMIT ?)"""
  83         cur.execute(sql, (channel_id, self.STATE_DOWNLOADED, channel_id, max_episodes))
  84
  85         cur.close()
  86         self.lock.release()
  87
  88     @property
  89     def db(self):
  90         if self._db is None:
  91             self._db = sqlite.connect(self.settings['database'], check_same_thread=False)
  92             self._db.create_collation("UNICODE", lambda a, b: cmp(a.lower().replace('the ', ''), b.lower().replace('the ', '')))
  93             self.log('Connected')
  94         return self._db
  95
  96     def cursor(self, lock=False):
  97         if lock:
  98             self.lock.acquire()
  99         return self.db.cursor()
 100
 101     def commit(self):
 102         self.lock.acquire()
 103         try:
 104             self.log("COMMIT")
 105             self.db.commit()
 106         except ProgrammingError, e:
 107             log('Error commiting changes: %s', e, sender=self, traceback=True)
 108         self.lock.release()
 109
 110     def __check_schema(self):
 111         """
 112         Creates all necessary tables and indexes that don't exist.
 113         """
 114         self.log('Setting up tables and views')
 115
 116         cur = self.cursor(lock=True)
 117
 118         self.upgrade_table("channels", (
 119             ("id", "INTEGER PRIMARY KEY"),
 120             ("url", "TEXT"),
 121             ("title", "TEXT"),
 122             ("override_title", "TEXT"),
 123             ("link", "TEXT"),
 124             ("description", "TEXT"),
 125             ("image", "TEXT"),
 126             ("pubDate", "INTEGER"),
 127             ("sync_to_devices", "INTEGER"),
 128             ("device_playlist_name", "TEXT"),
 129             ("username", "TEXT"),
 130             ("password", "TEXT"),
 131             ("last_modified", "TEXT"),
 132             ("etag", "TEXT"),
 133             ("deleted", "INTEGER"),
 134             ))
 135
 136         self.upgrade_table("episodes", (
 137             ("id", "INTEGER PRIMARY KEY"),
 138             ("channel_id", "INTEGER"),
 139             ("url", "TEXT"),
 140             ("title", "TEXT"),
 141             ("length", "INTEGER"),
 142             ("mimetype", "TEXT"),
 143             ("guid", "TEXT"),
 144             ("description", "TEXT"),
 145             ("link", "TEXT"),
 146             ("pubDate", "INTEGER"),
 147             ("state", "INTEGER"),
 148             ("played", "INTEGER"),
 149             ("locked", "INTEGER"),
 150             ))
 151
 152         cur.execute("""CREATE UNIQUE INDEX IF NOT EXISTS idx_url ON channels (url)""")
 153         cur.execute("""CREATE INDEX IF NOT EXISTS idx_sync_to_devices ON channels (sync_to_devices)""")
 154         cur.execute("""CREATE INDEX IF NOT EXISTS idx_title ON channels (title)""")
 155         cur.execute("""CREATE INDEX IF NOT EXISTS idx_deleted ON channels (deleted)""")
 156
 157         cur.execute("""CREATE UNIQUE INDEX IF NOT EXISTS idx_guid ON episodes (guid)""")
 158         cur.execute("""CREATE INDEX IF NOT EXISTS idx_channel_id ON episodes (channel_id)""")
 159         cur.execute("""CREATE INDEX IF NOT EXISTS idx_pubDate ON episodes (pubDate)""")
 160         cur.execute("""CREATE INDEX IF NOT EXISTS idx_state ON episodes (state)""")
 161         cur.execute("""CREATE INDEX IF NOT EXISTS idx_played ON episodes (played)""")
 162         cur.execute("""CREATE INDEX IF NOT EXISTS idx_locked ON episodes (locked)""")
 163
 164         cur.execute("""CREATE TEMPORARY VIEW episodes_downloaded AS SELECT channel_id, COUNT(*) AS count FROM episodes WHERE state = 1 GROUP BY channel_id""")
 165         cur.execute("""CREATE TEMPORARY VIEW episodes_new AS SELECT channel_id, COUNT(*) AS count FROM episodes WHERE state = 0 AND played = 0 GROUP BY channel_id""")
 166         cur.execute("""CREATE TEMPORARY VIEW episodes_unplayed AS SELECT channel_id, COUNT(*) AS count FROM episodes WHERE played = 0 AND state = %d GROUP BY channel_id""" % self.STATE_DOWNLOADED)
 167
 168         # Make sure deleted episodes are played, to simplify querying statistics.
 169         cur.execute("UPDATE episodes SET played = 1 WHERE state = ?", (self.STATE_DELETED, ))
 170
 171         cur.close()
 172         self.lock.release()
 173
 174     def get_channel_stat(self, url_or_id, state=None, is_played=None, is_locked=None):
 175         where, params = ((),())
 176
 177         if state is not None:
 178             where += ("state = ?", )
 179             params += (state, )
 180         if is_played is not None:
 181             where += ("played = ?", )
 182             params += (is_played, )
 183         if is_locked is not None:
 184             where += ("locked = ?", )
 185             params += (is_locked, )
 186         if isinstance(url_or_id, int):
 187             where += ("channel_id = ?", )
 188             params += (url_or_id, )
 189         else:
 190             where += ("channel_id IN (SELECT id FROM channels WHERE url = ?)", )
 191             params += (url_or_id, )
 192
 193         self.log("get_channel_stats(%s)", url_or_id)
 194
 195         if len(where):
 196             return self.__get__("SELECT COUNT(*) FROM episodes WHERE %s" % (' AND '.join(where)), params)
 197         else:
 198             return 0
 199
 200     def load_channels(self, factory=None, url=None):
 201         """
 202         Returns channel descriptions as a list of dictionaries or objects,
 203         returned by the factory() function, which receives the dictionary
 204         as the only argument.
 205         """
 206
 207         self.log("load_channels()")
 208
 209         cur = self.cursor(lock=True)
 210         cur.execute("""
 211             SELECT
 212                 id,
 213                 url,
 214                 title,
 215                 override_title,
 216                 link,
 217                 description,
 218                 image,
 219                 pubDate,
 220                 sync_to_devices,
 221                 device_playlist_name,
 222                 username,
 223                 password,
 224                 last_modified,
 225                 etag
 226             FROM
 227                 channels
 228             WHERE
 229                 (deleted IS NULL OR deleted = 0)
 230             ORDER BY
 231                 title COLLATE UNICODE
 232                 """)
 233
 234         stats = self.stats()
 235
 236         result = []
 237         for row in cur.fetchall():
 238             channel = {
 239                 'id': row[0],
 240                 'url': row[1],
 241                 'title': row[2],
 242                 'override_title': row[3],
 243                 'link': row[4],
 244                 'description': row[5],
 245                 'image': row[6],
 246                 'pubDate': self.__formatdate__(row[7]),
 247                 'sync_to_devices': row[8],
 248                 'device_playlist_name': row[9],
 249                 'username': row[10],
 250                 'password': row[11],
 251                 'last_modified': row[12],
 252                 'etag': row[13],
 253                 }
 254
 255             if row[0] in stats:
 256                 channel['count_downloaded'] = stats[row[0]][0]
 257                 channel['count_new'] = stats[row[0]][1]
 258                 channel['count_unplayed'] = stats[row[0]][2]
 259
 260             if url is None:
 261                 # Maintain url/id relation for faster updates (otherwise
 262                 # we'd need to issue an extra query to find the channel id).
 263                 self.channel_map[channel['url']] = channel['id']
 264
 265             if url is None or url == channel['url']:
 266                 if factory is None:
 267                     result.append(channel)
 268                 else:
 269                     result.append(factory(channel))
 270
 271         cur.close()
 272         self.lock.release()
 273
 274         return result
 275
 276     def stats(self):
 277         cur = self.cursor(lock=True)
 278         self.log("stats()")
 279         cur.execute("""
 280             SELECT c.id, d.count, n.count, u.count
 281             FROM channels c
 282             LEFT JOIN episodes_downloaded d ON d.channel_id = c.id
 283             LEFT JOIN episodes_new n ON n.channel_id = c.id
 284             LEFT JOIN episodes_unplayed u ON u.channel_id = c.id
 285             WHERE c.deleted = 0
 286             """)
 287
 288         data = {}
 289
 290         for row in cur.fetchall():
 291             data[row[0]] = (row[1] or 0, row[2] or 0, row[3] or 0)
 292
 293         cur.close()
 294         self.lock.release()
 295
 296         return data
 297
 298     def save_channel(self, c, bulk=False):
 299         if c.id is None:
 300             c.id = self.find_channel_id(c.url)
 301
 302         cur = self.cursor(lock=True)
 303         self.log("save_channel((%s)%s)", c.id or "new", c.url)
 304
 305         if c.id is None:
 306             cur.execute("INSERT INTO channels (url, title, override_title, link, description, image, pubDate, sync_to_devices, device_playlist_name, username, password, last_modified, etag) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (c.url, c.title, c.override_title, c.link, c.description, c.image, self.__mktime__(c.pubDate), c.sync_to_devices, c.device_playlist_name, c.username, c.password, c.last_modified, c.etag, ))
 307             self.channel_map[c.url] = cur.lastrowid
 308         else:
 309             cur.execute("UPDATE channels SET url = ?, title = ?, override_title = ?, link = ?, description = ?, image = ?, pubDate = ?, sync_to_devices = ?, device_playlist_name = ?, username = ?, password = ?, last_modified = ?, etag = ?, deleted = 0 WHERE id = ?", (c.url, c.title, c.override_title, c.link, c.description, c.image, self.__mktime__(c.pubDate), c.sync_to_devices, c.device_playlist_name, c.username, c.password, c.last_modified, c.etag, c.id, ))
 310
 311         cur.close()
 312         self.lock.release()
 313
 314     def delete_channel(self, channel, purge=False):
 315         if channel.id is None:
 316             channel.id = self.find_channel_id(channel.url)
 317
 318         cur = self.cursor(lock=True)
 319         self.log("delete_channel((%d)%s), purge=%d", channel.id, channel.url, purge)
 320
 321         if purge:
 322             cur.execute("DELETE FROM channels WHERE id = ?", (channel.id, ))
 323             cur.execute("DELETE FROM episodes WHERE channel_id = ?", (channel.id, ))
 324             if channel.url in self.channel_map:
 325                 del self.channel_map[channel.url]
 326         else:
 327             cur.execute("UPDATE channels SET deleted = 1 WHERE id = ?", (channel.id, ))
 328             cur.execute("DELETE FROM episodes WHERE channel_id = ? AND state <> ?", (channel.id, self.STATE_DELETED))
 329
 330         cur.close()
 331         self.lock.release()
 332
 333     def __read_episodes(self, factory=None, where=None, params=None, commit=True):
 334         sql = "SELECT url, title, length, mimetype, guid, description, link, pubDate, state, played, locked, id FROM episodes"
 335
 336         if where:
 337             sql = "%s %s" % (sql, where)
 338
 339         if params is None:
 340             params = ()
 341
 342         cur = self.cursor(lock=True)
 343         cur.execute(sql, params)
 344
 345         result = []
 346         for row in cur.fetchall():
 347             episode = {
 348                 'url': row[0],
 349                 'title': row[1],
 350                 'length': row[2],
 351                 'mimetype': row[3],
 352                 'guid': row[4],
 353                 'description': row[5],
 354                 'link': row[6],
 355                 'pubDate': row[7],
 356                 'state': row[8],
 357                 'is_played': row[9],
 358                 'is_locked': row[10],
 359                 'id': row[11],
 360                 }
 361             if episode['state'] is None:
 362                 episode['state'] = self.STATE_NORMAL
 363             if factory is None:
 364                 result.append(episode)
 365             else:
 366                 result.append(factory(episode))
 367
 368         cur.close()
 369         self.lock.release()
 370         return result
 371
 372     def load_episodes(self, channel, factory=None, limit=1000, state=None):
 373         if channel.id is None:
 374             channel.id = self.find_channel_id(channel.url)
 375
 376         self.log("load_episodes((%d)%s)", channel.id, channel.url)
 377
 378         if state is None:
 379             return self.__read_episodes(factory = factory, where = """
 380                 WHERE channel_id = ? AND state = ? OR id IN
 381                 (SELECT id FROM episodes WHERE channel_id = ?
 382                 ORDER BY pubDate DESC LIMIT ?)
 383                 ORDER BY pubDate DESC
 384                 """, params = (channel.id, self.STATE_DOWNLOADED, channel.id, limit, ))
 385         else:
 386             return self.__read_episodes(factory = factory, where = " WHERE channel_id = ? AND state = ? ORDER BY pubDate DESC LIMIT ?", params = (channel.id, state, limit, ))
 387
 388     def load_episode(self, url, factory=None):
 389         self.log("load_episode(%s)", url)
 390         list = self.__read_episodes(factory = factory, where = " WHERE url = ?", params = (url, ))
 391         if len(list):
 392             return list[0]
 393
 394     def save_episode(self, e, bulk=False):
 395         if not e.guid:
 396             log('Refusing to save an episode without guid: %s', e)
 397             return
 398
 399         self.lock.acquire()
 400
 401         self.log("save_episode((%s)%s)", e.id, e.guid)
 402
 403         try:
 404             cur = self.cursor()
 405             channel_id = self.find_channel_id(e.channel.url)
 406
 407             if e.id is None:
 408                 e.id = self.__get__("SELECT id FROM episodes WHERE guid = ?", (e.guid, ))
 409                 self.log("save_episode() -- looking up id")
 410
 411             if e.id is None:
 412                 cur.execute("INSERT INTO episodes (channel_id, url, title, length, mimetype, guid, description, link, pubDate, state, played, locked) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (channel_id, e.url, e.title, e.length, e.mimetype, e.guid, e.description, e.link, self.__mktime__(e.pubDate), e.state, e.is_played, e.is_locked, ))
 413                 e.id = cur.lastrowid
 414             else:
 415                 cur.execute("UPDATE episodes SET title = ?, length = ?, mimetype = ?, description = ?, link = ?, pubDate = ?, state = ?, played = ?, locked = ? WHERE id = ?", (e.title, e.length, e.mimetype, e.description, e.link, self.__mktime__(e.pubDate), e.state, e.is_played, e.is_locked, e.id, ))
 416         except Exception, e:
 417             log('save_episode() failed: %s', e, sender=self)
 418
 419         cur.close()
 420         self.lock.release()
 421
 422     def mark_episode(self, url, state=None, is_played=None, is_locked=None, toggle=False):
 423         cur = self.cursor(lock=True)
 424         cur.execute("SELECT state, played, locked FROM episodes WHERE url = ?", (url, ))
 425
 426         self.log("mark_episode(%s, state=%s, played=%s, locked=%s)", url, state, is_played, is_locked)
 427
 428         try:
 429             ( cur_state, cur_played, cur_locked ) = cur.fetchone()
 430         except:
 431             # This only happens when we try to mark an unknown episode,
 432             # which is typical for database upgrade, so we just ignore it.
 433             cur.close()
 434             self.lock.release()
 435             return
 436
 437         if toggle:
 438             if is_played:
 439                 cur_played = not cur_played
 440             if is_locked:
 441                 cur_locked = not cur_locked
 442         else:
 443             if state is not None:
 444                 cur_state = state
 445             if is_played is not None:
 446                 cur_played = is_played
 447             if is_locked is not None:
 448                 cur_locked = is_locked
 449
 450         cur.close()
 451
 452         cur = self.cursor()
 453         cur.execute("UPDATE episodes SET state = ?, played = ?, locked = ? WHERE url = ?", (cur_state, cur_played, cur_locked, url, ))
 454         cur.close()
 455
 456         self.lock.release()
 457
 458     def __get__(self, sql, params=None):
 459         """
 460         Returns the first cell of a query result, useful for COUNT()s.
 461         """
 462         cur = self.cursor(lock=True)
 463
 464         self.log("__get__(): %s", sql)
 465
 466         if params is None:
 467             cur.execute(sql)
 468         else:
 469             cur.execute(sql, params)
 470
 471         row = cur.fetchone()
 472         cur.close()
 473         self.lock.release()
 474
 475         if row is None:
 476             return None
 477         else:
 478             return row[0]
 479
 480     def __mktime__(self, date):
 481         if isinstance(date, float) or isinstance(date, int):
 482             return date
 483         if date is None or '' == date:
 484             return None
 485         try:
 486             return mktime_tz(parsedate_tz(date))
 487         except TypeError:
 488             log('Could not convert "%s" to a unix timestamp.', date)
 489             return None
 490
 491     def __formatdate__(self, date):
 492         try:
 493             return formatdate(date, localtime=1)
 494         except TypeError:
 495             log('Could not convert "%s" to a string date.', date)
 496             return None
 497
 498     def find_channel_id(self, url):
 499         """
 500         Looks up the channel id in the map (which lists all undeleted
 501         channels), then tries to look it up in the database, including
 502         deleted channels.
 503         """
 504         if url in self.channel_map.keys():
 505             return self.channel_map[url]
 506         else:
 507             self.log("find_channel_id(%s)", url)
 508             return self.__get__("SELECT id FROM channels WHERE url = ?", (url, ))
 509
 510     def force_last_new(self, channel):
 511         old = self.__get__("""SELECT COUNT(*) FROM episodes WHERE channel_id = ?
 512             AND state IN (?, ?)""", (channel.id, self.STATE_DOWNLOADED,
 513             self.STATE_DELETED))
 514
 515         cur = self.cursor(lock=True)
 516
 517         self.log("force_last_new((%d)%s)", channel.id, channel.url)
 518
 519         if old > 0:
 520             cur.execute("""
 521                 UPDATE episodes SET played = 1 WHERE channel_id = ?
 522                 AND played = 0 AND pubDate < (SELECT MAX(pubDate)
 523                 FROM episodes WHERE channel_id = ? AND state IN (?, ?))""",
 524                 (channel.id, channel.id, self.STATE_DOWNLOADED,
 525                 self.STATE_DELETED, ))
 526         else:
 527             cur.execute("""
 528                 UPDATE episodes SET played = 1 WHERE channel_id = ?
 529                 AND pubDate <> (SELECT MAX(pubDate) FROM episodes
 530                 WHERE channel_id = ?)""", (channel.id, channel.id, ))
 531
 532         cur.close()
 533         self.lock.release()
 534
 535     def upgrade_table(self, table_name, fields):
 536         """
 537         Creates a table or adds fields to it.
 538         """
 539         cur = self.cursor(lock=True)
 540
 541         cur.execute("PRAGMA table_info(%s)" % table_name)
 542         available = cur.fetchall()
 543
 544         if not len(available):
 545             log('Creating table %s', table_name, sender=self)
 546             sql = "CREATE TABLE %s (%s)" % (table_name, ", ".join([a+" "+b for (a,b) in fields]))
 547             cur.execute(sql)
 548         else:
 549             available = [row[1] for row in available]
 550
 551             for field_name, field_type in fields:
 552                 if field_name not in available:
 553                     log('Adding column %s to %s (%s)', table_name, field_name, field_type, sender=self)
 554                     cur.execute("ALTER TABLE %s ADD COLUMN %s %s" % (table_name, field_name, field_type))
 555
 556         self.lock.release()
 557
 558 db = Storage()