Page MenuHomec4science
Files F77654089

feed-bot.py
No OneTemporary
Actions

File Metadata

Created
Thu, Aug 15, 17:25

feed-bot.py
View Options

#!/usr/bin/env python3
import json, os, ssl, hmac, hashlib
from http.server import BaseHTTPRequestHandler, HTTPServer
from phabricator import Phabricator, APIError
from configparser import ConfigParser
from hypchat import HypChat
import logging
'''
Dependencies:
pip3 install phabricator hypchat configparser
'''
HOST = '0.0.0.0'
PORT = 9000
TXT = 'Nothing to see here'
PROJECT_USERS = ['scitas-all']
CHATROOM = 'Activity'
CERTFILE = '/etc/letsencrypt/live/myserver/combined.pem'
HMAC_KEY = 'abcdef'
def get_users():
proj = phab.project.query(names=PROJECT_USERS)
members = []
for p in proj['data']:
members.extend(proj['data'][p]['members'])
logging.info('Loaded {} members of {}'.format(len(members), PROJECT_USERS))
return members
def connect_chat():
config = ConfigParser()
config.read([os.path.expanduser('~/.hypchat')])
token = config.get('HipChat', 'token')
uri = config.get('HipChat', 'uri')
hc = HypChat(token, endpoint=uri)
logging.info('Connected to {}'.format(uri))
return hc
def send_message(msg, uri):
msg = '<a href="{uri}">{txt}</a>'.format(txt=msg, uri=uri)
room = chat.get_room(CHATROOM)
room.notification(msg, color='gray', format='html')
return msg
class MyHandler(BaseHTTPRequestHandler):
last_message = ''
def do_GET(self):
# HTTP Response
self.send_response(500)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(bytes(TXT, 'UTF-8'))
def do_POST(self):
# Only react to request with JSON data
if 'Content-Length' in self.headers and 'Content-Type' in self.headers:
if self.headers['Content-Type'] == 'application/json':
# Read request data
content_len = int(self.headers['Content-Length'])
content_data = self.rfile.read(content_len)
obj = json.loads(content_data.decode('utf-8'))
logging.info(obj)
# Check HMAC signature
h = hmac.new(
bytes(HMAC_KEY, 'utf-8'),
msg=content_data,
digestmod=hashlib.sha256).hexdigest()
if h == self.headers['X-Phabricator-Webhook-Signature']:
# Check if feed not silenced
if not obj['action']['secure'] and \
not obj['action']['silent']:
# Check transaction author
try:
tphid = [t['phid'] for t in obj['transactions']]
tsearch = phab.transaction.search(
objectIdentifier=obj['object']['phid'],
constraints={'phids': tphid}
)
transactions = [t['phid'] for t in tsearch['data']
if t['authorPHID'] in members]
except APIError as e:
logging.error(e)
transactions = []
# TODO: find a better way of getting stories according to
# transactions.
# Get last Feed stories for modified object
tlen = len(transactions)
slen = tlen if tlen > 1 else 1
stories = phab.feed.query(
filterPHIDs=[obj['object']['phid']],
view="text",
limit=slen)
stories_text = []
for s in stories:
if stories[s]['authorPHID'] in members:
if stories[s]['text'] not in stories_text and \
stories[s]['text'] != self.last_message:
# Get object URI
phid = stories[s]['objectPHID']
o = phab.phid.query(phids=[phid])
msg = send_message(
stories[s]['text'],
o[phid]['uri'])
stories_text.append(stories[s]['text'])
self.last_message = stories[s]['text']
logging.info('Posting: {}'.format(msg))
else:
logging.info('Same story already posted')
else:
logging.info('Author {} not a member of project {}'
.format(
stories[s]['authorPHID'],
PROJECT_USERS))
else:
logging.info('Matching story not found for {}'
.format(obj['object']['phid']))
else:
logging.info('Silenced or Secured object.')
else:
logging.info('HMAC key ({}) verification failed ({})'
.format(
h,
self.headers['X-Phabricator-Webhook-Signature']))
else:
logging.info('Invalid Content Type {}'
.format(self.headers['Content-Type']))
else:
logging.info('Request without content')
# HTTP Response
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(bytes(TXT, 'UTF-8'))
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
# Initialize APIs
phab = Phabricator()
members = get_users()
chat = connect_chat()
# Start web server
httpd = HTTPServer((HOST, PORT), MyHandler)
try:
httpd.socket = ssl.wrap_socket(httpd.socket,
server_side=True,
certfile=CERTFILE,
ssl_version=ssl.PROTOCOL_TLSv1_2)
except FileNotFoundError:
logging.error('Certificate not found. Not using TLS!')
try:
httpd.serve_forever()
except KeyboardInterrupt:
logging.info('Keyboard interruption')
httpd.server_close()

Event Timeline

c4science · Help