"""A MediaWiki site identified by its hostname.

""" # noqa: E501

api_limit = 500

def __init__(

self,

host: str,

path: str = '/w/',

ext: str = '.php',

pool: Optional[requests.Session] = None,

retry_timeout: int = 30,

max_retries: int = 25,

wait_callback: Callable[['Sleeper', int, Optional[Any]], Any] = lambda *x: None,

clients_useragent: Optional[str] = None,

max_lag: int = 3,

compress: bool = True,

force_login: bool = True,

do_init: bool = True,

httpauth: Union[

Tuple[Union[str, bytes], Union[str, bytes]],

requests.auth.AuthBase,

List[Union[str, bytes]],

None,

] = None,

connection_options: Optional[MutableMapping[str, Any]] = None,

consumer_token: Optional[str] = None,

consumer_secret: Optional[str] = None,

access_token: Optional[str] = None,

access_secret: Optional[str] = None,

client_certificate: Optional[Union[str, Tuple[str, str]]] = None,

custom_headers: Optional[Mapping[str, str]] = None,

scheme: str = 'https',

reqs: Optional[MutableMapping[str, Any]] = None

) -> None:

# Setup member variables

self.host = host

self.path = path

self.ext = ext

self.credentials = None # type: Optional[Tuple[str, str, Optional[str]]]

self.compress = compress

self.max_lag = str(max_lag)

self.force_login = force_login

self.logged_in = False

if reqs and connection_options:

raise ValueError(

"reqs is a deprecated alias of connection_options. Do not specify both."

)

if reqs:

warnings.warn(

"reqs is deprecated in mwclient 1.0.0. Use connection_options instead",

DeprecationWarning

)

connection_options = reqs

self.requests = connection_options or {}

self.scheme = scheme

if 'timeout' not in self.requests:

self.requests['timeout'] = 30 # seconds

if consumer_token is not None:

auth = OAuth1(consumer_token, consumer_secret, access_token, access_secret)

elif isinstance(httpauth, (list, tuple)):

# workaround weird requests default to encode as latin-1

# https://github.com/mwclient/mwclient/issues/315

# https://github.com/psf/requests/issues/4564

httpauth = [

it.encode("utf-8") if isinstance(it, str) else it for it in httpauth

]

auth = HTTPBasicAuth(*httpauth)

elif httpauth is None or isinstance(httpauth, (AuthBase,)):

auth = httpauth

else:

# FIXME: Raise a specific exception instead of a generic RuntimeError.

raise RuntimeError('Authentication is not a tuple or an instance of AuthBase')

self.sleepers = Sleepers(max_retries, retry_timeout, wait_callback)

# Site properties

self.blocked = False # type: Union[Tuple[str, str], bool] # Is user blocked?

self.hasmsg = False # Whether current user has new messages

self.groups = [] # type: List[str] # Groups current user is in

self.rights = [] # type: List[str] # Rights current user has

self.tokens = {} # type: Dict[str, str] # Edit tokens of the current user

self.version = None # type: Optional[VersionTuple]

self.namespaces = self.default_namespaces # type: Dict[int, str]

# Setup connection

if pool is None:

self.connection = requests.Session() # type: requests.Session

self.connection.auth = auth

if client_certificate:

self.connection.cert = client_certificate

# Set User-Agent header field

if clients_useragent:

ua = clients_useragent + ' ' + USER_AGENT

else:

ua = USER_AGENT

self.connection.headers['User-Agent'] = ua

if custom_headers:

self.connection.headers.update(custom_headers)

else:

self.connection = pool

# Page generators

self.pages = listing.PageList(self)

self.categories = listing.PageList(self, namespace=14)

self.images = listing.PageList(self, namespace=6)

# Compat page generators

self.Pages = self.pages

self.Categories = self.categories

self.Images = self.images

# Initialization status

self.initialized = False

# Upload chunk size in bytes

self.chunk_size = 1048576

if do_init:

try:

self.site_init()

except errors.APIError as e:

if e.args[0] == 'mwoauth-invalid-authorization':

raise errors.OAuthAuthorizationError(self, e.code, e.info)

# Private wiki, do init after login

if e.args[0] not in {'unknown_action', 'readapidenied'}:

raise

def site_init(self) -> None:

"""Populates the object with information about the current user and site. This is

if self.initialized:

info = self.get('query', meta='userinfo', uiprop='groups|rights')

userinfo = info['query']['userinfo']

self.username = userinfo['name']

self.groups = userinfo.get('groups', [])

self.rights = userinfo.get('rights', [])

self.tokens = {}

return

meta = self.get('query', meta='siteinfo|userinfo',

siprop='general|namespaces', uiprop='groups|rights',

retry_on_error=False)

# Extract site info

self.site = meta['query']['general']

self.namespaces = {

namespace['id']: namespace.get('*', '')

for namespace in meta['query']['namespaces'].values()

}

self.version = self.version_tuple_from_generator(self.site['generator'])

# Require MediaWiki version >= 1.16

self.require(1, 16)

# User info

userinfo = meta['query']['userinfo']

self.username = userinfo['name']

self.groups = userinfo.get('groups', [])

self.rights = userinfo.get('rights', [])

self.initialized = True

@staticmethod

def version_tuple_from_generator(

string: str, prefix: str = 'MediaWiki '

) -> VersionTuple:

"""Return a version tuple from a MediaWiki Generator string.

if not string.startswith(prefix):

raise errors.MediaWikiVersionError(f'Unknown generator {string}')

version = string[len(prefix):]

def _split_version(version: str) -> Iterator[str]:

"""Split a version string into segments.

current_segment = ''

for curr_char in version:

if curr_char in "-+_.":

yield current_segment

current_segment = ''

elif current_segment and (

(current_segment[-1].isdigit() and curr_char.isalpha())

or (current_segment[-1].isalpha() and curr_char.isdigit())

):

yield current_segment

current_segment = curr_char

else:

current_segment += curr_char

yield current_segment

version_tuple = tuple(

int(segment) if segment.isdigit() else segment

for segment in _split_version(version)

) # type: Tuple[Union[int, str], ...]

if len(version_tuple) < 2:

raise errors.MediaWikiVersionError(f'Unknown MediaWiki {".".join(version)}')

# Ensure the major and minor version components are integers.

# Non-integer values for these components are not supported and will

# cause comparison issues.

if not all(isinstance(segment, int) for segment in version_tuple[:2]):

raise errors.MediaWikiVersionError(

f'Unknown MediaWiki {".".join(version)}. '

'Major and minor version must be integers.'

)

return version_tuple

default_namespaces = {

0: '', 1: 'Talk', 2: 'User', 3: 'User talk', 4: 'Project',

5: 'Project talk', 6: 'Image', 7: 'Image talk', 8: 'MediaWiki',

9: 'MediaWiki talk', 10: 'Template', 11: 'Template talk', 12: 'Help',

13: 'Help talk', 14: 'Category', 15: 'Category talk',

-1: 'Special', -2: 'Media'

}

def __repr__(self) -> str:

return f"<{self.__class__.__name__} object '{self.host}{self.path}'>"

def get(self, action: str, *args: Tuple[str, Any], **kwargs: Any) -> Dict[str, Any]:

"""Perform a generic API call using GET.

return self.api(action, 'GET', *args, **kwargs)

def post(self, action: str, *args: Tuple[str, Any], **kwargs: Any) -> Dict[str, Any]:

"""Perform a generic API call using POST.

return self.api(action, 'POST', *args, **kwargs)

def api(

self,

action: str,

http_method: str = 'POST',

*args: Tuple[str, Any],

**kwargs: Any

) -> Dict[str, Any]:

"""Perform a generic API call and handle errors.

kwargs.update(args)

# this enables new-style continuation in mediawiki 1.21

# through 1.25, can be dropped when we bump baseline to 1.26

if action == 'query' and 'continue' not in kwargs:

kwargs['continue'] = ''

if action == 'query':

if 'meta' in kwargs:

kwargs['meta'] += '|userinfo'

else:

kwargs['meta'] = 'userinfo'

if 'uiprop' in kwargs:

kwargs['uiprop'] += '|blockinfo|hasmsg'

else:

kwargs['uiprop'] = 'blockinfo|hasmsg'

sleeper = self.sleepers.make()

while True:

info = self.raw_api(action, http_method, **kwargs)

if not info:

info = {}

if self.handle_api_result(info, sleeper=sleeper):

return info

def handle_api_result(

self,

info: Mapping[str, Any],

kwargs: Optional[Mapping[str, Any]] = None,

sleeper: Optional['Sleeper'] = None

) -> bool:

"""Checks the given API response, raising an appropriate exception or sleeping if

if sleeper is None:

sleeper = self.sleepers.make()

try:

userinfo = info['query']['userinfo']

except KeyError:

userinfo = {}

if 'blockedby' in userinfo:

self.blocked = (userinfo['blockedby'], userinfo.get('blockreason', ''))

else:

self.blocked = False

self.hasmsg = 'messages' in userinfo

if userinfo:

self.logged_in = 'anon' not in userinfo

if 'warnings' in info:

for module, warning in info['warnings'].items():

if '*' in warning:

log.warning(warning['*'])

if 'error' in info:

if info['error'].get('code') in {'internal_api_error_DBConnectionError',

'internal_api_error_DBQueryError'}:

sleeper.sleep()

return False

# cope with https://phabricator.wikimedia.org/T106066

if (

info['error'].get('code') == 'mwoauth-invalid-authorization'

and 'Nonce already used' in info['error'].get('info')

):

log.warning('Retrying due to nonce error, see'

'https://phabricator.wikimedia.org/T106066')

sleeper.sleep()

return False

if 'query' in info['error']:

# Semantic Mediawiki does not follow the standard error format

raise errors.APIError(None, info['error']['query'], kwargs)

if '*' in info['error']:

raise errors.APIError(info['error']['code'],

info['error']['info'], info['error']['*'])

raise errors.APIError(info['error']['code'],

info['error']['info'], kwargs)

return True

@staticmethod

def _query_string(*args: Tuple[str, Any], **kwargs: Any) -> Dict[str, Any]:

kwargs.update(args)

qs1 = [

(k, v) for k, v in kwargs.items() if k not in {'wpEditToken', 'token'}

]

qs2 = [

(k, v) for k, v in kwargs.items() if k in {'wpEditToken', 'token'}

]

return OrderedDict(qs1 + qs2)

def raw_call(

self,

script: str,

data: Mapping[str, Any],

files: Optional[Mapping[str, Union[BinaryIO, Tuple[str, BinaryIO]]]] = None,

retry_on_error: bool = True,

http_method: str = 'POST'

) -> str:

"""

headers = {}

if self.compress:

headers['Accept-Encoding'] = 'gzip'

sleeper = self.sleepers.make((script, data))

scheme = self.scheme

host = self.host

if isinstance(host, (list, tuple)): # type: ignore[unreachable]

warnings.warn( # type: ignore[unreachable]

'Specifying host as a tuple is deprecated as of mwclient 0.10.1. '

+ 'Please use the new scheme argument instead.',

DeprecationWarning

)

scheme, host = host

url = f'{scheme}://{host}{self.path}{script}{self.ext}'

while True:

toraise = None # type: Optional[Union[requests.RequestException, str]]

wait_time = 0

args = {'files': files, 'headers': headers} # type: Dict[str, Any]

for k, v in self.requests.items():

args[k] = v

if http_method == 'GET':

args['params'] = data

else:

args['data'] = data

try:

stream = self.connection.request(http_method, url, **args)

if stream.headers.get('x-database-lag'):

wait_time = int(

stream.headers.get('retry-after') # type: ignore[arg-type]

)

log.warning('Database lag exceeds max lag. '

'Waiting for %d seconds', wait_time)

# fall through to the sleep

elif stream.status_code == 200:

return stream.text

elif (

(stream.status_code < 500 or stream.status_code > 599)

and stream.status_code != 429 # 429 Too Many Requests is retryable

):

stream.raise_for_status()

else:

if not retry_on_error:

stream.raise_for_status()

log.warning('Received %d response: %s. Retrying in a moment.',

stream.status_code, stream.text)

toraise = "stream"

# fall through to the sleep

except (

requests.exceptions.ConnectionError,

requests.exceptions.Timeout

) as err:

# In the event of a network problem

# (e.g. DNS failure, refused connection, etc),

# Requests will raise a ConnectionError exception.

if not retry_on_error:

raise

log.warning('Connection error. Retrying in a moment.')

toraise = err

# proceed to the sleep

# all retry paths come here

try:

sleeper.sleep(wait_time)

except errors.MaximumRetriesExceeded:

if toraise == "stream":

stream.raise_for_status()

elif toraise and isinstance(toraise, BaseException):

raise toraise

else:

raise

def raw_api(

self,

action: str,

http_method: str = 'POST',

retry_on_error: bool = True,

*args: Tuple[str, Any],

**kwargs: Any

) -> Dict[str, Any]:

"""Send a call to the API.

kwargs['action'] = action

kwargs['format'] = 'json'

data = self._query_string(*args, **kwargs)

res = self.raw_call('api', data, retry_on_error=retry_on_error,

http_method=http_method)

try:

return cast(Dict[str, Any], json.loads(res, object_pairs_hook=OrderedDict))

except ValueError:

if res.startswith('MediaWiki API is not enabled for this site.'):

raise errors.APIDisabledError

raise errors.InvalidResponse(res)

def raw_index(

self,

action: str,

http_method: str = 'POST',

*args: Tuple[str, Any],

**kwargs: Any

) -> str:

"""Sends a call to index.php rather than the API.

kwargs['action'] = action

kwargs['maxlag'] = self.max_lag

data = self._query_string(*args, **kwargs)

return self.raw_call('index', data, http_method=http_method)

def require(

self,

major: int,

minor: int,

revision: Optional[int] = None,

raise_error: bool = True

) -> Optional[bool]:

"""Check whether the current wiki matches the required version.

if self.version is None:

if raise_error is None:

warnings.warn( # type: ignore[unreachable]

'Passing raise_error=None to require is deprecated and will be '

'removed in a future version. Use raise_error=False instead.',

DeprecationWarning

)

return None

elif raise_error is False:

return None

else:

# FIXME: Replace this with a specific error

raise RuntimeError(f'Site {repr(self)} has not yet been initialized')

if revision is None:

if self.version[:2] >= (major, minor):

return True

elif raise_error:

raise errors.MediaWikiVersionError(

f'Requires version {major}.{minor}, '

f'current version is {self.version[0]}.{self.version[1]}')

else:

return False

else:

raise NotImplementedError

# Actions

def email(

self, user: str, text: str, subject: str, cc: bool = False

) -> Dict[str, Any]:

"""

token = self.get_token('email')

try:

info = self.post('emailuser', target=user, subject=subject,

text=text, ccme=cc, token=token)

except errors.APIError as e:

if e.args[0] == 'noemail':

raise errors.NoSpecifiedEmail(user, e.args[1])

raise errors.EmailError(*e) # type: ignore[misc]

return info

def login(

self,

username: Optional[str] = None,

password: Optional[str] = None,

cookies: Optional[Cookies] = None,

domain: Optional[str] = None

) -> None:

"""

if username and password:

self.credentials = (username, password, domain)

if cookies:

self.connection.cookies.update(cookies)

if self.credentials:

sleeper = self.sleepers.make()

kwargs = {

'lgname': self.credentials[0],

'lgpassword': self.credentials[1]

}

if self.credentials[2]:

kwargs['lgdomain'] = self.credentials[2]

# Try to login using the scheme for MW 1.27+. If the wiki is read protected,

# it is not possible to get the wiki version upfront using the API, so we just

# have to try. If the attempt fails, we try the old method.

try:

kwargs['lgtoken'] = self.get_token('login')

except (errors.APIError, KeyError):

log.debug('Failed to get login token, MediaWiki is older than 1.27.')

while True:

login = self.post('login', **kwargs)

if login['login']['result'] == 'Success':

break

elif login['login']['result'] == 'NeedToken':

kwargs['lgtoken'] = login['login']['token']

elif login['login']['result'] == 'Throttled':

sleeper.sleep(int(login['login'].get('wait', 5)))

else:

raise errors.LoginError(self, login['login']['result'],

login['login']['reason'])

self.site_init()

def clientlogin(self, cookies: Optional[Cookies] = None, **kwargs: Any) -> Any:

"""

self.require(1, 27)

if cookies:

self.connection.cookies.update(cookies)

if not kwargs:

# TODO: Check if we should raise an error here. It's not clear what the

# expected behavior is when no kwargs are passed. To update the

# cookies, the user can update the connection object directly.

return

if 'logintoken' not in kwargs:

kwargs['logintoken'] = self.get_token('login')

if 'logincontinue' not in kwargs and 'loginreturnurl' not in kwargs:

kwargs['loginreturnurl'] = f'{self.scheme}://{self.host}'

response = self.post('clientlogin', **kwargs)

status = response['clientlogin'].get('status')

if status == 'PASS':

self.site_init()

return True

elif status in ('UI', 'REDIRECT'):

return response['clientlogin']

else:

raise errors.LoginError(self, status, response['clientlogin'].get('message'))

def get_token(

self, type: str, force: bool = False, title: Optional[str] = None

) -> str:

"""Request a MediaWiki access token of the given `type`.

if self.version is None or self.require(1, 24, raise_error=False):

# The 'csrf' (cross-site request forgery) token introduced in 1.24 replaces

# the majority of older tokens, like edittoken and movetoken.

if type not in {

'watch',

'patrol',

'rollback',

'userrights',

'login',

'createaccount',

}:

type = 'csrf'

if type not in self.tokens:

self.tokens[type] = '0'

if self.tokens.get(type, '0') == '0' or force:

if self.version is None or self.require(1, 24, raise_error=False):

# We use raw_api() rather than api() because api() is adding "userinfo"

# to the query and this raises a readapideniederror if the wiki is read

# protected, and we're trying to fetch a login token.

info = self.raw_api('query', 'GET', meta='tokens', type=type)

self.handle_api_result(info)

# Note that for read protected wikis, we don't know the version when

# fetching the login token. If it's < 1.27, the request below will

# raise a KeyError that we should catch.

self.tokens[type] = info['query']['tokens'][f'{type}token']