<s> """<STR_LIT>""" <EOL> import logging <EOL> import os . path <EOL> import pytz <EOL> from datetime import datetime <EOL> import asimap . utils <EOL> import asimap . constants <EOL> from asimap . exceptions import MailboxInconsistency <EOL> class BadSearchOp ( Exception ) : <EOL> def __init__ ( self , value = "<STR_LIT>" ) : <EOL> self . value = value <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % self . value <EOL> class SearchContext ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , mailbox , msg_key , msg_number , seq_max , uid_max , <EOL> sequences ) : <EOL> """<STR_LIT>""" <EOL> self . log = logging . getLogger ( "<STR_LIT>" % ( __name__ , <EOL> self . __class__ . __name__ , <EOL> mailbox . name , <EOL> msg_key ) ) <EOL> self . mailbox = mailbox <EOL> self . msg_key = msg_key <EOL> self . seq_max = seq_max <EOL> self . uid_max = uid_max <EOL> self . msg_number = msg_number <EOL> self . mailbox_sequences = sequences <EOL> self . path = os . path . join ( mailbox . mailbox . _path , str ( msg_key ) ) <EOL> self . internal_date = datetime . fromtimestamp ( os . path . getmtime ( self . path ) , <EOL> pytz . UTC ) <EOL> self . _msg = None <EOL> self . _uid_vv = None <EOL> self . _uid = None <EOL> self . _sequences = None <EOL> return <EOL> @ property <EOL> def msg ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _msg : <EOL> return self . _msg <EOL> self . _msg = self . mailbox . get_and_cache_msg ( self . msg_key ) <EOL> if self . _uid is None : <EOL> self . _uid_vv , self . _uid = asimap . utils . get_uidvv_uid ( self . msg [ '<STR_LIT>' ] ) <EOL> if self . _uid is None : <EOL> raise MailboxInconsistency ( mbox_name = self . mailbox . name , <EOL> msg_key = self . msg_key ) <EOL> else : <EOL> uid_vv , uid = asimap . utils . get_uidvv_uid ( self . msg [ '<STR_LIT>' ] ) <EOL> if self . _uid != uid or uid is None : <EOL> raise MailboxInconsistency ( mbox_name = self . mailbox . name , <EOL> msg_key = self . msg_key ) <EOL> return self . _msg <EOL> @ property <EOL> def uid ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _uid : <EOL> return self . _uid <EOL> try : <EOL> self . _uid = self . mailbox . uids [ self . msg_number - <NUM_LIT:1> ] <EOL> except IndexError : <EOL> self . _uid_vv , self . _uid = self . mailbox . get_uid_from_msg ( self . msg_key ) <EOL> return self . _uid <EOL> @ property <EOL> def uid_vv ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _uid_vv : <EOL> return self . _uid_vv <EOL> self . _uid_vv , self . _uid = self . mailbox . get_uid_from_msg ( self . msg_key ) <EOL> return self . _uid_vv <EOL> @ property <EOL> def sequences ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _sequences : <EOL> return self . _sequences <EOL> if self . _msg : <EOL> return self . _msg . get_sequences ( ) <EOL> self . _sequences = [ ] <EOL> for name , key_list in self . mailbox_sequences . iteritems ( ) : <EOL> if self . msg_key in key_list : <EOL> self . _sequences . append ( name ) <EOL> return self . _sequences <EOL> class IMAPSearch ( object ) : <EOL> """<STR_LIT>""" <EOL> OP_ALL = '<STR_LIT:all>' <EOL> OP_AND = '<STR_LIT>' <EOL> OP_BEFORE = '<STR_LIT>' <EOL> OP_BODY = '<STR_LIT:body>' <EOL> OP_HEADER = '<STR_LIT>' <EOL> OP_KEYWORD = '<STR_LIT>' <EOL> OP_LARGER = '<STR_LIT>' <EOL> OP_MESSAGE_SET = '<STR_LIT>' <EOL> OP_NOT = '<STR_LIT>' <EOL> OP_ON = '<STR_LIT>' <EOL> OP_OR = '<STR_LIT>' <EOL> OP_SENTBEFORE = '<STR_LIT>' <EOL> OP_SENTON = '<STR_LIT>' <EOL> OP_SENTSINCE = '<STR_LIT>' <EOL> OP_SINCE = '<STR_LIT>' <EOL> OP_SMALLER = '<STR_LIT>' <EOL> OP_TEXT = '<STR_LIT:text>' <EOL> OP_UID = '<STR_LIT>' <EOL> VALID_OPS = ( OP_ALL , OP_AND , OP_BEFORE , OP_BODY , OP_HEADER , OP_KEYWORD , <EOL> OP_LARGER , OP_MESSAGE_SET , OP_NOT , OP_ON , OP_OR , OP_SENTON , <EOL> OP_SENTBEFORE , OP_SENTSINCE , OP_SINCE , OP_SMALLER , OP_TEXT , <EOL> OP_UID ) <EOL> def __init__ ( self , op , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . log = logging . getLogger ( "<STR_LIT>" % ( __name__ , self . __class__ . __name__ ) ) <EOL> if op not in self . VALID_OPS : <EOL> raise BadSearchOp ( "<STR_LIT>" % op ) <EOL> self . op = op <EOL> self . args = kwargs <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % self . op <EOL> def __str__ ( self ) : <EOL> result = "<STR_LIT>" % self . op <EOL> if self . op in ( self . OP_AND , self . OP_OR ) : <EOL> elt = [ ] <EOL> for search in self . args [ '<STR_LIT>' ] : <EOL> elt . append ( str ( search ) ) <EOL> result += '<STR_LIT>' % '<STR_LIT:U+002CU+0020>' . join ( elt ) <EOL> elif self . op in ( self . OP_NOT ) : <EOL> result += '<STR_LIT>' % self . args [ '<STR_LIT>' ] <EOL> elif self . op in ( self . OP_BEFORE , self . OP_ON , self . OP_SENTON , <EOL> self . OP_SENTBEFORE , self . OP_SENTSINCE , self . OP_SINCE ) : <EOL> result += '<STR_LIT>' % self . args [ '<STR_LIT:date>' ] <EOL> elif self . op in ( self . OP_LARGER , self . OP_SMALLER ) : <EOL> result += '<STR_LIT>' % self . args [ '<STR_LIT:n>' ] <EOL> elif self . op in ( self . OP_TEXT , self . OP_BODY ) : <EOL> result += '<STR_LIT>' % self . args [ '<STR_LIT:string>' ] <EOL> elif self . op in ( self . OP_HEADER ) : <EOL> result += '<STR_LIT>' % ( self . args [ '<STR_LIT>' ] , <EOL> self . args [ '<STR_LIT:string>' ] ) <EOL> elif self . op in ( self . OP_KEYWORD ) : <EOL> result += '<STR_LIT>' % self . args [ '<STR_LIT>' ] <EOL> return result + "<STR_LIT:)>" <EOL> def match ( self , ctx ) : <EOL> """<STR_LIT>""" <EOL> self . ctx = ctx <EOL> return getattr ( self , '<STR_LIT>' % self . op ) ( ) <EOL> def _match_keyword ( self ) : <EOL> """<STR_LIT>""" <EOL> keyword = asimap . constants . flag_to_seq ( self . args [ '<STR_LIT>' ] ) <EOL> result = keyword in self . ctx . sequences <EOL> return result <EOL> def _match_header ( self ) : <EOL> """<STR_LIT>""" <EOL> header = self . args [ "<STR_LIT>" ] <EOL> return header in self . ctx . msg and self . ctx . msg [ header ] . lower ( ) . find ( self . args [ '<STR_LIT:string>' ] ) != - <NUM_LIT:1> <EOL> def _match_and ( self ) : <EOL> """<STR_LIT>""" <EOL> for search_op in self . args [ '<STR_LIT>' ] : <EOL> if not search_op . match ( self . ctx ) : <EOL> return False <EOL> return True <EOL> def _match_all ( self ) : <EOL> """<STR_LIT>""" <EOL> return True <EOL> def _match_or ( self ) : <EOL> """<STR_LIT>""" <EOL> for search_op in self . args [ '<STR_LIT>' ] : <EOL> if search_op . match ( self . ctx ) : <EOL> return True <EOL> return False <EOL> def _match_before ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . ctx . internal_date < self . args [ "<STR_LIT:date>" ] <EOL> def _match_body ( self ) : <EOL> """<STR_LIT>""" <EOL> text = self . args [ '<STR_LIT:string>' ] <EOL> for msg_part in self . ctx . msg . walk ( ) : <EOL> if msg_part . is_multipart ( ) : <EOL> continue <EOL> if msg_part . get_payload ( decode = True ) . lower ( ) . find ( text ) != - <NUM_LIT:1> : <EOL> return True <EOL> return False <EOL> def _match_larger ( self ) : <EOL> """<STR_LIT>""" <EOL> return os . path . getsize ( self . ctx . path ) > self . args [ "<STR_LIT:n>" ] <EOL> def _match_message_set ( self ) : <EOL> """<STR_LIT>""" <EOL> for elt in self . args [ "<STR_LIT>" ] : <EOL> if isinstance ( elt , str ) and elt == "<STR_LIT:*>" and self . ctx . msg_number == self . id_max : <EOL> return True <EOL> elif isinstance ( elt , int ) and elt == self . ctx . msg_number : <EOL> return True <EOL> elif isinstance ( elt , tuple ) and ( self . ctx . msg_number >= elt [ <NUM_LIT:0> ] and self . ctx . msg_number <= elt [ <NUM_LIT:1> ] ) : <EOL> return True <EOL> return False <EOL> def _match_not ( self ) : <EOL> """<STR_LIT>""" <EOL> return not self . args [ '<STR_LIT>' ] . match ( self . ctx ) <EOL> def _match_on ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . ctx . internal_date . date ( ) == self . args [ "<STR_LIT:date>" ] . date ( ) <EOL> def _match_sentbefore ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT:date>' in self . ctx . msg and self . args [ '<STR_LIT:date>' ] > asimap . utils . parsedate ( self . ctx . msg [ '<STR_LIT:date>' ] ) <EOL> def _match_senton ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT:date>' in self . ctx . msg and self . args [ '<STR_LIT:date>' ] . date ( ) == asimap . utils . parsedate ( self . ctx . msg [ '<STR_LIT:date>' ] ) . date ( ) <EOL> def _match_sentsince ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT:date>' in self . ctx . msg and self . args [ '<STR_LIT:date>' ] < asimap . utils . parsedate ( self . ctx . msg [ '<STR_LIT:date>' ] ) <EOL> def _match_since ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . ctx . internal_date > self . args [ "<STR_LIT:date>" ] or self . ctx . internal_date . date ( ) == self . args [ "<STR_LIT:date>" ] . date ( ) <EOL> def _match_smaller ( self ) : <EOL> """<STR_LIT>""" <EOL> return os . path . getsize ( self . ctx . path ) < self . args [ "<STR_LIT:n>" ] <EOL> def _match_text ( self ) : <EOL> """<STR_LIT>""" <EOL> text = self . args [ '<STR_LIT:string>' ] <EOL> for header in self . ctx . msg . values ( ) : <EOL> if header . lower ( ) . find ( text ) != - <NUM_LIT:1> : <EOL> return True <EOL> return self . _match_body ( ) <EOL> def _match_uid ( self ) : <EOL> """<STR_LIT>""" <EOL> for elt in self . args [ "<STR_LIT>" ] : <EOL> if isinstance ( elt , str ) and elt == "<STR_LIT:*>" : <EOL> if self . ctx . uid == self . ctx . uid_max : <EOL> return True <EOL> elif isinstance ( elt , int ) and elt == self . ctx . uid : <EOL> return True <EOL> elif isinstance ( elt , tuple ) and ( self . ctx . uid >= elt [ <NUM_LIT:0> ] and self . ctx . uid <= elt [ <NUM_LIT:1> ] ) : <EOL> return True <EOL> return False </s>
