jenkins-bot has submitted this change. ( https://gerrit.wikimedia.org/r/c/pywikibot/core/+/806854 )
Change subject: Fix regression for interwiki script
......................................................................
Fix regression for interwiki script
Commit 0a5f9135 introduces regression where interwiki script removes all
interwiki links instead adding/replacing ones.
Patch submitted by winterheart.
Bug: T310964
Change-Id: I3cccdbb4fb0115ec2270ea001826d9c61389ef9e
---
M scripts/interwiki.py
1 file changed, 1 insertion(+), 1 deletion(-)
Approvals:
Xqt: Looks good to me, approved
jenkins-bot: Verified
diff --git a/scripts/interwiki.py b/scripts/interwiki.py
index e217308..76e75ad 100755
--- a/scripts/interwiki.py
+++ b/scripts/interwiki.py
@@ -1224,7 +1224,7 @@
break
if not self.skipPage(page, linkedPage, counter) \
- and self.conf.followinterwiki or page == self.origin \
+ and (self.conf.followinterwiki or page == self.origin) \
and self.addIfNew(linkedPage, counter, page):
# It is new. Also verify whether it is the second on the
# same site
--
To view, visit https://gerrit.wikimedia.org/r/c/pywikibot/core/+/806854
To unsubscribe, or for help writing mail filters, visit https://gerrit.wikimedia.org/r/settings
Gerrit-Project: pywikibot/core
Gerrit-Branch: master
Gerrit-Change-Id: I3cccdbb4fb0115ec2270ea001826d9c61389ef9e
Gerrit-Change-Number: 806854
Gerrit-PatchSet: 1
Gerrit-Owner: Xqt <info(a)gno.de>
Gerrit-Reviewer: D3r1ck01 <xsavitar.wiki(a)aol.com>
Gerrit-Reviewer: Xqt <info(a)gno.de>
Gerrit-Reviewer: jenkins-bot
Gerrit-MessageType: merged
Xqt has submitted this change. ( https://gerrit.wikimedia.org/r/c/pywikibot/core/+/803232 )
Change subject: [fix] Add a memento_client fix to the framework
......................................................................
[fix] Add a memento_client fix to the framework
- add memento module and import MementoClient and MementoClientException.
Derive MementoClient and add timout parameters for several methods. The
implementation was originally made with this commit:
https://github.com/mementoweb/py-memento-client/commit/15dc9f520230aaa0793d…
- move weblinkchecker._get_closest_memento_url() to memento.py
as get_closest_memento_url() but keep the blame history
- use pywikibot logging system
- set memento_client requirement to ==0.6.1
- update weblinkchecker.py script
- update tests and rename weblinkchecker_tests to memento_tests
- update doc
- update license
Bug: T185561
Change-Id: I137a27ad198f0e0aae713c888401265f7aca187b
---
M docs/api_ref/pywikibot.data.rst
M docs/licenses.rst
M pywikibot/CONTENT.rst
M pywikibot/data/memento.py
M requirements.txt
R scripts/weblinkchecker.py
M setup.py
M tests/__init__.py
R tests/memento_tests.py
M tox.ini
10 files changed, 326 insertions(+), 771 deletions(-)
Approvals:
jenkins-bot: Verified
Xqt: Verified; Looks good to me, approved
diff --git a/docs/api_ref/pywikibot.data.rst b/docs/api_ref/pywikibot.data.rst
index 038c439..12b7d14 100644
--- a/docs/api_ref/pywikibot.data.rst
+++ b/docs/api_ref/pywikibot.data.rst
@@ -11,6 +11,11 @@
.. automodule:: pywikibot.data.api
+pywikibot.data.memento module
+-----------------------------
+
+.. automodule:: pywikibot.data.memento
+
pywikibot.data.mysql module
---------------------------
diff --git a/docs/licenses.rst b/docs/licenses.rst
index 965e28e..f07112a 100644
--- a/docs/licenses.rst
+++ b/docs/licenses.rst
@@ -7,7 +7,10 @@
:ref:`MIT license`; translations by translators and manual pages on
mediawiki.org are available under the `CC-BY-SA 3.0`_ license. The
Pywikibot logo is Public domain but it includes material that may be
-protected as a trademark.
+protected as a trademark. Parts of :mod:`memento<pywikibot.data.memento>`
+module is licenced under the `BSD`_ open source software license. You
+may obtain a copy of the License at
+http://mementoweb.github.io/SiteStory/license.html.
MIT License
@@ -18,4 +21,4 @@
.. _CC-BY-SA 3.0: https://creativecommons.org/licenses/by-sa/3.0/
-
+.. _BSD: https://github.com/mementoweb/py-memento-client/blob/master/LICENSE.txt
diff --git a/pywikibot/CONTENT.rst b/pywikibot/CONTENT.rst
index 5f100ee..617b52d 100644
--- a/pywikibot/CONTENT.rst
+++ b/pywikibot/CONTENT.rst
@@ -103,6 +103,8 @@
| +----------------+-------------------------------------+
| | _requests.py | API Requests interface |
+----------------------------+----------------+-------------------------------------+
+ | memento.py | memento_client 0.6.1 package fix |
+ +----------------------------+------------------------------------------------------+
| mysql.py | Miscellaneous helper functions for mysql queries |
+----------------------------+------------------------------------------------------+
| sparql.py | Objects representing SPARQL query API |
diff --git a/pywikibot/data/memento.py b/pywikibot/data/memento.py
index 398ba64..ecaf35e 100644
--- a/pywikibot/data/memento.py
+++ b/pywikibot/data/memento.py
@@ -1,185 +1,311 @@
-#!/usr/bin/python3
-"""
-This bot is used for checking external links found at the wiki.
+"""Fix ups for memento-client package version 0.6.1.
-It checks several pages at once, with a limit set by the config variable
-max_external_links, which defaults to 50.
-
-The bot won't change any wiki pages, it will only report dead links such that
-people can fix or remove the links themselves.
-
-The bot will store all links found dead in a .dat file in the deadlinks
-subdirectory. To avoid the removing of links which are only temporarily
-unavailable, the bot ONLY reports links which were reported dead at least
-two times, with a time lag of at least one week. Such links will be logged to a
-.txt file in the deadlinks subdirectory.
-
-The .txt file uses wiki markup and so it may be useful to post it on the
-wiki and then exclude that page from subsequent runs. For example if the
-page is named Broken Links, exclude it with '-titleregexnot:^Broken Links$'
-
-After running the bot and waiting for at least one week, you can re-check those
-pages where dead links were found, using the -repeat parameter.
-
-In addition to the logging step, it is possible to automatically report dead
-links to the talk page of the article where the link was found. To use this
-feature, set report_dead_links_on_talk = True in your user-config.py, or
-specify "-talk" on the command line. Adding "-notalk" switches this off
-irrespective of the configuration variable.
-
-When a link is found alive, it will be removed from the .dat file.
-
-These command line parameters can be used to specify which pages to work on:
-
--repeat Work on all pages were dead links were found before. This is
- useful to confirm that the links are dead after some time (at
- least one week), which is required before the script will report
- the problem.
-
--namespace Only process templates in the namespace with the given number or
- name. This parameter may be used multiple times.
-
--xml Should be used instead of a simple page fetching method from
- pagegenerators.py for performance and load issues
-
--xmlstart Page to start with when using an XML dump
-
--ignore HTTP return codes to ignore. Can be provided several times :
- -ignore:401 -ignore:500
-
-¶ms;
-
-Furthermore, the following command line parameters are supported:
-
--talk Overrides the report_dead_links_on_talk config variable, enabling
- the feature.
-
--notalk Overrides the report_dead_links_on_talk config variable, disabling
- the feature.
-
--day Do not report broken link if the link is there only since
- x days or less. If not set, the default is 7 days.
-
-The following config variables are supported:
-
- max_external_links The maximum number of web pages that should be
- loaded simultaneously. You should change this
- according to your Internet connection speed.
- Be careful: if it is set too high, the script
- might get socket errors because your network
- is congested, and will then think that the page
- is offline.
-
- report_dead_links_on_talk If set to true, causes the script to report dead
- links on the article's talk page if (and ONLY if)
- the linked page has been unavailable at least two
- times during a timespan of at least one week.
-
- weblink_dead_days sets the timespan (default: one week) after which
- a dead link will be reported
-
-Examples
---------
-
-Loads all wiki pages in alphabetical order using the Special:Allpages
-feature:
-
- python pwb.py weblinkchecker -start:!
-
-Loads all wiki pages using the Special:Allpages feature, starting at
-"Example page":
-
- python pwb.py weblinkchecker -start:Example_page
-
-Loads all wiki pages that link to www.example.org:
-
- python pwb.py weblinkchecker -weblink:www.example.org
-
-Only checks links found in the wiki page "Example page":
-
- python pwb.py weblinkchecker Example page
-
-Loads all wiki pages where dead links were found during a prior run:
-
- python pwb.py weblinkchecker -repeat
+.. versionadded:: 7.4
+.. seealso:: https://github.com/mementoweb/py-memento-client#readme
"""
#
-# (C) Pywikibot team, 2005-2022
+# (C) Shawn M. Jones, Harihar Shankar, Herbert Van de Sompel.
+# -- Los Alamos National Laboratory, 2013
+# Parts of MementoClient class codes are
+# licensed under the BSD open source software license.
+#
+# (C) Pywikibot team, 2015-2022
#
# Distributed under the terms of the MIT license.
#
-import codecs
-import datetime
-import pickle
-import re
-import threading
-import time
-import urllib.parse as urlparse
-from contextlib import suppress
-from functools import partial
-from http import HTTPStatus
+from datetime import datetime
+from typing import Optional
+
+from memento_client.memento_client import MementoClient as OldMementoClient
+from memento_client.memento_client import MementoClientException
import requests
+from requests.exceptions import InvalidSchema, MissingSchema
-import pywikibot
-from pywikibot import comms, config, i18n, pagegenerators, textlib
-from pywikibot.backports import Dict, removeprefix
-from pywikibot.bot import ExistingPageBot, SingleSiteBot, suggest_help
-from pywikibot.exceptions import (
- IsRedirectPageError,
- NoPageError,
- SpamblacklistError,
-)
-from pywikibot.pagegenerators import (
- XMLDumpPageGenerator as _XMLDumpPageGenerator,
-)
-from pywikibot.tools import ThreadList
+from pywikibot import config, debug, sleep, warning
+
+__all__ = ('MementoClient', 'MementoClientException')
-try:
- import memento_client
- from memento_client.memento_client import MementoClientException
- missing_dependencies = None
-except ImportError:
- missing_dependencies = ['memento_client']
+class MementoClient(OldMementoClient):
+
+ """A Memento Client.
+
+ It makes it straightforward to access the Web of the past as it is
+ to access the current Web.
+
+ .. versionchanged:: 7.4
+ `timeout` is used in several methods.
+
+ Basic usage:
+
+ >>> mc = MementoClient()
+ >>> dt = mc.convert_to_datetime("Sun, 01 Apr 2010 12:00:00 GMT")
+ >>> mc = mc.get_memento_info("http://www.bbc.com/", dt)
+ >>> print(mc['original_uri'])
+ http://www.bbc.com/
+ >>> print(mc['timegate_uri'])
+ http://timetravel.mementoweb.org/timegate/http://www.bbc.com/
+ >>> print(sorted(mc['mementos']))
+ ['closest', 'first', 'last', 'next', 'prev']
+ >>> del mc['mementos']['last']
+ >>> from pprint import pprint
+ >>> pprint(mc['mementos'])
+ {'closest': {'datetime': datetime.datetime(2010, 2, 28, 8, 5, 38),
+ 'http_status_code': 200,
+ 'uri': ['https://swap.stanford.edu/20100228080538/http://www.bbc.co.uk/']},
+ 'first': {'datetime': datetime.datetime(1998, 12, 2, 21, 26, 10),
+ 'uri': ['http://wayback.nli.org.il:8080/19981202212610/http://bbc.com/']},
+ 'next': {'datetime': datetime.datetime(2010, 5, 23, 13, 47, 38),
+ 'uri': ['https://web.archive.org/web/20100523134738/http://www.bbc.com/']},
+ 'prev': {'datetime': datetime.datetime(1998, 12, 2, 21, 26, 10),
+ 'uri': ['http://wayback.nli.org.il:8080/19981202212610/http://bbc.com/']}}
+
+ The output conforms to the Memento API format explained here:
+ http://timetravel.mementoweb.org/guide/api/#memento-json
+
+ By default, MementoClient uses the Memento Aggregator:
+ http://mementoweb.org/depot/
+
+ It is also possible to use different TimeGate, simply initialize
+ with a preferred timegate base uri. Toggle check_native_timegate to
+ see if the original uri has its own timegate. The native timegate,
+ if found will be used instead of the timegate_uri preferred. If no
+ native timegate is found, the preferred timegate_uri will be used.
+
+ :param str timegate_uri: A valid HTTP base uri for a timegate.
+ Must start with http(s):// and end with a /.
+ :param int max_redirects: the maximum number of redirects allowed
+ for all HTTP requests to be made.
+ :return: A :class:`MementoClient` obj.
+ """ # noqa: E501
+
+ def __init__(self, *args, **kwargs):
+ """Initializer."""
+ # To prevent documentation inclusion from inherited class
+ # because it is malformed.
+ super().__init__(*args, **kwargs)
+
+ def get_memento_info(self, request_uri: str,
+ accept_datetime: Optional[datetime] = None,
+ timeout: Optional[int] = None,
+ **kwargs) -> dict:
+ """Query the preferred timegate and return the closest memento uri.
+
+ Given an original uri and an accept datetime, this method
+ queries the preferred timegate and returns the closest memento
+ uri, along with prev/next/first/last if available.
+
+ .. seealso:: http://timetravel.mementoweb.org/guide/api/#memento-json
+ for the response format.
+
+ :param request_uri: The input http uri.
+ :param accept_datetime: The datetime object of the accept
+ datetime. The current datetime is used if none is provided.
+ :param timeout: the timeout value for the HTTP connection.
+ :return: A map of uri and datetime for the
+ closest/prev/next/first/last mementos.
+ """
+ # for reading the headers of the req uri to find uri_r
+ req_uri_response = kwargs.get('req_uri_response')
+ # for checking native tg uri in uri_r
+ org_response = kwargs.get('org_response')
+ tg_response = kwargs.get('tg_response')
+ if not tg_response:
+ native_tg = None
+
+ original_uri = self.get_original_uri(
+ request_uri, response=req_uri_response)
+
+ if self.check_native_timegate:
+ native_tg = self.get_native_timegate_uri(
+ original_uri, accept_datetime=accept_datetime,
+ response=org_response)
+
+ timegate_uri = native_tg if native_tg \
+ else self.timegate_uri + original_uri
+
+ http_acc_dt = MementoClient.convert_to_http_datetime(
+ accept_datetime)
+
+ tg_response = MementoClient.request_head(
+ timegate_uri,
+ accept_datetime=http_acc_dt,
+ follow_redirects=True,
+ session=self.session,
+ timeout=timeout
+ )
+
+ return super().get_memento_info(request_uri,
+ accept_datetime=accept_datetime,
+ tg_response=tg_response,
+ **kwargs)
+
+ def get_native_timegate_uri(self,
+ original_uri: str,
+ accept_datetime: Optional[datetime],
+ timeout: Optional[int] = None,
+ **kwargs) -> Optional[str]:
+ """Check the original uri whether the timegate uri is provided.
+
+ Given an original URL and an accept datetime, check the original uri
+ to see if the timegate uri is provided in the Link header.
+
+ :param original_uri: An HTTP uri of the original resource.
+ :param accept_datetime: The datetime object of the accept
+ datetime
+ :param timeout: the timeout value for the HTTP connection.
+ :return: The timegate uri of the original resource, if provided,
+ else None.
+ """
+ org_response = kwargs.pop('response', None)
+ if not org_response:
+ try:
+ org_response = MementoClient.request_head(
+ original_uri,
+ accept_datetime=MementoClient.convert_to_http_datetime(
+ accept_datetime),
+ session=self.session,
+ timeout=timeout
+ )
+ except (requests.exceptions.ConnectTimeout,
+ requests.exceptions.ConnectionError):
+ warning('Could not connect to URI {}, returning no native '
+ 'URI-G'.format(original_uri))
+ return None
+
+ debug('Request headers sent to search for URI-G: '
+ + str(org_response.request.headers))
+
+ return super().get_native_timegate_uri(original_uri, accept_datetime,
+ response=org_response, **kwargs)
+
+ @staticmethod
+ def is_timegate(uri: str,
+ accept_datetime: Optional[str] = None,
+ response: Optional[requests.Response] = None,
+ session: Optional[requests.Session] = None,
+ timeout: Optional[int] = None) -> bool:
+ """Checks if the given uri is a valid timegate according to the RFC.
+
+ :param uri: the http uri to check.
+ :param accept_datetime: the accept datetime string in http date
+ format.
+ :param response: the response object of the uri.
+ :param session: the requests session object.
+ :param timeout: the timeout value for the HTTP connection.
+ :return: True if a valid timegate, else False.
+ """
+ if not response:
+ if not accept_datetime:
+ accept_datetime = MementoClient.convert_to_http_datetime(
+ datetime.now())
+
+ response = MementoClient.request_head(
+ uri,
+ accept_datetime=accept_datetime,
+ session=session,
+ timeout=timeout
+ )
+ return old_is_timegate(
+ uri, accept_datetime, response=response, session=session)
+
+ @staticmethod
+ def is_memento(uri: str,
+ response: Optional[requests.Response] = None,
+ session: Optional[requests.Session] = None,
+ timeout: Optional[int] = None) -> bool:
+ """
+ Determines if the URI given is indeed a Memento.
+
+ The simple case is to look for a Memento-Datetime header in the
+ request, but not all archives are Memento-compliant yet.
+
+ :param uri: an HTTP URI for testing
+ :param response: the response object of the uri.
+ :param session: the requests session object.
+ :param timeout: (int) the timeout value for the HTTP connection.
+ :return: True if a Memento, False otherwise
+ """
+ if not response:
+ response = MementoClient.request_head(uri,
+ follow_redirects=False,
+ session=session,
+ timeout=timeout)
+ return old_is_memento(uri, response=response)
+
+ @staticmethod
+ def convert_to_http_datetime(dt: Optional[datetime]) -> str:
+ """Converts a datetime object to a date string in HTTP format.
+
+ :param dt: A datetime object.
+ :return: The date in HTTP format.
+ :raises TypeError: Expecting dt parameter to be of type datetime.
+ """
+ if dt and not isinstance(dt, datetime):
+ raise TypeError(
+ 'Expecting dt parameter to be of type datetime.')
+ return old_convert_to_http_datetime(dt)
+
+ @staticmethod
+ def request_head(uri: str,
+ accept_datetime: Optional[str] = None,
+ follow_redirects: bool = False,
+ session: Optional[requests.Session] = None,
+ timeout: Optional[int] = None) -> requests.Response:
+ """Makes HEAD requests.
+
+ :param uri: the uri for the request.
+ :param accept_datetime: the accept-datetime in the http format.
+ :param follow_redirects: Toggle to follow redirects. False by
+ default, so does not follow any redirects.
+ :param session: the request session object to avoid opening new
+ connections for every request.
+ :param timeout: the timeout for the HTTP requests.
+ :return: the response object.
+ :raises ValueError: Only HTTP URIs are supported
+ """
+ headers = {
+ 'Accept-Datetime': accept_datetime} if accept_datetime else {}
+
+ # create a session if not supplied
+ session_set = False
+ if not session:
+ session = requests.Session()
+ session_set = True
+ try:
+ response = session.head(uri,
+ headers=headers,
+ allow_redirects=follow_redirects,
+ timeout=timeout or 9)
+ except (InvalidSchema, MissingSchema):
+ raise ValueError('Only HTTP URIs are supported, '
+ 'URI {} unrecognized.'.format(uri))
+ if session_set:
+ session.close()
+
+ return response
-docuReplacements = {'¶ms;': pagegenerators.parameterHelp} # noqa: N816
-
-ignorelist = [
- # Officially reserved for testing, documentation, etc. in
- # https://datatracker.ietf.org/doc/html/rfc2606#page-2
- # top-level domains:
- re.compile(r'.*[\./(a)]test(/.*)?'),
- re.compile(r'.*[\./(a)]example(/.*)?'),
- re.compile(r'.*[\./(a)]invalid(/.*)?'),
- re.compile(r'.*[\./(a)]localhost(/.*)?'),
- # second-level domains:
- re.compile(r'.*[\./(a)]example\.com(/.*)?'),
- re.compile(r'.*[\./(a)]example\.net(/.*)?'),
- re.compile(r'.*[\./(a)]example\.org(/.*)?'),
-
- # Other special cases
- re.compile(r'.*[\./(a)]berlinonline\.de(/.*)?'),
- # above entry to be manually fixed per request at
- # [[de:Benutzer:BLueFiSH.as/BZ]]
- # bot can't handle their redirects:
-
- # bot rejected on the site, already archived
- re.compile(r'.*[\./(a)]web\.archive\.org(/.*)?'),
-
- # Ignore links containing * in domain name
- # as they are intentionally fake
- re.compile(r'https?\:\/\/\*(/.*)?'),
-]
+# Save old static methods and update static methods of parent class
+old_is_timegate = OldMementoClient.is_timegate
+old_is_memento = OldMementoClient.is_memento
+old_convert_to_http_datetime = OldMementoClient.convert_to_http_datetime
+OldMementoClient.is_timegate = MementoClient.is_timegate
+OldMementoClient.is_memento = MementoClient.is_memento
+OldMementoClient.convert_to_http_datetime \
+ = MementoClient.convert_to_http_datetime
+OldMementoClient.request_head = MementoClient.request_head
-def _get_closest_memento_url(url, when=None, timegate_uri=None):
+def get_closest_memento_url(url: str,
+ when: Optional[datetime] = None,
+ timegate_uri: Optional[str] = None):
"""Get most recent memento for url."""
if not when:
when = datetime.datetime.now()
- mc = memento_client.MementoClient()
+ mc = MementoClient()
if timegate_uri:
mc.timegate_uri = timegate_uri
@@ -191,561 +317,17 @@
except (requests.ConnectionError, MementoClientException) as e:
error = e
retry_count += 1
- pywikibot.sleep(config.retry_wait)
+ sleep(config.retry_wait)
else:
raise error
mementos = memento_info.get('mementos')
if not mementos:
- raise Exception(
- 'mementos not found for {} via {}'.format(url, timegate_uri))
- if 'closest' not in mementos:
- raise Exception(
- 'closest memento not found for {} via {}'.format(
- url, timegate_uri))
- if 'uri' not in mementos['closest']:
- raise Exception(
- 'closest memento uri not found for {} via {}'.format(
- url, timegate_uri))
- return mementos['closest']['uri'][0]
-
-
-def get_archive_url(url):
- """Get archive URL."""
- try:
- archive = _get_closest_memento_url(
- url,
- timegate_uri='http://web.archive.org/web/')
- except Exception:
- archive = _get_closest_memento_url(
- url,
- timegate_uri='http://timetravel.mementoweb.org/webcite/timegate/')
-
- # FIXME: Hack for T167463: Use https instead of http for archive.org links
- if archive.startswith('http://web.archive.org'):
- archive = archive.replace('http://', 'https://', 1)
- return archive
-
-
-def weblinks_from_text(
- text,
- without_bracketed: bool = False,
- only_bracketed: bool = False
-):
- """
- Yield web links from text.
-
- Only used as text predicate for XmlDumpPageGenerator to speed up
- generator.
-
- TODO: move to textlib
- """
- text = textlib.removeDisabledParts(text)
-
- # Ignore links in fullurl template
- text = re.sub(r'{{\s?fullurl:.[^}]*}}', '', text)
-
- # MediaWiki parses templates before parsing external links. Thus, there
- # might be a | or a } directly after a URL which does not belong to
- # the URL itself.
-
- # First, remove the curly braces of inner templates:
- nested_template_regex = re.compile(r'{{([^}]*?){{(.*?)}}(.*?)}}')
- while nested_template_regex.search(text):
- text = nested_template_regex.sub(r'{{\1 \2 \3}}', text)
-
- # Then blow up the templates with spaces so that the | and }} will not
- # be regarded as part of the link:.
- template_with_params_regex = re.compile(r'{{([^}]*?[^ ])\|([^ ][^}]*?)}}',
- re.DOTALL)
- while template_with_params_regex.search(text):
- text = template_with_params_regex.sub(r'{{ \1 | \2 }}', text)
-
- # Add <blank> at the end of a template
- # URL as last param of multiline template would not be correct
- text = text.replace('}}', ' }}')
-
- # Remove HTML comments in URLs as well as URLs in HTML comments.
- # Also remove text inside nowiki links etc.
- text = textlib.removeDisabledParts(text)
- link_regex = textlib.compileLinkR(without_bracketed, only_bracketed)
- for m in link_regex.finditer(text):
- if m.group('url'):
- yield m.group('url')
- else:
- yield m.group('urlb')
-
-
-XmlDumpPageGenerator = partial(
- _XMLDumpPageGenerator, text_predicate=weblinks_from_text)
-
-
-class NotAnURLError(BaseException):
-
- """The link is not an URL."""
-
-
-class LinkCheckThread(threading.Thread):
-
- """A thread responsible for checking one URL.
-
- After checking the page, it will die.
- """
-
- #: Collecting start time of a thread for any host
- hosts = {} # type: Dict[str, float]
- lock = threading.Lock()
-
- def __init__(self, page, url, history, http_ignores, day) -> None:
- """Initializer."""
- self.page = page
- self.url = url
- self.history = history
- self.header = {
- 'Accept': 'text/xml,application/xml,application/xhtml+xml,'
- 'text/html;q=0.9,text/plain;q=0.8,image/png,*/*;q=0.5',
- 'Accept-Language': 'de-de,de;q=0.8,en-us;q=0.5,en;q=0.3',
- 'Accept-Charset': 'ISO-8859-1,utf-8;q=0.7,*;q=0.7',
- 'Keep-Alive': '30',
- 'Connection': 'keep-alive',
- }
- # identification for debugging purposes
- self.http_ignores = http_ignores
- self._use_fake_user_agent = config.fake_user_agent_default.get(
- 'weblinkchecker', False)
- self.day = day
- super().__init__()
-
- @classmethod
- def get_delay(cls, name: str) -> float:
- """Determine delay from class attribute.
-
- Store the last call for a given hostname with an offset of
- 6 seconds to ensure there are no more than 10 calls per minute
- for the same host. Calculate the delay to start the run.
-
- :param name: The key for the hosts class attribute
- :return: The calulated delay to start the run
- """
- now = time.monotonic()
- with cls.lock:
- timestamp = cls.hosts.get(name, now)
- cls.hosts[name] = max(now, timestamp) + 6
- return max(0, timestamp - now)
-
- def run(self):
- """Run the bot."""
- time.sleep(self.get_delay(self.name))
- try:
- header = self.header
- r = comms.http.fetch(
- self.url, headers=header,
- use_fake_user_agent=self._use_fake_user_agent)
- except requests.exceptions.InvalidURL:
- message = i18n.twtranslate(self.page.site,
- 'weblinkchecker-badurl_msg',
- {'URL': self.url})
- except Exception:
- pywikibot.output('Exception while processing URL {} in page {}'
- .format(self.url, self.page.title()))
- raise
-
- if (
- r.status_code != HTTPStatus.OK
- or r.status_code in self.http_ignores
- ):
- message = HTTPStatus(r.status_code).phrase
- pywikibot.output('*{} links to {} - {}.'
- .format(self.page.title(as_link=True), self.url,
- message))
- self.history.set_dead_link(self.url, message, self.page,
- config.weblink_dead_days)
- elif self.history.set_link_alive(self.url):
- pywikibot.output(
- '*Link to {} in {} is back alive.'
- .format(self.url, self.page.title(as_link=True)))
-
-
-class History:
-
- """
- Store previously found dead links.
-
- The URLs are dictionary keys, and
- values are lists of tuples where each tuple represents one time the URL was
- found dead. Tuples have the form (title, date, error) where title is the
- wiki page where the URL was found, date is an instance of time, and error
- is a string with error code and message.
-
- We assume that the first element in the list represents the first time we
- found this dead link, and the last element represents the last time.
-
- Example::
-
- dict = {
- 'https://www.example.org/page': [
- ('WikiPageTitle', DATE, '404: File not found'),
- ('WikiPageName2', DATE, '404: File not found'),
- ]
- }
- """
-
- def __init__(self, report_thread, site=None) -> None:
- """Initializer."""
- self.report_thread = report_thread
- if not site:
- self.site = pywikibot.Site()
- else:
- self.site = site
- self.semaphore = threading.Semaphore()
- self.datfilename = pywikibot.config.datafilepath(
- 'deadlinks', 'deadlinks-{}-{}.dat'.format(self.site.family.name,
- self.site.code))
- # Count the number of logged links, so that we can insert captions
- # from time to time
- self.log_count = 0
- try:
- with open(self.datfilename, 'rb') as datfile:
- self.history_dict = pickle.load(datfile)
- except (OSError, EOFError):
- # no saved history exists yet, or history dump broken
- self.history_dict = {}
-
- def log(self, url, error, containing_page, archive_url) -> None:
- """Log an error report to a text file in the deadlinks subdirectory."""
- if archive_url:
- error_report = '* {} ([{} archive])\n'.format(url, archive_url)
- else:
- error_report = '* {}\n'.format(url)
- for (page_title, date, error) in self.history_dict[url]:
- # ISO 8601 formulation
- iso_date = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(date))
- error_report += '** In [[{}]] on {}, {}\n'.format(
- page_title, iso_date, error)
- pywikibot.output('** Logging link for deletion.')
- txtfilename = pywikibot.config.datafilepath('deadlinks',
- 'results-{}-{}.txt'
- .format(
- self.site.family.name,
- self.site.lang))
- with codecs.open(txtfilename, 'a', 'utf-8') as txtfile:
- self.log_count += 1
- if self.log_count % 30 == 0:
- # insert a caption
- txtfile.write('=== {} ===\n'
- .format(containing_page.title()[:3]))
- txtfile.write(error_report)
-
- if self.report_thread and not containing_page.isTalkPage():
- self.report_thread.report(url, error_report, containing_page,
- archive_url)
-
- def set_dead_link(self, url, error, page, weblink_dead_days) -> None:
- """Add the fact that the link was found dead to the .dat file."""
- with self.semaphore:
- now = time.time()
- if url in self.history_dict:
- time_since_first_found = now - self.history_dict[url][0][1]
- time_since_last_found = now - self.history_dict[url][-1][1]
- # if the last time we found this dead link is less than an hour
- # ago, we won't save it in the history this time.
- if time_since_last_found > 60 * 60:
- self.history_dict[url].append((page.title(), now, error))
- # if the first time we found this link longer than x day ago
- # (default is a week), it should probably be fixed or removed.
- # We'll list it in a file so that it can be removed manually.
- if time_since_first_found > 60 * 60 * 24 * weblink_dead_days:
- # search for archived page
- try:
- archive_url = get_archive_url(url)
- except Exception as e:
- pywikibot.warning(
- 'get_closest_memento_url({}) failed: {}'.format(
- url, e))
- archive_url = None
- self.log(url, error, page, archive_url)
- else:
- self.history_dict[url] = [(page.title(), now, error)]
-
- def set_link_alive(self, url) -> bool:
- """
- Record that the link is now alive.
-
- If link was previously found dead, remove it from the .dat file.
-
- :return: True if previously found dead, else returns False.
- """
- if url in self.history_dict:
- with self.semaphore, suppress(KeyError):
- del self.history_dict[url]
- return True
-
- return False
-
- def save(self) -> None:
- """Save the .dat file to disk."""
- with open(self.datfilename, 'wb') as f:
- pickle.dump(self.history_dict, f, protocol=config.pickle_protocol)
-
-
-class DeadLinkReportThread(threading.Thread):
-
- """
- A Thread that is responsible for posting error reports on talk pages.
-
- There is only one DeadLinkReportThread, and it is using a semaphore to make
- sure that two LinkCheckerThreads cannot access the queue at the same time.
- """
-
- def __init__(self) -> None:
- """Initializer."""
- super().__init__()
- self.semaphore = threading.Semaphore()
- self.queue = []
- self.finishing = False
- self.killed = False
-
- def report(self, url, error_report, containing_page, archive_url) -> None:
- """Report error on talk page of the page containing the dead link."""
- with self.semaphore:
- self.queue.append((url, error_report, containing_page,
- archive_url))
-
- def shutdown(self) -> None:
- """Finish thread."""
- self.finishing = True
-
- def kill(self) -> None:
- """Kill thread."""
- # TODO: remove if unneeded
- self.killed = True
-
- def run(self) -> None:
- """Run thread."""
- while not self.killed:
- if not self.queue:
- if self.finishing:
- break
- time.sleep(0.1)
- continue
-
- with self.semaphore:
- url, error_report, containing_page, archive_url = self.queue[0]
- self.queue = self.queue[1:]
- talk_page = containing_page.toggleTalkPage()
- pywikibot.output('<<lightaqua>>** Reporting dead link on {}...'
- '<<default>>'.format(talk_page))
- try:
- content = talk_page.get() + '\n\n\n'
- if url in content:
- pywikibot.output('<<lightaqua>>** Dead link seems to '
- 'have already been reported on {}'
- '<<default>>'.format(talk_page))
- continue
- except (NoPageError, IsRedirectPageError):
- content = ''
-
- if archive_url:
- archive_msg = '\n' + i18n.twtranslate(
- containing_page.site, 'weblinkchecker-archive_msg',
- {'URL': archive_url})
- else:
- archive_msg = ''
- # The caption will default to "Dead link". But if there
- # is already such a caption, we'll use "Dead link 2",
- # "Dead link 3", etc.
- caption = i18n.twtranslate(containing_page.site,
- 'weblinkchecker-caption')
- i = 1
- count = ''
- # Check if there is already such a caption on
- # the talk page.
- while re.search('= *{}{} *='
- .format(caption, count), content) is not None:
- i += 1
- count = ' ' + str(i)
- caption += count
- content += '== {0} ==\n\n{3}\n\n{1}{2}\n--~~~~'.format(
- caption, error_report, archive_msg,
- i18n.twtranslate(containing_page.site,
- 'weblinkchecker-report'))
-
- comment = '[[{}#{}|→]] {}'.format(
- talk_page.title(), caption,
- i18n.twtranslate(containing_page.site,
- 'weblinkchecker-summary'))
- try:
- talk_page.put(content, comment)
- except SpamblacklistError as error:
- pywikibot.output(
- '<<lightaqua>>** SpamblacklistError while trying to '
- 'change {}: {}<<default>>'
- .format(talk_page, error.url))
-
-
-class WeblinkCheckerRobot(SingleSiteBot, ExistingPageBot):
-
- """
- Bot which will search for dead weblinks.
-
- It uses several LinkCheckThreads at once to process pages from generator.
- """
-
- use_redirects = False
-
- def __init__(self, http_ignores=None, day: int = 7, **kwargs) -> None:
- """Initializer."""
- super().__init__(**kwargs)
-
- if config.report_dead_links_on_talk:
- pywikibot.log('Starting talk page thread')
- report_thread = DeadLinkReportThread()
- report_thread.start()
- else:
- report_thread = None
- self.history = History(report_thread, site=self.site)
- self.http_ignores = http_ignores or []
- self.day = day
-
- # Limit the number of threads started at the same time
- self.threads = ThreadList(limit=config.max_external_links,
- wait_time=config.retry_wait)
-
- def treat_page(self) -> None:
- """Process one page."""
- page = self.current_page
- for url in page.extlinks():
- for ignore_regex in ignorelist:
- if ignore_regex.match(url):
- break
- else:
- # Each thread will check one page, then die.
- thread = LinkCheckThread(page, url, self.history,
- self.http_ignores, self.day)
- # thread dies when program terminates
- thread.daemon = True
- # use hostname as thread.name
- thread.name = removeprefix(
- urlparse.urlparse(url).hostname, 'www.')
- self.threads.append(thread)
-
- def teardown(self) -> None:
- """Finish remaining threads and save history file."""
- num = self.count_link_check_threads()
- if num:
- pywikibot.info('<<lightblue>>Waiting for remaining {} threads '
- 'to finish, please wait...'.format(num))
-
- while self.count_link_check_threads():
- try:
- time.sleep(0.1)
- except KeyboardInterrupt:
- # Threads will die automatically because they are daemonic.
- if pywikibot.input_yn('There are {} pages remaining in the '
- 'queue. Really exit?'
- .format(self.count_link_check_threads()),
- default=False, automatic_quit=False):
- break
-
- num = self.count_link_check_threads()
- if num:
- pywikibot.info('<<yellow>>>Remaining {} threads will be killed.'
- .format(num))
-
- if self.history.report_thread:
- self.history.report_thread.shutdown()
- # wait until the report thread is shut down; the user can
- # interrupt it by pressing CTRL-C.
- try:
- while self.history.report_thread.is_alive():
- time.sleep(0.1)
- except KeyboardInterrupt:
- pywikibot.info('Report thread interrupted.')
- self.history.report_thread.kill()
-
- pywikibot.info('Saving history...')
- self.history.save()
-
- @staticmethod
- def count_link_check_threads() -> int:
- """Count LinkCheckThread threads.
-
- :return: number of LinkCheckThread threads
- """
- return sum(isinstance(thread, LinkCheckThread)
- for thread in threading.enumerate())
-
-
-def RepeatPageGenerator(): # noqa: N802
- """Generator for pages in History."""
- history = History(None)
- page_titles = set()
- for value in history.history_dict.values():
- for entry in value:
- page_titles.add(entry[0])
- for page_title in sorted(page_titles):
- page = pywikibot.Page(pywikibot.Site(), page_title)
- yield page
-
-
-def main(*args: str) -> None:
- """
- Process command line arguments and invoke bot.
-
- If args is an empty list, sys.argv is used.
-
- :param args: command line arguments
- """
- gen = None
- xml_filename = None
- http_ignores = []
-
- # Process global args and prepare generator args parser
- local_args = pywikibot.handle_args(args)
- gen_factory = pagegenerators.GeneratorFactory()
-
- for arg in local_args:
- if arg == '-talk':
- config.report_dead_links_on_talk = True
- elif arg == '-notalk':
- config.report_dead_links_on_talk = False
- elif arg == '-repeat':
- gen = RepeatPageGenerator()
- elif arg.startswith('-ignore:'):
- http_ignores.append(int(arg[8:]))
- elif arg.startswith('-day:'):
- config.weblink_dead_days = int(arg[5:])
- elif arg.startswith('-xmlstart'):
- if len(arg) == 9:
- xml_start = pywikibot.input(
- 'Please enter the dumped article to start with:')
- else:
- xml_start = arg[10:]
- elif arg.startswith('-xml'):
- if len(arg) == 4:
- xml_filename = i18n.input('pywikibot-enter-xml-filename')
- else:
- xml_filename = arg[5:]
- else:
- gen_factory.handle_arg(arg)
-
- if xml_filename:
- try:
- xml_start
- except NameError:
- xml_start = None
- gen = XmlDumpPageGenerator(xml_filename, xml_start,
- gen_factory.namespaces)
-
- if not gen:
- gen = gen_factory.getCombinedGenerator()
-
- if not suggest_help(missing_generator=not gen,
- missing_dependencies=missing_dependencies):
- bot = WeblinkCheckerRobot(http_ignores, config.weblink_dead_days,
- generator=gen)
- bot.run()
-
-
-if __name__ == '__main__':
- main()
+ err_msg = 'mementos not found for {} via {}'
+ elif 'closest' not in mementos:
+ err_msg = 'closest memento not found for {} via {}'
+ elif 'uri' not in mementos['closest']:
+ err_msg = 'closest memento uri not found for {} via {}'
+ else:
+ return mementos['closest']['uri'][0]
+ raise Exception(err_msg)
diff --git a/requirements.txt b/requirements.txt
index 5a4796f..10eddfb 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -58,4 +58,4 @@
beautifulsoup4
# scripts/weblinkchecker.py
-memento_client>=0.5.1,!=0.6.0
+memento_client==0.6.1
diff --git a/scripts/_weblinkchecker.py b/scripts/weblinkchecker.py
similarity index 94%
rename from scripts/_weblinkchecker.py
rename to scripts/weblinkchecker.py
index 398ba64..215b8d8 100755
--- a/scripts/_weblinkchecker.py
+++ b/scripts/weblinkchecker.py
@@ -109,7 +109,6 @@
# Distributed under the terms of the MIT license.
#
import codecs
-import datetime
import pickle
import re
import threading
@@ -137,9 +136,8 @@
try:
- import memento_client
- from memento_client.memento_client import MementoClientException
- missing_dependencies = None
+ from pywikibot.data.memento import get_closest_memento_url
+ missing_dependencies = []
except ImportError:
missing_dependencies = ['memento_client']
@@ -174,50 +172,13 @@
]
-def _get_closest_memento_url(url, when=None, timegate_uri=None):
- """Get most recent memento for url."""
- if not when:
- when = datetime.datetime.now()
-
- mc = memento_client.MementoClient()
- if timegate_uri:
- mc.timegate_uri = timegate_uri
-
- retry_count = 0
- while retry_count <= config.max_retries:
- try:
- memento_info = mc.get_memento_info(url, when)
- break
- except (requests.ConnectionError, MementoClientException) as e:
- error = e
- retry_count += 1
- pywikibot.sleep(config.retry_wait)
- else:
- raise error
-
- mementos = memento_info.get('mementos')
- if not mementos:
- raise Exception(
- 'mementos not found for {} via {}'.format(url, timegate_uri))
- if 'closest' not in mementos:
- raise Exception(
- 'closest memento not found for {} via {}'.format(
- url, timegate_uri))
- if 'uri' not in mementos['closest']:
- raise Exception(
- 'closest memento uri not found for {} via {}'.format(
- url, timegate_uri))
- return mementos['closest']['uri'][0]
-
-
def get_archive_url(url):
"""Get archive URL."""
try:
- archive = _get_closest_memento_url(
- url,
- timegate_uri='http://web.archive.org/web/')
+ archive = get_closest_memento_url(
+ url, timegate_uri='http://web.archive.org/web/')
except Exception:
- archive = _get_closest_memento_url(
+ archive = get_closest_memento_url(
url,
timegate_uri='http://timetravel.mementoweb.org/webcite/timegate/')
diff --git a/setup.py b/setup.py
index e6e0925..3ab222a 100755
--- a/setup.py
+++ b/setup.py
@@ -60,6 +60,7 @@
'isbn': ['python-stdnum>=1.17'],
'Graphviz': ['pydot>=1.2'],
'Google': ['google>=1.7'],
+ 'memento': ['memento_client==0.6.1'],
'mwparserfromhell': ['mwparserfromhell>=0.5.0'],
'wikitextparser': ['wikitextparser>=0.47.5; python_version < "3.6"',
'wikitextparser>=0.47.0; python_version >= "3.6"'],
@@ -99,7 +100,7 @@
script_deps = {
'commons_information.py': extra_deps['mwparserfromhell'],
'patrol.py': extra_deps['mwparserfromhell'],
- 'weblinkchecker.py': ['memento_client!=0.6.0,>=0.5.1'],
+ 'weblinkchecker.py': extra_deps['memento'],
}
extra_deps.update(script_deps)
diff --git a/tests/__init__.py b/tests/__init__.py
index 6606d55..39ce2b0 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -101,6 +101,7 @@
'logentries',
'login',
'mediawikiversion',
+ 'memento',
'mysql',
'namespace',
'oauth',
@@ -158,7 +159,6 @@
'script',
'template_bot',
'uploadscript',
- 'weblinkchecker'
}
disabled_test_modules = {
diff --git a/tests/weblinkchecker_tests.py b/tests/memento_tests.py
similarity index 88%
rename from tests/weblinkchecker_tests.py
rename to tests/memento_tests.py
index 9dd1109..a8e2768 100755
--- a/tests/weblinkchecker_tests.py
+++ b/tests/memento_tests.py
@@ -12,7 +12,6 @@
from requests.exceptions import ConnectionError as RequestsConnectionError
-from scripts import weblinkchecker
from tests.aspects import TestCase, require_modules
@@ -22,15 +21,17 @@
"""Test memento client."""
def _get_archive_url(self, url, date_string=None):
- from memento_client.memento_client import MementoClientException
+ from pywikibot.data.memento import (
+ MementoClientException,
+ get_closest_memento_url,
+ )
if date_string is None:
when = datetime.datetime.now()
else:
when = datetime.datetime.strptime(date_string, '%Y%m%d')
try:
- result = weblinkchecker._get_closest_memento_url(
- url, when, self.timegate_uri)
+ result = get_closest_memento_url(url, when, self.timegate_uri)
except (RequestsConnectionError, MementoClientException) as e:
self.skipTest(e)
return result
@@ -72,8 +73,7 @@
"""Test getting memento for invalid URL."""
# memento_client raises 'Exception', not a subclass.
with self.assertRaisesRegex(
- Exception,
- 'Only HTTP URIs are supported'):
+ ValueError, 'Only HTTP URIs are supported'):
self._get_archive_url('invalid')
diff --git a/tox.ini b/tox.ini
index 8278dfa..bc97202 100644
--- a/tox.ini
+++ b/tox.ini
@@ -70,6 +70,7 @@
nosetests --with-doctest pywikibot {[params]doctest_skip}
deps =
nose
+ .[memento]
.[mwparserfromhell]
[testenv:venv]
--
To view, visit https://gerrit.wikimedia.org/r/c/pywikibot/core/+/803232
To unsubscribe, or for help writing mail filters, visit https://gerrit.wikimedia.org/r/settings
Gerrit-Project: pywikibot/core
Gerrit-Branch: master
Gerrit-Change-Id: I137a27ad198f0e0aae713c888401265f7aca187b
Gerrit-Change-Number: 803232
Gerrit-PatchSet: 22
Gerrit-Owner: Xqt <info(a)gno.de>
Gerrit-Reviewer: D3r1ck01 <xsavitar.wiki(a)aol.com>
Gerrit-Reviewer: Dvorapa <dvorapa(a)seznam.cz>
Gerrit-Reviewer: Framawiki <framawiki(a)tools.wmflabs.org>
Gerrit-Reviewer: Matěj Suchánek <matejsuchanek97(a)gmail.com>
Gerrit-Reviewer: Shawnmjones <jones.shawn.m(a)gmail.com>
Gerrit-Reviewer: Xqt <info(a)gno.de>
Gerrit-Reviewer: Zhuyifei1999 <zhuyifei1999(a)gmail.com>
Gerrit-Reviewer: jenkins-bot
Gerrit-MessageType: merged
Xqt has submitted this change. ( https://gerrit.wikimedia.org/r/c/pywikibot/core/+/806546 )
Change subject: [fix] move get_closest_memento_url to data/memento.py
......................................................................
[fix] move get_closest_memento_url to data/memento.py
Change-Id: I53e7c99b93389907b985e5223bcc4d4134de42a7
---
C pywikibot/data/memento.py
R scripts/_weblinkchecker.py
2 files changed, 0 insertions(+), 0 deletions(-)
Approvals:
Xqt: Verified; Looks good to me, approved
diff --git a/scripts/weblinkchecker.py b/pywikibot/data/memento.py
old mode 100755
new mode 100644
similarity index 100%
copy from scripts/weblinkchecker.py
copy to pywikibot/data/memento.py
diff --git a/scripts/weblinkchecker.py b/scripts/_weblinkchecker.py
similarity index 100%
rename from scripts/weblinkchecker.py
rename to scripts/_weblinkchecker.py
--
To view, visit https://gerrit.wikimedia.org/r/c/pywikibot/core/+/806546
To unsubscribe, or for help writing mail filters, visit https://gerrit.wikimedia.org/r/settings
Gerrit-Project: pywikibot/core
Gerrit-Branch: master
Gerrit-Change-Id: I53e7c99b93389907b985e5223bcc4d4134de42a7
Gerrit-Change-Number: 806546
Gerrit-PatchSet: 1
Gerrit-Owner: Xqt <info(a)gno.de>
Gerrit-Reviewer: Xqt <info(a)gno.de>
Gerrit-Reviewer: jenkins-bot
Gerrit-MessageType: merged
Xqt has submitted this change. ( https://gerrit.wikimedia.org/r/c/pywikibot/core/+/806500 )
Change subject: [IMPR] move api.Request to api._requests.py file (part 3)
......................................................................
[IMPR] move api.Request to api._requests.py file (part 3)
Change-Id: Ie4f9eef90d27866011ac6edc3035d6dbd13a269f
---
M pywikibot/CONTENT.rst
M pywikibot/data/api/__init__.py
M pywikibot/data/api/_requests.py
M tests/api_tests.py
4 files changed, 9 insertions(+), 1,278 deletions(-)
Approvals:
Xqt: Verified; Looks good to me, approved
diff --git a/pywikibot/CONTENT.rst b/pywikibot/CONTENT.rst
index f74a2ef..5f100ee 100644
--- a/pywikibot/CONTENT.rst
+++ b/pywikibot/CONTENT.rst
@@ -100,6 +100,8 @@
| | _optionset.py | Boolean API option |
| +----------------+-------------------------------------+
| | _paraminfo.py | API information data object |
+ | +----------------+-------------------------------------+
+ | | _requests.py | API Requests interface |
+----------------------------+----------------+-------------------------------------+
| mysql.py | Miscellaneous helper functions for mysql queries |
+----------------------------+------------------------------------------------------+
diff --git a/pywikibot/data/api/__init__.py b/pywikibot/data/api/__init__.py
index c64d4ab..7f5b15c 100644
--- a/pywikibot/data/api/__init__.py
+++ b/pywikibot/data/api/__init__.py
@@ -1,29 +1,13 @@
"""Interface to Mediawiki's api.php."""
#
-# (C) Pywikibot team, 2007-2022
+# (C) Pywikibot team, 2014-2022
#
# Distributed under the terms of the MIT license.
#
-import datetime
-import hashlib
-import inspect
-import os
-import pickle
-import pprint
-import re
-import traceback
-from collections.abc import MutableMapping
from email.generator import BytesGenerator
from email.mime.multipart import MIMEMultipart as MIMEMultipartOrig
-from email.mime.nonmultipart import MIMENonMultipart
from io import BytesIO
-from typing import Any, Optional, Union
-from urllib.parse import unquote, urlencode
-from warnings import warn
-import pywikibot
-from pywikibot import config
-from pywikibot.backports import Callable, Dict, Match, Tuple, removeprefix
from pywikibot.comms import http
from pywikibot.data.api._generators import (
APIGenerator,
@@ -37,20 +21,8 @@
from pywikibot.data.api._login import LoginManager
from pywikibot.data.api._paraminfo import ParamInfo
from pywikibot.data.api._optionset import OptionSet
-from pywikibot.exceptions import (
- Error,
- FatalServerError,
- MaxlagTimeoutError,
- NoUsernameError,
- Server414Error,
- Server504Error,
- SiteDefinitionError,
- TimeoutError,
-)
+from pywikibot.data.api._requests import CachedRequest, Request, encode_url
from pywikibot.family import SubdomainFamily
-from pywikibot.login import LoginStatus
-from pywikibot.textlib import removeHTMLParts
-from pywikibot.tools import PYTHON_VERSION
__all__ = (
'APIGenerator',
@@ -69,10 +41,6 @@
)
-lagpattern = re.compile(
- r'Waiting for [\w.: ]+: (?P<lag>\d+(?:\.\d+)?) seconds? lagged')
-
-
def _invalidate_superior_cookies(family) -> None:
"""
Clear cookies for site's second level domain.
@@ -125,1242 +93,3 @@
MIMEMultipart = CTEBinaryMIMEMultipart
-
-
-class Request(MutableMapping):
-
- """A request to a Site's api.php interface.
-
- Attributes of this object (except for the special parameters listed
- below) get passed as commands to api.php, and can be get or set
- using the dict interface. All attributes must be strings. Use an
- empty string for parameters that don't require a value. For example,
- Request(action="query", titles="Foo bar", prop="info", redirects="")
- corresponds to the API request
- "api.php?action=query&titles=Foo%20bar&prop=info&redirects"
-
- This is the lowest-level interface to the API, and can be used for any
- request that a particular site's API supports. See the API documentation
- (https://www.mediawiki.org/wiki/API) and site-specific settings for
- details on what parameters are accepted for each request type.
-
- Uploading files is a special case: to upload, the parameter "mime" must
- contain a dict, and the parameter "file" must be set equal to a valid
- filename on the local computer, _not_ to the content of the file.
-
- Returns a dict containing the JSON data returned by the wiki. Normally,
- one of the dict keys will be equal to the value of the 'action'
- parameter. Errors are caught and raise an APIError exception.
-
- Example:
-
- >>> r = Request(parameters={'action': 'query', 'meta': 'userinfo'})
- >>> # This is equivalent to
- >>> # https://{path}/api.php?action=query&meta=userinfo&format=json
- >>> # change a parameter
- >>> r['meta'] = "userinfo|siteinfo"
- >>> # add a new parameter
- >>> r['siprop'] = "namespaces"
- >>> # note that "uiprop" param gets added automatically
- >>> str(r.action)
- 'query'
- >>> sorted(str(key) for key in r._params.keys())
- ['action', 'meta', 'siprop']
- >>> [str(key) for key in r._params['action']]
- ['query']
- >>> [str(key) for key in r._params['meta']]
- ['userinfo', 'siteinfo']
- >>> [str(key) for key in r._params['siprop']]
- ['namespaces']
- >>> data = r.submit()
- >>> isinstance(data, dict)
- True
- >>> set(['query', 'batchcomplete', 'warnings']).issuperset(data.keys())
- True
- >>> 'query' in data
- True
- >>> sorted(str(key) for key in data['query'].keys())
- ['namespaces', 'userinfo']
-
- """
-
- # To make sure the default value of 'parameters' can be identified.
- _PARAM_DEFAULT = object()
-
- def __init__(self, site=None,
- mime: Optional[dict] = None,
- throttle: bool = True,
- max_retries: Optional[int] = None,
- retry_wait: Optional[int] = None,
- use_get: Optional[bool] = None,
- parameters=_PARAM_DEFAULT, **kwargs) -> None:
- """
- Create a new Request instance with the given parameters.
-
- The parameters for the request can be defined via either the
- 'parameters' parameter or the keyword arguments. The keyword arguments
- were the previous implementation but could cause problems when there
- are arguments to the API named the same as normal arguments to this
- class. So the second parameter 'parameters' was added which just
- contains all parameters. When a Request instance is created it must use
- either one of them and not both at the same time. To have backwards
- compatibility it adds a parameter named 'parameters' to kwargs when
- both parameters are set as that indicates an old call and 'parameters'
- was originally supplied as a keyword parameter.
-
- If undefined keyword arguments were given AND the 'parameters'
- parameter was supplied as a positional parameter it still assumes
- 'parameters' were part of the keyword arguments.
-
- If a class is using Request and is directly forwarding the parameters,
- :py:obj:`Request.clean_kwargs` can be used to automatically
- convert the old kwargs mode into the new parameter mode. This
- normalizes the arguments so that when the API parameters are
- modified the changes can always be applied to the 'parameters'
- parameter.
-
- :param site: The Site to which the request will be submitted. If not
- supplied, uses the user's configured default Site.
- :param mime: If not None, send in "multipart/form-data" format (default
- None). Parameters which should only be transferred via mime
- mode are defined via this parameter (even an empty dict means
- mime shall be used).
- :param max_retries: Maximum number of times to retry after
- errors, defaults to config.max_retries.
- :param retry_wait: Minimum time in seconds to wait after an
- error, defaults to config.retry_wait seconds (doubles each retry
- until config.retry_max seconds is reached).
- :param use_get: Use HTTP GET request if possible. If False it
- uses a POST request. If None, it'll try to determine via
- action=paraminfo if the action requires a POST.
- :param parameters: The parameters used for the request to the API.
- :type parameters: dict
- :param kwargs: The parameters used for the request to the API.
- """
- if site is None:
- self.site = pywikibot.Site()
- warn('Request() invoked without a site; setting to {}'
- .format(self.site), RuntimeWarning, 2)
- else:
- self.site = site
-
- self.mime = mime
- if isinstance(mime, bool):
- raise TypeError('mime param in api.Request() must not be boolean')
-
- self.throttle = throttle
- self.use_get = use_get
- if max_retries is None:
- self.max_retries = pywikibot.config.max_retries
- else:
- self.max_retries = max_retries
- self.current_retries = 0
- if retry_wait is None:
- self.retry_wait = pywikibot.config.retry_wait
- else:
- self.retry_wait = retry_wait
- # The only problem with that system is that it won't detect when
- # 'parameters' is actually the only parameter for the request as it
- # then assumes it's using the new mode (and the parameters are actually
- # in the parameter 'parameters' not that the parameter 'parameters' is
- # actually a parameter for the request). But that is invalid anyway as
- # it MUST have at least an action parameter for the request which would
- # be in kwargs if it's using the old mode.
- if kwargs:
- if parameters is not self._PARAM_DEFAULT:
- # 'parameters' AND kwargs is set. In that case think of
- # 'parameters' being an old kwarg which is now filled in an
- # actual parameter
- self._warn_both()
- kwargs['parameters'] = parameters
- # When parameters wasn't set it's likely that kwargs-mode was used
- self._warn_kwargs()
- parameters = kwargs
- elif parameters is self._PARAM_DEFAULT:
- parameters = {}
- self._params = {}
- if 'action' not in parameters:
- raise ValueError("'action' specification missing from Request.")
- self.action = parameters['action']
- self.update(parameters) # also convert all parameter values to lists
- self._warning_handler = None # type: Optional[Callable[[str, str], Union[Match[str], bool, None]]] # noqa: E501
- # Actions that imply database updates on the server, used for various
- # things like throttling or skipping actions when we're in simulation
- # mode
- self.write = self.action in {
- 'block', 'clearhasmsg', 'createaccount', 'delete', 'edit',
- 'emailuser', 'filerevert', 'flowthank', 'imagerotate', 'import',
- 'managetags', 'mergehistory', 'move', 'options', 'patrol',
- 'protect', 'purge', 'resetpassword', 'revisiondelete', 'rollback',
- 'setnotificationtimestamp', 'setpagelanguage', 'tag', 'thank',
- 'unblock', 'undelete', 'upload', 'userrights', 'watch',
- 'wbcreateclaim', 'wbcreateredirect', 'wbeditentity',
- 'wblinktitles', 'wbmergeitems', 'wbremoveclaims',
- 'wbremovequalifiers', 'wbremovereferences', 'wbsetaliases',
- 'wbsetclaim', 'wbsetclaimvalue', 'wbsetdescription', 'wbsetlabel',
- 'wbsetqualifier', 'wbsetreference', 'wbsetsitelink',
- 'wbladdform', 'wbleditformelements', 'wblmergelexemes',
- 'wblremoveform',
- }
- # Client side verification that the request is being performed
- # by a logged in user, and warn if it isn't a config username.
- if self.write:
- try:
- username = self.site.userinfo['name']
- except KeyError:
- raise Error('API write action attempted without user name')
-
- if 'anon' in self.site.userinfo:
- raise Error("API write action attempted as IP '{}'"
- .format(username))
-
- if not self.site.user() or self.site.username() != username:
- pywikibot.warning(
- 'API write action by unexpected username {} commenced.\n'
- 'userinfo: {!r}'.format(username, self.site.userinfo))
-
- # Make sure user is logged in
- if self.write:
- pywikibot.debug('Adding user assertion')
- self['assert'] = 'user'
-
- @classmethod
- def create_simple(cls, req_site, **kwargs):
- """Create a new instance using all args except site for the API."""
- # This ONLY support site so that any caller can be sure there will be
- # no conflict with PWB parameters
- # req_site is needed to avoid conflicts with possible site keyword in
- # kwarg until positional-only parameters are supported, see T262926
- # TODO: Use ParamInfo request to determine valid parameters
- if isinstance(kwargs.get('parameters'), dict):
- warn('The request contains already a "parameters" entry which is '
- 'a dict.')
- return cls(site=req_site, parameters=kwargs)
-
- @classmethod
- def _warn_both(cls) -> None:
- """Warn that kwargs mode was used but parameters was set too."""
- warn('Both kwargs and parameters are set in Request.__init__. It '
- 'assumes that "parameters" is actually a parameter of the '
- 'Request and is added to kwargs.', DeprecationWarning, 3)
-
- @classmethod
- def _warn_kwargs(cls) -> None:
- """Warn that kwargs was used instead of parameters."""
- warn('Instead of using kwargs from Request.__init__, parameters '
- 'for the request to the API should be added via the '
- '"parameters" parameter.', DeprecationWarning, 3)
-
- @classmethod
- def clean_kwargs(cls, kwargs: dict) -> dict:
- """
- Convert keyword arguments into new parameters mode.
-
- If there are no other arguments in kwargs apart from the used arguments
- by the class' initializer it'll just return kwargs and otherwise remove
- those which aren't in the initializer and put them in a dict which is
- added as a 'parameters' keyword. It will always create a shallow copy.
-
- :param kwargs: The original keyword arguments which is not modified.
- :return: The normalized keyword arguments.
- """
- if 'expiry' in kwargs and kwargs['expiry'] is None:
- del kwargs['expiry']
-
- args = set()
- for super_cls in inspect.getmro(cls):
- if not super_cls.__name__.endswith('Request'):
- break
- args |= set(inspect.getfullargspec(super_cls.__init__).args)
- else:
- raise ValueError('Request was not a super class of '
- '{!r}'.format(cls))
- args -= {'self'}
- old_kwargs = set(kwargs)
- # all kwargs defined above but not in args indicate 'kwargs' mode
- if old_kwargs - args:
- # Move all kwargs into parameters
- parameters = {name: value for name, value in kwargs.items()
- if name not in args or name == 'parameters'}
- if 'parameters' in parameters:
- cls._warn_both()
- # Copy only arguments and not the parameters
- kwargs = {name: value for name, value in kwargs.items()
- if name in args or name == 'self'}
- kwargs['parameters'] = parameters
- # Make sure that all arguments have remained
- assert(old_kwargs | {'parameters'}
- == set(kwargs) | set(kwargs['parameters']))
- assert(('parameters' in old_kwargs)
- is ('parameters' in kwargs['parameters']))
- cls._warn_kwargs()
- else:
- kwargs = dict(kwargs)
- kwargs.setdefault('parameters', {})
- return kwargs
-
- def _format_value(self, value):
- """
- Format the MediaWiki API request parameter.
-
- Converts from Python datatypes to MediaWiki API parameter values.
-
- Supports:
- * datetime.datetime (using strftime and ISO8601 format)
- * pywikibot.page.BasePage (using title (+namespace; -section))
-
- All other datatypes are converted to string.
- """
- if isinstance(value, datetime.datetime):
- return value.strftime(pywikibot.Timestamp.ISO8601Format)
- if isinstance(value, pywikibot.page.BasePage):
- if value.site != self.site:
- raise RuntimeError(
- 'value.site {!r} is different from Request.site {!r}'
- .format(value.site, self.site))
- return value.title(with_section=False)
- return str(value)
-
- def __getitem__(self, key):
- """Implement dict interface."""
- return self._params[key]
-
- def __setitem__(self, key: str, value) -> None:
- """Set MediaWiki API request parameter.
-
- :param value: param value(s)
- :type value: str in site encoding
- (string types may be a `|`-separated list)
- iterable, where items are converted to string
- with special handling for datetime.datetime to convert it to a
- string using the ISO 8601 format accepted by the MediaWiki API.
- """
- if isinstance(value, bytes):
- value = value.decode(self.site.encoding())
-
- if isinstance(value, str):
- value = value.split('|')
-
- if hasattr(value, 'api_iter'):
- self._params[key] = value
- else:
- try:
- iter(value)
- except TypeError:
- # convert any non-iterable value into a single-element list
- self._params[key] = [value]
- else:
- self._params[key] = list(value)
-
- def __delitem__(self, key) -> None:
- """Implement dict interface."""
- del self._params[key]
-
- def keys(self):
- """Implement dict interface."""
- return list(self._params.keys())
-
- def __contains__(self, key) -> bool:
- """Implement dict interface."""
- return key in self._params
-
- def __iter__(self):
- """Implement dict interface."""
- return iter(self._params)
-
- def __len__(self) -> int:
- """Implement dict interface."""
- return len(self._params)
-
- def iteritems(self):
- """Implement dict interface."""
- return iter(self._params.items())
-
- def items(self):
- """Return a list of tuples containing the parameters in any order."""
- return list(self._params.items())
-
- def _add_defaults(self):
- """
- Add default parameters to the API request.
-
- This method will only add them once.
- """
- if hasattr(self, '__defaulted'):
- return
-
- if self.mime is not None \
- and set(self._params.keys()) & set(self.mime.keys()):
- raise ValueError('The mime and params shall not share the '
- 'same keys.')
-
- if self.action == 'query':
- meta = self._params.get('meta', [])
- # Special logic for private wikis (T153903).
- # If the wiki requires login privileges to read articles, pywikibot
- # will be blocked from accessing the userinfo.
- # Work around this by requiring userinfo only if 'tokens' and
- # 'login' are not both set.
- typep = self._params.get('type', [])
- if not ('tokens' in meta and 'login' in typep):
- if 'userinfo' not in meta:
- meta = set(meta + ['userinfo'])
- self['meta'] = sorted(meta)
- uiprop = self._params.get('uiprop', [])
- uiprop = set(uiprop + ['blockinfo', 'hasmsg'])
- self['uiprop'] = sorted(uiprop)
- if 'prop' in self._params \
- and self.site.has_extension('ProofreadPage'):
- prop = set(self['prop'] + ['proofread'])
- self['prop'] = sorted(prop)
- # When neither 'continue' nor 'rawcontinue' is present and the
- # version number is at least 1.25wmf5 we add a dummy rawcontinue
- # parameter. Querying siteinfo is save as it adds 'continue'
- # except for 'tokens' (T284577)
- if ('tokens' not in meta and 'continue' not in self._params
- and self.site.mw_version >= '1.25wmf5'):
- self._params.setdefault('rawcontinue', [''])
- elif self.action == 'help' and self.site.mw_version > '1.24':
- self['wrap'] = ''
-
- if config.maxlag:
- self._params.setdefault('maxlag', [str(config.maxlag)])
- self._params.setdefault('format', ['json'])
- if self['format'] != ['json']:
- raise TypeError(
- "Query format '{}' cannot be parsed.".format(self['format']))
-
- self.__defaulted = True # skipcq: PTC-W0037
-
- def _encoded_items(self) -> Dict[str, Union[str, bytes]]:
- """
- Build a dict of params with minimal encoding needed for the site.
-
- This helper method only prepares params for serialisation or
- transmission, so it only encodes values which are not ASCII,
- requiring callers to consider how to handle ASCII vs other values,
- however the output is designed to enable __str__ and __repr__ to
- do the right thing in most circumstances.
-
- Servers which use an encoding that is not a superset of ASCII
- are not supported.
-
- :return: Parameters either in the site encoding, or ASCII strings
- """
- params = {}
- for key, values in self._params.items():
- try:
- iterator = values.api_iter()
- except AttributeError:
- if len(values) == 1:
- value = values[0]
- if value is True:
- values = ['']
- elif value is False or value is None:
- # False and None are not included in the http URI
- continue
- iterator = iter(values)
- value = '|'.join(self._format_value(value) for value in iterator)
- # If the value is encodable as ascii, do not encode it.
- # This means that any value which can be encoded as ascii
- # is presumed to be ascii, and servers using a site encoding
- # which is not a superset of ascii may be problematic.
- try:
- value.encode('ascii')
- except UnicodeError:
- try:
- value = value.encode(self.site.encoding())
- except Exception:
- pywikibot.error(
- "_encoded_items: '{}' could not be encoded as '{}':"
- ' {!r}'.format(key, self.site.encoding(), value))
- assert key.encode('ascii')
- assert isinstance(key, str)
- params[key] = value
- return params
-
- def _http_param_string(self):
- """
- Return the parameters as a HTTP URL query fragment.
-
- URL encodes the parameters provided by _encoded_items()
-
- .. note:: Not all parameters are sorted, therefore for two given
- CachedRequest objects with equal _params, the result of
- _http_param_string() is not necessarily equal.
- """
- return encode_url(self._encoded_items())
-
- def __str__(self) -> str:
- """Return a string representation."""
- return unquote(self.site.scriptpath()
- + '/api.php?'
- + self._http_param_string())
-
- def __repr__(self) -> str:
- """Return internal representation."""
- return '{}.{}<{}->{!r}>'.format(self.__class__.__module__,
- self.__class__.__name__,
- self.site, str(self))
-
- def _simulate(self, action):
- """Simulate action."""
- if action and config.simulate and (
- self.write or action in config.actions_to_block):
- pywikibot.output('<<black;yellow>>SIMULATION: {} action blocked.'
- '<<default>>'.format(action))
- # for more realistic simulation
- if config.simulate is not True:
- pywikibot.sleep(float(config.simulate))
- return {
- action: {'result': 'Success', 'nochange': ''},
-
- # wikibase results
- 'entity': {'lastrevid': -1},
- 'pageinfo': {'lastrevid': -1},
- 'reference': {'hash': -1},
- }
- return None
-
- def _is_wikibase_error_retryable(self, error):
- # dict of error message and current action.
- # Value is True if action type is to be ignored
- err_msg = {
- 'edit-already-exists': 'wbeditentity',
- 'actionthrottledtext': True, # T192912, T268645
- }
- messages = error.get('messages')
- message = None
- # bug T68619; after Wikibase breaking change 1ca9cee we have a
- # list of messages
- if isinstance(messages, list):
- for item in messages:
- message = item['name']
- action = err_msg.get(message)
- if action is True or action == self.action:
- return True
-
- return False
-
- if isinstance(messages, dict):
- try: # behaviour before gerrit 124323 breaking change
- message = messages['0']['name']
- except KeyError: # unsure the new output is always a list
- message = messages['name']
- action = err_msg.get(message)
- return action is True or action == self.action
-
- @staticmethod
- def _generate_mime_part(key, content, keytype=None, headers=None):
- if not keytype:
- try:
- content.encode('ascii')
- keytype = ('text', 'plain')
- except (UnicodeError, AttributeError):
- keytype = ('application', 'octet-stream')
- submsg = MIMENonMultipart(*keytype)
- content_headers = {'name': key}
- if headers:
- content_headers.update(headers)
- submsg.add_header('Content-disposition', 'form-data',
- **content_headers)
-
- if keytype != ('text', 'plain'):
- submsg['Content-Transfer-Encoding'] = 'binary'
-
- submsg.set_payload(content)
- return submsg
-
- def _use_get(self):
- """Verify whether 'get' is to be used."""
- if (not config.enable_GET_without_SSL
- and self.site.protocol() != 'https'
- or self.site.is_oauth_token_available()): # T108182 workaround
- use_get = False
- elif self.use_get is None:
- if self.action == 'query':
- # for queries check the query module
- modules = set()
- for mod_type_name in ('list', 'prop', 'generator'):
- modules.update(self._params.get(mod_type_name, []))
- else:
- modules = {self.action}
- if modules:
- self.site._paraminfo.fetch(modules)
- use_get = all('mustbeposted' not in self.site._paraminfo[mod]
- for mod in modules)
- else:
- # If modules is empty, just 'meta' was given, which doesn't
- # require POSTs, and is required for ParamInfo
- use_get = True
- else:
- use_get = self.use_get
- return use_get
-
- @classmethod
- def _build_mime_request(cls, params: dict,
- mime_params: dict) -> Tuple[dict, bytes]:
- """
- Construct a MIME multipart form post.
-
- :param params: HTTP request params
- :param mime_params: HTTP request parts which must be sent in the body
- :type mime_params: dict of (content, keytype, headers) # noqa: DAR103
- :return: HTTP request headers and body
- """
- # construct a MIME message containing all API key/values
- container = MIMEMultipart(_subtype='form-data')
- for key, value in params.items():
- submsg = cls._generate_mime_part(key, value)
- container.attach(submsg)
- for key, value in mime_params.items():
- submsg = cls._generate_mime_part(key, *value)
- container.attach(submsg)
-
- # strip the headers to get the HTTP message body
- body = container.as_bytes()
- marker = b'\n\n' # separates headers from body
- eoh = body.find(marker)
- body = body[eoh + len(marker):]
- # retrieve the headers from the MIME object
- headers = dict(container.items())
- return headers, body
-
- def _get_request_params(self, use_get, paramstring):
- """Get request parameters."""
- uri = self.site.apipath()
- if self.mime is not None:
- (headers, body) = Request._build_mime_request(
- self._encoded_items(), self.mime)
- use_get = False # MIME requests require HTTP POST
- else:
- headers = {'Content-Type': 'application/x-www-form-urlencoded'}
- if (not self.site.maximum_GET_length()
- or self.site.maximum_GET_length() < len(paramstring)):
- use_get = False
- if use_get:
- uri = '{}?{}'.format(uri, paramstring)
- body = None
- else:
- body = paramstring
-
- pywikibot.debug('API request to {} (uses get: {}):\n'
- 'Headers: {!r}\nURI: {!r}\nBody: {!r}'
- .format(self.site, use_get, headers, uri, body))
- return use_get, uri, body, headers
-
- def _http_request(self, use_get: bool, uri: str, data, headers,
- paramstring) -> tuple:
- """Get or post a http request with exception handling.
-
- :return: a tuple containing requests.Response object from
- http.request and use_get value
- """
- try:
- response = http.request(self.site, uri=uri,
- method='GET' if use_get else 'POST',
- data=data, headers=headers)
- except Server504Error:
- pywikibot.log('Caught HTTP 504 error; retrying')
- except Server414Error:
- if use_get:
- pywikibot.log('Caught HTTP 414 error; retrying')
- use_get = False
- else:
- pywikibot.warning('Caught HTTP 414 error, although not '
- 'using GET.')
- raise
- except (ConnectionError, FatalServerError):
- # This error is not going to be fixed by just waiting
- pywikibot.error(traceback.format_exc())
- raise
- # TODO: what other exceptions can occur here?
- except Exception:
- # for any other error on the http request, wait and retry
- pywikibot.error(traceback.format_exc())
- pywikibot.log('{}, {}'.format(uri, paramstring))
- else:
- return response, use_get
- self.wait()
- return None, use_get
-
- def _json_loads(self, response) -> Optional[dict]:
- """Return a dict from requests.Response.
-
- :param response: a requests.Response object
- :type response: requests.Response
- :return: a data dict
- :raises pywikibot.exceptions.APIError: unknown action found
- :raises pywikibot.exceptions.APIError: unknown query result type
- """
- try:
- result = response.json()
- except ValueError:
- # if the result isn't valid JSON, there may be a server
- # problem. Wait a few seconds and try again
- # Show 20 lines of bare text
- text = '\n'.join(removeHTMLParts(response.text).splitlines()[:20])
- msg = """\
-Non-JSON response received from server {site} for url
-{resp.url}
-The server may be down.
-Status code: {resp.status_code}
-
-The text message is:
-{text}
-""".format(site=self.site, resp=response, text=text)
-
- # Do not retry for AutoFamily but raise a SiteDefinitionError
- # Note: family.AutoFamily is a function to create that class
- if self.site.family.__class__.__name__ == 'AutoFamily':
- pywikibot.debug(msg)
- raise SiteDefinitionError('Invalid AutoFamily({!r})'
- .format(self.site.family.domain))
-
- pywikibot.warning(msg)
-
- # there might also be an overflow, so try a smaller limit
- for param in self._params:
- if param.endswith('limit'):
- # param values are stored a list of str
- value = self[param][0]
- if value.isdigit():
- self[param] = [str(int(value) // 2)]
- pywikibot.output('Set {} = {}'
- .format(param, self[param]))
- else:
- return result or {}
- self.wait()
- return None
-
- def _relogin(self, message: str = '') -> None:
- """Force re-login and inform user."""
- pywikibot.error('{}{}Forcing re-login.'.format(message,
- ' ' if message else ''))
- self.site._relogin()
-
- def _userinfo_query(self, result) -> bool:
- """Handle userinfo query."""
- if self.action == 'query' and 'userinfo' in result.get('query', ()):
- # if we get passed userinfo in the query result, we can confirm
- # that we are logged in as the correct user. If this is not the
- # case, force a re-login.
- username = result['query']['userinfo']['name']
- if (self.site.user() is not None and self.site.user() != username
- and self.site._loginstatus != LoginStatus.IN_PROGRESS):
- message = ("Logged in as '{actual}' instead of '{expected}'."
- .format(actual=username, expected=self.site.user()))
- self._relogin(message)
- return True
- return False
-
- def _handle_warnings(self, result: Dict[str, Any]) -> bool:
- """Handle warnings; return True to retry request, False to resume.
-
- .. versionchanged:: 7.2
- Return True to retry the current request and Falso to resume.
- """
- retry = False
- if 'warnings' not in result:
- return retry
-
- for mod, warning in result['warnings'].items():
- if mod == 'info':
- continue
- if '*' in warning:
- text = warning['*']
- elif 'html' in warning:
- # bug T51978
- text = warning['html']['*']
- else:
- pywikibot.warning('API warning ({}) of unknown format: {}'
- .format(mod, warning))
- continue
-
- # multiple warnings are in text separated by a newline
- for single_warning in text.splitlines():
- if (not callable(self._warning_handler)
- or not self._warning_handler(mod, single_warning)):
- handled = self._default_warning_handler(mod,
- single_warning)
- if handled is None:
- pywikibot.warning('API warning ({}): {}'
- .format(mod, single_warning))
- else:
- retry = retry or handled
- return retry
-
- def _default_warning_handler(self, mode: str, msg: str) -> Optional[bool]:
- """A default warning handler to handle specific warnings.
-
- Return True to retry the request, False to resume and None if
- the warning is not handled.
-
- .. versionadded:: 7.2
- """
- warnings = {
- 'purge': ("You've exceeded your rate limit. "
- 'Please wait some time and try again.',
- '_ratelimited', True),
- }
- warning, handler, retry = warnings.get(mode, (None, None, None))
- if handler and msg == warning:
- # Only show the first warning part
- pywikibot.warning(msg.split('.')[0] + '.')
- # call the handler
- getattr(self, handler)()
- return retry
- return None
-
- def _logged_in(self, code) -> bool:
- """Check whether user is logged in.
-
- Older wikis returned an error instead of a warning when the request
- asked for too many values. If we get this error, assume we are not
- logged in (we can't check this because the userinfo data is not
- present) and force a re-login
- """
- if code.endswith('limit'):
- message = 'Received API limit error.'
-
- # If the user assertion failed, we're probably logged out as well.
- elif code == 'assertuserfailed':
- message = 'User assertion failed.'
-
- # Lastly, the purge module requires a POST if used as anonymous user,
- # but we normally send a GET request. If the API tells us the request
- # has to be POSTed, we're probably logged out.
- elif code == 'mustbeposted' and self.action == 'purge':
- message = "Received unexpected 'mustbeposted' error."
-
- else:
- return True
-
- self._relogin(message)
- return False
-
- def _internal_api_error(self, code, error, result) -> bool:
- """Check for internal_api_error_ or readonly and retry.
-
- :raises pywikibot.exceptions.APIMWError: internal_api_error or readonly
- """
- iae = 'internal_api_error_'
- if not (code.startswith(iae) or code == 'readonly'):
- return False
-
- # T154011
- class_name = code if code == 'readonly' else removeprefix(code, iae)
-
- del error['code'] # is added via class_name
- e = pywikibot.exceptions.APIMWError(class_name, **error)
-
- # If the error key is in this table, it is probably a temporary
- # problem, so we will retry the edit.
- # TODO: T154011: 'ReadOnlyError' seems replaced by 'readonly'
- retry = class_name in ['DBConnectionError', # T64974
- 'DBQueryError', # T60158
- 'DBQueryTimeoutError', # T297708
- 'ReadOnlyError', # T61227
- 'readonly', # T154011
- ]
-
- pywikibot.error('Detected MediaWiki API exception {}{}'
- .format(e, '; retrying' if retry else '; raising'))
- param_repr = str(self._params)
- pywikibot.log('MediaWiki exception {} details:\n'
- ' query=\n{}\n'
- ' response=\n{}'
- .format(class_name,
- pprint.pformat(param_repr),
- result))
- if not retry:
- raise e
-
- self.wait()
- return True
-
- def _ratelimited(self) -> None:
- """Handle ratelimited warning."""
- ratelimits = self.site.userinfo['ratelimits']
- delay = None
-
- ratelimit = ratelimits.get(self.action, {})
- # find the lowest wait time for the given action
- for limit in ratelimit.values():
- seconds = limit['seconds']
- hits = limit['hits']
- delay = min(delay or seconds, seconds / hits)
-
- if not delay:
- pywikibot.warning(
- 'No rate limit found for action {}'.format(self.action))
- self.wait(delay)
-
- def _bad_token(self, code) -> bool:
- """Check for bad token."""
- if code != 'badtoken': # Other code not handled here
- return False
-
- if self.site._loginstatus == LoginStatus.IN_PROGRESS:
- pywikibot.log('Login status: {}'
- .format(self.site._loginstatus.name))
- return False
-
- user_tokens = self.site.tokens._tokens[self.site.user()]
- # all token values mapped to their type
- tokens = {token: t_type for t_type, token in user_tokens.items()}
- # determine which tokens are bad
- invalid_param = {name: tokens[param[0]]
- for name, param in self._params.items()
- if len(param) == 1 and param[0] in tokens}
- # doesn't care about the cache so can directly load them
- if invalid_param:
- pywikibot.log(
- 'Bad token error for {}. Tokens for "{}" used in request; '
- 'invalidated them.'
- .format(self.site.user(),
- '", "'.join(sorted(set(invalid_param.values())))))
- # invalidate superior wiki cookies (T224712)
- _invalidate_superior_cookies(self.site.family)
- # request new token(s) instead of invalid
- self.site.tokens.load_tokens(set(invalid_param.values()))
- # fix parameters; lets hope that it doesn't mistake actual
- # parameters as tokens
- for name, t_type in invalid_param.items():
- self[name] = self.site.tokens[t_type]
- return True
-
- # otherwise couldn't find any … weird there is nothing what
- # can be done here because it doesn't know which parameters
- # to fix
- pywikibot.log(
- 'Bad token error for {} but no parameter is using a '
- 'token. Current tokens: {}'
- .format(self.site.user(),
- ', '.join('{}: {}'.format(*e)
- for e in user_tokens.items())))
- return False
-
- def submit(self) -> dict:
- """
- Submit a query and parse the response.
-
- :return: a dict containing data retrieved from api.php
- """
- self._add_defaults()
- use_get = self._use_get()
- retries = 0
- while True:
- paramstring = self._http_param_string()
-
- simulate = self._simulate(self.action)
- if simulate:
- return simulate
-
- if self.throttle:
- self.site.throttle(write=self.write)
- else:
- pywikibot.log(
- "Submitting unthrottled action '{}'.".format(self.action))
-
- use_get, uri, body, headers = self._get_request_params(use_get,
- paramstring)
- response, use_get = self._http_request(use_get, uri, body, headers,
- paramstring)
- if response is None:
- continue
-
- result = self._json_loads(response)
- if result is None:
- continue
-
- if self._userinfo_query(result):
- continue
-
- if self._handle_warnings(result):
- continue
-
- if 'error' not in result:
- return result
-
- error = result['error'].copy()
- for key in result:
- if key in ('error', 'warnings'):
- continue
- assert key not in error
- assert isinstance(result[key], str), \
- 'Unexpected {}: {!r}'.format(key, result[key])
- error[key] = result[key]
-
- if '*' in result['error']:
- # help text returned
- result['error']['help'] = result['error'].pop('*')
- code = result['error'].setdefault('code', 'Unknown')
- info = result['error'].setdefault('info', None)
-
- if not self._logged_in(code):
- continue
-
- if code == 'maxlag':
- retries += 1
- if retries > max(5, pywikibot.config.max_retries):
- break
- pywikibot.log('Pausing due to database lag: ' + info)
-
- try:
- lag = result['error']['lag']
- except KeyError:
- lag = lagpattern.search(info)
- lag = float(lag.group('lag')) if lag else 0.0
-
- self.site.throttle.lag(lag * retries)
- continue
-
- if code == 'help' and self.action == 'help':
- # The help module returns an error result with the complete
- # API information. As this data was requested, return the
- # data instead of raising an exception.
- return {'help': {'mime': 'text/plain',
- 'help': result['error']['help']}}
-
- pywikibot.warning('API error {}: {}'.format(code, info))
- pywikibot.log(' headers=\n{}'.format(response.headers))
-
- if self._internal_api_error(code, error, result):
- continue
-
- # Phab. tickets T48535, T64126, T68494, T68619
- if code == 'failed-save' \
- and self._is_wikibase_error_retryable(result['error']):
- self.wait()
- continue
-
- if code == 'ratelimited':
- self._ratelimited()
- continue
-
- # If readapidenied is returned try to login
- if code == 'readapidenied' \
- and self.site._loginstatus in (LoginStatus.NOT_ATTEMPTED,
- LoginStatus.NOT_LOGGED_IN):
- self.site.login()
- continue
-
- if self._bad_token(code):
- continue
-
- if 'mwoauth-invalid-authorization' in code:
- if 'Nonce already used' in info:
- pywikibot.error(
- 'Retrying failed OAuth authentication for {}: {}'
- .format(self.site, info))
- continue
- raise NoUsernameError('Failed OAuth authentication for {}: {}'
- .format(self.site, info))
- if code == 'cirrussearch-too-busy-error': # T170647
- self.wait()
- continue
-
- if code in ('search-title-disabled', 'search-text-disabled'):
- prefix = 'gsr' if 'gsrsearch' in self._params else 'sr'
- del self._params[prefix + 'what']
- # use intitle: search instead
- if code == 'search-title-disabled' \
- and self.site.has_extension('CirrusSearch'):
- key = prefix + 'search'
- self._params[key] = ['intitle:' + search
- for search in self._params[key]]
- continue
-
- if code == 'urlshortener-blocked': # T244062
- # add additional informations to result['error']
- result['error']['current site'] = self.site
- if self.site.user():
- result['error']['current user'] = self.site.user()
- else: # not logged in; show the IP
- uinfo = self.site.userinfo
- result['error']['current user'] = uinfo['name']
-
- # raise error
- try:
- param_repr = str(self._params)
- pywikibot.log('API Error: query=\n{}'
- .format(pprint.pformat(param_repr)))
- pywikibot.log(' response=\n{}'.format(result))
-
- raise pywikibot.exceptions.APIError(**result['error'])
- except TypeError:
- raise RuntimeError(result)
-
- msg = 'Maximum retries attempted due to maxlag without success.'
- if os.environ.get('PYWIKIBOT_TESTS_RUNNING', '0') == '1':
- import unittest
- raise unittest.SkipTest(msg)
-
- raise MaxlagTimeoutError(msg)
-
- def wait(self, delay=None):
- """Determine how long to wait after a failed request."""
- self.current_retries += 1
- if self.current_retries > self.max_retries:
- raise TimeoutError('Maximum retries attempted without success.')
-
- # double the next wait, but do not exceed config.retry_max seconds
- delay = delay or self.retry_wait
- delay *= 2 ** (self.current_retries - 1)
- delay = min(delay, config.retry_max)
-
- pywikibot.warning('Waiting {:.1f} seconds before retrying.'
- .format(delay))
- pywikibot.sleep(delay)
-
-
-class CachedRequest(Request):
-
- """Cached request."""
-
- def __init__(self, expiry, *args, **kwargs) -> None:
- """Initialize a CachedRequest object.
-
- :param expiry: either a number of days or a datetime.timedelta object
- """
- assert expiry is not None
- super().__init__(*args, **kwargs)
- if not isinstance(expiry, datetime.timedelta):
- expiry = datetime.timedelta(expiry)
- self.expiry = min(expiry, datetime.timedelta(config.API_config_expiry))
- self._data = None
- self._cachetime = None
-
- @classmethod
- def create_simple(cls, req_site, **kwargs):
- """Unsupported as it requires at least two parameters."""
- raise NotImplementedError('CachedRequest cannot be created simply.')
-
- @classmethod
- def _get_cache_dir(cls) -> str:
- """
- Return the base directory path for cache entries.
-
- The directory will be created if it does not already exist.
-
- :return: base directory path for cache entries
- """
- path = os.path.join(config.base_dir,
- 'apicache-py{:d}'.format(PYTHON_VERSION[0]))
- cls._make_dir(path)
- cls._get_cache_dir = classmethod(lambda c: path) # cache the result
- return path
-
- @staticmethod
- def _make_dir(dir_name: str) -> str:
- """Create directory if it does not exist already.
-
- .. versionchanged:: 7.0
- Only `FileExistsError` is ignored but other OS exceptions can
- be still raised
-
- :param dir_name: directory path
- :return: unmodified directory name for test purpose
- """
- os.makedirs(dir_name, exist_ok=True)
- return dir_name
-
- def _uniquedescriptionstr(self) -> str:
- """Return unique description for the cache entry.
-
- If this is modified, please also update
- scripts/maintenance/cache.py to support
- the new key and all previous keys.
- """
- login_status = self.site._loginstatus
-
- if login_status >= LoginStatus.AS_USER:
- # This uses the format of Page.__repr__, without performing
- # config.console_encoding as done by Page.__repr__.
- # The returned value can't be encoded to anything other than
- # ascii otherwise it creates an exception when _create_file_name()
- # tries to encode it as utf-8.
- user_key = 'User(User:{})'.format(self.site.userinfo['name'])
- else:
- user_key = repr(LoginStatus(LoginStatus.NOT_LOGGED_IN))
-
- request_key = repr(sorted(self._encoded_items().items()))
- return '{!r}{}{}'.format(self.site, user_key, request_key)
-
- def _create_file_name(self) -> str:
- """Return a unique ascii identifier for the cache entry."""
- return hashlib.sha256(
- self._uniquedescriptionstr().encode('utf-8')
- ).hexdigest()
-
- def _cachefile_path(self):
- return os.path.join(CachedRequest._get_cache_dir(),
- self._create_file_name())
-
- def _expired(self, dt):
- return dt + self.expiry < datetime.datetime.utcnow()
-
- def _load_cache(self) -> bool:
- """Load cache entry for request, if available.
-
- :return: Whether the request was loaded from the cache
- """
- self._add_defaults()
- try:
- filename = self._cachefile_path()
- with open(filename, 'rb') as f:
- uniquedescr, self._data, self._cachetime = pickle.load(f)
- if uniquedescr != self._uniquedescriptionstr():
- raise RuntimeError('Expected unique description for the cache '
- 'entry is different from file entry.')
- if self._expired(self._cachetime):
- self._data = None
- return False
- pywikibot.debug('{}: cache hit ({}) for API request: {}'
- .format(self.__class__.__name__, filename,
- uniquedescr))
- return True
- except OSError:
- # file not found
- return False
- except Exception as e:
- pywikibot.output('Could not load cache: {!r}'.format(e))
- return False
-
- def _write_cache(self, data) -> None:
- """Write data to self._cachefile_path()."""
- data = (self._uniquedescriptionstr(), data, datetime.datetime.utcnow())
- with open(self._cachefile_path(), 'wb') as f:
- pickle.dump(data, f, protocol=config.pickle_protocol)
-
- def submit(self):
- """Submit cached request."""
- cached_available = self._load_cache()
- if not cached_available:
- self._data = super().submit()
- self._write_cache(self._data)
- else:
- self._handle_warnings(self._data)
- return self._data
-
-
-def encode_url(query) -> str:
- """Encode parameters to pass with a url.
-
- Reorder parameters so that token parameters go last and call wraps
- :py:obj:`urlencode`. Return an HTTP URL query fragment which
- complies with :api:`Edit#Parameters` (See the 'token' bullet.)
-
- :param query: keys and values to be uncoded for passing with a url
- :type query: mapping object or a sequence of two-element tuples
- :return: encoded parameters with token parameters at the end
- """
- if hasattr(query, 'items'):
- query = list(query.items())
-
- # parameters ending on 'token' should go last
- # wpEditToken should go very last
- query.sort(key=lambda x: x[0].lower().endswith('token')
- + (x[0] == 'wpEditToken'))
- return urlencode(query)
diff --git a/pywikibot/data/api/_requests.py b/pywikibot/data/api/_requests.py
index 710eff5..9f779ef 100644
--- a/pywikibot/data/api/_requests.py
+++ b/pywikibot/data/api/_requests.py
@@ -1,4 +1,4 @@
-"""Interface to Mediawiki's api.php."""
+"""Objects representing API requests."""
#
# (C) Pywikibot team, 2007-2022
#
diff --git a/tests/api_tests.py b/tests/api_tests.py
index 5e37368..0a1390c 100755
--- a/tests/api_tests.py
+++ b/tests/api_tests.py
@@ -959,7 +959,7 @@
cached = False
def test_valid_lagpattern(self):
- """Test whether api.lagpattern is valid."""
+ """Test whether api lagpattern is valid."""
mysite = self.get_site()
if ('dbrepllag' not in mysite.siteinfo
or mysite.siteinfo['dbrepllag'][0]['lag'] == -1):
@@ -979,7 +979,7 @@
pass # expected exception from DummyThrottle instance
except APIError as e:
pywikibot.warning(
- 'Wrong api.lagpattern regex, cannot retrieve lag value')
+ 'Wrong api lagpattern regex, cannot retrieve lag value')
raise e
self.assertIsInstance(mythrottle._lagvalue, (int, float))
self.assertGreaterEqual(mythrottle._lagvalue, 0)
@@ -987,7 +987,7 @@
self.assertGreaterEqual(mythrottle.retry_after, 0)
def test_individual_patterns(self):
- """Test api.lagpattern with example patterns."""
+ """Test api lagpattern with example patterns."""
patterns = {
'Waiting for 10.64.32.115: 0.14024019241333 seconds lagged':
0.14024019241333,
@@ -995,7 +995,7 @@
'Waiting for 127.0.0.1: 1.7 seconds lagged': 1.7
}
for info, time in patterns.items():
- lag = api.lagpattern.search(info)
+ lag = api._requests.lagpattern.search(info)
self.assertIsNotNone(lag)
self.assertEqual(float(lag.group('lag')), time)
--
To view, visit https://gerrit.wikimedia.org/r/c/pywikibot/core/+/806500
To unsubscribe, or for help writing mail filters, visit https://gerrit.wikimedia.org/r/settings
Gerrit-Project: pywikibot/core
Gerrit-Branch: master
Gerrit-Change-Id: Ie4f9eef90d27866011ac6edc3035d6dbd13a269f
Gerrit-Change-Number: 806500
Gerrit-PatchSet: 3
Gerrit-Owner: Xqt <info(a)gno.de>
Gerrit-Reviewer: Xqt <info(a)gno.de>
Gerrit-Reviewer: jenkins-bot
Gerrit-MessageType: merged