piccreate/libs/webscan.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
""" erzeugt Vorschaubilder von Webseiten (Ersatz für CutyCapt + convert)

Eingabe ist eine Datei mit Zeilen der Form  Kundennummer;URL
Für jede Zeile wird die Seite in einem Browser ohne Fenster geladen und der
sichtbare Bereich (Breite x Höhe aus der Sektion [webscan] der Konfiguration)
als <Kundennummer>.<format> gespeichert.
"""

__version__ = '0.1.0'
__license__ = 'BSD'

import asyncio
import configparser
import csv
import optparse
import os
import sys

PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
VENV = os.path.join(PROJECT_DIR, '.venv')

INSTALL_HINTS = {
    'ubuntu': '''Ubuntu / Debian:
    sudo apt install python3-venv
    python3 -m venv {venv}
    {venv}/bin/pip install -r {project}/requirements.txt
    sudo {venv}/bin/playwright install-deps chromium
    {venv}/bin/playwright install chromium''',
    'arch': '''Arch Linux:
    python -m venv {venv}
    {venv}/bin/pip install -r {project}/requirements.txt
  then either the browser from Arch (recommended, Playwright does not
  officially support Arch):
    sudo pacman -S chromium
    export PICCREATE_BROWSER=/usr/bin/chromium
  or the Playwright browser:
    {venv}/bin/playwright install chromium''',
    'windows': r'''Windows (or simply run bin\setup-venv.bat):
    py -m venv {venv}
    {venv}\Scripts\pip install -r {project}\requirements.txt
    {venv}\Scripts\playwright install chromium''',
}

BROWSER_HINTS = {
    'ubuntu': '''Ubuntu / Debian:
    sudo {venv}/bin/playwright install-deps chromium
    {venv}/bin/playwright install chromium''',
    'arch': '''Arch Linux:
    sudo pacman -S chromium
    export PICCREATE_BROWSER=/usr/bin/chromium
  or the Playwright browser:
    {venv}/bin/playwright install chromium''',
    'windows': r'''Windows:
    {venv}\Scripts\playwright install chromium''',
}


def platform_name():
    ''' liefert 'windows', 'ubuntu' oder 'arch' (aus /etc/os-release),
    sonst None '''
    if sys.platform == 'win32':
        return 'windows'
    try:
        with open('/etc/os-release', encoding='utf-8') as fh:
            info = dict(line.rstrip('\n').split('=', 1) for line in fh if '=' in line)
    except OSError:
        return None
    ids = (info.get('ID', '') + ' ' + info.get('ID_LIKE', '')).replace('"', '').split()
    if 'arch' in ids:
        return 'arch'
    if 'ubuntu' in ids or 'debian' in ids:
        return 'ubuntu'
    return None


def print_hint(hints):
    ''' gibt die Installationshinweise für dieses System aus
    (bei unbekanntem Linux die für alle Linux-Distributionen) '''
    distro = platform_name()
    if distro in hints:
        keys = [distro]
    else:
        keys = sorted(k for k in hints if k != 'windows')
    for key in keys:
        print(hints[key].format(venv=VENV, project=PROJECT_DIR), file=sys.stderr)


try:
    from playwright.async_api import async_playwright, Error as PlaywrightError
except ImportError:
    print('Playwright is not installed for', sys.executable, file=sys.stderr)
    print('Install it into the project environment:', file=sys.stderr)
    print_hint(INSTALL_HINTS)
    sys.exit(2)

from utils import CompleteEnding


def read_config(cfg_name, section):
    ''' liest Breite, Höhe und Bildformat aus der Konfiguration '''
    cfg_path = os.path.expandvars(os.path.join('$PICCREATE_CFG',
                                               CompleteEnding(cfg_name, 'cfg')))
    if not os.path.exists(cfg_path):
        raise SystemExit('"' + cfg_path + '" does not exist. Please give the name of a config file!')
    config = configparser.RawConfigParser(strict=False)
    config.read(cfg_path, encoding='utf-8')
    if not config.has_section(section):
        raise SystemExit('section [' + section + '] is missing in ' + cfg_path)
    return {
        'width': config.getint(section, 'width'),
        'height': config.getint(section, 'height'),
        'format': config.get(section, 'format', fallback='jpg'),
    }


def read_urls(infile):
    ''' liest die Paare (Kundennummer, URL) aus der Eingabedatei.
    Eine Kopfzeile wird übersprungen, wenn ihre zweite Spalte keine URL ist.
    '''
    in_path = infile
    if not os.path.exists(in_path):
        in_path = os.path.expandvars(os.path.join('$PICCREATE_IN',
                                                  CompleteEnding(infile, 'csv')))
    if not os.path.exists(in_path):
        raise SystemExit('"' + in_path + '" does not exist. Please give the name to a input file!')

    jobs = []
    with open(in_path, encoding='utf-8', newline='') as fh:
        for n, row in enumerate(csv.reader(fh, delimiter=';')):
            row = [field.strip() for field in row]
            if not row or not row[0] or row[0].startswith('#'):
                continue
            url = row[-1] if len(row) > 1 else ''
            if n == 0 and '.' not in url:
                continue
            if url and '://' not in url:
                url = 'http://' + url
            jobs.append((row[0], url))
    return jobs


async def capture(browser, customer_id, url, out_path, pic_config, timeout):
    ''' lädt eine Seite und speichert den sichtbaren Bereich als Bild '''
    page = await browser.new_page(viewport={'width': pic_config['width'],
                                            'height': pic_config['height']})
    try:
        await page.goto(url, timeout=timeout, wait_until='load')
        kwargs = {'path': out_path}
        if pic_config['format'] in ('jpg', 'jpeg'):
            kwargs['type'] = 'jpeg'
            kwargs['quality'] = 90
        else:
            kwargs['type'] = 'png'
        await page.screenshot(**kwargs)
    finally:
        await page.close()


async def run(jobs, pic_config, options):
    out_dir = options.outdir
    timeout = int(options.timeout) * 1000
    semaphore = asyncio.Semaphore(int(options.jobs))
    failed = []

    async with async_playwright() as p:
        launch_args = {}
        if options.browser:
            if not os.path.isfile(options.browser):
                print('browser', options.browser, 'does not exist.', file=sys.stderr)
                print_hint(BROWSER_HINTS)
                return None
            launch_args['executable_path'] = options.browser
        try:
            browser = await p.chromium.launch(**launch_args)
        except PlaywrightError as e:
            print('could not start the browser:', str(e).splitlines()[0], file=sys.stderr)
            print_hint(BROWSER_HINTS)
            return None

        async def worker(customer_id, url):
            out_path = os.path.join(out_dir, customer_id + '.' + pic_config['format'])
            if not url:
                print(customer_id, 'has no url - skipped', file=sys.stderr)
                failed.append(customer_id)
                return
            if os.path.exists(out_path) and not options.force:
                if options.verbose:
                    print(out_path, 'exists')
                return
            async with semaphore:
                if options.verbose:
                    print('loading', customer_id, url)
                try:
                    await capture(browser, customer_id, url, out_path, pic_config, timeout)
                    print(out_path)
                except PlaywrightError as e:
                    first_line = str(e).splitlines()[0]
                    print(customer_id, url, 'failed:', first_line, file=sys.stderr)
                    failed.append(customer_id)

        await asyncio.gather(*(worker(cid, url) for cid, url in jobs))
        await browser.close()
    return failed


def main():
    if not os.path.isdir(options.outdir):
        print('"' + options.outdir + '" does not exist. Please give the path to the directory for the pictures!')
        return 1
    pic_config = read_config(options.config, options.section)
    if options.format:
        pic_config['format'] = options.format
    jobs = read_urls(options.infile)
    if not jobs:
        print('no urls found in', options.infile)
        return 1
    failed = asyncio.run(run(jobs, pic_config, options))
    if failed is None:
        return 2
    if failed:
        print(len(failed), 'of', len(jobs), 'pages failed:', ', '.join(failed), file=sys.stderr)
        return 1
    return 0


if __name__ == '__main__':
    parser = optparse.OptionParser(usage="""%prog [options]

Example usage
-------------

Screenshots of all urls in in/kunden.csv (lines: customerid;url):
 $ %prog -c webscan -i kunden
""", version='%prog ' + __version__)

    parser.add_option("-c", "--config", "--cfg", dest="config", default="webscan",
                      help="config file with a [webscan] section (width, height, format)",
                      metavar="CONFIG_FILE")
    parser.add_option("--section", dest="section", default="webscan",
                      help="section in the config file. default is webscan")
    parser.add_option("-i", "--infile", dest="infile",
                      help="file with lines customerid;url")
    parser.add_option("-o", "--outdir", dest="outdir", default=os.path.expandvars('$PICCREATE_OUT'),
                      help="directory where the pictures will be saved")
    parser.add_option("--format", dest="format",
                      help="jpg or png, overrides the config")
    parser.add_option("-j", "--jobs", dest="jobs", default=4,
                      help="number of pages loaded in parallel. default is 4")
    parser.add_option("--timeout", dest="timeout", default=20,
                      help="seconds to wait for a page. default is 20")
    parser.add_option("-f", "--force", dest="force", action='store_true',
                      help="create the pictures, no matter if they already exist")
    parser.add_option("--browser", dest="browser",
                      default=os.environ.get('PICCREATE_BROWSER'),
                      help="path to a chromium executable instead of the playwright one "
                           "(default: $PICCREATE_BROWSER)")
    parser.add_option("-v", "--verbose", dest="verbose", action='store_true',
                      help="print more information")

    (options, args) = parser.parse_args()
    if not options.infile:
        parser.error('please give an input file with -i')
    sys.exit(main())

# vim: set tw=80 ft=python ts=4 sw=4: