ki-backend/ki/actions/seed.py

# SPDX-FileCopyrightText: WTF Kooperative eG <https://wtf-eg.de/>
# SPDX-License-Identifier: AGPL-3.0-or-later

import csv

from app import app, db
from ki.models import Address, Contact, ContactType, Language, Skill, Profile, ProfileLanguage, ProfileSearchtopic, \
    ProfileSkill, User


def seed_contacttypes():
    contacttypes_seed_file_path = app.config["KI_DATA_DIR"] + "/seed_data/contacttypes.csv"

    app.logger.info("importing contacttypes")

    with open(contacttypes_seed_file_path) as file:
        csv_reader = csv.DictReader(file)

        for contacttype in csv_reader:
            id = int(contacttype["id"])
            db_contacttype = db.session.get(ContactType, id)

            if db_contacttype is None:
                db.session.add(ContactType(id=int(contacttype["id"]), name=contacttype["name"]))


def seed_user(auth_id,
              nickname=None,
              pronouns="",
              visible=True,
              volunteerwork="",
              availability_status=False,
              freetext="",
              availability_text="",
              availability_hours_per_week=42,
              skills=[],
              searchtopics=[],
              languages=[],
              address=None,
              contacts=[]):
    if not nickname:
        nickname = auth_id

    app.logger.info(f"seeding {nickname} \\o/")

    user = User(auth_id=auth_id)
    db.session.add(user)

    profile = Profile(nickname=nickname,
                      pronouns=pronouns,
                      volunteerwork=volunteerwork,
                      availability_status=availability_status,
                      availability_text=availability_text,
                      availability_hours_per_week=availability_hours_per_week,
                      freetext=freetext,
                      visible=visible,
                      user=user)

    for skill_data in skills:
        skill = ProfileSkill(profile=profile, skill_id=skill_data[0], level=skill_data[1])
        db.session.add(skill)

    for skill_id in searchtopics:
        searchtopic = ProfileSearchtopic(profile=profile, skill_id=skill_id)
        db.session.add(searchtopic)

    for language_data in languages:
        language = ProfileLanguage(profile=profile, language_id=language_data[0], level=language_data[1])
        db.session.add(language)

    if address:
        _address = Address(name=address[0],
                           street=address[1],
                           house_number=address[2],
                           additional=address[3],
                           postcode=address[4],
                           city=address[5],
                           country=address[6],
                           profile=profile)
        db.session.add(_address)

    for contact_data in contacts:
        contact = Contact(profile=profile, contacttype_id=contact_data[0], content=contact_data[1])
        db.session.add(contact)

    db.session.add(profile)


def seed(dev: bool):
    seed_contacttypes()

    skill_seed_file_path = app.config["KI_DATA_DIR"] + "/seed_data/skills.csv"

    app.logger.info("importing skills")

    with open(skill_seed_file_path) as skills_file:
        skills_csv_reader = csv.DictReader(skills_file)

        for skill in skills_csv_reader:
            id = int(skill["id"])
            db_skill = db.session.get(Skill, id)

            if db_skill is None:
                db.session.add(Skill(id=int(skill["id"]), name=skill["name"]))

    app.logger.info("importing languages")

    iso_seed_file_path = app.config["KI_DATA_DIR"] + "/seed_data/iso_639_1.csv"

    with open(iso_seed_file_path) as iso_file:
        iso_csv_reader = csv.DictReader(iso_file)

        for iso in iso_csv_reader:
            id = iso["639-1"]
            db_language = db.session.get(Language, id)

            if db_language is None:
                db.session.add(Language(id=iso["639-1"], name=iso["Sprache"]))

    if dev:
        seed_user("klaus", visible=False)

        for i in range(1, 20):
            seed_user(f"babsi{i}")

        seed_user("peter",
                  nickname="peternichtlustig",
                  visible=False,
                  pronouns="Herr Dr. Dr.",
                  volunteerwork="Gartenverein",
                  availability_status=True,
                  availability_hours_per_week=42,
                  availability_text="Immer",
                  freetext="Ich mag Kaffee",
                  skills=[(3, 3), (1, 5)],
                  searchtopics=[3, 1],
                  languages=[("de", 5), ("fr", 3)],
                  address=("Peter Nichtlustig", "Waldweg", "23i", "Hinterhaus", "13337", "Bielefeld", "Deutschland"),
                  contacts=[(4, "@peter:wtf-eg.de"), (1, "peter@wtf-eg.de")])

        seed_user("dirtydieter",
                  volunteerwork="Müll sammeln",
                  availability_status=True,
                  availability_hours_per_week=24,
                  availability_text="Nur Nachts!",
                  freetext="1001010010111!!!",
                  skills=[(1, 5)],
                  address=("Friedrich Witzig", "", "", "", "", "", ""))
        # all_skills = [(skill.id, 3) for skill in Skill.query.all()] # query causes problems
        # seed_user("jutta", languages=[("fr", 5)], skills=all_skills)

        seed_user("giesela", skills=[(9, 3), (10, 5)])
        seed_user("bertha", visible=False, skills=[(11, 3), (10, 5)])
        seed_user("monique", languages=[("fr", 4)])
    print("seeding done")
    with app.app_context():
        db.session.commit()  # also problematic
    print("commit done")