urltopdf.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create pdf archive of urls from url files\n",
    "# 2018/12/14\n",
    "\n",
    "chromiumbrowser_path = ''\n",
    "#chromiumbrowser_path = \"/usr/lib/chromium-browser/chromium-browser\"\n",
    "#chromiumbrowser_path = \"/usr/lib/chromium/chromium\"\n",
    "\n",
    "import logging, os, tempfile, asyncio, urllib.parse\n",
    "import requests, dns.resolver, validators, pyppeteer\n",
    "\n",
    "\n",
    "def setup_pydrive (client_id = ''): # uses stored secrets, otherwise uses id\n",
    "    # check if secret file exists already, if not create and delete when done\n",
    "    del_client_secrets = False\n",
    "    if not os.path.exists('client_secrets.json'):\n",
    "        if client_id == '':\n",
    "            raise Exception('Need client_secrets.json in cwd or client_id param')\n",
    "        with open('client_secrets.json', 'w') as client_secrets:\n",
    "            client_secrets.write(client_id)\n",
    "        del_client_secrets = True\n",
    "    import pydrive.auth\n",
    "    gauth = pydrive.auth.GoogleAuth()\n",
    "    gauth.CommandLineAuth() # provides link, paste code\n",
    "    if del_client_secrets: os.remove('client_secrets.json')\n",
    "    import pydrive.drive\n",
    "    return pydrive.drive.GoogleDrive(gauth)\n",
    "\n",
    "\n",
    "def move_file (file, dest_id, token): # update file replacing parent id\n",
    "    new_parent = [{\n",
    "        'id': dest_id,\n",
    "        'isRoot': False,\n",
    "        'kind': 'drive#parentReference',\n",
    "        'parentLink':\n",
    "        'https://www.googleapis.com/drive/v2/files/' + dest_id,\n",
    "        'selfLink':\n",
    "        f'https://www.googleapis.com/drive/v2/files/{token}/parents/' +\n",
    "        dest_id } ]\n",
    "    file['parents'] = new_parent\n",
    "    try:\n",
    "        file.Upload()\n",
    "    except Exception as error:\n",
    "        raise Exception(error)\n",
    "\n",
    "\n",
    "def get_file (drive, folder_id): # get files from folder on Google pydrive\n",
    "    param = {}\n",
    "    param['maxResults'] = 1\n",
    "    param['q'] = \"('\" + folder_id + \"' in parents) and (mimeType != 'application/vnd.google-apps.folder') and trashed=false\"\n",
    "    for file_list in drive.ListFile({**param}):\n",
    "        for file in file_list:\n",
    "            yield file\n",
    "\n",
    "\n",
    "def extract_url (url_file): # extract URL from '.url' internet shortcut file\n",
    "    return url_file.GetContentString().splitlines()[1][4:]\n",
    "\n",
    "\n",
    "def check_url_dns (url):\n",
    "    url_resolver = dns.resolver.Resolver()\n",
    "    url_resolver.nameservers = ['8.8.8.8', '8.8.4.4']\n",
    "    try: url_resolver.query(urllib.parse.urlsplit(url).netloc)\n",
    "    except: return False\n",
    "    return True\n",
    "\n",
    "\n",
    "def check_url (url):\n",
    "    mLog.debug('validating url: {}'.format(url))\n",
    "    if not validators.url(url):\n",
    "        mLog.info('url did not validate')\n",
    "        return False\n",
    "    mLog.debug('checking dns')\n",
    "    if not check_url_dns(url):\n",
    "        mLog.info('unable to resolve dns')\n",
    "        return False\n",
    "    mLog.debug('requesting http head')\n",
    "    try:\n",
    "        r = requests.head(url, allow_redirects=True)\n",
    "        r.raise_for_status()\n",
    "    except:\n",
    "        mLog.info('http head request failed', exc_info=True)\n",
    "        return False\n",
    "    else:\n",
    "        mLog.info('no connection error')\n",
    "        if r.ok: \n",
    "            return True\n",
    "        else:\n",
    "            mLog.info('response not ok: {}'.format(r))\n",
    "            return False\n",
    "    raise Exception('check_url - this should be unreachable')\n",
    "\n",
    "\n",
    "def guess_protocol (url):\n",
    "    schemes = ['https:', 'http:']\n",
    "    for scheme in schemes:\n",
    "        mLog.debug('trying protocol: {}'.format(scheme))\n",
    "        if check_url(scheme + url):\n",
    "            return scheme\n",
    "    mLog.info('no schemes passed check_url')\n",
    "    return ''\n",
    "\n",
    "\n",
    "async def fetch_url (url, pdf_out, html_out):\n",
    "    try:\n",
    "        mLog.info('fetching url')\n",
    "        options = {'args': [\n",
    "                '--no-sandbox',\n",
    "                '--disable-setuid-sandbox']}\n",
    "        options['ignoreHTTPSErrors'] = True\n",
    "        options['headless'] = True \n",
    "        if chromiumbrowser_path:\n",
    "            options['executablePath'] = chromiumbrowser_path\n",
    "            mLog.debug('chromium browser executable path set: {}'.format(chromiumbrowser_path))\n",
    "        browser = await pyppeteer.launch( )\n",
    "        page = await browser.newPage()\n",
    "        mLog.info('getting page')\n",
    "        await page.goto(url)\n",
    "        mLog.debug('getting html')\n",
    "        if html_out != None:\n",
    "            html = await page.content()\n",
    "            html_out.write(html.encode('utf-8'))\n",
    "            html_out.seek(0)\n",
    "        mLog.debug('getting pdf')\n",
    "        if pdf_out != None:\n",
    "            pdf = await page.pdf()\n",
    "            pdf_out.write(pdf)\n",
    "            pdf_out.seek(0)\n",
    "        mLog.debug('closing browser')\n",
    "        await browser.close()\n",
    "        return True\n",
    "    except:\n",
    "        mLog.error(error, exc_info=True)\n",
    "        raise Exception(error)\n",
    "\n",
    "\n",
    "async def get_result (url, drive, name, parent_id):\n",
    "    try:\n",
    "        pdf_tmp = tempfile.NamedTemporaryFile(delete=False)\n",
    "        html_tmp = tempfile.NamedTemporaryFile(delete=False)\n",
    "        await fetch_url(url, pdf_tmp, html_tmp)\n",
    "        mLog.debug('saving pdf')\n",
    "        pdf_out = drive.CreateFile({'title': name + '.pdf'})\n",
    "        pdf_out.SetContentFile(pdf_tmp.name)\n",
    "        move_file(pdf_out, vars_input['outs_id'], vars_input['token']) # also Uploads\n",
    "        pdf_tmp.close()\n",
    "        mLog.debug('saving html')\n",
    "        html_out = drive.CreateFile({'title': name + '.html'})\n",
    "        html_out.SetContentFile(html_tmp.name)\n",
    "        move_file(html_out, vars_input['outs_id'], vars_input['token']) # also Uploads\n",
    "        html_tmp.close()\n",
    "        return True\n",
    "    except:\n",
    "        mLog.info('error in get_result', exc_info=True)\n",
    "        return False\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "    vars_input = {\n",
    "        'src_id': '',\n",
    "        'proc_id': '',\n",
    "        'fail_id': '',\n",
    "        'done_id': '',\n",
    "        'outs_id': '',\n",
    "        'token': ''\n",
    "    }\n",
    "    client_id = ''\n",
    "\n",
    "    src = ''\n",
    "    prc = vars_input['proc_id']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "drive = setup_pydrive(client_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}