requests_cache explorer

cache-expl.py

#! /usr/bin/env python3

import sys
import requests_cache
import logging
import argparse
import os
from urllib.parse import urlparse
import datetime
import email.utils


# logger options
if sys.stdout.isatty():
    logging.addLevelName(logging.DEBUG, "\033[0;32m%s\033[0m" % logging.getLevelName(logging.DEBUG))
    logging.addLevelName(logging.INFO, "\033[1;33m%s\033[0m" % logging.getLevelName(logging.INFO))
    logging.addLevelName(logging.WARNING, "\033[1;35m%s\033[1;0m" % logging.getLevelName(logging.WARNING))
    logging.addLevelName(logging.ERROR, "\033[1;41m%s\033[1;0m" % logging.getLevelName(logging.ERROR))


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-v", "--verbose", help="increase verbosity", action='store_true')
    parser.add_argument("-f", "--file", help="cache file", default="cache.sqlite")

    parser.add_argument("-H", "--headers", help="print headers", action='store_true')
    parser.add_argument("--request", help="print request body", action='store_true')
    parser.add_argument("--content", help="print response content", action='store_true')
    parser.add_argument("-x", "--extract", help="extract content, with path like wget -x", action='store_true')

    args = parser.parse_args()

    if args.verbose:
        logging.basicConfig(format='%(asctime)s:%(levelname)s:%(message)s', level=logging.DEBUG, datefmt='%H:%M:%S')
    else:
        logging.basicConfig(format='%(asctime)s:%(levelname)s:%(message)s', level=logging.ERROR, datefmt='%H:%M:%S')

    db = requests_cache.backends.storage.dbdict.DbPickleDict(args.file, 'responses')
    for k, v in db.items():
        data = v[0]
        date = v[1]

        # attr = list(s for s in dir(data) if not re.match(r'^__.*__', s))
        # ['_content', 'cookies', 'encoding', 'headers', 'history', 'raw', 'reason', 'request', 'status_code', 'url']
        # print(attr)

        if args.extract:
            o = urlparse(data.url)
            name = o.netloc + o.path
            if not os.path.exists(name):
                os.makedirs(os.path.dirname(name), exist_ok=True)
                with open(name, 'wb') as f:
                    f.write(getattr(data, '_content'))
                print(name)

                if data.headers.get('last-modified'):
                    d = datetime.datetime(*email.utils.parsedate(data.headers['last-modified'])[:6])
                    ts = d.timestamp()
                    os.utime(name, (ts, ts))
            continue

        print(data.url, date)
        # print(data.status_code, data.reason)

        if args.headers:
            fmt = "{:>%d}: {}" % (2 + max(len(k) for k in data.headers.keys()))
            for k, v in data.headers.items():
                print(fmt.format(k, v))

        # print(data.encoding, len(data._content))
        if args.request:
            print('>>>', data.request.body)
        if args.content:
            print('<<<', getattr(data, '_content'))


if __name__ == '__main__':
    main()


"""
requests_cache.install_cache(
                             allowable_codes=(200,404),
                            fast_save=True,
                             extension='.sqlite',
                            )

requests.get('http://livebox.home/toto')
"""