{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# default_exp core"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# core\n",
    "\n",
    "> API details."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "import json,tweepy,hmac,hashlib,traceback\n",
    "\n",
    "from ipaddress import ip_address,ip_network\n",
    "from http.server import HTTPServer, BaseHTTPRequestHandler\n",
    "from fastcore.imports import *\n",
    "from fastcore.foundation import *\n",
    "from fastcore.utils import *\n",
    "from fastcore.script import *\n",
    "from configparser import ConfigParser"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "def tweet_text(payload):\n",
    "    \"Send a tweet announcing release based on `payload`\"\n",
    "    rel_json = payload['release']\n",
    "    url = rel_json['url']\n",
    "    owner,repo = re.findall(r'https://api.github.com/repos/([^/]+)/([^/]+)/', url)[0]\n",
    "    tweet_tmpl = \"\"\"New #{repo} release: v{tag_name}. {html_url}\n",
    "\n",
    "    {body}\"\"\"\n",
    "    res = tweet_tmpl.format(repo=repo, tag_name=rel_json['tag_name'], html_url=rel_json['html_url'], body=rel_json['body'])\n",
    "    if len(res)<=280: return res\n",
    "    return res[:279] + \"…\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "def check_sig(content, headers, secret):\n",
    "    digest = hmac.new(secret, content, hashlib.sha1).hexdigest()\n",
    "    assert f'sha1={digest}' == headers.get('X-Hub-Signature')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "class _RequestHandler(BaseHTTPRequestHandler):\n",
    "    def _post(self):\n",
    "        if self.server.check_ip:\n",
    "            src_ip = ip_address(self.client_address[0])\n",
    "            assert any((src_ip in wl) for wl in self.server.whitelist)\n",
    "        self.send_response(200)\n",
    "        self.end_headers()\n",
    "        content = self.rfile.read(int(self.headers.get('content-length')))\n",
    "        if self.server.debug:\n",
    "            print(self.headers)\n",
    "            return\n",
    "        payload = json.loads(content.decode())\n",
    "        if payload['action']=='released':\n",
    "            check_sig(content, self.headers, self.server.gh_secret)\n",
    "            tweet = tweet_text(payload)\n",
    "            stat = self.server.api.update_status(tweet)\n",
    "            print(stat.id)\n",
    "        self.wfile.write('ok'.encode(encoding='utf_8'))\n",
    "\n",
    "    def do_POST(self):\n",
    "        try: self._post()\n",
    "        except Exception as e: sys.stderr.write(traceback.format_exc())\n",
    "\n",
    "    def log_message(self, fmt, *args): sys.stderr.write(fmt%args)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "@call_parse\n",
    "def run_server(hostname: Param(\"Host name or IP\", str)='localhost',\n",
    "               port:     Param(\"Port to listen on\", int)=8000,\n",
    "               debug:    Param(\"If True, do not trigger actions, just print\", bool_arg)=False,\n",
    "               inifile:  Param(\"Path to settings ini file\", str)='twitter.ini',\n",
    "               check_ip: Param(\"Check source IP against GitHub list\", bool_arg)=True):\n",
    "    \"Run a GitHub webhook server that tweets about new releases\"\n",
    "    os.environ['PYTHONUNBUFFERED'] = '1'\n",
    "    sys.stdout.reconfigure(line_buffering=True)\n",
    "    sys.stderr.reconfigure(line_buffering=True)\n",
    "    print(f\"Listening on {hostname}:{port}\")\n",
    "    assert os.path.exists(inifile), f\"{inifile} not found\"\n",
    "    cfg = ConfigParser(interpolation=None)\n",
    "    cfg.read([inifile])\n",
    "    cfg = cfg['DEFAULT']\n",
    "    auth = tweepy.OAuthHandler(cfg['consumer_key'], cfg['consumer_secret'])\n",
    "    auth.set_access_token(cfg['access_token'], cfg['access_token_secret'])\n",
    "    with HTTPServer((hostname, port), _RequestHandler) as httpd:\n",
    "        httpd.gh_secret = bytes(cfg['gh_secret'], 'utf-8')\n",
    "        httpd.api = tweepy.API(auth)\n",
    "        httpd.whitelist = L(urljson('https://api.github.com/meta')['hooks']).map(ip_network)\n",
    "        httpd.check_ip,httpd.debug = check_ip,debug\n",
    "        httpd.serve_forever()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# run_server(check_ip=False, debug=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#hide\n",
    "# with HTTPServer(server_address, RequestHandler) as httpd: httpd.handle_request()\n",
    "# rel = Path('release.json').read_text()\n",
    "# wh_json = json.loads(Path('ping.json').read_text())\n",
    "# _api.destroy_status(1311413699366678529);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Installer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "import shutil"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "@call_parse\n",
    "def install_service(hostname: Param(\"Host name or IP\", str)='0.0.0.0',\n",
    "                    port:     Param(\"Port to listen on\", int)=8000,\n",
    "                    inifile:  Param(\"Path to settings ini file\", str)='twitter.ini',\n",
    "                    check_ip: Param(\"Check source IP against GitHub list\", bool_arg)=True,\n",
    "                    service_path: Param(\"Directory to write service file to\", str)=\"/etc/systemd/system/\"):\n",
    "    \"Install fastwebhook as a service\"\n",
    "    script_loc = shutil.which('fastwebhook')\n",
    "    inifile = Path(inifile).absolute()\n",
    "    _unitfile = f\"\"\"[Unit]\n",
    "    Description=fastwebhook\n",
    "    Wants=network-online.target\n",
    "    After=network-online.target\n",
    "\n",
    "    [Service]\n",
    "    ExecStart={script_loc} --inifile {inifile} --check_ip {check_ip} --hostname {hostname} --port {port}\n",
    "    Restart=always\n",
    "\n",
    "    [Install]\n",
    "    WantedBy=multi-user.target\"\"\"\n",
    "    Path(\"fastwebhook.service\").write_text(_unitfile)\n",
    "    run_proc(\"sudo\", \"cp\", \"fastwebhook.service\", service_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Export -"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Converted 00_core.ipynb.\n",
      "Converted index.ipynb.\n"
     ]
    }
   ],
   "source": [
    "#hide\n",
    "from nbdev.export import notebook2script\n",
    "notebook2script()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}