Wiki script - Dump sorgenti delle guide (Python 3)

Versioni Compatibili Debian 8 "jessie" Debian 9 "stretch" Debian 10 "buster"

Questo semplice script Python 3 si occupa di effettuare il download dei sorgenti delle guide di questo Wiki.
Il file creato si chiamerà "wikidump.txt" (o con qualsivoglia altro nome specificato nella variabile "NOMEFILE" del codice sorgente dello script) e, se si desidera, è possibile modificare il numero di thread (variabile "NUMTHREAD") in base al numero di core/CPU della propria macchina.

Salvare il codice in un file di testo con nome scelto a proprio piacimento e poi eseguirlo. Ad esempio, dopo averlo chiamato "wikidump.py":

$ python3 wikidump.py

Script

#!/usr/bin/python3
# -*- coding: utf8 -*-

import urllib.parse
import urllib.error
import urllib.request
import json
import queue
from threading import Thread
from collections import OrderedDict

URL = 'https://guide.debianizzati.org/api.php'
NOMEFILE = 'wikidump.txt'
NUMTHREAD = 4

# dizionario ordinato
mydict = OrderedDict()


def api(parametri):

        # codifica parametri
        parametri_enc = urllib.parse.urlencode(parametri).encode('utf-8')
        # creazione URL
        url_req = urllib.request.Request(URL, parametri_enc)

        try:
            # dati risposta dal Wiki
            risposta = urllib.request.urlopen(url_req)
            # decodifica risposta
            risposta_dec = risposta.read().decode('utf-8')
            # decodifica in JSON convertendo gli interi in stringhe
            dati_json = json.loads(risposta_dec, parse_int=str)

            return dati_json

        except urllib.error.URLError as e:
            print('Errore di connessione al Wiki')
            print(e)


def get_pid_and_titles():
        # Restituisce un dizionario con i pageid (chiavi) e, come valori,
        # una lista di titolo + contenuto della guida (inizialmente vuoto)
        # esempio:
        # {
        #    'pageid1' : ['titolo1', ''],
        #    'pageid2' : ['titolo2', ''],
        #    'pageid3' : ['titolo3', ''],
        #    ...
        #    ...
        # }

        all_titles_params = {
                            'action': 'query',
                            'list': 'allpages',
                            'aplimit': '500',
                            'apfilterredir': 'nonredirects',
                            'format': 'json',
                            'apfrom': ''
                            }

        while True:
            dati = api(all_titles_params)

            for x in dati['query']['allpages']:
                mydict[x['pageid']] = [x['title'], '']

            # esiste un blocco successivo? Il limite di un blocco è 500.
            try:
                last = dati['continue']['apcontinue']

            except KeyError:
                break

            # se esiste un blocco successivo, allora 'apfrom' assume il valore corrispondente
            # al titolo della prima guida del blocco successivo
            all_titles_params['apfrom'] = last


def get_content():
    # accetta in ingresso il dizionario incompleto (senza contenuto delle guide)
    # e restituisce il dizionario completo, cioè qualcosa del tipo:
    # {
    #    'pageid1' : ['titolo1', 'contenuto1'],
    #    'pageid2' : ['titolo2', 'contenuto2'],
    #    'pageid3' : ['titolo3', 'contenuto3'],
    #    ...
    #    ...
    # }

    # parametri
    content_ts_params = {
                        'action': 'query',
                        'prop': 'revisions',
                        'rvprop': 'content',
                        'pageids': '',
                        'format': 'json'
                        }

    # lista delle chiavi del dizionario (pageids)
    lista_pageid = list(mydict.keys())

    # converte la lista appena ottenuta in una lista di liste (di max 50 valori ciascuna in
    # quanto 50 è il max valore consentito dall'API del Wiki)
    # Qui viene usato il valore 40 per evitare possibili problemi
    listFifty = [lista_pageid[start:start + 40] for start in range(0, len(lista_pageid), 40)]

    q = queue.Queue()

    for listaid in listFifty:
        # inserisce le liste di id in una coda
        q.put(listaid)

    def mythreaded_func():
        while not q.empty():
            # preleva un valore dalla coda
            mystr = q.get()
            # modifica il valore del dizionario con gli id separati da '|'
            content_ts_params['pageids'] = '|'.join(mystr)

            d_reply = api(content_ts_params)

            for _id in mystr:
                mycontent = d_reply['query']['pages'][_id]['revisions'][0]['*']
                mydict[_id][1] = mycontent

            q.task_done()

    for i in range(NUMTHREAD):
        t = Thread(target=mythreaded_func)
        t.start()

    q.join()


print('Download delle guide...')
get_pid_and_titles()
get_content()

print('Scrittura sul file...')
with open(NOMEFILE, 'w', encoding='utf8') as f:
    for k, v in mydict.items():
        # parsing di caratteri non consentiti in URL (i.e. il carattere "+")
        name_parsed = urllib.parse.quote(v[0])

        url = ''.join(('guide.debianizzati.org/index.php?title=', name_parsed, '&action=edit'))
        f.write('\n'.join(('þþþþ', url, v[1], 'øøøø', '')))

print("Script terminato")