import requests
import time
import datetime
import sys
S = requests.Session()
URL = "https://ja.wikiwiki.li/api.php"
def getLogs(usr=False, action=False, ns=None):
PARAMS = [{
"action": "query",
"list": "logevents",
"leprop": "title|type|user",
"lelimit": 500,
"format": "json"
}]
if usr:
PARAMS[0]["leuser"] = usr
if action:
PARAMS[0]["leaction"] = action
if ns != None:
PARAMS[0]["lenamespace"] = ns
R = [S.get(url=URL, params=PARAMS[0])]
DATA = [R[0].json()]
output = DATA[0]["query"]["logevents"]
flg = False
if "continue" in DATA[0]:
flg = True
while flg:
PARAMS.append({
"action": "query",
"list": "logevents",
"lecontinue": DATA[-1]["continue"]["lecontinue"],
"leprop": "title|type|user",
"lelimit": 500,
"format": "json"
})
if usr:
PARAMS[-1]["leuser"] = usr
if action:
PARAMS[-1]["leaction"] = action
R.append(S.get(url=URL, params=PARAMS[-1]))
DATA.append(R[-1].json())
output += DATA[-1]["query"]["logevents"]
if "continue" not in DATA[-1]:
flg = False
return output
def isRedir(page):
PARAMS = {
"action": "parse",
"page": page,
"prop": "wikitext",
"format": "json"
}
R = S.get(url=URL, params=PARAMS)
DATA = R.json()
cont = DATA["parse"]["wikitext"]["*"].lower()
if "#転送" in cont or "#redirect" in cont:
return True
else:
return False
def getSize(page):
PARAMS = {
"action": "query",
"prop": "revisions",
"titles": page,
"rvprop": "size",
"format": "json"
}
R = S.get(url=URL, params=PARAMS)
DATA = R.json()
pageid = list(DATA["query"]["pages"])[0]
return DATA["query"]["pages"][pageid]["revisions"][0]["size"]
IPUsers = {
"172.70.223.86": "ひしょう", "60.145.16.169": "芯"
}
createLogs = getLogs(action="create/create", ns=0)
createLogsLen = len(createLogs)
pageDataByUser = {
"Yuito": [], "MediaWiki default": [], "キュアラプラプ": [], "せうゆ": [],
"Mapilaplap": [], "芯": [], "Popbob": [], "Notorious": [], "しんたろう": [], "Long谷": [],
"神座麟": [], "ケツアゴコロロ": [], "ひしょう": [], "いせ": [], "デデ二オン": [],
"MagnoliaWoolery": [], "210.151.113.170": []
}
processed = 0
for lg in createLogs:
creator = lg["user"]
if creator not in pageDataByUser:
if creator in IPUsers:
creator = IPUsers[creator]
else:
raise Exception("An unknown IP user.")
if not isRedir(lg["title"]):
pageDataByUser[creator].append({"title": lg["title"], "size": getSize(lg["title"])})
time.sleep(0.5)
processed += 1
print("\rpageDataByUser: {:.2f}%".format(processed/createLogsLen*100), end="")
byteCnt = {}
for usr, data in pageDataByUser.items():
byteCnt[usr] = []
for page in data:
byteCnt[usr].append(page["size"])
minMaxTtlAvg = {}
for usr, data in byteCnt.items():
if data:
ttl = sum(data)
avg = round(ttl/len(data),2)
sortedData = sorted(data)
minim, maxim = sortedData[0], sortedData[-1]
minMaxTtlAvg[usr] = [
"{:,}".format(minim) + "バイト",
"{:,}".format(maxim) + "バイト",
"{:,}".format(ttl) + "バイト",
"{:,}".format(avg) + "バイト"
]
else:
minMaxTtlAvg[usr] = None
wTable = "{| class=\"wikitable sortable\" style=\"text-align: right;"
wTable += "\"\n! 利用者名 !! style=\"width:7em;\" | 最小値 !! style=\"width:7em;\" | 最大値 !! style=\"width:7em;\" | 合計値 !! style=\"width:7em;\" | 平均値"
for usr, data in minMaxTtlAvg.items():
wTable += "\n|-\n! [[利用者:" + usr + "|" + usr + "]]"
if data:
wTable += "\n| " + data[0] + "\n| " + data[1] + "\n| " + data[2] + "\n| " + data[3]
else:
for i in range(4):
wTable += "\n| style=\"text-align: center;\" | -"
wTable += "\n|}"
now = datetime.datetime.now()
revision = "最終更新日時: " + str(now.year) + "/" + str(now.month) + "/" + str(now.day) + "/" + str(now.hour) + ":" + str(now.minute) + " 頃"
wText = "==利用者別バイト数データ==\n本節では、各々の利用者が作成してきた標準記事（リダイレクトを除く）のバイト数について、その代表値を表示する。\n"
wText += wTable + "\n" + revision
PARAMS_0 = {
"action": "query",
"meta": "tokens",
"type": "login",
"format": "json"
}
R = S.get(url=URL, params=PARAMS_0)
DATA = R.json()
LOGIN_TOKEN = DATA["query"]["tokens"]["logintoken"]
myname = (自分の利用者名を引用符で囲って置換せよ)
mypass = (自分の利用者アカウントのパスワードを引用符で囲って置換せよ)
PARAMS_1 = {
"action": "login",
"lgname": myname,
"lgpassword": mypass,
"lgtoken": LOGIN_TOKEN,
"format": "json"
}
R = S.post(URL, data=PARAMS_1)
DATA = R.json()
if DATA["login"]["result"] == "Failed":
choice = input("\rLogin faild. Still continue with your IP? [y/N]: ").lower()
if choice in ["yes", "ye", "y"]:
pass
elif choice in ["no", "n"]:
print("Processing has been aborted.")
sys.exit()
PARAMS_2 = {
"action": "query",
"meta": "tokens",
"format": "json"
}
R = S.get(url=URL, params=PARAMS_2)
DATA = R.json()
CSRF_TOKEN = DATA["query"]["tokens"]["csrftoken"]
PARAMS_3 = {
"action": "edit",
"title": "麻薬:データ",
"section":1,
"text": wText,
"summary": "利用者別バイト数データを更新",
"token": CSRF_TOKEN,
"format": "json"
}
R = S.post(URL, data=PARAMS_3)
print(R.json())
|