Revision: 5874
Author: russblau
Date: 2008-09-04 18:35:41 +0000 (Thu, 04 Sep 2008)
Log Message:
-----------
More Site methods, and minor fixes in data.api
Modified Paths:
--------------
branches/rewrite/pywikibot/data/api.py
branches/rewrite/pywikibot/site.py
Modified: branches/rewrite/pywikibot/data/api.py
===================================================================
--- branches/rewrite/pywikibot/data/api.py 2008-09-04 15:30:59 UTC (rev 5873)
+++ branches/rewrite/pywikibot/data/api.py 2008-09-04 18:35:41 UTC (rev 5874)
@@ -282,12 +282,10 @@
except KeyError:
self.site = pywikibot.Site()
# make sure request type is valid, and get limit key if any
- if "generator" in kwargs:
- self.module = kwargs["generator"]
- elif "list" in kwargs:
- self.module = kwargs["list"]
- elif "prop" in kwargs:
- self.module = kwargs["prop"]
+ for modtype in ("generator", "list", "prop", "meta"):
+ if modtype in kwargs:
+ self.module = kwargs[modtype]
+ break
else:
raise Error("%s: No query module name found in arguments."
% self.__class__.__name__)
Modified: branches/rewrite/pywikibot/site.py
===================================================================
--- branches/rewrite/pywikibot/site.py 2008-09-04 15:30:59 UTC (rev 5873)
+++ branches/rewrite/pywikibot/site.py 2008-09-04 18:35:41 UTC (rev 5874)
@@ -350,6 +350,7 @@
15: [u"Category talk"],
}
self.sitelock = threading.Lock()
+ self._msgcache = {}
return
# ANYTHING BELOW THIS POINT IS NOT YET IMPLEMENTED IN __init__()
@@ -417,7 +418,8 @@
- blockinfo: present if user is blocked (dict)
"""
- if not hasattr(self, "_userinfo") or "rights" not in self._userinfo:
+ if not hasattr(self, "_userinfo") or "rights" not in self._userinfo \
+ or self._userinfo['name'] != self.user():
uirequest = api.Request(
site=self,
action="query",
@@ -432,6 +434,79 @@
self._userinfo = uidata['query']['userinfo']
return self._userinfo
+ def is_blocked(self, sysop=False):
+ """Return true if and only if user is blocked.
+
+ @param sysop: If true, log in to sysop account (if available)
+
+ """
+ if not self.logged_in(sysop):
+ self.login(sysop)
+ return 'blockinfo' in self._userinfo
+
+ def isBlocked(self, sysop=False):
+ """Deprecated; retained for backwards-compatibility"""
+ logger.debug("Site.isBlocked() method is deprecated; use is_blocked()")
+ return self.is_blocked(sysop)
+
+ def has_right(self, right, sysop=False):
+ """Return true if and only if the user has a specific right.
+
+ Possible values of 'right' may vary depending on wiki settings,
+ but will usually include:
+
+ * Actions: edit, move, delete, protect, upload
+ * User levels: autoconfirmed, sysop, bot
+
+ """
+ if not self.logged_in(sysop):
+ self.login(sysop)
+ return right.lower() in self._userinfo['rights']
+
+ def isAllowed(self, right, sysop=False):
+ """Deprecated; retained for backwards-compatibility"""
+ logger.debug("Site.isAllowed() method is deprecated; use has_right()")
+ return self.has_right(right, sysop)
+
+ def has_group(self, group, sysop=False):
+ """Return true if and only if the user is a member of specified group.
+
+ Possible values of 'group' may vary depending on wiki settings,
+ but will usually include bot.
+
+ """
+ if not self.logged_in(sysop):
+ self.login(sysop)
+ return group.lower() in self._userinfo['groups']
+
+ def messages(self, sysop=False):
+ """Returns true if the user has new messages, and false otherwise."""
+ if not self.logged_in(sysop):
+ self.login(sysop)
+ return 'hasmsg' in self._userinfo
+
+ def mediawiki_message(self, key):
+ """Return the MediaWiki message text for key "key" """
+ if not key in self._msgcache:
+ msg_query = api.QueryGenerator(site=self, meta="allmessages",
+ amfilter=key)
+ for msg in msg_query:
+ if msg['name'] == key and not 'missing' in msg:
+ self._msgcache[key] = msg['*']
+ break
+ else:
+ raise KeyError("Site %(self)s has no message '%(key)s'"
+ % locals())
+ return self._msgcache[key]
+
+ def has_mediawiki_message(self, key):
+ """Return True iff this site defines a MediaWiki message for 'key'."""
+ try:
+ v = self.mediawiki_message(key)
+ return True
+ except KeyError:
+ return False
+
def getcurrenttimestamp(self):
"""Return (Mediawiki) timestamp, {{CURRENTTIMESTAMP}}, the server time.
@@ -1039,7 +1114,7 @@
logger.debug("allpages: the 'throttle' parameter is deprecated.")
if includeRedirects is not None:
logger.debug(
- "allpages: the 'includeRedirect' parameter is deprecated.")
+ "allpages: the 'includeRedirects' parameter is deprecated.")
if includeRedirects:
if includeRedirects == "only":
filterredirs = True
@@ -1074,6 +1149,17 @@
apgen.request["gapdir"] = "descending"
return apgen
+ def prefixindex(self, prefix, namespace=0, includeredirects=True):
+ """Yield all pages with a given prefix. Deprecated.
+
+ Use allpages() with the prefix= parameter instead of this method.
+
+ """
+ logger.debug("Site.prefixindex() is deprecated; use allpages instead.")
+ return self.allpages(prefix=prefix, namespace=namespace,
+ includeRedirects=includeredirects)
+
+
def alllinks(self, start="!", prefix="", namespace=0, unique=False,
limit=None, fromids=False):
"""Iterate all links to pages (which need not exist) in one namespace.
@@ -2023,34 +2109,7 @@
#### METHODS NOT IMPLEMENTED YET (but may be delegated to Family object) ####
class NotImplementedYet:
- def isBlocked(self, sysop=False):
- """Check if the user is blocked."""
- try:
- text = self.getUrl(u'%saction=query&meta=userinfo&uiprop=blockinfo'
- % self.api_address(), sysop=sysop)
- return text.find('blockedby=') > -1
- except NotImplementedError:
- return False
-
- def isAllowed(self, right, sysop = False):
- """Check if the user has a specific right.
- Among possible rights:
- * Actions: edit, move, delete, protect, upload
- * User levels: autoconfirmed, sysop, bot, empty string (always true)
- """
- if right == '' or right == None:
- return True
- else:
- self._load(sysop = sysop)
- index = self._userIndex(sysop)
- return right in self._rights[index]
-
- def messages(self, sysop = False):
- """Returns true if the user has new messages, and false otherwise."""
- self._load(sysop = sysop)
- index = self._userIndex(sysop)
- return self._messages[index]
-
+ # TODO: is this needed any more? can it be obtained from the http module?
def cookies(self, sysop = False):
"""Return a string containing the user's current cookies."""
self._loadCookies(sysop = sysop)
@@ -2089,447 +2148,6 @@
self._cookies[index] = '; '.join([x.strip() for x in f.readlines()])
f.close()
- def urlEncode(self, query):
- """Encode a query so that it can be sent using an http POST request."""
- if not query:
- return None
- if hasattr(query, 'iteritems'):
- iterator = query.iteritems()
- else:
- iterator = iter(query)
- l = []
- wpEditToken = None
- for key, value in iterator:
- if isinstance(key, unicode):
- key = key.encode('utf-8')
- if isinstance(value, unicode):
- value = value.encode('utf-8')
- key = urllib.quote(key)
- value = urllib.quote(value)
- if key == 'wpEditToken':
- wpEditToken = value
- continue
- l.append(key + '=' + value)
-
- # wpEditToken is explicicmy added as last value.
- # If a premature connection abort occurs while putting, the server will
- # not have received an edit token and thus refuse saving the page
- if wpEditToken != None:
- l.append('wpEditToken=' + wpEditToken)
- return '&'.join(l)
-
- def postForm(self, address, predata, sysop=False, useCookie=True):
- """Post http form data to the given address at this site.
-
- address is the absolute path without hostname.
- predata is a dict or any iterable that can be converted to a dict,
- containing keys and values for the http form.
-
- Return a (response, data) tuple, where response is the HTTP
- response object and data is a Unicode string containing the
- body of the response.
-
- """
- data = self.urlEncode(predata)
- try:
- return self.postData(address, data, sysop=sysop,
- useCookie=useCookie)
- except socket.error, e:
- raise ServerError(e)
-
- def postData(self, address, data,
- contentType='application/x-www-form-urlencoded',
- sysop=False, useCookie=True, compress=True):
- """Post encoded data to the given http address at this site.
-
- address is the absolute path without hostname.
- data is an ASCII string that has been URL-encoded.
-
- Returns a (response, data) tuple where response is the HTTP
- response object and data is a Unicode string containing the
- body of the response.
- """
-
- # TODO: add the authenticate stuff here
-
- if False: #self.persistent_http:
- conn = self.conn
- else:
- # Encode all of this into a HTTP request
- if self.protocol() == 'http':
- conn = httplib.HTTPConnection(self.hostname())
- elif self.protocol() == 'https':
- conn = httplib.HTTPSConnection(self.hostname())
- # otherwise, it will crash, as other protocols are not supported
-
- conn.putrequest('POST', address)
- conn.putheader('Content-Length', str(len(data)))
- conn.putheader('Content-type', contentType)
- conn.putheader('User-agent', useragent)
- if useCookie and self.cookies(sysop = sysop):
- conn.putheader('Cookie', self.cookies(sysop = sysop))
- if False: #self.persistent_http:
- conn.putheader('Connection', 'Keep-Alive')
- if compress:
- conn.putheader('Accept-encoding', 'gzip')
- conn.endheaders()
- conn.send(data)
-
- # Prepare the return values
- # Note that this can raise network exceptions which are not
- # caught here.
- try:
- response = conn.getresponse()
- except httplib.BadStatusLine:
- # Blub.
- conn.close()
- conn.connect()
- return self.postData(address, data, contentType, sysop, useCookie)
-
- data = response.read()
-
- if compress and response.getheader('Content-Encoding') == 'gzip':
- data = decompress_gzip(data)
-
- data = data.decode(self.encoding())
- response.close()
-
- if True: #not self.persistent_http:
- conn.close()
-
- # If a wiki page, get user data
- self._getUserData(data, sysop = sysop)
-
- return response, data
-
- def getUrl(self, path, retry = True, sysop = False, data = None, compress = True):
- """
- Low-level routine to get a URL from the wiki.
-
- Parameters:
- path - The absolute path, without the hostname.
- retry - If True, retries loading the page when a network error
- occurs.
- sysop - If True, the sysop account's cookie will be used.
- data - An optional dict providing extra post request parameters
-
- Returns the HTML text of the page converted to unicode.
- """
- if False: #self.persistent_http and not data:
- self.conn.putrequest('GET', path)
- self.conn.putheader('User-agent', useragent)
- self.conn.putheader('Cookie', self.cookies(sysop = sysop))
- self.conn.putheader('Connection', 'Keep-Alive')
- if compress:
- self.conn.putheader('Accept-encoding', 'gzip')
- self.conn.endheaders()
-
- # Prepare the return values
- # Note that this can raise network exceptions which are not
- # caught here.
- try:
- response = self.conn.getresponse()
- except httplib.BadStatusLine:
- # Blub.
- self.conn.close()
- self.conn.connect()
- return self.getUrl(path, retry, sysop, data, compress)
-
- text = response.read()
- headers = dict(response.getheaders())
-
- else:
- if self.hostname() in config.authenticate.keys():
- uo = authenticateURLopener
- else:
- uo = MyURLopener()
- if self.cookies(sysop = sysop):
- uo.addheader('Cookie', self.cookies(sysop = sysop))
- if compress:
- uo.addheader('Accept-encoding', 'gzip')
-
- url = '%s://%s%s' % (self.protocol(), self.hostname(), path)
- data = self.urlEncode(data)
-
- # Try to retrieve the page until it was successfully loaded (just in
- # case the server is down or overloaded).
- # Wait for retry_idle_time minutes (growing!) between retries.
- retry_idle_time = 1
- retrieved = False
- while not retrieved:
- try:
- if self.hostname() in config.authenticate.keys():
- if False: # compress:
- request = urllib2.Request(url, data)
- request.add_header('Accept-encoding', 'gzip')
- opener = urllib2.build_opener()
- f = opener.open(request)
- else:
- f = urllib2.urlopen(url, data)
- else:
- f = uo.open(url, data)
- retrieved = True
- except KeyboardInterrupt:
- raise
- except Exception, e:
- if retry:
- # We assume that the server is down. Wait some time, then try again.
- output(u"%s" % e)
- output(u"""\
-WARNING: Could not open '%s://%s%s'. Maybe the server or
-your connection is down. Retrying in %i minutes..."""
- % (self.protocol(), self.hostname(), path,
- retry_idle_time))
- time.sleep(retry_idle_time * 60)
- # Next time wait longer, but not longer than half an hour
- retry_idle_time *= 2
- if retry_idle_time > 30:
- retry_idle_time = 30
- else:
- raise
- text = f.read()
-
- headers = f.info()
-
- contentType = headers.get('content-type', '')
- contentEncoding = headers.get('content-encoding', '')
-
- # Ensure that all sent data is received
- if int(headers.get('content-length', '0')) != len(text) and 'content-length' in headers:
- output(u'Warning! len(text) does not match content-length: %s != %s' % \
- (len(text), headers.get('content-length')))
- if False: #self.persistent_http
- self.conn.close()
- self.conn.connect()
- return self.getUrl(path, retry, sysop, data, compress)
-
- if compress and contentEncoding == 'gzip':
- text = decompress_gzip(text)
-
- R = re.compile('charset=([^\'\";]+)')
- m = R.search(contentType)
- if m:
- charset = m.group(1)
- else:
- output(u"WARNING: No character set found.")
- # UTF-8 as default
- charset = 'utf-8'
- # Check if this is the charset we expected
- self.checkCharset(charset)
- # Convert HTML to Unicode
- try:
- text = unicode(text, charset, errors = 'strict')
- except UnicodeDecodeError, e:
- print e
- output(u'ERROR: Invalid characters found on %s://%s%s, replaced by \\ufffd.' % (self.protocol(), self.hostname(), path))
- # We use error='replace' in case of bad encoding.
- text = unicode(text, charset, errors = 'replace')
-
- # If a wiki page, get user data
- self._getUserData(text, sysop = sysop)
-
- return text
-
- def _getUserData(self, text, sysop = False):
- """
- Get the user data from a wiki page data.
-
- Parameters:
- * text - the page text
- * sysop - is the user a sysop?
- """
- if '<div id="globalWrapper">' not in text:
- # Not a wiki page
- return
-
- index = self._userIndex(sysop)
-
- # Check for blocks - but only if version is 1.11 (userinfo is available)
- # and the user data was not yet loaded
- if self.versionnumber() >= 11 and not self._userData[index]:
- blocked = self.isBlocked(sysop = sysop)
- if blocked and not self._isBlocked[index]:
- # Write a warning if not shown earlier
- if sysop:
- account = 'Your sysop account'
- else:
- account = 'Your account'
- output(u'WARNING: %s on %s is blocked. Editing using this account will stop the run.' % (account, self))
- self._isBlocked[index] = blocked
-
- # Check for new messages
- if '<div class="usermessage">' in text:
- if not self._messages[index]:
- # User has *new* messages
- if sysop:
- output(u'NOTE: You have new messages in your sysop account on %s' % self)
- else:
- output(u'NOTE: You have new messages on %s' % self)
- self._messages[index] = True
- else:
- self._messages[index] = False
-
- # Don't perform other checks if the data was already loaded
- if self._userData[index]:
- return
-
- # Search for the the user page link at the top.
- # Note that the link of anonymous users (which doesn't exist at all
- # in Wikimedia sites) has the ID pt-anonuserpage, and thus won't be
- # found here.
- userpageR = re.compile('<li id="pt-userpage"><a href=".+?">(?P<username>.+?)</a></li>')
- m = userpageR.search(text)
- if m:
- self._isLoggedIn[index] = True
- self._userName[index] = m.group('username')
- else:
- self._isLoggedIn[index] = False
- # No idea what is the user name, and it isn't important
- self._userName[index] = None
-
- # Check user groups, if possible (introduced in 1.10)
- groupsR = re.compile(r'var wgUserGroups = \[\"(.+)\"\];')
- m = groupsR.search(text)
- if m:
- rights = m.group(1)
- rights = rights.split('", "')
- if '*' in rights:
- rights.remove('*')
- self._rights[index] = rights
- # Warnings
- # Don't show warnings for not logged in users, they will just fail to
- # do any action
- if self._isLoggedIn[index]:
- if 'bot' not in self._rights[index]:
- if sysop:
- output(u'Note: Your sysop account on %s does not have a bot flag. Its edits will be visible in the recent changes.' % self)
- else:
- output(u'WARNING: Your account on %s does not have a bot flag. Its edits will be visible in the recent changes and it may get blocked.' % self)
- if sysop and 'sysop' not in self._rights[index]:
- output(u'WARNING: Your sysop account on %s does not seem to have sysop rights. You may not be able to perform any sysop-restricted actions using it.' % self)
- else:
- # We don't have wgUserGroups, and can't check the rights
- self._rights[index] = []
- if self._isLoggedIn[index]:
- # Logged in user
- self._rights[index].append('user')
- # Assume bot, and thus autoconfirmed
- self._rights[index].extend(['bot', 'autoconfirmed'])
- if sysop:
- # Assume user reported as a sysop indeed has the sysop rights
- self._rights[index].append('sysop')
- # Assume the user has the default rights
- self._rights[index].extend(['read', 'createaccount', 'edit', 'upload', 'createpage', 'createtalk', 'move', 'upload'])
- if 'bot' in self._rights[index] or 'sysop' in self._rights[index]:
- self._rights[index].append('apihighlimits')
- if 'sysop' in self._rights[index]:
- self._rights[index].extend(['delete', 'undelete', 'block', 'protect', 'import', 'deletedhistory', 'unwatchedpages'])
-
- # Search for a token
- tokenR = re.compile(r"\<input type='hidden' value=\"(.*?)\" name=\"wpEditToken\"")
- tokenloc = tokenR.search(text)
- if tokenloc:
- self._token[index] = tokenloc.group(1)
- if self._rights[index] is not None:
- # In this case, token and rights are loaded - user data is now loaded
- self._userData[index] = True
- else:
- # Token not found
- # Possible reason for this is the user is blocked, don't show a
- # warning in this case, otherwise do show a warning
- # Another possible reason is that the page cannot be edited - ensure
- # there is a textarea and the tab "view source" is not shown
- if u'<textarea' in text and u'<li id="ca-viewsource"' not in text and not self._isBlocked[index]:
- # Token not found
- output(u'WARNING: Token not found on %s. You will not be able to edit any page.' % self)
-
- def mediawiki_message(self, key):
- """Return the MediaWiki message text for key "key" """
- global mwpage, tree
- if key.lower() not in self._mediawiki_messages.keys() \
- and not hasattr(self, "_phploaded"):
- get_throttle()
- mwpage = self.getUrl("%s?title=%s:%s&action=edit"
- % (self.path(), urllib.quote(
- self.namespace(8).replace(' ', '_').encode(
- self.encoding())),
- key))
- tree = BeautifulSoup(mwpage,
- convertEntities=BeautifulSoup.HTML_ENTITIES,
- parseOnlyThese=SoupStrainer("textarea"))
- if tree.textarea is not None and tree.textarea.string is not None:
- value = tree.textarea.string.strip()
- else:
- value = None
- if value:
- self._mediawiki_messages[key.lower()] = value
- else:
- self._mediawiki_messages[key.lower()] = None
- # Fallback in case MediaWiki: page method doesn't work
- if verbose:
- output(
- u"Retrieving mediawiki messages from Special:Allmessages")
- retry_idle_time = 1
- while True:
- get_throttle()
- phppage = self.getUrl(self.get_address("Special:Allmessages")
- + "&ot=php")
- Rphpvals = re.compile(r"(?ms)'([^']*)' => '(.*?[^\\])',")
- count = 0
- for (phpkey, phpval) in Rphpvals.findall(phppage):
- count += 1
- self._mediawiki_messages[str(phpkey).lower()] = phpval
- if count == 0:
- # No messages could be added.
- # We assume that the server is down.
- # Wait some time, then try again.
- output('WARNING: No messages found in Special:Allmessages. Maybe the server is down. Retrying in %i minutes...' % retry_idle_time)
- time.sleep(retry_idle_time * 60)
- # Next time wait longer, but not longer than half an hour
- retry_idle_time *= 2
- if retry_idle_time > 30:
- retry_idle_time = 30
- continue
- break
- self._phploaded = True
-
- key = key.lower()
- if self._mediawiki_messages[key] is None:
- raise KeyError("MediaWiki key '%s' does not exist on %s"
- % (key, self))
- return self._mediawiki_messages[key]
-
- def has_mediawiki_message(self, key):
- """Return True iff this site defines a MediaWiki message for 'key'."""
- try:
- v = self.mediawiki_message(key)
- return True
- except KeyError:
- return False
-
- def _load(self, sysop = False):
- """
- Loads user data.
- This is only done if we didn't do get any page yet and the information
- is requested, otherwise we should already have this data.
-
- Parameters:
- * sysop - Get sysop user data?
- """
- index = self._userIndex(sysop)
- if self._userData[index]:
- return
-
- if verbose:
- output(u'Getting information for site %s' % self)
-
- # Get data
- url = self.edit_address('Non-existing_page')
- text = self.getUrl(url, sysop = sysop)
-
- # Parse data
- self._getUserData(text, sysop = sysop)
-
# TODO: avoid code duplication for the following methods
def newpages(self, number = 10, get_redirect = False, repeat = False):
"""Yield new articles (as Page objects) from Special:Newpages.
@@ -2850,29 +2468,6 @@
if not repeat:
break
- def prefixindex(self, prefix, namespace=0, includeredirects=True):
- """Yield all pages with a given prefix.
-
- Parameters:
- prefix The prefix of the pages.
- namespace Namespace number; defaults to 0.
- MediaWiki software will only return pages in one namespace
- at a time.
-
- If includeredirects is False, redirects will not be found.
- If includeredirects equals the string 'only', only redirects
- will be found. Note that this has not been tested on older
- versions of the MediaWiki code.
-
- It is advised not to use this directly, but to use the
- PrefixingPageGenerator from pagegenerators.py instead.
- """
- for page in self.allpages(start = prefix, namespace = namespace, includeredirects = includeredirects):
- if page.titleWithoutNamespace().startswith(prefix):
- yield page
- else:
- break
-
def linksearch(self, siteurl):
"""Yield Pages from results of Special:Linksearch for 'siteurl'."""
if siteurl.startswith('*.'):
@@ -3062,14 +2657,3 @@
self.namespace(14)+':'+self.family.disambcatname[self.code])
except KeyError:
raise NoPage(u'No page %s.' % page)
-
- def getToken(self, getalways = True, getagain = False, sysop = False):
- index = self._userIndex(sysop)
- if getagain or (getalways and self._token[index] is None):
- output(u'Getting a token.')
- self._load(sysop = sysop)
- if self._token[index] is not None:
- return self._token[index]
- else:
- return False
-
