Wiki script - Dump sorgenti delle guide (Python 3): differenze tra le versioni
Vai alla navigazione
Vai alla ricerca
S3v (discussione | contributi) (http->https nell'URL) |
S3v (discussione | contributi) (modificato per seguire cambiamenti della nuova API di MW) |
||
Riga 25: | Riga 25: | ||
NOMEFILE = 'wikidump.txt' | NOMEFILE = 'wikidump.txt' | ||
NUMTHREAD = 4 | NUMTHREAD = 4 | ||
# dizionario ordinato | |||
mydict = OrderedDict() | |||
def api(parametri): | def api(parametri): | ||
# codifica parametri | # codifica parametri | ||
parametri_enc = urllib.parse.urlencode(parametri).encode('utf-8') | parametri_enc = urllib.parse.urlencode(parametri).encode('utf-8') | ||
# creazione URL | # creazione URL | ||
url_req = urllib.request.Request(URL, parametri_enc) | url_req = urllib.request.Request(URL, parametri_enc) | ||
try: | try: | ||
# dati risposta dal Wiki | # dati risposta dal Wiki | ||
Riga 42: | Riga 46: | ||
return dati_json | return dati_json | ||
except urllib.error.URLError as e: | except urllib.error.URLError as e: | ||
print('Errore di connessione al Wiki') | print('Errore di connessione al Wiki') | ||
Riga 61: | Riga 65: | ||
all_titles_params = { | all_titles_params = { | ||
'action' : 'query', | 'action': 'query', | ||
'list' : 'allpages', | 'list': 'allpages', | ||
'aplimit' : '500', | 'aplimit': '500', | ||
'apfilterredir' : 'nonredirects', | 'apfilterredir': 'nonredirects', | ||
'format' : 'json', | 'format': 'json', | ||
'apfrom' : '' | 'apfrom': '' | ||
} | } | ||
while True: | while True: | ||
dati = api(all_titles_params) | dati = api(all_titles_params) | ||
Riga 80: | Riga 81: | ||
# esiste un blocco successivo? Il limite di un blocco è 500. | # esiste un blocco successivo? Il limite di un blocco è 500. | ||
try: | try: | ||
last = dati[' | last = dati['continue']['apcontinue'] | ||
except KeyError: | except KeyError: | ||
break | break | ||
# se esiste un blocco successivo, allora 'apfrom' assume il valore corrispondente | # se esiste un blocco successivo, allora 'apfrom' assume il valore corrispondente | ||
# al titolo della prima guida del blocco successivo | # al titolo della prima guida del blocco successivo | ||
all_titles_params['apfrom'] = | all_titles_params['apfrom'] = last | ||
def get_content( | def get_content(): | ||
# accetta in ingresso il dizionario incompleto (senza contenuto delle guide) | # accetta in ingresso il dizionario incompleto (senza contenuto delle guide) | ||
# e restituisce il dizionario completo, cioè qualcosa del tipo: | # e restituisce il dizionario completo, cioè qualcosa del tipo: | ||
Riga 102: | Riga 101: | ||
# ... | # ... | ||
# } | # } | ||
# parametri | # parametri | ||
content_ts_params = { | content_ts_params = { | ||
'action' : 'query', | 'action': 'query', | ||
'prop' : 'revisions', | 'prop': 'revisions', | ||
'rvprop' : 'content', | 'rvprop': 'content', | ||
'pageids' : '', | 'pageids': '', | ||
'format' : 'json' | 'format': 'json' | ||
} | } | ||
# lista delle chiavi del dizionario (pageids) | # lista delle chiavi del dizionario (pageids) | ||
lista_pageid = list( | lista_pageid = list(mydict.keys()) | ||
# converte la lista appena ottenuta in una lista di liste (di max 50 valori ciascuna in | # 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) | # quanto 50 è il max valore consentito dall'API del Wiki) | ||
# Qui viene usato il valore 40 per evitare possibili problemi | # 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)] | listFifty = [lista_pageid[start:start + 40] for start in range(0, len(lista_pageid), 40)] | ||
q = queue.Queue() | q = queue.Queue() | ||
Riga 125: | Riga 124: | ||
# inserisce le liste di id in una coda | # inserisce le liste di id in una coda | ||
q.put(listaid) | q.put(listaid) | ||
def mythreaded_func(): | def mythreaded_func(): | ||
while not q.empty(): | while not q.empty(): | ||
Riga 132: | Riga 131: | ||
# modifica il valore del dizionario con gli id separati da '|' | # modifica il valore del dizionario con gli id separati da '|' | ||
content_ts_params['pageids'] = '|'.join(mystr) | content_ts_params['pageids'] = '|'.join(mystr) | ||
d_reply = api(content_ts_params) | d_reply = api(content_ts_params) | ||
for _id in mystr: | for _id in mystr: | ||
mycontent = d_reply['query']['pages'][_id]['revisions'][0]['*'] | mycontent = d_reply['query']['pages'][_id]['revisions'][0]['*'] | ||
mydict[_id][1] = mycontent | |||
q.task_done() | q.task_done() | ||
for i in range(NUMTHREAD): | for i in range(NUMTHREAD): | ||
t = Thread(target=mythreaded_func) | t = Thread(target=mythreaded_func) | ||
t.start() | t.start() | ||
q.join() | q.join() | ||
print('Download delle guide...') | print('Download delle guide...') | ||
get_pid_and_titles() | |||
get_content() | |||
print('Scrittura sul file...') | print('Scrittura sul file...') | ||
with open(NOMEFILE, 'w', encoding='utf8') as f: | with open(NOMEFILE, 'w', encoding='utf8') as f: | ||
for k,v in mydict.items(): | for k, v in mydict.items(): | ||
# parsing di caratteri non consentiti in URL (i.e. il carattere "+") | # parsing di caratteri non consentiti in URL (i.e. il carattere "+") | ||
name_parsed = urllib.parse.quote(v[0]) | name_parsed = urllib.parse.quote(v[0]) | ||
Riga 172: | Riga 160: | ||
f.write('\n'.join(('þþþþ', url, v[1], 'øøøø', ''))) | f.write('\n'.join(('þþþþ', url, v[1], 'øøøø', ''))) | ||
print("Script terminato") | print("Script terminato") | ||
</pre> | </pre> |
Versione delle 15:02, 27 dic 2020
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")