<s> """<STR_LIT>""" <EOL> from __future__ import with_statement <EOL> import contextlib <EOL> try : <EOL> from urllib . request import Request as URLRequest , urlopen <EOL> from urllib . error import HTTPError <EOL> except ImportError : <EOL> from urllib2 import Request as URLRequest , urlopen <EOL> from urllib2 import HTTPError <EOL> import logging <EOL> from io import open <EOL> from webassets . six . moves import filter <EOL> from . utils import cmp_debug_levels , StringIO , hash_func <EOL> __all__ = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> log . addHandler ( logging . StreamHandler ( ) ) <EOL> import os <EOL> if os . environ . get ( '<STR_LIT>' ) : <EOL> log . setLevel ( logging . DEBUG ) <EOL> else : <EOL> log . setLevel ( logging . ERROR ) <EOL> class BaseHunk ( object ) : <EOL> """<STR_LIT>""" <EOL> def mtime ( self ) : <EOL> raise NotImplementedError ( ) <EOL> def id ( self ) : <EOL> return hash_func ( self . data ( ) ) <EOL> def __eq__ ( self , other ) : <EOL> if isinstance ( other , BaseHunk ) : <EOL> return hash_func ( self ) == hash_func ( other ) <EOL> return False <EOL> def data ( self ) : <EOL> raise NotImplementedError ( ) <EOL> def save ( self , filename ) : <EOL> with open ( filename , '<STR_LIT:w>' , encoding = '<STR_LIT:utf-8>' ) as f : <EOL> f . write ( self . data ( ) ) <EOL> class FileHunk ( BaseHunk ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , filename ) : <EOL> self . filename = filename <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . filename ) <EOL> def mtime ( self ) : <EOL> pass <EOL> def data ( self ) : <EOL> f = open ( self . filename , '<STR_LIT:r>' , encoding = '<STR_LIT:utf-8>' ) <EOL> try : <EOL> return f . read ( ) <EOL> finally : <EOL> f . close ( ) <EOL> class UrlHunk ( BaseHunk ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url , env = None ) : <EOL> self . url = url <EOL> self . env = env <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . url ) <EOL> def data ( self ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> request = URLRequest ( self . url ) <EOL> if self . env and self . env . cache : <EOL> headers = self . env . cache . get ( <EOL> ( '<STR_LIT:url>' , '<STR_LIT>' , self . url ) ) <EOL> if headers : <EOL> etag , lmod = headers <EOL> if etag : request . add_header ( '<STR_LIT>' , etag ) <EOL> if lmod : request . add_header ( '<STR_LIT>' , lmod ) <EOL> try : <EOL> response = urlopen ( request ) <EOL> except HTTPError as e : <EOL> if e . code != <NUM_LIT> : <EOL> raise <EOL> self . _data = self . env . cache . get ( ( '<STR_LIT:url>' , '<STR_LIT>' , self . url ) ) <EOL> else : <EOL> with contextlib . closing ( response ) : <EOL> self . _data = response . read ( ) <EOL> if self . env and self . env . cache : <EOL> self . env . cache . set ( <EOL> ( '<STR_LIT:url>' , '<STR_LIT>' , self . url ) , <EOL> ( response . headers . get ( "<STR_LIT>" ) , <EOL> response . headers . get ( "<STR_LIT>" ) ) ) <EOL> self . env . cache . set ( ( '<STR_LIT:url>' , '<STR_LIT>' , self . url ) , self . _data ) <EOL> return self . _data <EOL> class MemoryHunk ( BaseHunk ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , data , files = None ) : <EOL> self . _data = data <EOL> self . files = files or [ ] <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , hash_func ( self ) ) <EOL> def mtime ( self ) : <EOL> pass <EOL> def data ( self ) : <EOL> if hasattr ( self . _data , '<STR_LIT>' ) : <EOL> return self . _data . read ( ) <EOL> return self . _data <EOL> def save ( self , filename ) : <EOL> f = open ( filename , '<STR_LIT:w>' , encoding = '<STR_LIT:utf-8>' ) <EOL> try : <EOL> f . write ( self . data ( ) ) <EOL> finally : <EOL> f . close ( ) <EOL> def merge ( hunks , separator = None ) : <EOL> """<STR_LIT>""" <EOL> if not separator : <EOL> separator = '<STR_LIT:\n>' <EOL> return MemoryHunk ( separator . join ( [ h . data ( ) for h in hunks ] ) ) <EOL> class MoreThanOneFilterError ( Exception ) : <EOL> def __init__ ( self , message , filters ) : <EOL> Exception . __init__ ( self , message ) <EOL> self . filters = filters <EOL> class NoFilters ( Exception ) : <EOL> pass <EOL> class FilterTool ( object ) : <EOL> """<STR_LIT>""" <EOL> VALID_TRANSFORMS = ( '<STR_LIT:input>' , '<STR_LIT>' , ) <EOL> VALID_FUNCS = ( '<STR_LIT>' , '<STR_LIT>' , ) <EOL> def __init__ ( self , cache = None , no_cache_read = False , kwargs = None ) : <EOL> self . cache = cache <EOL> self . no_cache_read = no_cache_read <EOL> self . kwargs = kwargs or { } <EOL> def _wrap_cache ( self , key , func ) : <EOL> """<STR_LIT>""" <EOL> if self . cache : <EOL> if not self . no_cache_read : <EOL> log . debug ( '<STR_LIT>' , key ) <EOL> content = self . cache . get ( key ) <EOL> if not content in ( False , None ) : <EOL> log . debug ( '<STR_LIT>' , key ) <EOL> return MemoryHunk ( content ) <EOL> content = func ( ) . getvalue ( ) <EOL> if self . cache : <EOL> log . debug ( '<STR_LIT>' , key , ) <EOL> self . cache . set ( key , content ) <EOL> return MemoryHunk ( content ) <EOL> def apply ( self , hunk , filters , type , kwargs = None ) : <EOL> """<STR_LIT>""" <EOL> assert type in self . VALID_TRANSFORMS <EOL> log . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' , type , filters , hunk , kwargs ) <EOL> filters = [ f for f in filters if getattr ( f , type , None ) ] <EOL> if not filters : <EOL> log . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( type , ) ) <EOL> return hunk <EOL> kwargs_final = self . kwargs . copy ( ) <EOL> kwargs_final . update ( kwargs or { } ) <EOL> def func ( ) : <EOL> data = StringIO ( hunk . data ( ) ) <EOL> for filter in filters : <EOL> log . debug ( '<STR_LIT>' , <EOL> type , filter , kwargs_final ) <EOL> out = StringIO ( u'<STR_LIT>' ) <EOL> getattr ( filter , type ) ( data , out , ** kwargs_final ) <EOL> data = out <EOL> data . seek ( <NUM_LIT:0> ) <EOL> return data <EOL> additional_cache_keys = [ ] <EOL> if kwargs_final : <EOL> for filter in filters : <EOL> additional_cache_keys += filter . get_additional_cache_keys ( ** kwargs_final ) <EOL> key = ( "<STR_LIT>" , hunk , tuple ( filters ) , type , additional_cache_keys ) <EOL> return self . _wrap_cache ( key , func ) <EOL> def apply_func ( self , filters , type , args , kwargs = None , cache_key = None ) : <EOL> """<STR_LIT>""" <EOL> assert type in self . VALID_FUNCS <EOL> log . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' , type , filters , args , kwargs ) <EOL> filters = [ f for f in filters if getattr ( f , type , None ) ] <EOL> if not filters : <EOL> log . debug ( '<STR_LIT>' % type ) <EOL> raise NoFilters ( ) <EOL> if len ( filters ) > <NUM_LIT:1> : <EOL> raise MoreThanOneFilterError ( <EOL> '<STR_LIT>' % ( <EOL> '<STR_LIT:U+002CU+0020>' . join ( [ f . name for f in filters ] ) ) , filters ) <EOL> kwargs_final = self . kwargs . copy ( ) <EOL> kwargs_final . update ( kwargs or { } ) <EOL> def func ( ) : <EOL> filter = filters [ <NUM_LIT:0> ] <EOL> out = StringIO ( u'<STR_LIT>' ) <EOL> log . debug ( '<STR_LIT>' , <EOL> type , filter , args , kwargs ) <EOL> getattr ( filter , type ) ( out , * args , ** kwargs_final ) <EOL> return out <EOL> additional_cache_keys = [ ] <EOL> if kwargs_final : <EOL> for filter in filters : <EOL> additional_cache_keys += filter . get_additional_cache_keys ( ** kwargs_final ) <EOL> key = ( "<STR_LIT>" , args , tuple ( filters ) , type , cache_key or [ ] , additional_cache_keys ) <EOL> return self . _wrap_cache ( key , func ) <EOL> def merge_filters ( filters1 , filters2 ) : <EOL> """<STR_LIT>""" <EOL> result = list ( filters1 [ : ] ) <EOL> if filters2 : <EOL> for f in filters2 : <EOL> if not f in result : <EOL> result . append ( f ) <EOL> return result <EOL> def select_filters ( filters , level ) : <EOL> """<STR_LIT>""" <EOL> return [ f for f in filters <EOL> if f . max_debug_level is None or <EOL> cmp_debug_levels ( level , f . max_debug_level ) <= <NUM_LIT:0> ] </s>
