nizika_ai/queries_to_answers.py

from __future__ import annotations

import random
from datetime import datetime
from typing import cast

from eloquent import DatabaseManager, Model  # type: ignore

from config import CONFIG
from consts import Character
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 = query.user
    user_name: str | None = None
    if query.user_id 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 ()
    query.answered = True
    query.save ()


if __name__ == '__main__':
    main ()