nizika_ai/queries_to_answers.py

from __future__ import annotations

import random
from datetime import datetime
from enum import Enum

from eloquent import DatabaseManager, Model

from config import CONFIG
from models import Answer, Query, User
from talk import Talk


def main (
) -> None:
    db = DatabaseManager (CONFIG)
    Model.set_connection_resolver (db)
    queries: list[Query] = Query.where ('answered', False).get ()
    if not queries:
        return
    query: Query = random.choice (queries)
    message: str | list[dict[str, str | dict[str, str]]]
    if query.image_url is None:
        message = query.content
    else:
        message = [{ 'type': 'text', 'text': query.content },
                   { 'type': 'image_url', 'image_url': query.image_url }]
    user: User | None = query.user
    user_name: str | None = None
    if user is not None:
        user_name = user.name
    histories: list[dict[str, str]] = []
    for history in query.answer_histories:
        if history.query is not None:
            histories.append ({ 'role': 'user', 'content': history.query.content })
        histories.append ({ 'role': 'assistant', 'content': history.content })
    if query.target_character & Character.DEERJIKA.value:
        answer = Answer ()
        answer.query_id = query.id
        answer.character = Character.DEERJIKA.value
        answer.content = Talk.main (query.content, user_name, histories)
        answer.answer_type = query.query_type
        answer.sent_at = datetime.now ()
        answer.save ()
    if query.target_character & Character.GOATOH.value:
        answer = Answer ()
        answer.query_id = query.id
        answer.character = Character.GOATOH.value
        answer.content = Talk.main (query.content, user_name, histories, True)
        answer.answer_type = query.query_type
        answer.sent_at = datetime.now ()
        answer.save ()


class Character (Enum):
    DEERJIKA    = 1
    GOATOH      = 2


if __name__ == '__main__':
    main ()