nizika_ai/migration.py

from __future__ import annotations

import os
from typing import TypedDict

from eloquent import DatabaseManager, Schema

CONFIG: dict[str, DbConfig] = { 'mysql': { 'driver':    'mysql',
                                           'host':      'localhost',
                                           'database':  'nizika_ai',
                                           'user':      os.environ['MYSQL_USER'],
                                           'password':  os.environ['MYSQL_PASS'],
                                           'prefix':    '' } }
DB = DatabaseManager (CONFIG)
SCHEMA = Schema (DB)


def main (
) -> None:
    create_queries ()
    create_answers ()
    create_users ()
    create_query_answer_histories ()
    add_constraints_to_answers ()
    add_constraints_to_query_answer_histories ()


def create_queries (
) -> None:
    with SCHEMA.create ('queries') as table:
        table.big_increments ('id')
        table.big_integer ('user_id').nullable ().comment ('クエリ主')
        table.integer ('target_character').comment ('クエリ先キャラクタ')
        table.text ('content').comment ('クエリ内容')
        table.string ('image_url').nullable ().default (None).comment ('添附画像 URL')
        table.integer ('query_type').comment ('クエリ区分')
        table.integer ('model').comment ('GPT のモデル')
        table.datetime ('sent_at').comment ('送信日時')
        table.boolean ('answered').default (False).comment ('回答済')


def create_answers (
) -> None:
    with SCHEMA.create ('answers') as table:
        table.big_increments ('id')
        table.big_integer ('query_id').nullable ().comment ('クエリ')
        table.integer ('character').comment ('キャラクタ区分')
        table.text ('content').comment ('回答内容')
        table.integer ('answer_type').comment ('回答区分')
        table.datetime ('sent_at').comment ('送信日時')
        table.boolean ('answered').default (False).comment ('回答済')


def add_constraints_to_answers (
) -> None:
    with SCHEMA.table ('answers') as table:
        table.foreign ('query_id').references ('id').on ('queries').on_update ('cascade').on_delete ('restrict')


def create_users (
) -> None:
    with SCHEMA.create ('users') as table:
        table.big_increments ('id')
        table.integer ('platform').comment ('プラットフォーム区分')
        table.string ('code').comment ('ユーザ・コード（プラットフォーム依存）')
        table.string ('name').comment ('ユーザ名（プラットフォーム内）')
        table.binary ('icon').nullable ().comment ('アイコン')
        table.unique (['platform', 'code'])


def create_query_answer_histories (
) -> None:
    with SCHEMA.create ('query_answer_histories') as table:
        table.big_increments ('id')
        table.big_integer ('query_id')
        table.big_integer ('answer_id')
        table.unique (['query_id', 'answer_id'])


def add_constraints_to_query_answer_histories (
) -> None:
    with SCHEMA.table ('query_answer_histories') as table:
        table.foreign ('query_id').references ('id').on ('queries').on_update ('cascade').on_delete ('cascade')
        table.foreign ('answer_id').references ('id').on ('answers').on_update ('cascade').on_delete ('cascade')


class DbConfig (TypedDict):
    driver:     str
    host:       str
    database:   str
    user:       str
    password:   str
    prefix:     str


if __name__ == '__main__':
    main ()