python_skripte/movie-db/moviedb_func.py

import DBcm
from mariadb import ProgrammingError
import pandas as pd
import sqlite3


def create_movie_database(db_name="movie_db.db"):

    create_my_list = """
    create table if not exists movie_list (
         id integer not null primary key autoincrement,
        titel varchar(64) not null,
        genre_id integer not null,
        regie_id integer not null,
        medium_id integer not null
    )
    """

    create_genre = """
    create table if not exists genre (
         id integer not null primary key autoincrement,
        name varchar(64) not null
    )
    """

    create_regie = """
    create table if not exists regie (
         id integer not null primary key autoincrement,
        surname varchar(64) not null,
        lastname varchar(64) not null
    )
    """

    create_medium = """
    create table if not exists medium (
         id integer not null primary key autoincrement,
         medium varchar(64) not null
    )
    """

    ADD_GENRE_VALUE = """
    INSERT INTO genre(name)
    SELECT ?
    WHERE NOT EXISTS (SELECT 1 FROM genre WHERE name = ?);
    """

    ADD_MEDIUM_VALUE = """
    INSERT INTO medium(medium)
    SELECT ?
    WHERE NOT EXISTS (SELECT 1 FROM medium WHERE medium = ?);
    """

    ADD_REGIE_VALUE = """
    INSERT INTO regie (surname, lastname )
    SELECT ?, ?
    WHERE NOT EXISTS
        (SELECT surname, lastname
            FROM regie
            WHERE surname = ? AND lastname = ?);
    """

    with DBcm.UseDatabase(db_name) as db:
        db.execute(create_my_list)
        db.execute(create_genre)
        db.execute(create_regie)
        db.execute(create_medium)

    with open("genre_list", "r") as fs:
        for genre_value in fs.readlines():
            with DBcm.UseDatabase(db_name) as db:
                db.execute(ADD_GENRE_VALUE,
                           (genre_value.strip(), genre_value.strip()))

    usecols = ["Name", "Vorname"]
    air = pd.read_csv("regie_name.csv", usecols=usecols)
    for count, (reg_name, reg_vorname) in enumerate(zip(air["Name"], air["Vorname"])):
        # print(count, reg_vorname, reg_name)
        with DBcm.UseDatabase(db_name) as db:
            db.execute(ADD_REGIE_VALUE, (reg_vorname,
                       reg_name, reg_vorname, reg_name))

    LISTE_MEDIUM = ["BlueRay", "DVD", "Datei",
                    "BlueRay Steelbook", "DVD Steelbook"]
    with DBcm.UseDatabase(db_name) as db:
        for MEDIUM in LISTE_MEDIUM:
            db.execute(ADD_MEDIUM_VALUE, (MEDIUM, MEDIUM))


def all_select(db_name="movie_db.db", what_select="genre"):
    ALL_SELECT = "SELECT * from " + what_select
    if what_select == "genre" or what_select == "medium":
        with DBcm.UseDatabase(db_name) as db:
            db.execute(ALL_SELECT)
            all_value = [i[1] for i in db.fetchall()]
        return all_value
    elif what_select == 'regie':
        all_value = []
        with DBcm.UseDatabase(db_name) as db:
            db.execute(ALL_SELECT)
            for i in db.fetchall():
                all_value.append(i[1] + " " + i[2])
        return all_value
    else:
        return "Wrong Value !!!"


def search_id(db_name="movie_db.db", search_name=str, select_from="genre", select_where="name"):
    if select_from == "regie":
        split_search = search_name.split(" ")
        GENRE_QUERY = f"""select id from {select_from}
                      where surname = ? and lastname =  ?"""
        with DBcm.UseDatabase(db_name) as db:
            db.execute(GENRE_QUERY, (split_search[0], split_search[1],))
            regie_id = db.fetchone()[0]
        return int(regie_id)
    else:
        try:
            GENRE_QUERY = f"""select id from {select_from}
                         where {select_where} = ?"""
            with DBcm.UseDatabase(db_name) as db:
                db.execute(GENRE_QUERY, (search_name,))
                genre_id = db.fetchone()[0]
            return int(genre_id)
        except:
            return int(0)


def add_movie_to_list(db_name="movie_db.db", movie_name=str, genre_id=int, regie_id=int, medium_id=int):
    SQL_PARAM = f"""
    INSERT INTO movie_list (titel, genre_id, regie_id, medium_id )
    SELECT ?, ?, ?, ?
    WHERE NOT EXISTS
        (SELECT titel FROM movie_list WHERE titel = ?);
    """
    try:
        with DBcm.UseDatabase(db_name) as db:
            db.execute(SQL_PARAM, (movie_name.lower(), genre_id,
                       regie_id, medium_id, movie_name.lower(),))
    except ProgrammingError:
        raise ProgrammingError("Konnte nicht in die DB schreiben")
    return True


def show_movie_list(db_name="movie_db.db"):
    SQL_PARAM = f"""SELECT
                    movie_list.id,
                    titel,
                    genre.name AS genre,
                    regie.surname as regie_surname,
                    regie.lastname as regie_lastname,
                    medium.medium
        FROM movie_list
        INNER JOIN genre ON movie_list.genre_id=genre.id
        INNER JOIN regie ON movie_list.regie_id=regie.id
        INNER JOIN medium ON movie_list.medium_id=medium.id
        ;
    """
    db = sqlite3.connect(db_name)
    SELCET_VALUE = pd.read_sql(SQL_PARAM, db)
    return_list_dict = []
    for id, titel, genre, regie_surname, regie_lastname in zip(SELCET_VALUE["id"], SELCET_VALUE["titel"], SELCET_VALUE["genre"], SELCET_VALUE["regie_surname"], SELCET_VALUE["regie_lastname"]):
        return_list_dict.append(
            {"id": id, "titel": titel, "genre": genre, "regie": regie_surname + " " + regie_lastname})
    return return_list_dict


if __name__ == "__main__":
    create_movie_database()
    # print(all_select())
    # id_genre = search_id(
    #    search_name="DVD", select_from="medium", select_where="medium")
    add_movie_to_list(movie_name="Schlumpfland",
                      genre_id=1, regie_id=1, medium_id=1)
    print(show_movie_list())