PCv5/master.py

#! /usr/bin/python3

from app import app, db
from app.models.user import Member, Group, GroupPrivilege
from app.models.priv import SpecialPrivilege
from app.models.trophy import Trophy, Title
from app.models.forum import Forum
from app.utils import unicode_names
import os
import sys
import yaml
import slugify
from PIL import Image


help_msg = """
This is the Planète Casio master shell. Type 'exit' or C-D to leave.

Type a category name to see a list of elements. Available categories are:

  'members'          Registered community members
  'groups'           Privilege groups
  'trophies'         Trophies
  'trophy-members'   Trophies owned by members
  'forums'           Forum tree

For each category, an argument can be specified:
* 'clear' will remove all entries in the category (destroys a lot of data!)
* 'update' will update from the model in app/data/, when applicable
  (currently available on: forums)

Type 'create-groups-and-privs' to recreate all groups and privileges to the
default. This function generates a minimal set of groups and members to prepare
the database.
1. Deletes all groups
2. Creates groups 'Administrateur', 'Modérateur', 'Développeur', 'Rédacteur',
   'Responsable Communauté', 'Partenaire', 'Compte communautaire', 'Robot', and
   'Membre de CreativeCalc'
3. Grants privileges related to these groups
4. Recreates common accounts: 'Planète Casio' (community account) and 'GLaDOS'
   (robot)

Type 'add-group <member> #<group-id>' to add a new member to a group.

Type 'create-trophies' to reset trophies and titles and their icons.

Type 'enable-user' to enable a email-disabled account.
"""

#
#  Category viewers
#

def members(*args):
    if args == ("clear",):
        for m in Member.query.all():
            m.delete()
        db.session.commit()
        print("Removed all members.")
        return

    for m in Member.query.all():
        print(m)

def groups(*args):
    if args == ("clear",):
        for g in Group.query.all():
            g.delete()
        db.session.commit()
        print("Removed all groups.")
        return

    for g in Group.query.all():
         print(f"#{g.id}  {g.name}")

def trophies(*args):
    if args == ("clear",):
        for t in Trophy.query.all():
            db.session.delete(t)
        db.session.commit()
        print("Removed all trophies.")
        return

    for t in Trophy.query.all():
        print(t)

def trophy_members(*args):
    for t in Trophy.query.all():
        if t.owners == []:
            continue

        print(t)
        for m in t.owners:
            print(f"  {m}")

def forums(*args):
    if args == ("clear",):
        for f in Forum.query.all():
            db.session.delete(f)
        db.session.commit()
        print("Removed all forums.")
        return

    if args == ("update",):
        update_forums()
        return

    for f in Forum.query.all():
        parent = f"in {f.parent.url}" if f.parent is not None else "root"
        print(f"{f.url} ({parent}) [{f.prefix}]: {f.name}")
        print(f"  {f.descr}")

#
#  Creation and edition
#

def create_groups_and_privs():
    # Clean up groups
    groups("clear")

    # Create base groups
    gr = []
    with open(os.path.join(app.root_path, "data", "groups.yaml")) as fp:
        gr = yaml.safe_load(fp.read())

    for g in gr:
        g["obj"] = Group(g["name"], g["css"], g["descr"])
        db.session.add(g["obj"])
    db.session.commit()

    for g in gr:
        for priv in g.get("privs", "").split():
            db.session.add(GroupPrivilege(g["obj"], priv))
    db.session.commit()

    print(f"Created {len(gr)} groups.")

    # Clean up test members
    for name in "PlanèteCasio GLaDOS".split():
        m = Member.query.filter_by(name=name).first()
        if m is not None:
            m.delete()

    print("Removed test members.")

    # Create template members

    def addgroup(member, group):
        g = Group.query.filter_by(name=group).first()
        if g is not None:
            member.groups.append(g)

    m = Member("PlanèteCasio", "contact@planet-casio.com", "nologin")
    addgroup(m, "Compte communautaire")
    addgroup(m, "No login")
    db.session.add(m)

    m = Member("GLaDOS", "glados@aperture.science", "nologin")
    m.xp = 1338
    addgroup(m, "Robot")
    addgroup(m, "No login")
    db.session.add(m)
    db.session.commit()

    db.session.add(SpecialPrivilege(m, "edit-posts"))
    db.session.add(SpecialPrivilege(m, "shoutbox-ban"))

    db.session.commit()

    print(f"Created 2 test members with some privileges.")


def create_trophies():
    # Clean up trophies
    trophies("clear")

    # Create base trophies
    tr = []
    with open(os.path.join(app.root_path, "data", "trophies.yaml")) as fp:
        tr = yaml.safe_load(fp.read())

    for t in tr:
        description = t.get("description", "")

        if t["is_title"]:
            trophy = Title(t["name"], description, t["hidden"],
                t.get("css", ""))
        else:
            trophy = Trophy(t["name"], description, t["hidden"])
        db.session.add(trophy)
    db.session.commit()

    print(f"Created {len(tr)} trophies.")
    # Create their icons
    create_trophies_icons()


def create_trophies_icons():
    tr = []
    with open(os.path.join(app.root_path, "data", "trophies.yaml")) as fp:
        tr = yaml.safe_load(fp.read())

    names = [slugify.slugify(t["name"]) for t in tr]
    src = os.path.join(app.root_path, "data", "trophies.png")
    dst = os.path.join(app.root_path, "static", "images", "trophies")

    try:
        os.mkdir(dst)
    except FileExistsError:
        pass

    img = Image.open(src)

    def trophy_iterator(img):
        for y in range(img.height // 26):
            for x in range(img.width // 26):
                icon = img.crop((26*x+1, 26*y+1, 26*x+25, 26*y+25))
                # Skip blank squares in the source image
                if len(icon.getcolors()) > 1:
                    yield icon.resize((48,48))

    for (name, icon) in zip(names, trophy_iterator(img)):
        icon.save(os.path.join(dst, f"{name}.png"))


def update_forums():
    # Get current forums
    existing = Forum.query.all()

    # Get the list of forums we want to end up with
    fr = []
    success = 0
    with open(os.path.join(app.root_path, "data", "forums.yaml")) as fp:
        fr = yaml.safe_load(fp.read())

    for url, info in fr.items():
        if url == "/":
            parent = None
        else:
            parent_url = url.rsplit('/', 1)[0]
            if parent_url == "":
                parent_url = "/"
            parent = Forum.query.filter_by(url=parent_url).first()

            if parent is None:
                print(f"error: no parent with url {parent_url} for {url}")
                continue

        descr = info.get("descr", "")

        # Either change an existing forum endpoint (same URL) or create one
        f = Forum.query.filter_by(url=url).first()
        if f is not None:
            # No need to change the parent (same URL implies same parent)
            changes = (f.name != info["name"]) or (f.prefix != info["prefix"])\
                or (f.descr != descr)
            f.name = info["name"]
            f.prefix = info["prefix"]
            f.descr = descr
            if changes:
                print(f"[forum] Updated {url}")
        else:
            f = Forum(url, info["name"], info["prefix"], descr, parent)
            print(f"[forum] Created {url}")

        db.session.add(f)
        success += 1

    # Remove old forums
    for f in existing:
        if f.url not in fr:
            f.delete()
            print(f"[forum] Removed {f.url}")

    db.session.commit()

def add_group(member, group):
    if group[0] != "#":
        print(f"error: group id {group} should start with '#'")
        return
    gid = int(group[1:])

    norm = unicode_names.normalize(member)

    g = Group.query.filter_by(id=gid).first()
    m = Member.query.filter_by(norm=norm).first()

    if m is None:
        print(f"error: no member has a normalized name of '{norm}'")
        return

    m.groups.append(g)
    db.session.add(m)
    db.session.commit()

def enable_user(member):
    norm = unicode_names.normalize(member)
    m = Member.query.filter_by(norm=norm).first()
    if m is None:
            print(f"error: no member has a normalized name of '{norm}'")
            return

    m.email_confirmed = True
    db.session.add(m)
    db.session.commit()

#
#  Main program
#

commands = {
    "exit":                     lambda: sys.exit(0),
    "members":                  members,
    "groups":                   groups,
    "trophies":                 trophies,
    "trophy-members":           trophy_members,
    "forums":                   forums,
    "create-groups-and-privs":  create_groups_and_privs,
    "create-trophies":          create_trophies,
    "create-trophies-icons":    create_trophies_icons,
    "add-group":                add_group,
    "enable-user":              enable_user,
}

def execute(cmd):
    if cmd[0] not in commands:
        print(f"error: unknown command '{cmd[0]}'")
    else:
        commands[cmd[0]](*cmd[1:])

# If a command is specified on the command-line, use it and do not prompt
if len(sys.argv) > 1:
    execute(sys.argv[1:])
    sys.exit(0)
# Otherwise, prompt interactively
else:
    print(help_msg)

    while True:
        try:
            print("@> ", end="")
            cmd = input().split()
        except EOFError:
            sys.exit(0)

        if cmd:
            execute(cmd)