198 lines
7.1 KiB
Python
198 lines
7.1 KiB
Python
# Echte xStream-Handler (Standard-Lib only)
|
|
import re
|
|
from urllib.parse import quote, urlencode
|
|
import gzip, ssl
|
|
from urllib.request import Request, urlopen, HTTPCookieProcessor, build_opener, HTTPSHandler
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
class _FakeConfig:
|
|
_settings = {
|
|
'cacheTime': '1', 'requestTimeout': '15',
|
|
'bypassDNSlock': 'false', 'volatileHtmlCache': 'false',
|
|
}
|
|
def getAddonInfo(self, k):
|
|
return {'profile': '/tmp', 'id': 'xstream', 'name': 'xStream',
|
|
'version': '1.0.0', 'icon': '', 'author': '', 'description': ''}.get(k, '')
|
|
def getSetting(self, k): return self._settings.get(k, '')
|
|
def setSetting(self, k, v): self._settings[k] = str(v)
|
|
def getLocalizedString(self, k): return f'[{k}]'
|
|
|
|
class _FakeCache:
|
|
def get(self, key, cache_time): return None
|
|
def set(self, key, data): pass
|
|
def clear(self): pass
|
|
|
|
class _FakeLogger:
|
|
def info(self, *a, **k): pass
|
|
def debug(self, *a, **k): pass
|
|
def error(self, *a, **k): pass
|
|
|
|
def _infoDialog(*a, **k): pass
|
|
|
|
|
|
class cRequestHandler:
|
|
persistent_openers = {}
|
|
|
|
UA_LIST = [
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:152.0) Gecko/20100101 Firefox/152.0',
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36',
|
|
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/26.5 Safari/605.1.15',
|
|
]
|
|
|
|
def __init__(self, sUrl='', caching=True, ignoreErrors=False, method='GET',
|
|
data=None, compression=True, jspost=False, ssl_verify=False):
|
|
self._sUrl = str(sUrl) if sUrl else ''
|
|
self._sRealUrl = ''
|
|
self._USER_AGENT = self.UA_LIST[0]
|
|
self._aParameters = {}
|
|
self._headerEntries = {}
|
|
self._Status = ''
|
|
self._sResponseHeader = ''
|
|
self._ssl_verify = ssl_verify
|
|
self.caching = caching
|
|
self.method = method
|
|
self.data = data
|
|
self.ignoreErrors = ignoreErrors
|
|
self.compression = compression
|
|
self.jspost = jspost
|
|
self.requestTimeout = 15
|
|
self.__bRemoveNewLines = True
|
|
self.__bRemoveBreakLines = True
|
|
self.isMemoryCacheActive = False
|
|
self.__setDefaultHeader()
|
|
|
|
def __setDefaultHeader(self):
|
|
self.addHeaderEntry('User-Agent', self._USER_AGENT)
|
|
self.addHeaderEntry('Accept', '*/*')
|
|
self.addHeaderEntry('Accept-Language', 'en-US,en;q=0.5')
|
|
if self.compression:
|
|
self.addHeaderEntry('Accept-Encoding', 'gzip, deflate')
|
|
|
|
def removeNewLines(self, b): self.__bRemoveNewLines = b
|
|
def removeBreakLines(self, b): self.__bRemoveBreakLines = b
|
|
|
|
def addHeaderEntry(self, k, v): self._headerEntries[k] = v
|
|
def addParameters(self, k, v, Quote=False):
|
|
self._aParameters[k] = v if not Quote else quote(str(v))
|
|
def getHeaderEntry(self, k): return self._headerEntries.get(k)
|
|
def getResponseHeader(self): return self._sResponseHeader
|
|
def getRealUrl(self): return self._sRealUrl or self._sUrl
|
|
def getStatus(self): return self._Status
|
|
def getUrl(self): return self._sUrl
|
|
def setUrl(self, url):
|
|
self._sUrl = str(url)
|
|
return self
|
|
def ignoreDiscard(self, v): pass
|
|
def ignoreExpired(self, v): pass
|
|
|
|
def getHtml(self, encoding='utf-8'):
|
|
if not self._sUrl or self._sUrl in ('False', 'None', ''):
|
|
return ''
|
|
try:
|
|
url = self._sUrl
|
|
if self._aParameters and self.method == 'GET':
|
|
sep = '&' if '?' in url else '?'
|
|
url = url + sep + urlencode(self._aParameters)
|
|
|
|
req = Request(url, data=self.data, method=self.method)
|
|
for k, v in self._headerEntries.items():
|
|
try: req.add_header(k, v)
|
|
except: pass
|
|
|
|
ctx = None
|
|
if self._ssl_verify is False:
|
|
ctx = ssl._create_unverified_context()
|
|
|
|
opener = build_opener(HTTPCookieProcessor())
|
|
with opener.open(req, timeout=self.requestTimeout) as resp:
|
|
self._sRealUrl = resp.geturl()
|
|
self._sResponseHeader = str(resp.headers)
|
|
data = resp.read()
|
|
|
|
if self.compression and 'gzip' in resp.headers.get('Content-Encoding', ''):
|
|
try: data = gzip.decompress(data)
|
|
except: pass
|
|
|
|
text = data.decode(encoding, errors='replace')
|
|
|
|
if self.__bRemoveBreakLines:
|
|
text = re.sub(r'[\r\n]+', '\n', text)
|
|
if self.__bRemoveNewLines:
|
|
text = re.sub(r'\n+', ' ', text)
|
|
|
|
return text
|
|
|
|
except HTTPError as e:
|
|
self._Status = str(e.code)
|
|
return f'<!-- HTTPError {e.code} -->' if not self.ignoreErrors else ''
|
|
except URLError as e:
|
|
self._Status = str(e.reason)
|
|
return f'<!-- URLError {e.reason} -->' if not self.ignoreErrors else ''
|
|
except Exception as e:
|
|
self._Status = str(e)
|
|
return f'<!-- Error {e} -->' if not self.ignoreErrors else ''
|
|
|
|
# Alias for getHtml() - used by original xStream site plugins
|
|
def request(self, encoding='utf-8'):
|
|
return self.getHtml(encoding)
|
|
|
|
def getJson(self):
|
|
import json
|
|
text = self.getHtml()
|
|
if text.strip().startswith('<!--'): return {}
|
|
try: return json.loads(text)
|
|
except: return {}
|
|
|
|
|
|
class cParser:
|
|
@staticmethod
|
|
def urlEncode(params):
|
|
if isinstance(params, dict): return urlencode(params)
|
|
return quote(str(params) if params else '')
|
|
|
|
@staticmethod
|
|
def parse(source, pattern, flags=0):
|
|
return re.findall(pattern, source, flags)
|
|
|
|
@staticmethod
|
|
def replace(pattern, replacement, source, count=0):
|
|
return re.sub(pattern, replacement, source, count)
|
|
|
|
@staticmethod
|
|
def searchSingle(result, searchPattern, outputParam=None, source=None):
|
|
text = source if source is not None else (str(result) if result else '')
|
|
match = re.search(searchPattern, text, re.S)
|
|
if not match: return None
|
|
return match.group(outputParam) if outputParam and outputParam <= match.lastindex else match.group(0)
|
|
|
|
@staticmethod
|
|
def htmlParse(html):
|
|
if not html: return []
|
|
text = re.sub(r'<script[^>]*>.*?</script>', '', html, flags=re.S|re.I)
|
|
text = re.sub(r'<style[^>]*>.*?</style>', '', text, flags=re.S|re.I)
|
|
text = re.sub(r'<[^>]+>', ' ', text)
|
|
text = re.sub(r'\s+', ' ', text).strip()
|
|
return [t.strip() for t in text.split(' ') if t.strip()]
|
|
|
|
|
|
class ParameterHandler:
|
|
def __init__(self, url=''):
|
|
self._params = {}
|
|
if url and '?' in url:
|
|
_, qs = url.split('?', 1)
|
|
for part in qs.split('&'):
|
|
if '=' in part:
|
|
k, v = part.split('=', 1)
|
|
self._params[k] = v
|
|
|
|
def getValue(self, key='', default=''):
|
|
return self._params.get(key, default)
|
|
|
|
def setParam(self, key, value):
|
|
self._params[str(key)] = str(value)
|
|
|
|
def getAll(self):
|
|
return self._params
|