Source code for scripts.maintenance.cache

#!/usr/bin/env python3
This script runs commands on each entry in the API caches.


    python cache [-password] [-delete] [-c "..."] [-o "..."] [dir ...]

If no directory are specified, it will detect the API caches.

If no command is specified, it will print the filename of all entries.
If only -delete is specified, it will delete all entries.

The following parameters are supported:

-delete           Delete each command filtered. If that option is set the
                  default output will be nothing.

-c                Filter command in python syntax. It must evaluate to True to
                  output anything.

-o                Output command which is output when the filter evaluated to
                  True. If it returns None it won't output anything.


  Print the filename of any entry with 'wikidata' in the key:

    -c "wikidata" in entry._uniquedescriptionstr()

  Customised output if the site code is 'ar':

    -c == "ar"
    -o uniquedesc(entry)

  Or the state of the login:

    -c == LoginStatus.NOT_ATTEMPTED
    -o uniquedesc(entry)

  If the function only uses one parameter for the entry it can be omitted:

    -c has_password
    -o uniquedesc

Available filter commands:


There are helper functions which can be part of a command:

    older_than(entry, interval)
    newer_than(entry, interval)

Available output commands:

# (C) Pywikibot team, 2014-2024
# Distributed under the terms of the MIT license.
from __future__ import annotations

import datetime
import hashlib
import os
import pickle
import sys
from pathlib import Path
from random import sample

import pywikibot
from import api

# The follow attributes are used by eval()
from pywikibot.login import LoginStatus  # noqa: F401
from import User  # noqa: F401
from import APISite, ClosedSite, DataSite  # noqa: F401
from import PYTHON_VERSION

[docs] class ParseError(Exception): """Error parsing."""
[docs] class CacheEntry(api.CachedRequest): """A Request cache entry.""" def __init__(self, directory: str, filename: str): """Initializer.""" = directory self.filename = filename def __str__(self): """Return string equivalent of object.""" return self.filename def __repr__(self): """Representation of object.""" return str(self._cachefile_path()) def _create_file_name(self): """Filename of the cached entry.""" return self.filename def _get_cache_dir(self) -> Path: """Directory of the cached entry. .. versionchanged:: 8.0 return a `pathlib.Path` object. """ return Path( def _cachefile_path(self) -> Path: """Return cache file path. .. versionchanged:: 8.0 return a `pathlib.Path` object. """ return self._get_cache_dir() / self._create_file_name() def _load_cache(self): """Load the cache entry.""" with self._cachefile_path().open('rb') as f: self.key, self._data, self._cachetime = pickle.load(f) return True
[docs] def parse_key(self): """Parse the key loaded from the cache entry.""" # find the start of the first parameter start = self.key.index('(') # find the end of the first object end = self.key.index(')') if not end: raise ParseError(f'End of Site() keyword not found: {self.key}') if 'Site' not in self.key[0:start]: raise ParseError( f'Site() keyword not found at start of key: {self.key}') site = self.key[0:end + 1] if site[0:5] == 'Site(': site = 'APISite(' + site[5:] username = None login_status = None start = end + 1 if self.key[start:start + 5] == 'User(': # The addition of user to the cache key used: # repr(User) # which includes namespaces resulting in: # User(User:<username>) # This also accepts User(<username>) if self.key[start:start + 10] == 'User(User:': start += 10 else: start += 5 end = self.key.index(')', start + 5) if not end: raise ParseError( f'End of User() keyword not found: {self.key}') username = self.key[start:end] elif self.key[start:start + 12] == 'LoginStatus(': end = self.key.index(')', start + 12) if not end: raise ParseError( f'End of LoginStatus() keyword not found: {self.key}') login_status = self.key[start:end + 1] # If the key does not contain User(..) or LoginStatus(..), # it must be the old key format which only contains Site and params elif self.key[start:start + 3] != "[('": raise ParseError( f'Keyword after Site not recognised: {self.key}...') start = end + 1 params = self.key[start:] self._parsed_key = (site, username, login_status, params) return self._parsed_key
def _rebuild(self): """Reconstruct the original Request from the key.""" if hasattr(self, '_parsed_key'): (site, username, login_status, params) = self._parsed_key else: (site, username, login_status, params) = self.parse_key() if not site: raise ParseError('No Site') = eval(site) if login_status: = eval(login_status) if username: = username if not params: raise ParseError('No request params') self._params = {} for key, value in eval(params): if isinstance(value, bytes): value = value.decode( self._params[key] = value.split('|') def _delete(self): """Delete the cache entry.""" self._cachefile_path().unlink()
[docs] def process_entries(cache_path, func, use_accesstime: bool | None = None, output_func=None, action_func=None, *, tests: int | None = None): """Check the contents of the cache. This program tries to use file access times to determine whether cache files are being used. However file access times are not always usable. On many modern filesystems, they have been disabled. On Unix, check the filesystem mount options. You may need to remount with 'strictatime'. .. versionchanged:: 9.0 default cache path to 'apicache' without Python main version. :param use_accesstime: Whether access times should be used. `None` for detect, `False` for don't use and `True` for always use. :param tests: Only process a test sample of files """ if not cache_path: cache_path = os.path.join(pywikibot.config.base_dir, 'apicache') if not os.path.exists(cache_path): pywikibot.error(f'{cache_path}: no such file or directory') return if os.path.isdir(cache_path): filenames = [os.path.join(cache_path, filename) for filename in os.listdir(cache_path)] else: filenames = [cache_path] if tests: filenames = sample(filenames, min(len(filenames), tests)) for filepath in filenames: filename = os.path.basename(filepath) cache_dir = os.path.dirname(filepath) if use_accesstime is not False: stinfo = os.stat(filepath) entry = CacheEntry(cache_dir, filename) # Deletion is chosen only, abbreviate this request if func is None and output_func is None \ and action_func == CacheEntry._delete: action_func(entry) continue # Skip foreign python specific directory *_, version = cache_path.partition('-') if version and version[-1] != str(PYTHON_VERSION[0]): pywikibot.error( "Skipping {} directory, can't read content with python {}" .format(cache_path, PYTHON_VERSION[0])) continue try: entry._load_cache() except ValueError: pywikibot.error(f'Failed loading {entry._cachefile_path()}') pywikibot.exception() continue if use_accesstime is None: stinfo2 = os.stat(filepath) use_accesstime = stinfo.st_atime != stinfo2.st_atime if use_accesstime: # Reset access times to values before loading cache entry. os.utime(filepath, (stinfo.st_atime, stinfo.st_mtime)) entry.stinfo = stinfo try: entry.parse_key() except ParseError as e: pywikibot.error( f'Problems parsing {entry.filename} with key {entry.key}') pywikibot.error(e) continue try: entry._rebuild() except Exception: pywikibot.error('Problems loading {} with key {}, {!r}' .format(entry.filename, entry.key, entry._parsed_key)) pywikibot.exception() continue if func is None or func(entry): if output_func or action_func is None: if output_func is None: output = entry else: output = output_func(entry) if output is not None: if action_func: action_func(entry)
def _parse_command(command, name): """Parse command.""" obj = globals().get(command) if callable(obj): return obj try: return eval('lambda entry: ' + command) except Exception as e: pywikibot.error(e) pywikibot.error( f'Cannot compile {name} command: {command}') return None # Filter commands
[docs] def has_password(entry): """Entry has a password in the entry.""" return entry if 'lgpassword' in entry._uniquedescriptionstr() else None
[docs] def is_logout(entry): """Entry is a logout entry.""" return entry if not entry._data and 'logout' in entry.key else None
[docs] def empty_response(entry): """Entry has no data.""" return entry if not entry._data and 'logout' not in entry.key else None
[docs] def not_accessed(entry): """Entry has never been accessed.""" if not hasattr(entry, 'stinfo'): return None if entry.stinfo.st_atime <= entry.stinfo.st_mtime: return entry return None
[docs] def incorrect_hash(entry): """Incorrect hash.""" if hashlib.sha256(entry.key.encode('utf-8')).hexdigest() != entry.filename: return entry return None
[docs] def older_than(entry, interval): """Find older entries.""" if entry._cachetime.tzinfo is not None \ and entry._cachetime + interval < pywikibot.Timestamp.nowutc(): return entry # old apicache-py2/3 cache if entry._cachetime.tzinfo is None \ and entry._cachetime + interval < pywikibot.Timestamp.utcnow(): return entry return None
[docs] def newer_than(entry, interval): """Find newer entries.""" if entry._cachetime.tzinfo is not None \ and entry._cachetime + interval >= pywikibot.Timestamp.nowutc(): return entry # old apicache-py2/3 cache if entry._cachetime.tzinfo is None \ and entry._cachetime + interval >= pywikibot.Timestamp.utcnow(): return entry return None
[docs] def older_than_one_day(entry): """Find more than one day old entries.""" if older_than(entry, datetime.timedelta(days=1)): return entry return None
[docs] def recent(entry): """Find entries newer than on hour.""" return entry if newer_than(entry, datetime.timedelta(hours=1)) else None
# Output commands
[docs] def uniquedesc(entry): """Return the unique description string.""" return entry._uniquedescriptionstr()
[docs] def parameters(entry): """Return a pretty formatted parameters list.""" lines = '' for key, items in sorted(entry._params.items()): lines += f"{key}={', '.join(items)}\n" return lines
[docs] def main(): """Process command line arguments and invoke bot.""" local_args = pywikibot.handle_args() cache_paths = None delete = False command = None output = None for arg in local_args: if command == '': command = arg elif output == '': output = arg elif arg == '-delete': delete = True elif arg == '-password': command = 'has_password(entry)' elif arg == '-c': if command: sys.exit('Only one command may be executed.') command = '' elif arg == '-o': if output: sys.exit('Only one output may be defined.') output = '' else: if not cache_paths: cache_paths = [arg] else: cache_paths.append(arg) if not cache_paths: folders = ('apicache', 'apicache-py2', 'apicache-py3') cache_paths = list(folders) # Add tests folders cache_paths += [os.path.join('tests', f) for f in folders] # Also process the base directory, if it isn't the current directory if os.path.abspath(os.getcwd()) != pywikibot.config.base_dir: cache_paths += [ os.path.join(pywikibot.config.base_dir, f) for f in folders] # Also process the user home cache, if it isn't the config directory userpath = os.path.expanduser(os.path.join('~', '.pywikibot')) if userpath != pywikibot.config.base_dir: cache_paths += [ os.path.join(userpath, f) for f in folders] if delete: action_func = CacheEntry._delete else: action_func = None if output: output_func = _parse_command(output, 'output') if output_func is None: return else: output_func = None if command: filter_func = _parse_command(command, 'filter') if filter_func is None: return else: filter_func = None for cache_path in cache_paths: if len(cache_paths) > 1:'Processing {cache_path}') process_entries(cache_path, filter_func, output_func=output_func, action_func=action_func)
if __name__ == '__main__': main()