# default_exp core
API details.
#hide
from nbdev.showdoc import *
#export
import json,tweepy,hmac,hashlib
from pdb import set_trace
from http.server import HTTPServer, BaseHTTPRequestHandler
from fastcore.imports import *
from fastcore.script import *
from configparser import ConfigParser
#export
_cfg = ConfigParser(interpolation=None)
_cfg.read(['twitter.ini'])
_cfg = _cfg['DEFAULT']
globals().update(**_cfg)
_auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
_auth.set_access_token(access_token, access_token_secret)
_api = tweepy.API(_auth)
#export
def tweet_text(payload):
"Send a tweet announcing release based on `payload`"
rel_json = payload['release']
url = rel_json['url']
owner,repo = re.findall(r'https://api.github.com/repos/([^/]+)/([^/]+)/', url)[0]
tweet_tmpl = """New #{repo} release: v{tag_name}. {html_url}
{body}"""
res = tweet_tmpl.format(repo=repo, tag_name=rel_json['tag_name'], html_url=rel_json['html_url'], body=rel_json['body'])
if len(res)<=280: return res
return res[:279] + "…"
gh_secret = b'something'
#export
def check_secret(content, headers):
digest = hmac.new(gh_secret, content, hashlib.sha1).hexdigest()
assert f'sha1={digest}' == headers.get('X-Hub-Signature')
#export
class _RequestHandler(BaseHTTPRequestHandler):
def do_POST(self):
self.send_response(200)
self.end_headers()
content = self.rfile.read(int(self.headers.get('content-length')))
payload = json.loads(content.decode())
if payload['action']=='released':
check_secret(content, self.headers)
tweet = tweet_text(payload)
stat = _api.update_status(tweet)
print(stat.id)
self.wfile.write('ok'.encode(encoding='utf_8'))
#export
@call_parse
def run_server(hostname: Param("Host name or IP" , str)='localhost',
port: Param("Port to listen on", int)=8000):
"Run a GitHub webhook server that tweets about new releases"
print(f"Listening on {hostname}:{port}")
with HTTPServer((hostname, port), _RequestHandler) as httpd: httpd.serve_forever()
# run_server()
#hide
# with HTTPServer(server_address, RequestHandler) as httpd: httpd.handle_request()
# rel = Path('release.json').read_text()
# wh_json = json.loads(Path('ping.json').read_text())
# _api.destroy_status(1311413699366678529);
#hide
from nbdev.export import notebook2script
notebook2script()
Converted 00_core.ipynb. Converted index.ipynb.