<s> """<STR_LIT>""" <EOL> import contextlib <EOL> from datetime import datetime , timedelta <EOL> from os import makedirs , path <EOL> import sqlite3 <EOL> from dateutil import parser <EOL> import icalendar <EOL> import pytz <EOL> import xdg . BaseDirectory <EOL> from . event import Event , EventStandIn <EOL> from . import aux <EOL> from . . import log <EOL> from . exceptions import CouldNotCreateDbDir , OutdatedDbVersionError , UpdateFailed <EOL> logger = log . logger <EOL> DB_VERSION = <NUM_LIT:5> <EOL> RECURRENCE_ID = '<STR_LIT>' <EOL> THISANDFUTURE = '<STR_LIT>' <EOL> THISANDPRIOR = '<STR_LIT>' <EOL> DATE = <NUM_LIT:0> <EOL> DATETIME = <NUM_LIT:1> <EOL> PROTO = '<STR_LIT>' <EOL> def sort_key ( vevent ) : <EOL> assert isinstance ( vevent , icalendar . Event ) <EOL> uid = str ( vevent [ '<STR_LIT>' ] ) <EOL> rec_id = vevent . get ( RECURRENCE_ID ) <EOL> if rec_id is None : <EOL> return uid , <NUM_LIT:0> <EOL> rrange = rec_id . params . get ( '<STR_LIT>' ) <EOL> if rrange == THISANDFUTURE : <EOL> return uid , aux . to_unix_time ( rec_id . dt ) <EOL> else : <EOL> return uid , <NUM_LIT:1> <EOL> class SQLiteDb ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , calendars , db_path , locale ) : <EOL> if db_path is None : <EOL> db_path = xdg . BaseDirectory . save_data_path ( '<STR_LIT>' ) + '<STR_LIT>' <EOL> self . calendars = calendars <EOL> self . db_path = path . expanduser ( db_path ) <EOL> self . _create_dbdir ( ) <EOL> self . locale = locale <EOL> self . _at_once = False <EOL> self . conn = sqlite3 . connect ( self . db_path ) <EOL> self . cursor = self . conn . cursor ( ) <EOL> self . _create_default_tables ( ) <EOL> self . _check_calendars_exists ( ) <EOL> self . _check_table_version ( ) <EOL> @ property <EOL> def _select_calendars ( self ) : <EOL> return '<STR_LIT:U+002CU+0020>' . join ( [ '<STR_LIT>' + cal + '<STR_LIT>' for cal in self . calendars ] ) <EOL> @ contextlib . contextmanager <EOL> def at_once ( self ) : <EOL> assert not self . _at_once <EOL> self . _at_once = True <EOL> try : <EOL> yield self <EOL> except : <EOL> raise <EOL> else : <EOL> self . conn . commit ( ) <EOL> finally : <EOL> self . _at_once = False <EOL> def _create_dbdir ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . db_path == '<STR_LIT>' : <EOL> return None <EOL> dbdir = self . db_path . rsplit ( '<STR_LIT:/>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> if not path . isdir ( dbdir ) : <EOL> try : <EOL> logger . debug ( '<STR_LIT>' ) <EOL> makedirs ( dbdir , mode = <NUM_LIT> ) <EOL> logger . debug ( '<STR_LIT:success>' ) <EOL> except OSError as error : <EOL> logger . fatal ( '<STR_LIT>' . format ( dbdir , error ) ) <EOL> raise CouldNotCreateDbDir ( ) <EOL> def _check_table_version ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cursor . execute ( '<STR_LIT>' ) <EOL> result = self . cursor . fetchone ( ) <EOL> if result is None : <EOL> self . cursor . execute ( '<STR_LIT>' , <EOL> ( DB_VERSION , ) ) <EOL> self . conn . commit ( ) <EOL> elif not result [ <NUM_LIT:0> ] == DB_VERSION : <EOL> raise OutdatedDbVersionError ( <EOL> str ( self . db_path ) + <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def _create_default_tables ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cursor . execute ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> logger . debug ( u"<STR_LIT>" ) <EOL> self . cursor . execute ( '''<STR_LIT>''' ) <EOL> self . cursor . execute ( '''<STR_LIT>''' ) <EOL> self . cursor . execute ( '''<STR_LIT>''' ) <EOL> self . cursor . execute ( '''<STR_LIT>''' ) <EOL> self . conn . commit ( ) <EOL> def _check_calendars_exists ( self ) : <EOL> """<STR_LIT>""" <EOL> for cal in self . calendars : <EOL> self . cursor . execute ( '''<STR_LIT>''' , ( cal , ) ) <EOL> result = self . cursor . fetchone ( ) <EOL> if result [ <NUM_LIT:0> ] != <NUM_LIT:0> : <EOL> logger . debug ( u"<STR_LIT>" . format ( cal ) ) <EOL> else : <EOL> sql_s = '<STR_LIT>' <EOL> stuple = ( cal , '<STR_LIT>' ) <EOL> self . sql_ex ( sql_s , stuple ) <EOL> def sql_ex ( self , statement , stuple = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> self . cursor . execute ( statement , stuple ) <EOL> result = self . cursor . fetchall ( ) <EOL> if not self . _at_once : <EOL> self . conn . commit ( ) <EOL> return result <EOL> def update ( self , vevent_str , href , etag = '<STR_LIT>' , calendar = None ) : <EOL> """<STR_LIT>""" <EOL> assert calendar is not None <EOL> if href is None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> ical = icalendar . Event . from_ical ( vevent_str ) <EOL> check_for_errors ( ical , calendar , href ) <EOL> vevents = ( aux . sanitize ( c , self . locale [ '<STR_LIT>' ] , href , calendar ) for <EOL> c in ical . walk ( ) if c . name == '<STR_LIT>' ) <EOL> self . delete ( href , calendar = calendar ) <EOL> for vevent in sorted ( vevents , key = sort_key ) : <EOL> check_for_errors ( vevent , calendar , href ) <EOL> check_support ( vevent , href , calendar ) <EOL> self . _update_impl ( vevent , href , calendar ) <EOL> sql_s = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> stuple = ( vevent_str , etag , href , calendar ) <EOL> self . sql_ex ( sql_s , stuple ) <EOL> def update_birthday ( self , vevent , href , etag = '<STR_LIT>' , calendar = None ) : <EOL> """<STR_LIT>""" <EOL> assert calendar is not None <EOL> if href is None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> ical = icalendar . Event . from_ical ( vevent ) <EOL> vcard = ical . walk ( ) [ <NUM_LIT:0> ] <EOL> if '<STR_LIT>' in vcard . keys ( ) : <EOL> bday = vcard [ '<STR_LIT>' ] <EOL> if isinstance ( bday , list ) : <EOL> logger . warn ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( href , calendar ) ) <EOL> return <EOL> try : <EOL> if bday [ <NUM_LIT:0> : <NUM_LIT:2> ] == '<STR_LIT>' and bday [ <NUM_LIT:3> ] != '<STR_LIT:->' : <EOL> bday = '<STR_LIT>' + bday [ <NUM_LIT:2> : ] <EOL> orig_bday = False <EOL> else : <EOL> orig_bday = True <EOL> bday = parser . parse ( bday ) . date ( ) <EOL> except ValueError : <EOL> logger . warn ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( href , calendar ) ) <EOL> return <EOL> if '<STR_LIT>' in vcard : <EOL> name = vcard [ '<STR_LIT>' ] <EOL> else : <EOL> n = vcard [ '<STR_LIT:N>' ] . split ( '<STR_LIT:;>' ) <EOL> name = '<STR_LIT:U+0020>' . join ( [ n [ <NUM_LIT:1> ] , n [ <NUM_LIT:2> ] , n [ <NUM_LIT:0> ] ] ) <EOL> event = icalendar . Event ( ) <EOL> event . add ( '<STR_LIT>' , bday ) <EOL> event . add ( '<STR_LIT>' , bday + timedelta ( days = <NUM_LIT:1> ) ) <EOL> event . add ( '<STR_LIT>' , '<STR_LIT>' . format ( name ) ) <EOL> event . add ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> if orig_bday : <EOL> event . add ( '<STR_LIT>' , <EOL> '<STR_LIT>' . format ( bday . year , bday . month , bday . day ) ) <EOL> event . add ( '<STR_LIT>' , name ) <EOL> event . add ( '<STR_LIT>' , href ) <EOL> event_str = event . to_ical ( ) . decode ( '<STR_LIT:utf-8>' ) <EOL> self . _update_impl ( event , href , calendar ) <EOL> sql_s = ( '<STR_LIT>' ) <EOL> stuple = ( event_str , etag , href , calendar ) <EOL> self . sql_ex ( sql_s , stuple ) <EOL> def _update_impl ( self , vevent , href , calendar ) : <EOL> """<STR_LIT>""" <EOL> rec_id = vevent . get ( RECURRENCE_ID ) <EOL> if rec_id is None : <EOL> rrange = None <EOL> else : <EOL> rrange = rec_id . params . get ( '<STR_LIT>' ) <EOL> if not isinstance ( vevent [ '<STR_LIT>' ] . dt , datetime ) : <EOL> dtype = DATE <EOL> else : <EOL> dtype = DATETIME <EOL> if ( '<STR_LIT>' in vevent [ '<STR_LIT>' ] . params and dtype == DATETIME ) or getattr ( vevent [ '<STR_LIT>' ] . dt , '<STR_LIT>' , None ) : <EOL> recs_table = '<STR_LIT>' <EOL> else : <EOL> recs_table = '<STR_LIT>' <EOL> thisandfuture = ( rrange == THISANDFUTURE ) <EOL> if thisandfuture : <EOL> start_shift , duration = calc_shift_deltas ( vevent ) <EOL> start_shift = start_shift . days * <NUM_LIT> * <NUM_LIT> + start_shift . seconds <EOL> duration = duration . days * <NUM_LIT> * <NUM_LIT> + duration . seconds <EOL> dtstartend = aux . expand ( vevent , href ) <EOL> if not dtstartend : <EOL> return <EOL> for dtstart , dtend in dtstartend : <EOL> if dtype == DATE : <EOL> dbstart = aux . to_unix_time ( dtstart ) <EOL> dbend = aux . to_unix_time ( dtend ) <EOL> if rec_id is not None : <EOL> rec_inst = aux . to_unix_time ( rec_id . dt ) <EOL> ref = rec_inst <EOL> else : <EOL> rec_inst = dbstart <EOL> ref = PROTO <EOL> else : <EOL> dbstart = aux . to_unix_time ( dtstart ) <EOL> dbend = aux . to_unix_time ( dtend ) <EOL> if rec_id is not None : <EOL> ref = rec_inst = str ( aux . to_unix_time ( rec_id . dt ) ) <EOL> else : <EOL> rec_inst = dbstart <EOL> ref = PROTO <EOL> if thisandfuture : <EOL> recs_sql_s = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( recs_table ) ) <EOL> stuple = ( start_shift , start_shift + duration , ref , rec_inst , href , calendar ) <EOL> else : <EOL> recs_sql_s = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( recs_table ) ) <EOL> stuple = ( dbstart , dbend , href , ref , dtype , rec_inst , calendar ) <EOL> self . sql_ex ( recs_sql_s , stuple ) <EOL> def get_ctag ( self , calendar ) : <EOL> stuple = ( calendar , ) <EOL> sql_s = '<STR_LIT>' <EOL> try : <EOL> ctag = self . sql_ex ( sql_s , stuple ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> return ctag <EOL> except IndexError : <EOL> return None <EOL> def set_ctag ( self , ctag , calendar ) : <EOL> stuple = ( ctag , calendar , ) <EOL> sql_s = '<STR_LIT>' <EOL> self . sql_ex ( sql_s , stuple ) <EOL> self . conn . commit ( ) <EOL> def get_etag ( self , href , calendar ) : <EOL> """<STR_LIT>""" <EOL> sql_s = '<STR_LIT>' <EOL> try : <EOL> etag = self . sql_ex ( sql_s , ( href , calendar ) ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> return etag <EOL> except IndexError : <EOL> return None <EOL> def delete ( self , href , etag = None , calendar = None ) : <EOL> """<STR_LIT>""" <EOL> assert calendar is not None <EOL> for table in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> sql_s = '<STR_LIT>' . format ( table ) <EOL> self . sql_ex ( sql_s , ( href , calendar ) ) <EOL> sql_s = '<STR_LIT>' <EOL> self . sql_ex ( sql_s , ( href , calendar ) ) <EOL> def list ( self , calendar ) : <EOL> """<STR_LIT>""" <EOL> sql_s = '<STR_LIT>' <EOL> return list ( set ( self . sql_ex ( sql_s , ( calendar , ) ) ) ) <EOL> def get_localized ( self , start , end , minimal = False ) : <EOL> """<STR_LIT>""" <EOL> assert start . tzinfo is not None <EOL> assert end . tzinfo is not None <EOL> start = aux . to_unix_time ( start ) <EOL> end = aux . to_unix_time ( end ) <EOL> if minimal : <EOL> sql_s = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> sql_s = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> stuple = ( start , end , start , end , start , end ) <EOL> result = self . sql_ex ( sql_s . format ( self . _select_calendars ) , stuple ) <EOL> if minimal : <EOL> for calendar in result : <EOL> yield EventStandIn ( calendar [ <NUM_LIT:0> ] ) <EOL> else : <EOL> for item , href , start , end , ref , etag , dtype , calendar in result : <EOL> start = pytz . UTC . localize ( datetime . utcfromtimestamp ( start ) ) <EOL> end = pytz . UTC . localize ( datetime . utcfromtimestamp ( end ) ) <EOL> yield self . construct_event ( item , href , start , end , ref , etag , calendar , dtype ) <EOL> def get_floating ( self , start , end , minimal = False ) : <EOL> """<STR_LIT>""" <EOL> assert start . tzinfo is None <EOL> assert end . tzinfo is None <EOL> strstart = aux . to_unix_time ( start ) <EOL> strend = aux . to_unix_time ( end ) <EOL> if minimal : <EOL> sql_s = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> sql_s = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> stuple = ( strstart , strend , strstart , strend , strstart , strend ) <EOL> result = self . sql_ex ( sql_s . format ( self . _select_calendars ) , stuple ) <EOL> if minimal : <EOL> for calendar in result : <EOL> yield EventStandIn ( calendar [ <NUM_LIT:0> ] ) <EOL> else : <EOL> for item , href , start , end , ref , etag , dtype , calendar in result : <EOL> start = datetime . utcfromtimestamp ( start ) <EOL> end = datetime . utcfromtimestamp ( end ) <EOL> yield self . construct_event ( item , href , start , end , ref , etag , calendar , dtype ) <EOL> def get_localized_at ( self , dtime ) : <EOL> """<STR_LIT>""" <EOL> assert dtime . tzinfo is not None <EOL> dtime = aux . to_unix_time ( dtime ) <EOL> sql_s = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> stuple = ( dtime , dtime ) <EOL> result = self . sql_ex ( sql_s . format ( self . _select_calendars ) , stuple ) <EOL> for item , href , start , end , ref , etag , dtype , calendar in result : <EOL> start = pytz . UTC . localize ( datetime . utcfromtimestamp ( start ) ) <EOL> end = pytz . UTC . localize ( datetime . utcfromtimestamp ( end ) ) <EOL> yield self . construct_event ( item , href , start , end , ref , etag , calendar , dtype ) <EOL> def get_floating_at ( self , dtime ) : <EOL> """<STR_LIT>""" <EOL> assert dtime . tzinfo is None <EOL> dtime = aux . to_unix_time ( dtime ) <EOL> sql_s = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> stuple = ( dtime , dtime ) <EOL> result = self . sql_ex ( sql_s . format ( self . _select_calendars ) , stuple ) <EOL> for item , href , start , end , ref , etag , dtype , calendar in result : <EOL> start = datetime . utcfromtimestamp ( start ) <EOL> end = datetime . utcfromtimestamp ( end ) <EOL> yield self . construct_event ( item , href , start , end , ref , etag , calendar , dtype ) <EOL> def get ( self , href , start = None , end = None , ref = None , dtype = None , calendar = None ) : <EOL> """<STR_LIT>""" <EOL> assert calendar is not None <EOL> sql_s = '<STR_LIT>' <EOL> result = self . sql_ex ( sql_s , ( href , calendar ) ) <EOL> href , etag , item = result [ <NUM_LIT:0> ] <EOL> if dtype == DATE : <EOL> start = start . date ( ) <EOL> end = end . date ( ) <EOL> return Event . fromString ( item , <EOL> locale = self . locale , <EOL> href = href , <EOL> calendar = calendar , <EOL> etag = etag , <EOL> start = start , <EOL> end = end , <EOL> ref = ref , <EOL> ) <EOL> def construct_event ( self , item , href , start , end , ref , etag , calendar , dtype = None ) : <EOL> if dtype == DATE : <EOL> start = start . date ( ) <EOL> end = end . date ( ) <EOL> return Event . fromString ( item , <EOL> locale = self . locale , <EOL> href = href , <EOL> calendar = calendar , <EOL> etag = etag , <EOL> start = start , <EOL> end = end , <EOL> ref = ref , <EOL> ) <EOL> def search ( self , search_string ) : <EOL> """<STR_LIT>""" <EOL> sql_s = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> stuple = ( '<STR_LIT>' . format ( search_string ) , ) <EOL> result = self . sql_ex ( sql_s . format ( self . _select_calendars ) , stuple ) <EOL> for href , calendar in result : <EOL> event = self . get ( href , calendar = calendar ) <EOL> yield event <EOL> def check_support ( vevent , href , calendar ) : <EOL> """<STR_LIT>""" <EOL> rec_id = vevent . get ( RECURRENCE_ID ) <EOL> if rec_id is not None and rec_id . params . get ( '<STR_LIT>' ) == THISANDPRIOR : <EOL> raise UpdateFailed ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( href , calendar ) <EOL> ) <EOL> rdate = vevent . get ( '<STR_LIT>' ) <EOL> if rdate is not None and rdate . params . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> raise UpdateFailed ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( href , calendar ) <EOL> ) <EOL> def check_for_errors ( component , calendar , href ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( component , '<STR_LIT>' ) and component . errors : <EOL> logger . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( calendar , href ) ) <EOL> for error in component . errors : <EOL> logger . error ( error ) <EOL> logger . error ( '<STR_LIT>' ) <EOL> def calc_shift_deltas ( vevent ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( vevent , icalendar . Event ) <EOL> start_shift = vevent [ '<STR_LIT>' ] . dt - vevent [ '<STR_LIT>' ] . dt <EOL> try : <EOL> duration = vevent [ '<STR_LIT>' ] . dt - vevent [ '<STR_LIT>' ] . dt <EOL> except KeyError : <EOL> duration = vevent [ '<STR_LIT>' ] . dt <EOL> return start_shift , duration </s>
