Usuari:TronaBot/Python/EstatUsuari.py
< Usuari:TronaBot | Python
#! /usr/bin/env python
# -*- coding: utf-8 -*-
import codecs as cs, os, re, sys, unicodedata, time
from platform import system as platfsys
on_win = platfsys().lower() == "windows"
home = on_win and r"E:\\iShare\SugarSync\My Python scripts" \
or "/home/pasqual/public_html/"
for folder in ("pywikilib", "pyuserlib"):
sys.path.append(os.path.join(home, folder))
#pyuserlib
from common import File, Date, ArgumentHandler
from common import has_repeated_string, get_diffs, get_line_diffs
from common import format_string, green, red
#pywikilib
import query as api, wikipedia as pywikilib
def get_users_by_group(group="bot"):
users = []
params = {
"action": "query",
"list": "allusers",
"augroup": group,
"aulimit": "max"
}
offset=True
while offset:
if isinstance(offset, basestring):
params['aufrom']=offset
data = api.GetData(params)
offset=data.has_key("query-continue") and data['query-continue']['allusers']['aufrom'] or None
users += data['query']['allusers']
return [user['name'] for user in users]
def load_users_by_group(group="bot"):
f = File("%ss" % group, pref="estatusuaris")
if f.exists():
return f.load()
else:
users = get_users_by_group(group)
(group)
f.backup(users)
return users
def recent_changes(beat):
last_ts = Date(Date().substract(seconds=beat+10)).to_api()
print last_ts
params = {
"action": "query",
"list": ["recentchanges", "logevents"],
"rctype": ["new", "edit"],
"rcprop": ["user","timestamp"],
"rcend": last_ts,
"leprop": ["user","timestamp"],
"leend": last_ts,
}
offset=True
while offset:
if isinstance(offset, dict):
if offset.has_key("logevents"):
params['lestart']=offset['logevents']
if offset.has_key("recentchanges"):
params['rccontinue']=offset['recentchanges']
data = api.GetData(params)
if data.has_key("query-continue"):
offset={}
if data['query-continue'].has_key("recentchanges"):
offset['recentchanges'] = data['query-continue']['recentchanges']['rccontinue']
elif data['query-continue'].has_key("logevents"):
offset['logevents']=data['query-continue']['logevents']['lestart']
rc_events = data['query']['recentchanges'] if data.has_key("recentchanges") else []
rc_events += data['query']['logevents'] if data.has_key("logevents") else []
for rc_evt in rc_events:
yield rc_evt
def make_delta(lapse):
hours, remainder = divmod(lapse, 3600)
minutes, seconds = divmod(remainder, 60)
return (0, hours, minutes, seconds, 0)
class Subsciption(object):
def __init__(self):
self.subscriptors = []
self.page = pywikilib.Page(site, "Usuari:EstatBot/Detector d'estat/Llista d'usuaris")
self.regexp = re.compile(ur"\{\{/Assigna estat\|(?P<user>[^|]+?)\|(?P<time>\d+)(?P<url>\|[^}]+?)?\}\}")
self.available_status = (
"connectat", "desconnectat", "desconegut", "ocupat",
"inactiu", "actiu", "Funcionant", "Aturat"
)
self.beat=60
def get_subscriptors(self):
f = File("estatusuaris-subscriptors")
users = self.regexp.finditer(self.page.get())
for user_data in users:
user, lapse, url = user_data.groups()
if not lapse.isdigit():
lapse=30
else:
lapse=int(lapse)
if user in bots:
lapse_min=1
lapse_max=30
else:
lapse_min=30
lapse_max=120
if lapse < lapse_min or lapse > lapse_max:
lapse=30
self.subscriptors.append((user, int(time.time()), lapse*60, "desconnectat"))
f.backup(self.subscriptors)
return self.subscriptors
def load_subscriptors(self):
if args.force:
self.subscriptors = self.get_subscriptors()
else:
f = File("estatusuaris-subscriptors")
if f.exists():
self.subscriptors = f.load()
if not self.subscriptors:
print "updating..."
self.subscriptors = self.get_subscriptors()
else:
self.subscriptors = self.get_subscriptors()
self.beat = min([user_data[2] for user_data in self.subscriptors])
print self.beat
return self.subscriptors
def get_last_edit(self, users=None):
user_dict={}
if not users:
users = self.users
elif not isinstance(users, list):
users = [users]
for user in users:
params = {
"action": "query",
"list": "usercontribs",
"ucuser": user,
"ucprop": "timestamp",
}
data = api.GetData(params)
if data['query']['usercontribs']:
user = data['query']['usercontribs'][0]
user_dict[user['user']]=user['timestamp']
return user_dict
def start(self):
self.subscriptors = self.load_subscriptors()
self.users = [user_data[0] for user_data in self.subscriptors]
user_info = self.get_last_edit()
for user, timestamp, lapse, status in self.subscriptors:
status_page = pywikilib.Page(site, u"Usuari:%s/Estat" % user)
if not status_page.exists():
self.subscriptors.remove((user,timestamp,lapse,status))
continue
cur_status = status_page.get()
if cur_status not in self.available_status: continue
last_edit = Date(user_info[user]).to_cest()
delta = Date(last_edit).time_delta()
if delta > make_delta(lapse):
status="desconnectat"
if delta[0]>30*6:
status = "desconegut"
else:
status="connectat"
s = "&g" if status=="connectat" else "&r" if status == "desconnectat" else "&p"
if user in bots:
status = status.replace("desconnectat", "Aturat")
status = status.replace("connectat", "Funcionant")
pywikilib.output(format_string(u"&b(:$1:) $2 $3", user, "%s%s" %(s,status), delta[:3]))
if cur_status!=status:
pywikilib.showDiff(cur_status, status)
if args.edit:
status_page.put(status, "Bot: actualitzant l'"
"[[Usuari:EstatBot/Usuari:EstatBot/"
"Detector_d'estat/Llista_d'usuaris|"
"estat de l'usuari]].")
def check_status(self, user):
status_page = pywikilib.Page(site, u"Usuari:%s/Estat" % user)
cur_status = status_page.get()
if cur_status not in self.available_status: return
last_edit = Date(self.get_last_edit(user)).to_cest()
delta = Date(last_edit).time_delta()
if delta > make_delta(lapse):
status="desconnectat"
if delta[0]>30*6:
status = "desconegut"
else:
status="connectat"
s = "&g" if status=="connectat" else "&r" if status == "desconnectat" else "&p"
if user in bots:
status = status.replace("desconnectat", "Aturat")
status = status.replace("connectat", "Funcionant")
pywikilib.output(format_string(u"&b(:$1:) $2 $3", user, "%s%s" %(s, status), delta[:3]))
if cur_status!=status:
pywikilib.showDiff(cur_status, status)
if args.edit:
status_page.put(status, "Bot: actualitzant l'"
"[[Usuari:EstatBot/Usuari:EstatBot/"
"Detector_d'estat/Llista_d'usuaris|"
"estat de l'usuari]].")
def loop(self):
while True:
pywikilib.output(format_string("[&y$1] next time: [&p$2]", time.strftime("%H:%M:%S"), Date().add(seconds=self.beat)))
time.sleep(self.beat)
for rc_evt in recent_changes(self.beat):
if rc_evt['user'] in self.users:
self.check_status(rc_evt['user'])
if __name__ == '__main__':
args = ArgumentHandler()
args.parse_arguments()
site = pywikilib.getSite(fam="wikipedia_e")
bots = load_users_by_group()
sysops = load_users_by_group("sysop")
print "bots: %i, sysop: %i" %(len(bots), len(sysops))
subscr = Subsciption()
subscr.start()
subscr.loop()