Source code for scripts.maintenance.unidata

#!/usr/bin/env python3
#
# (C) Pywikibot team, 2018-2026
#
# Distributed under the terms of the MIT license.
#
"""Script to update :mod:`pywikibot.tools._unidata`.

This script is for updating ``_first_upper_exception_dict``.

.. note:: I seems that running under the latest version of Python gives
   a superse of the older version and should be enough. But this is not
   tested completely.

.. version-added:: 8.4
.. version-changed:: 11.3
   wikinews is no longer supported by this script because all wikinews
   sites are closed.
"""
from __future__ import annotations

from json import dump, load
from queue import Queue
from re import findall
from sys import maxunicode
from threading import Thread

from pywikibot import Site
from pywikibot.family import Family
from pywikibot.tools._unidata import _first_upper_exception_dict


NUMBER_OF_THREADS = 26
FILEPATH = '/data/firstup_excepts.json'

# supported families by this script
families_list = [
    'wikibooks',
    'wikipedia',
    'wikiquote',
    'wikisource',
    'wikiversity',
    'wikivoyage',
    'wiktionary',
]






[docs] def process_site(fam_name, site_code): """Process title for a single site.""" site = Site(site_code, fam_name) result = site.simple_request( action='parse', text=wikilinks, contentmodel='wikitext', prop='text' ).submit() parsed_text = result['parse']['text']['*'] titles = findall(r'title="[^:]*:(.)', parsed_text) site_excepts = { orig: title for orig, upper, title in zip(chars, uppers, titles) if upper != title } return site_excepts
[docs] def threads_target(q) -> None: """Thread processing a single site.""" global families_excepts while True: try: fam, code = q.get() except TypeError: # non-iterable NoneType object break site_excepts = process_site(fam, code) families_excepts[fam].setdefault(code, {}).update(site_excepts) q.task_done()
[docs] def spawn_threads(q): """Prepare several threads.""" # TODO: use ThreadList instead threads = [] for _ in range(NUMBER_OF_THREADS): t = Thread(target=threads_target, args=(q,)) t.start() threads.append(t) return threads
[docs] def stop_threads(q, threads) -> None: """Stop threads.""" for _ in range(NUMBER_OF_THREADS): q.put(None) for t in threads: t.join()
[docs] def main() -> None: """Main loop processing sites.""" global families_excepts q = Queue() threads = spawn_threads(q) for fam_name in families_list: family = Family.load(fam_name) families_excepts.setdefault(fam_name, {}) for site_code in family.codes: site = Site(site_code, family) if site.namespaces[8].case != 'first-letter': raise ValueError('MW namespace case is not first-letter') fam_code = (fam_name, site_code) if fam_code in { ('wikisource', 'www'), ('wikisource', 'mul'), ('wikiversity', 'test'), }: continue # the API of these codes does not respond as expected q.put(fam_code) # block until all tasks are done q.join() stop_threads(q, threads)
[docs] def save_json(obj, path) -> None: """Save data to file.""" with open(path, 'w', encoding='utf8') as f: dump(obj, f)
[docs] def load_json(path): """Load data from file.""" try: with open(path, encoding='utf8') as f: return load(f) except OSError: print('File not found:', path) # noqa: T201 return {}
if __name__ == '__main__': chars, uppers, wikilinks = chars_uppers_wikilinks() # save_json({'chars': chars, 'uppers': uppers, 'wikilinks': wikilinks}, # 'user-temp-save.json') # j = load_json('user-temp-save.json') # chars, uppers, wikilinks = j['chars'], j['uppers'], j['wikilinks'] # families_excepts = load_json(FILEPATH) # main() # save_json(families_excepts, FILEPATH) mapping = process_site('wiktionary', 'fr') print(len(mapping), 'entries found') # noqa: T201 print(mapping) # noqa: T201