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
import logging

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"

    logging.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 = ContactType.query.get(id)

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


def seed_user(nickname,
              visible=False,
              skills=[],
              languages=[],
              volunteerwork="",
              availabilityStatus=False,
              freetext="",
              availabilityText="",
              availabilityHoursPerWeek=42):
    app.logger.info(f"seeding {nickname} \\o/")

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

    profile = Profile(nickname=nickname,
                      pronouns="",
                      volunteerwork=volunteerwork,
                      availabilityStatus=availabilityStatus,
                      availabilityText=availabilityText,
                      availabilityHoursPerWeek=availabilityHoursPerWeek,
                      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 language_data in languages:
        language = ProfileLanguage(profile=profile, language_id=language_data[0], level=language_data[1])
        db.session.add(language)

    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 = Skill.query.get(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 = Language.query.get(id)

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

    if dev:
        app.logger.info("seeding peter :)")

        peter = User(auth_id="peter")
        db.session.add(peter)

        peters_profile = Profile(nickname="peternichtlustig",
                                 pronouns="Herr Dr. Dr.",
                                 volunteerwork="Gartenverein",
                                 availabilityStatus=True,
                                 availabilityHoursPerWeek=42,
                                 availabilityText="Immer",
                                 freetext="Ich mag Kaffee",
                                 user=peter)
        db.session.add(peters_profile)

        matrix_contact = Contact(profile=peters_profile, contacttype_id=4, content="@peter:wtf-eg.de")
        db.session.add(matrix_contact)

        email_contact = Contact(profile=peters_profile, contacttype_id=1, content="peter@wtf-eg.de")
        db.session.add(email_contact)

        peters_address = Address(name="Peter Nichtlustig",
                                 street="Waldweg",
                                 house_number="23i",
                                 additional="Hinterhaus",
                                 postcode="13337",
                                 city="Bielefeld",
                                 country="Deutschland",
                                 profile=peters_profile)
        db.session.add(peters_address)

        peters_python_skill = ProfileSkill(profile=peters_profile, skill_id=3, level=3)
        db.session.add(peters_python_skill)

        peters_php_skill = ProfileSkill(profile=peters_profile, skill_id=1, level=5)
        db.session.add(peters_php_skill)

        peters_python_searchtopic = ProfileSearchtopic(profile=peters_profile, skill_id=3)
        db.session.add(peters_python_searchtopic)

        peters_php_searchtopic = ProfileSearchtopic(profile=peters_profile, skill_id=1)
        db.session.add(peters_php_searchtopic)

        peter_de = ProfileLanguage(profile=peters_profile, language_id="de", level=5)
        db.session.add(peter_de)

        peter_fr = ProfileLanguage(profile=peters_profile, language_id="fr", level=3)
        db.session.add(peter_fr)

        seed_user("klaus")

        seed_user("dirtydieter",
                  visible=True,
                  volunteerwork="Müll sammeln",
                  availabilityStatus=True,
                  availabilityHoursPerWeek=24,
                  availabilityText="Nur Nachts!",
                  freetext="1001010010111!!!",
                  skills=[(Skill.skill_id_php, 5)])

        all_skills = Skill.query.all()
        all_profile_skills = []
        for skill in all_skills:
            all_profile_skills.append((skill.id, 3))

        seed_user("jutta", visible=True, languages=[("fr", 5)], skills=all_profile_skills)
        seed_user("giesela", visible=True, skills=[(Skill.skill_id_mysql, 3), (Skill.skill_id_postgresql, 5)])
        seed_user("bertha", visible=False, skills=[(Skill.skill_id_sqlite, 3), (Skill.skill_id_postgresql, 5)])

        seed_user("monique", visible=True, languages=[("fr", 4)])

    db.session.commit()