Source code for site_detect

"""Classes for detecting a MediaWiki site."""
# (C) Pywikibot team, 2010-2024
# Distributed under the terms of the MIT license.
from __future__ import annotations

import json
import re
from contextlib import suppress
from html.parser import HTMLParser
from http import HTTPStatus
from urllib.parse import urljoin, urlparse

from requests.exceptions import RequestException

import pywikibot
from pywikibot.backports import removesuffix
from pywikibot.comms.http import fetch
from pywikibot.exceptions import ClientError, ServerError
from import MediaWikiVersion

    from requests import JSONDecodeError
except ImportError:  # requests < 2.27.0
    from json import JSONDecodeError

    '<h1>Sorry! This site is experiencing technical difficulties.</h1>'

MIN_VERSION = MediaWikiVersion('1.27')

[docs] class MWSite: """Minimal wiki site class.""" def __init__(self, fromurl, **kwargs) -> None: """ Initializer. :raises pywikibot.exceptions.ServerError: a server error occurred while loading the site :raises Timeout: a timeout occurred while loading the site :raises RuntimeError: Version not found or version less than 1.27 """ fromurl = removesuffix(fromurl, '$1') r = fetch(fromurl, **kwargs) check_response(r) if fromurl != r.url: pywikibot.log(f'{fromurl} redirected to {r.url}') fromurl = r.url self.fromurl = fromurl data = r.text wp = WikiHTMLPageParser(fromurl) wp.feed(data) self.version = wp.version self.server = wp.server self.scriptpath = wp.scriptpath self.articlepath = None if self.api: try: self._parse_site() except (ServerError, RequestException): raise except Exception as e: pywikibot.log(f'MW detection failed: {e!r}') if not self.version: self._fetch_old_version() if not self.api: raise RuntimeError(f'Unsupported url: {self.fromurl}') if not self.version or self.version < MIN_VERSION: raise RuntimeError(f'Unsupported version: {self.version}') if not self.articlepath: if self.private_wiki: if self.api != self.fromurl and self.private_wiki: self.articlepath = self.fromurl.rsplit('/', 1)[0] + '/$1' else: raise RuntimeError( 'Unable to determine articlepath because the wiki is ' 'private. Use the Main Page URL instead of the API.') else: raise RuntimeError( f'Unable to determine articlepath: {self.fromurl}') def __repr__(self) -> str: return f'{type(self).__name__}("{self.fromurl}")' @property def langs(self): """Build interwikimap.""" response = fetch( self.api + '?action=query&meta=siteinfo&siprop=interwikimap' '&sifilteriw=local&format=json') iw = response.json() error = iw.get('error') if error: raise RuntimeError(f"{error['code']} - {error['info']}") return [wiki for wiki in iw['query']['interwikimap'] if 'language' in wiki] def _fetch_old_version(self) -> None: """Extract the version from API help with ?version enabled.""" if self.version is None: try: r = fetch(self.api + '?version&format=json') try: d = r.json() except JSONDecodeError: # Fallback for old versions which didn't wrap help in json d = {'error': {'*': r.text}} self.version = list(filter( lambda x: x.startswith('MediaWiki'), (line.strip() for line in d['error']['*'].split('\n'))))[0].split()[1] except Exception: pass else: self.version = MediaWikiVersion(self.version) def _parse_site(self) -> None: """Parse siteinfo data.""" response = fetch(self.api + '?action=query&meta=siteinfo&format=json') check_response(response) # remove preleading newlines and Byte Order Mark (BOM), see T128992 content = response.text.strip().lstrip('\uFEFF') info = json.loads(content) self.private_wiki = ('error' in info and info['error']['code'] == 'readapidenied') if self.private_wiki: # user config is not loaded because PYWIKIBOT_NO_USER_CONFIG # is set to '2' by # Prepare a temporary config for login. username = pywikibot.input( 'Private wiki detected. Login is required.\n' 'Please enter your username?') # Setup a dummy family so that we can create a site object fam ='temporary_family', self.server + self.scriptpath) site = pywikibot.Site(fam.code, fam, username) site.version = lambda: str(self.version) # Now the site object is able to login info = site.siteinfo else: info = info['query']['general'] self.version = MediaWikiVersion.from_generator(info['generator']) if self.version < MIN_VERSION: return self.server = urljoin(self.fromurl, info['server']) for item in ['scriptpath', 'articlepath', 'lang']: setattr(self, item, info[item]) def __eq__(self, other): """Return True if equal to other.""" return (self.server + self.scriptpath == other.server + other.scriptpath) def __hash__(self): """Get hashable representation.""" return hash(self.server + self.scriptpath) @property def api(self) -> str | None: """Get api URL.""" if self.server is None or self.scriptpath is None: return None return self.server + self.scriptpath + '/api.php' @property def iwpath(self): """Get article path URL.""" return self.server + self.articlepath
[docs] class WikiHTMLPageParser(HTMLParser): """Wiki HTML page parser.""" def __init__(self, url) -> None: """Initializer.""" super().__init__(convert_charrefs=True) self.url = urlparse(url) self.generator = None self.version = None self._parsed_url = None self.server = None self.scriptpath = None
[docs] def set_version(self, value) -> None: """Set highest version.""" if self.version and value < self.version: return self.version = value
[docs] def set_api_url(self, url) -> None: """Set api_url.""" url = url.split('.php', 1)[0] try: value, script_name = url.rsplit('/', 1) except ValueError: return if script_name not in ('api', 'load', 'opensearch_desc'): return if script_name == 'load': self.set_version(MediaWikiVersion('1.17.0')) if self._parsed_url: # A Resource Loader link is less reliable than other links. # Resource Loader can load resources from a different site. # e.g.$1 # loads resources from return new_parsed_url = urlparse(value) if self._parsed_url: assert new_parsed_url.path == self._parsed_url.path if not new_parsed_url.scheme or not new_parsed_url.netloc: new_parsed_url = urlparse( '{}://{}{}'.format( new_parsed_url.scheme or self.url.scheme, new_parsed_url.netloc or self.url.netloc, new_parsed_url.path)) else: if self._parsed_url: # allow upgrades to https, but not downgrades if self._parsed_url.scheme == 'https' \ and new_parsed_url.scheme != self._parsed_url.scheme: return # allow vs if (new_parsed_url.netloc in self._parsed_url.netloc or self._parsed_url.netloc in new_parsed_url.netloc): return assert new_parsed_url == self._parsed_url, '{} != {}'.format( self._parsed_url, new_parsed_url) self._parsed_url = new_parsed_url self.server = '{url.scheme}://{url.netloc}'.format( url=self._parsed_url) self.scriptpath = self._parsed_url.path
[docs] def handle_starttag(self, tag, attrs) -> None: """Handle an opening tag.""" attrs = dict(attrs) if tag == 'meta': if attrs.get('name') == 'generator': self.generator = attrs['content'] with suppress(ValueError): self.version = MediaWikiVersion.from_generator( self.generator) elif tag == 'link' and 'rel' in attrs and 'href' in attrs: if attrs['rel'] in ('EditURI', 'stylesheet', 'search'): self.set_api_url(attrs['href']) elif tag == 'script' and 'src' in attrs: self.set_api_url(attrs['src'])
[docs] def check_response(response): """Raise ClientError or ServerError depending on http status. .. versionadded:: 3.0 .. versionchanged:: 7.0 Raise a generic :class:`exceptions.ServerError` if http status code is not IANA-registered but unofficial code .. versionchanged:: 8.1 Raise a :class:`exceptions.ClientError` if status code is 4XX """ for status_code, err_class, err_type in [ (HTTPStatus.INTERNAL_SERVER_ERROR, ServerError, 'Server'), (HTTPStatus.BAD_REQUEST, ClientError, 'Client') ]: # highest http status code first if response.status_code >= status_code: try: status = HTTPStatus(response.status_code) except ValueError as err: m ='\d{3}', err.args[0], flags=re.ASCII) if not m: raise msg = f'Generic {err_type} Error ({})' else: msg = f'({status}) {status.description}' raise err_class(msg) if response.status_code == HTTPStatus.OK \ and SERVER_DB_ERROR_MSG in response.text: raise ServerError('Server cannot access the database')