60 行
1.9 KiB
Python
60 行
1.9 KiB
Python
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 ()
|
|
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 ()
|