gitea-registration-scripts/register.py

from swagger_client.api.admin_api import AdminApi
from swagger_client.api.user_api import UserApi
from swagger_client.api.organization_api import  OrganizationApi
from swagger_client.api_client import ApiClient
from swagger_client.models import CreateUserOption, Team, User, Organization
from swagger_client.rest import ApiException
from sys import stderr
from typing import *
import click
import secrets
import termcolor


def text_red(text) -> str:
    return termcolor.colored(text, 'red')


def text_green(text) -> str:
    return termcolor.colored(text, 'green')


def text_orange(text) -> str:
    return termcolor.colored(text, 'yellow')


def does_user_exist(user_api: UserApi, username: str) -> bool:
    try:
        user: User = user_api.user_get(username)
        return user is not None
    except ApiException as e:
        if e.status == 404:
            # uzivatel nebyl nalezen
            return False
        else:
            raise e


def does_organization_exist(org_api: OrganizationApi, orgname: str) -> bool:
    try:
        org: Organization = org_api.org_get(orgname)
        return org is not None
    except ApiException as e:
        if e.status == 404:
            # organizace nebyla nalezena
            return False
        else:
            raise e


def does_organization_have_org_team(org_api: OrganizationApi, orgname: str) -> bool:
    teams: List[Team] = org_api.org_list_teams(orgname)
    for t in teams:
        if 'org' == t.name:
            return True
    return False


def get_org_team_id(org_api: OrganizationApi, orgname: str) ->  int:
    teams: List[Team] = org_api.org_list_teams(orgname)
    for t in teams:
        if 'org' == t.name:
            return t.id
    raise AssertionError('Tento kod mel byt unreachable. Zmizel tym orgu za behu skriptu?')


def validate(func_check, message: str):
    print(message, end="")
    try:
        if func_check():
            print(text_green(' [OK]'))
            return
    except ApiException as e:
        print(text_red(' [FAIL]'))
        print(text_red(e.body))
        print("Request se nepodaril! Duvod:", e.reason, file=stderr)
        exit(1)

    print(text_red(' [FAIL]'))
    print("Neocekavany stav systemu!", file=stderr)
    exit(1)


def checked_api_action(func_action, message: str):
    print(message, end='')
    try:
        res = func_action()
        print(text_green('[OK]'))
        return res
    except ApiException as e:
        print(text_red('[FAIL]'))
        print(text_red('Server odpovedel:'))
        print(text_red(e.body))
        exit(1)


if __name__ == '__main__':
    # default configuration
    HOST = "https://git.vakabus.cz/api/v1"
    try:
        with open('config.py') as f:
            conf = f.read()
            exec(conf)
    except Exception:
        print(text_orange('Nepodarilo se nacist konfiguraci ze souboru \'config.py\''))


@click.command()
@click.option('-a', '--admin-token', help=f'Admin access token', required=True)
@click.option('-h', '--host', help=f'Gitea URL (default: {HOST})')
@click.option('-e','--email', help='Emailova adresa noveho uzivatele.', required=True)
@click.option('-u','--username', help='Uzivatelske jmeno noveho uzivatele.', required=True)
@click.option('-f','--fullname', help='Plne jmeno noveho uzivatele.')
@click.option('-s','--seminar', help='Seminar, kde uzivatel orguje.', required=True)
def register(username: str, seminar: str, fullname: str, email: str, host, admin_token):
    """
    Nastroj pro automatizaci zakladani uctu na serveru Gitea.

    Prvni probehne kontrola stavu systemu, zda je mozne uzivatele bezpecne zalozit. Pote se provede pokus o zalozeni
    nasledovan pridanim uzivatele do skupiny 'org' v pozadovane organizaci (jmeno dle seminare).

    Vytvoreny uzivatel ma nastavene nahodne heslo. Zadny email o vytvoreni uctu neni zaslan.
    """

    api_client = ApiClient()
    api_client.configuration.host = HOST
    api_client.configuration.api_key = {'token': admin_token}

    admin_api = AdminApi(api_client)
    user_api = UserApi(api_client)
    org_api = OrganizationApi(api_client)

    # kontrola predpokladu o stavu systemu
    validate(lambda: not does_user_exist(user_api, username), "Uzivatelske jmeno je volne...")
    validate(lambda: does_organization_exist(org_api, seminar), "Organizace pro seminar existuje...")
    validate(lambda: does_organization_have_org_team(org_api, seminar), "Team se jmenem 'org' v danem seminari existuje...")
    print(text_green('Zakladni predpoklady pro uspesne zalozeni uctu splneny!\n'))

    # nasbirame vsechny potrebne informace
    org_team_id = get_org_team_id(org_api, seminar)

    # pripravime si uzivatele na zalozeni
    create_user_req = CreateUserOption(email=email, full_name=fullname, username=username, password=secrets.token_urlsafe(20))

    # provedeme zmeny v systemu
    user = checked_api_action(lambda: admin_api.admin_create_user(body=create_user_req), "Zakladam noveho uzivatele...")
    checked_api_action(lambda: org_api.org_add_team_member(org_team_id, username), f"Pridavam do teamu 'org' v seminari {seminar}...")
    print(text_green('Uzivatel zalozen!'))


if __name__ == '__main__':
    register()