<s> from __future__ import unicode_literals <EOL> from email . utils import parseaddr <EOL> import logging <EOL> import mimetypes <EOL> import os <EOL> from os import path <EOL> import errno <EOL> from zipfile import ZipFile <EOL> import email <EOL> from . . compat import to_unicode , string_types , to_native , formataddr as compat_formataddr <EOL> from . . loader . helpers import decode_text <EOL> from . . message import Message <EOL> from . . utils import decode_header <EOL> class FileNotFound ( Exception ) : <EOL> pass <EOL> def split_template_path ( template ) : <EOL> """<STR_LIT>""" <EOL> pieces = [ ] <EOL> for piece in template . split ( '<STR_LIT:/>' ) : <EOL> if path . sep in piece or ( path . altsep and path . altsep in piece ) or piece == path . pardir : <EOL> raise FileNotFound ( template ) <EOL> elif piece and piece != '<STR_LIT:.>' : <EOL> pieces . append ( piece ) <EOL> return pieces <EOL> def open_if_exists ( filename , mode = '<STR_LIT:rb>' ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return open ( filename , mode ) <EOL> except IOError as e : <EOL> if e . errno not in ( errno . ENOENT , errno . EISDIR ) : <EOL> raise <EOL> class BaseLoader ( object ) : <EOL> def __getitem__ ( self , filename ) : <EOL> try : <EOL> contents , _ = self . get_file ( filename ) <EOL> return contents <EOL> except FileNotFound : <EOL> return None <EOL> def get_file ( self , name ) : <EOL> raise NotImplementedError <EOL> def list_files ( self ) : <EOL> raise NotImplementedError <EOL> def content ( self , filename , is_html = False , decode = True , guess_charset = False , charset = '<STR_LIT:utf-8>' ) : <EOL> data = self [ filename ] <EOL> if decode : <EOL> data , encoding = decode_text ( data , <EOL> is_html = is_html , <EOL> guess_charset = guess_charset , <EOL> try_common_charsets = False , <EOL> fallback_charset = charset ) <EOL> return data <EOL> def find_index_file ( self , filename = None , extensions = ( '<STR_LIT>' , '<STR_LIT>' ) , stop_names = ( '<STR_LIT:index>' , ) , raise_if_not_found = True ) : <EOL> if filename : <EOL> if self [ filename ] : <EOL> return filename <EOL> else : <EOL> raise FileNotFound ( filename ) <EOL> found_files = [ ] <EOL> for filename in self . list_files ( ) : <EOL> bn = os . path . basename ( filename ) <EOL> if bn . startswith ( '<STR_LIT:.>' ) : <EOL> continue <EOL> name , ext = os . path . splitext ( bn ) <EOL> if ext in extensions : <EOL> if stop_names and name in stop_names : <EOL> return filename <EOL> found_files . append ( filename ) <EOL> if found_files : <EOL> return found_files [ <NUM_LIT:0> ] <EOL> elif raise_if_not_found : <EOL> raise FileNotFound ( '<STR_LIT>' % "<STR_LIT:|>" . join ( extensions ) ) <EOL> def find_index_html ( self , filename = None ) : <EOL> return self . find_index_file ( filename = filename ) <EOL> def find_index_text ( self , filename = None ) : <EOL> return self . find_index_file ( filename = filename , <EOL> extensions = ( '<STR_LIT>' , ) , <EOL> stop_names = ( '<STR_LIT:index>' , ) , <EOL> raise_if_not_found = False ) <EOL> class FileSystemLoader ( BaseLoader ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , searchpath , encoding = '<STR_LIT:utf-8>' , base_path = None ) : <EOL> if isinstance ( searchpath , string_types ) : <EOL> searchpath = [ searchpath ] <EOL> self . searchpath = list ( searchpath ) <EOL> self . encoding = encoding <EOL> self . base_path = base_path <EOL> def get_file ( self , filename ) : <EOL> if self . base_path : <EOL> filename = path . join ( self . base_path , filename ) <EOL> pieces = split_template_path ( filename ) <EOL> for searchpath in self . searchpath : <EOL> filename = path . join ( searchpath , * pieces ) <EOL> f = open_if_exists ( filename ) <EOL> if f is None : <EOL> continue <EOL> try : <EOL> contents = f . read ( ) <EOL> finally : <EOL> f . close ( ) <EOL> return contents , filename <EOL> raise FileNotFound ( filename ) <EOL> def list_files ( self ) : <EOL> found = set ( ) <EOL> for searchpath in self . searchpath : <EOL> for dirpath , dirnames , filenames in os . walk ( searchpath ) : <EOL> for filename in filenames : <EOL> template = path . join ( dirpath , filename ) [ len ( searchpath ) : ] . strip ( path . sep ) . replace ( path . sep , '<STR_LIT:/>' ) <EOL> if template [ : <NUM_LIT:2> ] == '<STR_LIT>' : <EOL> template = template [ <NUM_LIT:2> : ] <EOL> if template not in found : <EOL> yield template <EOL> class ZipLoader ( BaseLoader ) : <EOL> """<STR_LIT>""" <EOL> common_filename_charsets = [ '<STR_LIT:ascii>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:utf-8>' ] <EOL> def __init__ ( self , file , encoding = '<STR_LIT:utf-8>' , base_path = None ) : <EOL> if not isinstance ( file , ZipFile ) : <EOL> file = ZipFile ( file , '<STR_LIT:r>' ) <EOL> self . zipfile = file <EOL> self . encoding = encoding <EOL> self . base_path = base_path <EOL> self . _decoded_filenames = None <EOL> self . _original_filenames = None <EOL> def _decode_filename ( self , name ) : <EOL> for enc in self . common_filename_charsets : <EOL> try : <EOL> return to_unicode ( name , enc ) <EOL> except UnicodeDecodeError : <EOL> pass <EOL> return name <EOL> def _unpack ( self ) : <EOL> if self . _decoded_filenames is None : <EOL> self . _original_filenames = set ( self . zipfile . namelist ( ) ) <EOL> self . _decoded_filenames = dict ( [ ( self . _decode_filename ( name ) , name ) for name in self . _original_filenames ] ) <EOL> def get_file ( self , name ) : <EOL> if self . base_path : <EOL> name = path . join ( self . base_path , name ) <EOL> name = path . join ( * split_template_path ( name ) ) <EOL> self . _unpack ( ) <EOL> if isinstance ( name , str ) : <EOL> name = to_unicode ( name , '<STR_LIT:utf-8>' ) <EOL> if name not in self . _original_filenames : <EOL> name = self . _decoded_filenames . get ( name ) <EOL> if name is None : <EOL> raise FileNotFound ( name ) <EOL> return self . zipfile . read ( name ) , name <EOL> def list_files ( self ) : <EOL> self . _unpack ( ) <EOL> return sorted ( self . _decoded_filenames ) <EOL> class MsgLoader ( BaseLoader ) : <EOL> """<STR_LIT>""" <EOL> common_charsets = [ '<STR_LIT:ascii>' , '<STR_LIT:utf-8>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> def __init__ ( self , msg , base_path = None ) : <EOL> if isinstance ( msg , string_types ) : <EOL> self . msg = email . message_from_string ( msg ) <EOL> elif isinstance ( msg , bytes ) : <EOL> self . msg = email . message_from_string ( to_native ( msg ) ) <EOL> else : <EOL> self . msg = msg <EOL> self . base_path = base_path <EOL> self . _html_parts = [ ] <EOL> self . _text_parts = [ ] <EOL> self . _files = { } <EOL> self . _content_ids = { } <EOL> self . _parsed = False <EOL> self . headers = { } <EOL> def decode_text ( self , text , charset = None ) : <EOL> if charset : <EOL> try : <EOL> return text . decode ( charset ) , charset <EOL> except UnicodeError : <EOL> pass <EOL> for charset in self . common_charsets : <EOL> try : <EOL> return text . decode ( charset ) , charset <EOL> except UnicodeError : <EOL> pass <EOL> return text , None <EOL> def clean_content_id ( self , content_id ) : <EOL> if content_id . startswith ( '<STR_LIT:<>' ) : <EOL> content_id = content_id [ <NUM_LIT:1> : ] <EOL> if content_id . endswith ( '<STR_LIT:>>' ) : <EOL> content_id = content_id [ : - <NUM_LIT:1> ] <EOL> return content_id <EOL> def extract_part_text ( self , part ) : <EOL> return self . decode_text ( part . get_payload ( decode = True ) , charset = part . get_param ( '<STR_LIT>' ) ) [ <NUM_LIT:0> ] <EOL> def add_html_part ( self , part ) : <EOL> self . _html_parts . append ( { '<STR_LIT:data>' : self . extract_part_text ( part ) , <EOL> '<STR_LIT>' : part . get_content_type ( ) } ) <EOL> def add_text_part ( self , part ) : <EOL> self . _text_parts . append ( { '<STR_LIT:data>' : self . extract_part_text ( part ) , <EOL> '<STR_LIT>' : part . get_content_type ( ) } ) <EOL> def add_attachment_part ( self , part ) : <EOL> counter = <NUM_LIT:1> <EOL> f = { } <EOL> filename = part . get_filename ( ) <EOL> if not filename : <EOL> ext = mimetypes . guess_extension ( part . get_content_type ( ) ) <EOL> if not ext : <EOL> ext = '<STR_LIT>' <EOL> filename = '<STR_LIT>' % ( counter , ext ) <EOL> counter += <NUM_LIT:1> <EOL> f [ '<STR_LIT:filename>' ] = filename <EOL> f [ '<STR_LIT>' ] = part . get_content_type ( ) <EOL> content_id = part [ '<STR_LIT>' ] <EOL> if content_id : <EOL> f [ '<STR_LIT>' ] = self . clean_content_id ( content_id ) <EOL> f [ '<STR_LIT>' ] = True <EOL> self . _content_ids [ f [ '<STR_LIT>' ] ] = f [ '<STR_LIT:filename>' ] <EOL> f [ '<STR_LIT:data>' ] = part . get_payload ( decode = True ) <EOL> self . _files [ f [ '<STR_LIT:filename>' ] ] = f <EOL> def _parse ( self ) : <EOL> for part in self . msg . walk ( ) : <EOL> content_type = part . get_content_type ( ) <EOL> if content_type . startswith ( '<STR_LIT>' ) : <EOL> continue <EOL> if content_type == '<STR_LIT>' : <EOL> self . add_html_part ( part ) <EOL> continue <EOL> if content_type == '<STR_LIT>' : <EOL> self . add_text_part ( part ) <EOL> continue <EOL> self . add_attachment_part ( part ) <EOL> def parse ( self ) : <EOL> if not self . _parsed : <EOL> self . _parse ( ) <EOL> self . _parsed = True <EOL> def get_file ( self , name ) : <EOL> self . parse ( ) <EOL> if name . startswith ( '<STR_LIT>' ) : <EOL> name = self . _content_ids . get ( name [ <NUM_LIT:4> : ] ) <EOL> f = self . _files . get ( name ) <EOL> if f : <EOL> return f [ '<STR_LIT:data>' ] , name <EOL> raise FileNotFound ( name ) <EOL> def list_files ( self ) : <EOL> self . parse ( ) <EOL> return self . _files <EOL> @ property <EOL> def attachments ( self ) : <EOL> self . parse ( ) <EOL> return self . _files . values ( ) <EOL> @ property <EOL> def html ( self ) : <EOL> self . parse ( ) <EOL> return self . _html_parts and self . _html_parts [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] or None <EOL> @ property <EOL> def text ( self ) : <EOL> self . parse ( ) <EOL> return self . _text_parts and self . _text_parts [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] or None <EOL> def decode_header_value ( self , v ) : <EOL> if isinstance ( v , bytes ) : <EOL> v = self . decode_text ( v ) [ <NUM_LIT:0> ] <EOL> return decode_header ( v ) <EOL> def decode_address_header_value ( self , value , skip_invalid = False ) : <EOL> r = [ ] <EOL> if isinstance ( value , bytes ) : <EOL> value = self . decode_text ( value ) [ <NUM_LIT:0> ] <EOL> for token in value . split ( '<STR_LIT:U+002C>' ) : <EOL> name , email = parseaddr ( token . strip ( ) ) <EOL> if not name and '<STR_LIT:@>' not in email : <EOL> if not skip_invalid : <EOL> r . append ( decode_header ( email ) ) <EOL> else : <EOL> r . append ( compat_formataddr ( [ decode_header ( name ) , email ] ) ) <EOL> return r <EOL> def filter_header ( self , name ) : <EOL> return name == '<STR_LIT>' or name in Message . ADDRESS_HEADERS <EOL> def copy_header_to_message ( self , message , name , value ) : <EOL> """<STR_LIT>""" <EOL> if name == '<STR_LIT>' : <EOL> message . subject = self . decode_header_value ( value ) <EOL> elif name == '<STR_LIT:to>' : <EOL> r = self . decode_address_header_value ( value ) <EOL> if r : <EOL> message . mail_to = r [ <NUM_LIT:0> ] <EOL> elif name == '<STR_LIT>' : <EOL> r = self . decode_address_header_value ( value ) <EOL> if r : <EOL> message . mail_from = r [ <NUM_LIT:0> ] <EOL> elif name in Message . ADDRESS_HEADERS : <EOL> message . _headers [ name ] = "<STR_LIT:U+002C>" . join ( self . decode_address_header_value ( value ) ) <EOL> else : <EOL> message . _headers [ name ] = self . decode_header_value ( value ) <EOL> def copy_headers_to_message ( self , message ) : <EOL> """<STR_LIT>""" <EOL> for k , v in self . msg . items ( ) : <EOL> k = k . lower ( ) <EOL> if self . filter_header ( k ) : <EOL> self . copy_header_to_message ( message , k , v ) </s>
