#
# (C) Pywikibot team, 2008-2026
#
# Distributed under the terms of the MIT license.
#
"""Objects representing API interface to MediaWiki site extensions."""
from __future__ import annotations
from collections.abc import Generator, Iterable
from typing import TYPE_CHECKING, Protocol
import pywikibot
from pywikibot.data import api
from pywikibot.echo import Notification
from pywikibot.exceptions import (
APIError,
Error,
InconsistentTitleError,
NoPageError,
SiteDefinitionError,
UnexpectedAPIDataError,
)
from pywikibot.site._decorators import need_extension, need_right
from pywikibot.tools import merge_unique_dicts
if TYPE_CHECKING:
from pywikibot.site import NamespacesDict
[docs]
class BaseSiteProtocol(Protocol):
_proofread_levels: dict[int, str]
tokens: dict[str, str]
def _generator(self, *args, **kwargs) -> api.Request:
...
def _request(self, **kwargs) -> api.Request:
...
def _update_page(self, *args, **kwargs) -> None:
...
[docs]
def encoding(self) -> str:
...
@property
def namespaces(self) -> NamespacesDict:
...
[docs]
def simple_request(self, **kwargs) -> api.Request:
...
[docs]
def querypage(
self, *args, **kwargs
) -> Generator[tuple[pywikibot.Page, int]]:
...
[docs]
class EchoMixin:
"""APISite mixin for Echo extension."""
[docs]
@need_extension('Echo')
def notifications(self, **kwargs):
"""Yield Notification objects from the Echo extension.
.. seealso:: :api:`Notifications` for other keywords.
:keyword str | None format: Notification output format.
Possible values are ``model``, ``special``, or ``None``.
The default is ``special``.
"""
params = {
'action': 'query',
'meta': 'notifications',
'notformat': 'special',
}
for key, value in kwargs.items():
params['not' + key] = value
data = self.simple_request(**params).submit()
notifications = data['query']['notifications']['list']
return (Notification.fromJSON(self, notification)
for notification in notifications)
[docs]
@need_extension('Echo')
def notifications_mark_read(self: BaseSiteProtocol, **kwargs) -> bool:
"""Mark selected notifications as read.
.. seealso:: :api:`echomarkread`
:return: Whether the action was successful
"""
kwargs = merge_unique_dicts(kwargs, action='echomarkread',
token=self.tokens['csrf'])
req = self.simple_request(**kwargs)
data = req.submit()
try:
return data['query']['echomarkread']['result'] == 'success'
except KeyError:
return False
[docs]
class ProofreadPageMixin:
"""APISite mixin for ProofreadPage extension."""
@need_extension('ProofreadPage')
def _cache_proofreadinfo(self: BaseSiteProtocol, expiry=False) -> None:
"""Retrieve proofreadinfo from site and cache response.
Applicable only to sites with ProofreadPage extension installed.
The following info is returned by the query and cached:
- self._proofread_index_ns: Index Namespace
- self._proofread_page_ns: Page Namespace
- self._proofread_levels: a dictionary with::
keys: int in the range [0, 1, ..., 4]
values: category name corresponding to the 'key' quality level
e.g. on en.wikisource:
.. code-block:: python
{0: 'Without text', 1: 'Not proofread', 2: 'Problematic',
3: 'Proofread', 4: 'Validated'}
:param expiry: Either a number of days or a datetime.timedelta object
:type expiry: int (days), :py:obj:`datetime.timedelta`, False (config)
:return: A tuple containing _proofread_index_ns,
self._proofread_page_ns and self._proofread_levels.
:rtype: Namespace, Namespace, dict
"""
if (not hasattr(self, '_proofread_index_ns')
or not hasattr(self, '_proofread_page_ns')
or not hasattr(self, '_proofread_levels')):
pirequest = self._request(
expiry=pywikibot.config.API_config_expiry
if expiry is False else expiry,
parameters={'action': 'query', 'meta': 'proofreadinfo'}
)
pidata = pirequest.submit()
ns_id = pidata['query']['proofreadnamespaces']['index']['id']
self._proofread_index_ns = self.namespaces[ns_id]
ns_id = pidata['query']['proofreadnamespaces']['page']['id']
self._proofread_page_ns = self.namespaces[ns_id]
self._proofread_levels = {}
for ql in pidata['query']['proofreadqualitylevels']:
self._proofread_levels[ql['id']] = ql['category']
@property
def proofread_index_ns(self):
"""Return Index namespace for the ProofreadPage extension."""
if not hasattr(self, '_proofread_index_ns'):
self._cache_proofreadinfo()
return self._proofread_index_ns
@property
def proofread_page_ns(self):
"""Return Page namespace for the ProofreadPage extension."""
if not hasattr(self, '_proofread_page_ns'):
self._cache_proofreadinfo()
return self._proofread_page_ns
@property
def proofread_levels(self):
"""Return Quality Levels for the ProofreadPage extension."""
if not hasattr(self, '_proofread_levels'):
self._cache_proofreadinfo()
return self._proofread_levels
[docs]
@need_extension('ProofreadPage')
def loadpageurls(self: BaseSiteProtocol,
page: pywikibot.page.BasePage) -> None:
"""Load URLs from api and store in page attributes.
Load URLs to images for a given page in the "Page:" namespace.
No effect for pages in other namespaces.
.. version-added:: 8.6
.. seealso:: :api:`imageforpage`
"""
title = page.title(with_section=False)
# responsiveimages: server would try to render the other images as well
# let's not load the server unless needed.
prppifpprop = 'filename|size|fullsize'
query = self._generator(api.PropertyGenerator,
type_arg='imageforpage',
titles=title.encode(self.encoding()),
prppifpprop=prppifpprop)
self._update_page(page, query)
[docs]
class GeoDataMixin:
"""APISite mixin for GeoData extension."""
[docs]
@need_extension('GeoData')
def loadcoordinfo(self: BaseSiteProtocol, page) -> None:
"""Load [[mw:Extension:GeoData]] info."""
title = page.title(with_section=False)
query = self._generator(api.PropertyGenerator,
type_arg='coordinates',
titles=title.encode(self.encoding()),
coprop=['type', 'name', 'dim',
'country', 'region',
'globe'],
coprimary='all')
self._update_page(page, query)
[docs]
class PageImagesMixin:
"""APISite mixin for PageImages extension."""
[docs]
@need_extension('PageImages')
def loadpageimage(self: BaseSiteProtocol, page) -> None:
"""Load [[mw:Extension:PageImages]] info.
:param page: The page for which to obtain the image
:type page: pywikibot.Page
:raises APIError: PageImages extension is not installed
"""
title = page.title(with_section=False)
query = self._generator(api.PropertyGenerator,
type_arg='pageimages',
titles=title.encode(self.encoding()),
piprop=['name'])
self._update_page(page, query)
[docs]
class GlobalUsageMixin:
"""APISite mixin for Global Usage extension."""
[docs]
@need_extension('Global Usage')
def globalusage(self, page, total=None):
"""Iterate global image usage for a given FilePage.
:param page: The page to return global image usage for.
:type page: pywikibot.FilePage
:param total: Iterate no more than this number of pages in
total.
:raises TypeError: Input page is not a FilePage.
:raises pywikibot.exceptions.SiteDefinitionError: Site could not
be defined for a returned entry in API response.
"""
if not isinstance(page, pywikibot.FilePage):
raise TypeError(f'Page {page} must be a FilePage.')
title = page.title(with_section=False)
args = {'titles': title,
'gufilterlocal': False,
}
query = self._generator(api.PropertyGenerator,
type_arg='globalusage',
guprop=['url', 'pageid', 'namespace'],
total=total, # will set gulimit=total in api,
**args)
for pageitem in query:
if not self.sametitle(pageitem['title'],
page.title(with_section=False)):
raise InconsistentTitleError(page, pageitem['title'])
api.update_page(page, pageitem, query.props)
assert 'globalusage' in pageitem, \
"API globalusage response lacks 'globalusage' key"
for entry in pageitem['globalusage']:
try:
gu_site = pywikibot.Site(url=entry['url'])
except SiteDefinitionError:
pywikibot.warning('Site could not be defined for global '
f'usage for {page}: {entry}.')
continue
gu_page = pywikibot.Page(gu_site, entry['title'])
yield gu_page
[docs]
class WikibaseClientMixin:
"""APISite mixin for WikibaseClient extension."""
[docs]
@need_extension('WikibaseClient')
def unconnected_pages(
self: BaseSiteProtocol,
total: int | None = None,
*,
strict: bool = False
) -> Generator[pywikibot.Page]:
"""Yield Page objects from Special:UnconnectedPages.
.. warning:: The retrieved pages may be connected in meantime.
To avoid this, use *strict* parameter to check.
.. version-changed:: 10.4.0
The *strict* parameter was added.
:param total: Maximum number of pages to return, or ``None`` for
all.
:param strict: If ``True``, verify that each page still has no
data item before yielding it.
"""
if total is not None and total <= 0:
return
if not strict:
yield from self.querypage('UnconnectedPages', total)
return
count = 0
for page in self.querypage('UnconnectedPages'):
if total is not None and count >= total:
break
try:
page.data_item()
except NoPageError:
yield page
count += 1
[docs]
class LinterMixin:
"""APISite mixin for Linter extension."""
[docs]
@need_extension('Linter')
def linter_pages(
self: BaseSiteProtocol,
lint_categories: Iterable[str] | str | None = None,
total: int | None = None,
namespaces=None,
pageids: Iterable[str | int] | str | int | None = None,
lint_from: str | int | None = None
) -> Iterable[pywikibot.Page]:
"""Return a generator to pages containing linter errors.
.. seealso:: :ext:`Linter`
:param lint_categories: Categories of lint errors. Must be an
iterable of lint categories, or a pipe-separated string of
lint categories.
:param total: If not None, yield this many items in total.
:param namespaces: Only iterate pages in these namespaces.
:type namespaces: Iterable of str or Namespace key, or a single
instance of those types. May be a '|' separated list of
namespace identifiers.
:param pageids: Only include lint errors from the specified
pageids. Must be given as an iterable of page ids, or a
pipe-separated string of page ids
(e.g. '945097|483753|956608').
:param lint_from: Lint ID to start querying from
:return: Pages with Linter errors.
"""
query = self._generator(api.ListGenerator, type_arg='linterrors',
total=total, namespaces=namespaces,
lntfrom=lint_from)
if lint_categories:
if isinstance(lint_categories, str):
lint_categories = lint_categories.replace(' ', '')
query.request['lntcategories'] = lint_categories
if pageids:
if isinstance(pageids, str):
pageids = pageids.replace(' ', '')
query.request['lntpageid'] = pageids
for pageitem in query:
page = pywikibot.Page(self, pageitem['title'])
api.update_page(page, pageitem)
yield page
[docs]
class ThanksMixin:
"""APISite mixin for Thanks extension."""
[docs]
@need_extension('Thanks')
def thank_revision(self, revid: int, source: str | None = None):
"""Corresponding method to the 'action=thank' API action.
:param revid: Revision ID for the revision to be thanked.
:param source: A source for the thanking operation.
:raise APIError: On thanking oneself or other API errors.
:return: The API response.
"""
token = self.tokens['csrf']
req = self.simple_request(action='thank', rev=revid, token=token,
source=source)
data = req.submit()
if data['result']['success'] != 1:
raise APIError('Thanking unsuccessful', '')
return data
[docs]
class UrlShortenerMixin:
"""APISite mixin for UrlShortener extension."""
[docs]
@need_extension('UrlShortener')
def create_short_link(self, url: str) -> str:
"""Return a shortened link.
Note that on Wikimedia wikis only metawiki supports this action,
and this wiki can process links to all WM domains.
:param url: The link to reduce, with protocol prefix.
:return: The reduced link, without protocol prefix.
"""
req = self.simple_request(action='shortenurl', url=url)
data = req.submit()
return data['shortenurl']['shorturl']
[docs]
class FlaggedRevsMixin:
"""APISite mixin for the FlaggesRevs extension.
.. version-added:: 11.7
.. seealso:: :ext:`FlaggedRevs`
"""
[docs]
@need_extension('FlaggedRevs')
def stable_revid(self: BaseSiteProtocol,
page: pywikibot.Page) -> int | None:
"""Return the stable (reviewed) revision id for a page, if any.
:param page: The page to inspect.
:return: The stable revision id or None if not available.
:raises UnknownExtensionError: FlaggedRevs not available
"""
req = self.simple_request(
action='query',
prop='flagged',
titles=page.title(with_section=False),
formatversion=2
)
data = req.submit()
pages = data.get('query', {}).get('pages', [])
if not pages:
return None
return pages[0].get('flagged', {}).get('stable_revid')
[docs]
@need_extension('FlaggedRevs')
@need_right('review')
def review_revision(
self,
revid: int,
*,
comment: str | None = None,
unapprove: bool = False,
flag: int | None = None,
) -> None:
"""Review a revision using the FlaggedRevs ``action=review`` API.
.. note::
Reviewing or unapproving a revision may change the stable
revision of the associated page. Cached
:attr:`BasePage.stable_revision_id
<page.BasePage.stable_revision_id>` values are not
invalidated automatically.
:param revid: Revision ID to review.
:param comment: Optional review comment.
:param unapprove: If True, the revision will be *unapproved*.
:param flag: Set the review flag value.
:raises APIError: On API failure.
:raises UnexpectedAPIDataError: Unexpected API data for review
parameters or review result.
:raises UnknownExtensionError: FlaggedRevs not available.
:raises UserRightsError: User has insufficient rights.
:raises ValueError: Unsupported *flag* parameter.
"""
try:
review_params = self._paraminfo['review']['parameters']
except KeyError as e:
raise UnexpectedAPIDataError(
'Unexpected API data: no param info found') from e
names = {item['name'] for item in review_params}
flag_param = next((p for p in names if p.startswith('flag_')), None)
params = {
'action': 'review',
'token': self.tokens['csrf'],
'revid': revid,
'comment': comment,
}
if flag is not None:
if flag_param is None:
raise ValueError(
"The 'flag' parameter is not supported by this wiki")
params[flag_param] = flag
if unapprove:
params['unapprove'] = '1'
request = self.simple_request(**params, formatversion=2)
data = request.submit()
if data.get('review', {}).get('result') != 'Success':
raise UnexpectedAPIDataError(
f'Unexpected review result:\n{data!r}')