src/gpodder/dbsqlite.py

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