UK – įrankis įkelti Aleph vartotojų duomenims per users.library.lt

aleph_upload_patrons.py

#!/usr/bin/env python3
import argparse
import bs4
import os
import requests
import subprocess
import sys

def validate_xml_file(path, schema):
    r = subprocess.call(["xmllint", "--noout", "--schema", schema, path])
    return (r == 0)

class ServerError(Exception):
    pass

class AlephUploader():
    def __init__(self):
        self.ua = requests.Session()

    def _raise_for_errors(self, body, is_upload):
        page = bs4.BeautifulSoup(body, "lxml")
        errors = page.select(".error")
        if is_upload:
            errors += page.select(".checkResultsTable li")
        errors = [tag.get_text(strip=True) for tag in errors]
        errors = [msg for msg in errors if msg]
        if errors:
            raise ServerError(errors)

    def login(self, username, password):
        print("Prisijungiama vardu %r" % (username,))

        resp = self.ua.post("https://users.library.lt/login.php",
                            data={"_action": "jungtis",
                                  "username": username,
                                  "password": password,
                                  "Jungtis": "Prisijungti"})
        resp.raise_for_status()
        self._raise_for_errors(resp.content, False)
        return True

    def upload(self, file, remote_name=None):
        if not remote_name:
            remote_name = os.path.basename(file)

        print("Įkeliamas failas %r kaip %r" % (file, remote_name))

        with open(file, "rb") as fh:
            resp = self.ua.post("https://users.library.lt/upload.php",
                                files={"_action": (None, "upload"),
                                       "failas": (remote_name, fh, "text/xml")})
        resp.raise_for_status()
        self._raise_for_errors(resp.content, True)
        return True

script_dir = sys.path[0]
schema_file = os.path.join(script_dir, "Aleph.xsd")

parser = argparse.ArgumentParser()
parser.add_argument("-u", "--username",
                    default=os.environ.get("UPLOAD_USER"),
                    help="users.library.lt vartotojo vardas")
parser.add_argument("-p", "--password",
                    default=os.environ.get("UPLOAD_PASS"),
                    help="users.library.lt slaptažodis")
parser.add_argument("-n", "--upload-as",
                    metavar="FILENAME",
                    help="XML failui serveryje suteikiamas pavadinimas")
#parser.add_argument("--insecure",
#                    action="store_true",
#                    help="ignoruoti HTTPS sertifikato problemas")
parser.add_argument("xml_file",
                    help="įkeliamas vartotojų XML failas")
args = parser.parse_args()

if not os.path.exists(schema_file):
    exit("error: Nerastas Aleph XSD Schema failas %r." % schema_file)

if not (args.username and args.password):
    exit("error: Nenurodytas prisijungimo vardas ar slaptažodis.")

up = AlephUploader()
#if args.insecure:
#    print("HTTPS tikrinimas išjungtas, duomenys gali būti perduodami nesaugiai.")
#    up.ua.verify = False
up.login(args.username, args.password)

for file in [args.xml_file]:
    if not file.endswith(".xml"):
        exit("error: Failas %r nėra XML failas." % file)

    if not validate_xml_file(file, schema_file):
        exit("error: Failas %r neatitinka XML Schema formato." % file)

    if not up.upload(args.xml_file, args.upload_as):
        exit("error: Failas %r buvo atmestas serverio." % file)