<s> """<STR_LIT>""" <EOL> import threading <EOL> import re <EOL> from xml . sax . saxutils import escape , quoteattr <EOL> from . constants import STANZA_NAMESPACES , STREAM_NS , XML_NS <EOL> __docformat__ = "<STR_LIT>" <EOL> STANDARD_PREFIXES = { <EOL> STREAM_NS : '<STR_LIT>' , <EOL> XML_NS : '<STR_LIT>' , <EOL> } <EOL> EVIL_CHARACTERS_RE = re . compile ( r"<STR_LIT>" , re . UNICODE ) <EOL> def remove_evil_characters ( data ) : <EOL> """<STR_LIT>""" <EOL> return EVIL_CHARACTERS_RE . sub ( "<STR_LIT>" , data ) <EOL> class XMPPSerializer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , stanza_namespace , extra_prefixes = None ) : <EOL> """<STR_LIT>""" <EOL> self . stanza_namespace = stanza_namespace <EOL> self . _prefixes = { } <EOL> if extra_prefixes : <EOL> self . _prefixes . update ( extra_prefixes ) <EOL> self . _root_prefixes = None <EOL> self . _head_emitted = False <EOL> self . _next_id = <NUM_LIT:1> <EOL> def add_prefix ( self , namespace , prefix ) : <EOL> """<STR_LIT>""" <EOL> if prefix == "<STR_LIT>" and namespace != XML_NS : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . _prefixes [ namespace ] = prefix <EOL> def emit_head ( self , stream_from , stream_to , stream_id = None , <EOL> version = '<STR_LIT:1.0>' , language = None ) : <EOL> """<STR_LIT>""" <EOL> self . _root_prefixes = dict ( STANDARD_PREFIXES ) <EOL> self . _root_prefixes [ self . stanza_namespace ] = None <EOL> for namespace , prefix in list ( self . _root_prefixes . items ( ) ) : <EOL> if not prefix or prefix == "<STR_LIT>" : <EOL> continue <EOL> if namespace in STANDARD_PREFIXES or namespace in STANZA_NAMESPACES : <EOL> continue <EOL> self . _root_prefixes [ namespace ] = prefix <EOL> tag = "<STR_LIT>" . format ( STANDARD_PREFIXES [ STREAM_NS ] , <EOL> quoteattr ( version ) ) <EOL> if stream_from : <EOL> tag += "<STR_LIT>" . format ( quoteattr ( stream_from ) ) <EOL> if stream_to : <EOL> tag += "<STR_LIT>" . format ( quoteattr ( stream_to ) ) <EOL> if stream_id is not None : <EOL> tag += "<STR_LIT>" . format ( quoteattr ( stream_id ) ) <EOL> if language is not None : <EOL> tag += "<STR_LIT>" . format ( quoteattr ( language ) ) <EOL> for namespace , prefix in list ( self . _root_prefixes . items ( ) ) : <EOL> if prefix == "<STR_LIT>" : <EOL> continue <EOL> if prefix : <EOL> tag += '<STR_LIT>' . format ( prefix , quoteattr ( namespace ) ) <EOL> else : <EOL> tag += '<STR_LIT>' . format ( prefix , quoteattr ( namespace ) ) <EOL> tag += "<STR_LIT:>>" <EOL> self . _head_emitted = True <EOL> return tag <EOL> def emit_tail ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" . format ( self . _root_prefixes [ STREAM_NS ] ) <EOL> def _split_qname ( self , name , is_element ) : <EOL> """<STR_LIT>""" <EOL> if name . startswith ( "<STR_LIT:{>" ) : <EOL> namespace , name = name [ <NUM_LIT:1> : ] . split ( "<STR_LIT:}>" , <NUM_LIT:1> ) <EOL> if namespace in STANZA_NAMESPACES : <EOL> namespace = self . stanza_namespace <EOL> elif is_element : <EOL> raise ValueError ( "<STR_LIT>" . format ( name ) ) <EOL> else : <EOL> namespace = None <EOL> return namespace , name <EOL> def _make_prefix ( self , declared_prefixes ) : <EOL> """<STR_LIT>""" <EOL> used_prefixes = set ( self . _prefixes . values ( ) ) <EOL> used_prefixes |= set ( declared_prefixes . values ( ) ) <EOL> while True : <EOL> prefix = "<STR_LIT>" . format ( self . _next_id ) <EOL> self . _next_id += <NUM_LIT:1> <EOL> if prefix not in used_prefixes : <EOL> break <EOL> return prefix <EOL> def _make_prefixed ( self , name , is_element , declared_prefixes , declarations ) : <EOL> """<STR_LIT>""" <EOL> namespace , name = self . _split_qname ( name , is_element ) <EOL> if namespace is None : <EOL> prefix = None <EOL> elif namespace in declared_prefixes : <EOL> prefix = declared_prefixes [ namespace ] <EOL> elif namespace in self . _prefixes : <EOL> prefix = self . _prefixes [ namespace ] <EOL> declarations [ namespace ] = prefix <EOL> declared_prefixes [ namespace ] = prefix <EOL> else : <EOL> if is_element : <EOL> prefix = None <EOL> else : <EOL> prefix = self . _make_prefix ( declared_prefixes ) <EOL> declarations [ namespace ] = prefix <EOL> declared_prefixes [ namespace ] = prefix <EOL> if prefix : <EOL> return prefix + "<STR_LIT::>" + name <EOL> else : <EOL> return name <EOL> @ staticmethod <EOL> def _make_ns_declarations ( declarations , declared_prefixes ) : <EOL> """<STR_LIT>""" <EOL> result = [ ] <EOL> for namespace , prefix in list ( declarations . items ( ) ) : <EOL> if prefix : <EOL> result . append ( '<STR_LIT>' . format ( prefix , quoteattr ( <EOL> namespace ) ) ) <EOL> else : <EOL> result . append ( '<STR_LIT>' . format ( prefix , quoteattr ( <EOL> namespace ) ) ) <EOL> for d_namespace , d_prefix in list ( declared_prefixes . items ( ) ) : <EOL> if ( not prefix and not d_prefix ) or d_prefix == prefix : <EOL> if namespace != d_namespace : <EOL> del declared_prefixes [ d_namespace ] <EOL> return "<STR_LIT:U+0020>" . join ( result ) <EOL> def _emit_element ( self , element , level , declared_prefixes ) : <EOL> """<STR_LIT>""" <EOL> declarations = { } <EOL> declared_prefixes = dict ( declared_prefixes ) <EOL> name = element . tag <EOL> prefixed = self . _make_prefixed ( name , True , declared_prefixes , <EOL> declarations ) <EOL> start_tag = "<STR_LIT>" . format ( prefixed ) <EOL> end_tag = "<STR_LIT>" . format ( prefixed ) <EOL> for name , value in list ( element . items ( ) ) : <EOL> prefixed = self . _make_prefixed ( name , False , declared_prefixes , <EOL> declarations ) <EOL> start_tag += '<STR_LIT>' . format ( prefixed , quoteattr ( value ) ) <EOL> declarations = self . _make_ns_declarations ( declarations , <EOL> declared_prefixes ) <EOL> if declarations : <EOL> start_tag += "<STR_LIT:U+0020>" + declarations <EOL> children = [ ] <EOL> for child in element : <EOL> children . append ( self . _emit_element ( child , level + <NUM_LIT:1> , <EOL> declared_prefixes ) ) <EOL> if not children and not element . text : <EOL> start_tag += "<STR_LIT>" <EOL> end_tag = "<STR_LIT>" <EOL> text = "<STR_LIT>" <EOL> else : <EOL> start_tag += "<STR_LIT:>>" <EOL> if level > <NUM_LIT:0> and element . text : <EOL> text = escape ( element . text ) <EOL> else : <EOL> text = "<STR_LIT>" <EOL> if level > <NUM_LIT:1> and element . tail : <EOL> tail = escape ( element . tail ) <EOL> else : <EOL> tail = "<STR_LIT>" <EOL> return start_tag + text + '<STR_LIT>' . join ( children ) + end_tag + tail <EOL> def emit_stanza ( self , element ) : <EOL> """<STR_LIT>""" <EOL> if not self . _head_emitted : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> string = self . _emit_element ( element , level = <NUM_LIT:1> , <EOL> declared_prefixes = self . _root_prefixes ) <EOL> return remove_evil_characters ( string ) <EOL> _THREAD = threading . local ( ) <EOL> _THREAD . serializer = None <EOL> def serialize ( element ) : <EOL> """<STR_LIT>""" <EOL> if _THREAD . serializer is None : <EOL> _THREAD . serializer = XMPPSerializer ( "<STR_LIT>" ) <EOL> _THREAD . serializer . emit_head ( None , None ) <EOL> return _THREAD . serializer . emit_stanza ( element ) </s>
