#!/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 . gespeichert. """ __version__ = '0.1.0' __license__ = 'BSD' import asyncio import configparser import csv import optparse import os import sys from playwright.async_api import async_playwright, Error as PlaywrightError 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: launch_args['executable_path'] = options.browser browser = await p.chromium.launch(**launch_args) 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: 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", "--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("-t", "--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("-b", "--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: