<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import functools <EOL> import itertools <EOL> import sys <EOL> import time <EOL> from PyQt5 . QtCore import ( QAbstractEventDispatcher , QDateTime , QObject , <EOL> QUrl , QVariant ) <EOL> from PyQt5 . QtWidgets import QApplication <EOL> from PyQt5 . QtNetwork import QNetworkAccessManager , QNetworkReply , QNetworkProxy <EOL> from PyQt5 . QtWebKit import QWebSettings <EOL> from PyQt5 . QtWebKitWidgets import QWebFrame <EOL> from twisted . python import log <EOL> import six <EOL> from splash . utils import truncated , to_bytes <EOL> OPERATION_NAMES = { <EOL> QNetworkAccessManager . HeadOperation : '<STR_LIT>' , <EOL> QNetworkAccessManager . GetOperation : '<STR_LIT:GET>' , <EOL> QNetworkAccessManager . PostOperation : '<STR_LIT:POST>' , <EOL> QNetworkAccessManager . PutOperation : '<STR_LIT>' , <EOL> QNetworkAccessManager . DeleteOperation : '<STR_LIT>' , <EOL> } <EOL> OPERATION_QT_CONSTANTS = { v : k for k , v in OPERATION_NAMES . items ( ) } <EOL> REQUEST_ERRORS = { <EOL> QNetworkReply . NoError : '<STR_LIT>' , <EOL> QNetworkReply . ConnectionRefusedError : '<STR_LIT>' , <EOL> QNetworkReply . RemoteHostClosedError : '<STR_LIT>' , <EOL> QNetworkReply . HostNotFoundError : '<STR_LIT>' , <EOL> QNetworkReply . TimeoutError : '<STR_LIT>' , <EOL> QNetworkReply . OperationCanceledError : '<STR_LIT>' , <EOL> QNetworkReply . SslHandshakeFailedError : '<STR_LIT>' , <EOL> QNetworkReply . TemporaryNetworkFailureError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyConnectionRefusedError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyConnectionClosedError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyNotFoundError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyTimeoutError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyAuthenticationRequiredError : '<STR_LIT>' , <EOL> QNetworkReply . ContentAccessDenied : '<STR_LIT>' , <EOL> QNetworkReply . ContentOperationNotPermittedError : '<STR_LIT>' , <EOL> QNetworkReply . ContentNotFoundError : '<STR_LIT>' , <EOL> QNetworkReply . AuthenticationRequiredError : '<STR_LIT>' , <EOL> QNetworkReply . ContentReSendError : '<STR_LIT>' , <EOL> QNetworkReply . ProtocolUnknownError : '<STR_LIT>' , <EOL> QNetworkReply . ProtocolInvalidOperationError : '<STR_LIT>' , <EOL> QNetworkReply . UnknownNetworkError : '<STR_LIT>' , <EOL> QNetworkReply . UnknownProxyError : '<STR_LIT>' , <EOL> QNetworkReply . UnknownContentError : '<STR_LIT>' , <EOL> QNetworkReply . ProtocolFailure : '<STR_LIT>' , <EOL> } <EOL> REQUEST_ERRORS_SHORT = { <EOL> QNetworkReply . NoError : '<STR_LIT:OK>' , <EOL> QNetworkReply . OperationCanceledError : '<STR_LIT>' , <EOL> QNetworkReply . ConnectionRefusedError : '<STR_LIT>' , <EOL> QNetworkReply . RemoteHostClosedError : '<STR_LIT>' , <EOL> QNetworkReply . HostNotFoundError : '<STR_LIT>' , <EOL> QNetworkReply . TimeoutError : '<STR_LIT>' , <EOL> QNetworkReply . SslHandshakeFailedError : '<STR_LIT>' , <EOL> QNetworkReply . TemporaryNetworkFailureError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyConnectionRefusedError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyConnectionClosedError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyNotFoundError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyTimeoutError : '<STR_LIT>' , <EOL> QNetworkReply . ProxyAuthenticationRequiredError : '<STR_LIT>' , <EOL> QNetworkReply . ContentAccessDenied : '<STR_LIT>' , <EOL> QNetworkReply . ContentOperationNotPermittedError : '<STR_LIT>' , <EOL> QNetworkReply . ContentNotFoundError : '<STR_LIT>' , <EOL> QNetworkReply . AuthenticationRequiredError : '<STR_LIT>' , <EOL> QNetworkReply . ContentReSendError : '<STR_LIT>' , <EOL> QNetworkReply . ProtocolUnknownError : '<STR_LIT>' , <EOL> QNetworkReply . ProtocolInvalidOperationError : '<STR_LIT>' , <EOL> QNetworkReply . UnknownNetworkError : '<STR_LIT>' , <EOL> QNetworkReply . UnknownProxyError : '<STR_LIT>' , <EOL> QNetworkReply . UnknownContentError : '<STR_LIT>' , <EOL> QNetworkReply . ProtocolFailure : '<STR_LIT>' , <EOL> } <EOL> PROXY_TYPES = { <EOL> '<STR_LIT>' : QNetworkProxy . HttpProxy , <EOL> '<STR_LIT>' : QNetworkProxy . Socks5Proxy , <EOL> } <EOL> _qtapp = None <EOL> def init_qt_app ( verbose ) : <EOL> """<STR_LIT>""" <EOL> global _qtapp <EOL> if _qtapp : <EOL> log . msg ( "<STR_LIT>" ) <EOL> return _qtapp <EOL> class QApp ( QApplication ) : <EOL> blockedAt = <NUM_LIT:0> <EOL> def __init__ ( self , * args ) : <EOL> super ( QApp , self ) . __init__ ( * args ) <EOL> if verbose : <EOL> disp = QAbstractEventDispatcher . instance ( ) <EOL> disp . aboutToBlock . connect ( self . aboutToBlock ) <EOL> disp . awake . connect ( self . awake ) <EOL> def aboutToBlock ( self ) : <EOL> self . blockedAt = time . time ( ) <EOL> log . msg ( "<STR_LIT>" , system = "<STR_LIT>" ) <EOL> def awake ( self ) : <EOL> diff = time . time ( ) - self . blockedAt <EOL> log . msg ( "<STR_LIT>" % diff , system = "<STR_LIT>" ) <EOL> _qtapp = QApp ( sys . argv ) <EOL> return _qtapp <EOL> def get_qt_app ( ) : <EOL> """<STR_LIT>""" <EOL> assert _qtapp is not None , "<STR_LIT>" <EOL> return _qtapp <EOL> def qurl2ascii ( url ) : <EOL> """<STR_LIT>""" <EOL> url = six . text_type ( url . toString ( ) ) . encode ( '<STR_LIT>' ) . decode ( '<STR_LIT:ascii>' ) <EOL> if url . lower ( ) . startswith ( '<STR_LIT>' ) : <EOL> return truncated ( url , <NUM_LIT> , '<STR_LIT>' ) <EOL> return url <EOL> def to_qurl ( s ) : <EOL> if isinstance ( s , QUrl ) : <EOL> return s <EOL> return QUrl . fromEncoded ( to_bytes ( s , encoding = '<STR_LIT:utf8>' ) ) <EOL> def set_request_url ( request , url ) : <EOL> """<STR_LIT>""" <EOL> request . setUrl ( to_qurl ( url ) ) <EOL> def drop_request ( request ) : <EOL> """<STR_LIT>""" <EOL> set_request_url ( request , "<STR_LIT>" ) <EOL> def validate_proxy_type ( typename ) : <EOL> if typename . upper ( ) not in PROXY_TYPES : <EOL> alllowed = "<STR_LIT:U+002CU+0020>" . join ( PROXY_TYPES . keys ( ) ) <EOL> raise ValueError ( <EOL> "<STR_LIT>" % ( typename , alllowed ) <EOL> ) <EOL> def create_proxy ( host , port , username = None , password = None , type = None ) : <EOL> """<STR_LIT>""" <EOL> if type is None : <EOL> type = '<STR_LIT>' <EOL> validate_proxy_type ( type ) <EOL> proxy_type = PROXY_TYPES [ type . upper ( ) ] <EOL> port = int ( port ) <EOL> if username is not None and password is not None : <EOL> proxy = QNetworkProxy ( proxy_type , host , port , username , password ) <EOL> else : <EOL> proxy = QNetworkProxy ( proxy_type , host , port ) <EOL> return proxy <EOL> def request_repr ( request , operation = None ) : <EOL> """<STR_LIT>""" <EOL> method = OPERATION_NAMES . get ( operation , '<STR_LIT:?>' ) <EOL> url = qurl2ascii ( request . url ( ) ) <EOL> return "<STR_LIT>" % ( method , url ) <EOL> def qt2py ( obj , max_depth = <NUM_LIT:100> ) : <EOL> """<STR_LIT>""" <EOL> if max_depth <= <NUM_LIT:0> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if isinstance ( obj , QVariant ) : <EOL> obj = obj . toPyObject ( ) <EOL> if isinstance ( obj , QDateTime ) : <EOL> return obj . toPyDateTime ( ) <EOL> if isinstance ( obj , dict ) : <EOL> return { <EOL> qt2py ( key , max_depth - <NUM_LIT:1> ) : qt2py ( value , max_depth - <NUM_LIT:1> ) <EOL> for key , value in obj . items ( ) <EOL> } <EOL> if isinstance ( obj , list ) : <EOL> return [ qt2py ( v , max_depth - <NUM_LIT:1> ) for v in obj ] <EOL> if isinstance ( obj , tuple ) : <EOL> return tuple ( [ qt2py ( v , max_depth - <NUM_LIT:1> ) for v in obj ] ) <EOL> if isinstance ( obj , set ) : <EOL> return { qt2py ( v , max_depth - <NUM_LIT:1> ) for v in obj } <EOL> assert not isinstance ( obj , QObject ) , ( obj , obj . __class__ ) <EOL> return obj <EOL> class WrappedSignal ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , signal ) : <EOL> self . ids = itertools . count ( ) <EOL> self . callbacks = { } <EOL> self . signal = signal <EOL> def connect ( self , func , ** kwargs ) : <EOL> callback_id = next ( self . ids ) <EOL> cb = functools . partial ( func , callback_id = callback_id , ** kwargs ) <EOL> self . callbacks [ callback_id ] = cb <EOL> self . signal . connect ( cb ) <EOL> return callback_id <EOL> def disconnect ( self , callback_id ) : <EOL> cb = self . callbacks . pop ( callback_id ) <EOL> self . signal . disconnect ( cb ) <EOL> def clear_caches ( ) : <EOL> QWebSettings . clearMemoryCaches ( ) <EOL> def get_request_webframe ( request ) : <EOL> """<STR_LIT>""" <EOL> web_frame = request . originatingObject ( ) <EOL> if isinstance ( web_frame , QWebFrame ) : <EOL> return web_frame <EOL> return None <EOL> def get_versions ( ) : <EOL> """<STR_LIT>""" <EOL> from sip import SIP_VERSION_STR <EOL> from PyQt5 . QtCore import PYQT_VERSION_STR , QT_VERSION_STR <EOL> from PyQt5 . QtWebKit import qWebKitVersion <EOL> return { <EOL> '<STR_LIT>' : QT_VERSION_STR , <EOL> '<STR_LIT>' : PYQT_VERSION_STR , <EOL> '<STR_LIT>' : qWebKitVersion ( ) , <EOL> '<STR_LIT>' : SIP_VERSION_STR <EOL> } <EOL> def qt_551_plus ( ) : <EOL> """<STR_LIT>""" <EOL> from distutils . version import LooseVersion <EOL> from PyQt5 . QtCore import QT_VERSION_STR <EOL> return LooseVersion ( QT_VERSION_STR ) >= LooseVersion ( "<STR_LIT>" ) <EOL> def get_headers_dict ( request_or_reply ) : <EOL> """<STR_LIT>""" <EOL> return { bytes ( k ) : bytes ( v ) for k , v in qt_header_items ( request_or_reply ) } <EOL> def qt_header_items ( request_or_reply ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( request_or_reply , '<STR_LIT>' ) : <EOL> return request_or_reply . rawHeaderPairs ( ) <EOL> return [ <EOL> ( name , request_or_reply . rawHeader ( name ) ) <EOL> for name in request_or_reply . rawHeaderList ( ) <EOL> ] </s>
