monitor_websocket_wikidata.py

# -*- coding: utf-8 -*-

# some code from Twisted Matrix's irc_test.py
fromtwisted.words.protocolsimportirc
fromtwisted.internetimportreactor, protocol
fromtwisted.internet.protocolimportReconnectingClientFactory
fromtwisted.web.clientimportgetPage
fromautobahn.websocketimport (WebSocketServerFactory,
WebSocketServerProtocol,
listenWS)
importre
importsocket

importwapiti
fromjsonimportdumps, loads

DEBUG=False

LOCAL_GEOIP='http://localhost:7999'

importlogging
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s\t%(name)s\t %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
bcast_log=logging.getLogger('bcast_log')
irc_log=logging.getLogger('irc_log')
api_log=logging.getLogger('api_log')


DEFAULT_LANG='en'
DEFAULT_PROJECT='wikipedia'
DEFAULT_BCAST_PORT=9000

COLOR_RE=re.compile(r"\x03(?:\d{1,2}(?:,\d{1,2})?)?",
re.UNICODE) # remove IRC color codes
PARSE_EDIT_RE=re.compile(r'(\[\[(?P<page_title>.*?)\]\])'
r' +((?P<flags>[A-Z\!]+) )?'
r'(?P<url>\S*)'
r' +\* (?P<user>.*?)'
r' \* (\((?P<change_size>[\+\-][0-9]+)\))?'
r' ?(?P<summary>.+)?')
HASHTAG_RE=re.compile("(?:^|\s)[＃#]{1}(\w+)", re.UNICODE)

NON_MAIN_NS= ['Talk',
'User',
'User talk',
'Wikipedia',
'Wikipedia talk',
'File',
'File talk',
'MediaWiki',
'MediaWiki talk',
'Template',
'Template talk',
'Help',
'Help talk',
'Category',
'Category talk',
'Portal',
'Portal talk',
'Book',
'Book talk',
'Education Program',
'Education Program talk',
'TimedText',
'TimedText talk',
'Module',
'Module talk',
'Special',
'Media']


defis_ip(addr):
try:
socket.inet_aton(addr)
exceptException:
returnFalse
returnTrue


defprocess_message(message, non_main_ns=NON_MAIN_NS, bcast_callback=None):
no_color=COLOR_RE.sub('', message)
ret=PARSE_EDIT_RE.match(no_color)
msg_dict= {'is_new': False,
'is_bot': False,
'is_unpatrolled': False,
'is_anon': False}
ifret:
msg_dict.update(ret.groupdict())
else:
msg_dict= {}
'''
 Special logs:
 - Special:Log/abusefilter
 - Special:Log/block
 - Special:Log/newusers
 - Special:Log/move
 - Special:Log/pagetriage-curation
 - Special:Log/delete
 - Special:Log/upload
 - Special:Log/patrol
 '''
ns, _, title=msg_dict['page_title'].rpartition(':')
ifnsnotinnon_main_ns:
msg_dict['ns'] ='Main'
else:
msg_dict['ns'] =ns
flags=msg_dict.get('flags') or''

msg_dict['is_new'] ='N'inflags
msg_dict['is_bot'] ='B'inflags
msg_dict['is_minor'] ='M'inflags
msg_dict['is_unpatrolled'] ='!'inflags

username=msg_dict.get('user')
is_anon=is_ip(username)

summary=msg_dict.get('summary')

ifsummary:
# This is missing some of the features from wikimon/parsers.py
msg_dict['hashtags'] =HASHTAG_RE.findall(summary)
else:
msg_dict['hashtags'] = []

defbroadcast(geo_json=None):
ifgeo_jsonisnotNone:
geo_dict=loads(geo_json)
msg_dict['geo_ip'] =geo_dict
bcast_callback(msg_dict)

defreport_failure_broadcast(error):
bcast_log.debug("could not fetch from local geoip: %s", error)
broadcast()

ifis_anon:
msg_dict['is_anon'] =True

ifbcast_callback:
try:
geo_url=str(LOCAL_GEOIP+'/json/'+username)
exceptUnicodeError:
pass
else:
getPage(geo_url).addCallbacks(callback=broadcast,
errback=report_failure_broadcast)
elifbcast_callback:
broadcast()
returnmsg_dict


classMonitor(irc.IRCClient):
def__init__(self, bsf, nmns, factory):
self.broadcaster=bsf
self.non_main_ns=nmns
self.factory=factory
irc_log.info('created IRC monitor...')

defconnectionMade(self):
irc.IRCClient.connectionMade(self)
irc_log.info('connected to IRC server...')

defsignedOn(self):
self.join(self.factory.channel)
irc_log.info('joined %s ...', self.factory.channel)

defprivmsg(self, user, channel, msg):
try:
msg=msg.decode('utf-8')
exceptUnicodeErrorasue:
bcast_log.warn('UnicodeError: %r on IRC message %r', ue, msg)
return
process_message(msg, self.non_main_ns, self._bc_callback)

def_bc_callback(self, msg_dict):
# Which revisions to broadcast?
json_msg_dict=dumps(msg_dict)
self.broadcaster.broadcast(json_msg_dict)


classMonitorFactory(ReconnectingClientFactory):
def__init__(self, channel, bsf, nmns=NON_MAIN_NS):
self.channel=channel
self.bsf=bsf
self.nmns=nmns

defbuildProtocol(self, addr):
irc_log.info('monitor IRC connected to %s', self.channel)
self.resetDelay()
returnMonitor(self.bsf, self.nmns, self)

defstartConnecting(self, connector):
irc_log.info('monitor IRC starting connection to %s', self.channel)
protocol.startConnecting(self, connector)

defclientConnectionLost(self, connector, reason):
irc_log.error('lost monitor IRC connection: %s', reason)
ReconnectingClientFactory.clientConnectionLost(self, connector, reason)

defclientConnectionFailed(self, connector, reason):
irc_log.error('failed monitor IRC connection: %s', reason)
ReconnectingClientFactory.clientConnectionFailed(self, connector, reason)


classBroadcastServerProtocol(WebSocketServerProtocol):
defonOpen(self):
self.factory.register(self)

defonMessage(self, msg, binary):
ifnotbinary:
self.factory.broadcast("'%s' from %s"% (msg, self.peerstr))

defconnectionLost(self, reason):
WebSocketServerProtocol.connectionLost(self, reason)
self.factory.unregister(self)


classBroadcastServerFactory(WebSocketServerFactory):
def__init__(self, url, lang, project, *a, **kw):
WebSocketServerFactory.__init__(self, url, *a, **kw)
self.clients=set()
self.tickcount=0

start_monitor(self, lang, project) # blargh

deftick(self):
self.tickcount+=1
self.broadcast("'tick %d' from server"%self.tickcount)
reactor.callLater(1, self.tick)

defregister(self, client):
ifnotclientinself.clients:
bcast_log.info("registered client %s", client.peerstr)
self.clients.add(client)

defunregister(self, client):
try:
self.clients.remove(client)
bcast_log.info("unregistered client %s", client.peerstr)
exceptKeyError:
pass

defbroadcast(self, msg):
bcast_log.info("broadcasting message %r", msg)
forcinself.clients:
c.sendMessage(msg)
bcast_log.debug("message sent to %s", c.peerstr)


classBroadcastPreparedServerFactory(BroadcastServerFactory):
defbroadcast(self, msg):
preparedMsg=self.prepareMessage(msg)
forcinself.clients:
c.sendPreparedMessage(preparedMsg)
bcast_log.info("prepared message sent to %s", c.peerstr)


defstart_monitor(broadcaster, lang=DEFAULT_LANG, project=DEFAULT_PROJECT):
channel='%s.%s'% (lang, project)
try: 
api_url='http://%s.%s.org/w/api.php'% (lang, project)
api_log.info('fetching namespaces from %r', api_url)
wc=wapiti.WapitiClient('wikimon@hatnote.com', api_url=api_url)
api_log.info('successfully fetched namespaces from %r', api_url)
page_info=wc.get_source_info()
nmns= [ns.titlefornsinpage_info[0].namespace_mapifns.title]
irc_log.info('connecting to %s...', channel)
f=MonitorFactory(channel, broadcaster, nmns)
except:
irc_log.info('connecting to %s without ns', channel)
f=MonitorFactory(channel, broadcaster)
reactor.connectTCP("irc.wikimedia.org", 6667, f)


defcreate_parser():
fromargparseimportArgumentParser
desc="broadcast realtime edits to a Mediawiki project over websockets"
prs=ArgumentParser(description=desc)
prs.add_argument('--project', default=DEFAULT_PROJECT)
prs.add_argument('--lang', default=DEFAULT_LANG)
prs.add_argument('--port', default=DEFAULT_BCAST_PORT, type=int,
help='listen port for websocket connections')
prs.add_argument('--debug', default=DEBUG, action='store_true')
prs.add_argument('--loglevel', default='WARN',
help='e.g., DEBUG, INFO, WARN, etc.')
returnprs


defmain():
parser=create_parser()
args=parser.parse_args()
try:
bcast_log.setLevel(getattr(logging, args.loglevel.upper()))
except:
print'warning: invalid log level'
bcast_log.setLevel(logging.WARN)
ws_listen_addr='ws://localhost:%d'% (args.port,)
ServerFactory=BroadcastServerFactory
factory=ServerFactory(ws_listen_addr,
project=args.project,
lang=args.lang,
debug=DEBUG,
debugCodePaths=DEBUG)
factory.protocol=BroadcastServerProtocol
factory.setProtocolOptions(allowHixie76=True)
listenWS(factory)
reactor.run()

if__name__=='__main__':
main()