Wiki script - Dump sorgenti delle guide (Python 3): differenze tra le versioni
Vai alla navigazione
Vai alla ricerca
S3v (discussione | contributi) (file scritto con codifica UTF8 e modifiche minori) |
m (syntaxhighlight) |
||
(4 versioni intermedie di un altro utente non mostrate) | |||
Riga 1: | Riga 1: | ||
{{Versioni compatibili|Jessie| | {{Versioni compatibili|Jessie|Stretch|Buster}} | ||
Questo semplice [[script]] Python 3 si occupa di effettuare il download dei sorgenti delle guide di questo Wiki.<br/> | Questo semplice [[script]] Python 3 si occupa di effettuare il download dei sorgenti delle guide di questo Wiki.<br/> | ||
Riga 10: | Riga 10: | ||
== Script == | == Script == | ||
< | <syntaxhighlight lang="python"> | ||
#!/usr/bin/python3 | #!/usr/bin/python3 | ||
# -*- coding: utf8 -*- | |||
import urllib.parse | import urllib.parse | ||
Riga 21: | Riga 22: | ||
from collections import OrderedDict | from collections import OrderedDict | ||
URL = ' | URL = 'https://guide.debianizzati.org/api.php' | ||
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 37: | Riga 42: | ||
# decodifica risposta | # decodifica risposta | ||
risposta_dec = risposta.read().decode('utf-8') | risposta_dec = risposta.read().decode('utf-8') | ||
# decodifica in JSON | # decodifica in JSON convertendo gli interi in stringhe | ||
dati_json = json.loads(risposta_dec, parse_int=str) | dati_json = json.loads(risposta_dec, parse_int=str) | ||
return | 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 60: | 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 79: | 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 101: | 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() | ||
for | for listaid in listFifty: | ||
# inserisce le liste di id in una coda | |||
# inserisce | |||
q.put(listaid) | q.put(listaid) | ||
def mythreaded_func(): | def mythreaded_func(): | ||
while not q.empty(): | while not q.empty(): | ||
# preleva un valore dalla coda | # preleva un valore dalla coda | ||
mystr = q.get() | mystr = q.get() | ||
# modifica il valore del dizionario | # modifica il valore del dizionario con gli id separati da '|' | ||
content_ts_params['pageids'] = mystr | content_ts_params['pageids'] = '|'.join(mystr) | ||
d_reply = api(content_ts_params) | d_reply = api(content_ts_params) | ||
for | for _id in mystr: | ||
mycontent = d_reply['query']['pages'][ | 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]) | ||
url = ''.join(('guide.debianizzati.org/index.php?title=', name_parsed, '&action=edit')) | |||
f.write('\n'.join(('þþþþ', url, v[1], 'øøøø', ''))) | |||
print("Script terminato") | print("Script terminato") | ||
</ | </syntaxhighlight> | ||
[[Categoria: Wiki]] | [[Categoria: Wiki]] | ||
[[Categoria: Python]] | [[Categoria: Python]] |
Versione attuale delle 18:30, 29 mar 2021
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")