abdicate

#!/usr/bin/env python3

# Copyright © 2014-2024 Jakub Wilk <jwilk@jwilk.net>
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the “Software”), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import argparse
import os
import re
import shlex
import shutil
import subprocess as ipc
import sys
import tempfile
import time
import urllib.parse
import urllib.request

try:
    import readline
except ImportError:
    pass
else:
    del readline

try:
    import PIL.Image
    import PIL.ImageOps
except ImportError:
    PIL = None

try:
    import aalib
except ImportError:
    aalib = None

int(0_0)  # Python >= 3.6 is required

wait_re = re.compile(br'\bwaitAndRedirect[(]"[^"]+", ([0-9]+)[)]')

captcha_image_re = re.compile(
    br"<img src='([\w+.]+[?]PHPSESSID=(\w+))' id='captcha'>"
)

server_error_re = re.compile(r'>\s*Błąd serwisu[.]'.encode('UTF-8'))

class ScrapingError(RuntimeError):
    pass

def show_image_mailcap(path, *, options):
    cmdline = shlex.split('run-mailcap --action=view')
    if options.debug:
        cmdline += ['--debug']
    cmdline += [path]
    ipc.check_call(cmdline)

def clean_image(path):
    image = PIL.Image.open(path)
    image = image.convert('L')
    image = PIL.ImageOps.invert(image)
    bbox = image.getbbox()
    image = image.crop(bbox)
    return image

def show_image_aalib(path, *, options):
    del options
    (twidth, theight) = shutil.get_terminal_size()
    if os.getenv('TERM') == 'linux':
        screen = aalib.LinuxScreen
    else:
        screen = aalib.AnsiScreen
    screen = screen(
        width=twidth,
        height=(theight - 4)
    )
    image = clean_image(path)
    image = image.resize(screen.virtual_size)
    screen.put_image((0, 0), image)
    print(screen.render())

def show_image_chafa(path, *, options):
    image = clean_image(path)
    with tempfile.NamedTemporaryFile(prefix='aero2.', suffix='.png') as imgfile:
        image.save(imgfile.name)
        cmdline = shlex.split('chafa --margin-bottom 4 --stretch')
        cmdline += options.chafa_opts
        cmdline += [imgfile.name]
        if options.debug:
            print('$', *(shlex.quote(arg) for arg in cmdline))
        ipc.check_call(cmdline)

if PIL and shutil.which('chafa'):
    show_image = show_image_chafa
elif PIL and aalib:
    show_image = show_image_aalib
else:
    show_image = show_image_mailcap

def wait(response):
    match = wait_re.search(response)
    if match is None:
        return None
    timeout = int(match.group(1))
    print('waiting', end='')
    sys.stdout.flush()
    for _i in range(timeout):
        time.sleep(1)
        print('.', end='')
        sys.stdout.flush()
    print()
    return True

class UserAgent:

    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:128.0) Gecko/20100101 Firefox/128.0'
    }
    debug_level = 0

    def __init__(self, base_url):
        self.base_url = base_url

    def _build_opener(self):
        # Work-around for <https://github.com/python/cpython/issues/99352>
        # ("urllib.request.urlopen() no longer respects the
        # http.client.HTTPConnection.debuglevel").
        # TODO: Get rid of this once Python < 3.12 is no longer supported.
        handlers = [
            Handler(debuglevel=self.debug_level)
            for Handler in [urllib.request.HTTPHandler, urllib.request.HTTPSHandler]
        ]
        return urllib.request.build_opener(*handlers)

    def _request(self, url, *, method, data=None):
        if isinstance(url, bytes):
            url = url.decode('ASCII')
        url = urllib.parse.urljoin(self.base_url, url)
        if isinstance(data, dict):
            data = urllib.parse.urlencode(data)
        if isinstance(data, str):
            data = data.encode('ASCII')
        request = urllib.request.Request(
            url,
            method=method,
            data=data,
            headers=self.headers,
        )
        opener = self._build_opener()
        with opener.open(request) as fp:
            return fp.read()

    def get(self, url):
        return self._request(url, method='GET')

    def post(self, url, *, data):
        return self._request(url, method='POST', data=data)

# pylint: disable=too-many-statements,too-many-branches
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--debug', action='store_true', help=argparse.SUPPRESS)
    ap.add_argument('--test-image', metavar='FILE', help=argparse.SUPPRESS)
    ap.add_argument('--chafa-opts', metavar='OPTS', default='', help='chafa(1) options')
    ap.add_argument('--copy-to', metavar='DIR', help='copy solved captchas to DIR')
    options = ap.parse_args()
    try:
        options.chafa_opts = shlex.split(options.chafa_opts)
    except ValueError as exc:
        ap.error(f'argument --chafa-opts: {exc}')
    if options.test_image is not None:
        show_image(options.test_image, options=options)
        return
    ua = UserAgent(base_url='http://bdi.free.aero2.net.pl:8080/')
    if options.debug:
        ua.debug_level = 1
    captcha = None
    while True:
        response = ua.post('/', data=dict(viewForm='true'))
        if wait(response):
            break
        match = captcha_image_re.search(response)
        if match is None:
            if server_error_re.search(response):
                print(f'{ap.prog}: internal server error', file=sys.stderr)
                sys.exit(1)
            raise ScrapingError
        url = match.group(1)
        session_id = match.group(2)
        img_response = ua.get(url)
        with tempfile.NamedTemporaryFile(prefix='aero2.', suffix='.jpeg') as img:
            img.write(img_response)
            img.flush()
            while True:
                show_image(img.name, options=options)
                try:
                    captcha = input('captcha: ')
                except EOFError:
                    continue
                else:
                    break
            captcha = str.join(' ', captcha.split())
        if not captcha:
            continue
        response = ua.post('/', data=dict(
            viewForm=True,
            captcha=captcha,
            PHPSESSID=session_id,
        ))
        if wait(response):
            break
    if captcha and options.copy_to:
        if os.path.basename(captcha) != captcha:
            raise RuntimeError
        path = os.path.join(options.copy_to, captcha) + '.jpeg'
        with open(path, 'wb') as img:
            img.write(img_response)

if __name__ == '__main__':
    main()

# vim:ts=4 sts=4 sw=4 et