131 lines
5.0 KiB
Python
131 lines
5.0 KiB
Python
import json
|
||
from typing import List, Dict, Any, Optional
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from app.services.gigachat_service import gigachat_service
|
||
from app.core.config import settings
|
||
from app.crud import schedule as crud_schedule, task as crud_task
|
||
from app.schemas.schedule import ScheduleCreate
|
||
from app.schemas.task import TaskCreate
|
||
from datetime import date
|
||
|
||
|
||
SCHEDULE_GENERATION_PROMPT = """Ты планета Земля, друг детей с расстройством аутистического спектра (РАС).
|
||
|
||
Создай расписание на {date} для ребенка {age} лет.
|
||
Предпочтения ребенка: {preferences}
|
||
|
||
Важные правила:
|
||
1. Задания должны быть простыми и понятными
|
||
2. Каждое задание имеет четкие временные рамки
|
||
3. Используй визуальные описания
|
||
4. Избегай резких переходов между активностями
|
||
5. Включи время на отдых
|
||
6. Учитывай особенности РАС
|
||
|
||
Верни ТОЛЬКО валидный JSON формат без дополнительного текста:
|
||
{{
|
||
"title": "Название расписания",
|
||
"description": "Краткое описание",
|
||
"tasks": [
|
||
{{
|
||
"title": "Название задания",
|
||
"description": "Подробное описание",
|
||
"duration_minutes": 30,
|
||
"category": "утренняя_рутина",
|
||
"order": 0
|
||
}}
|
||
]
|
||
}}"""
|
||
|
||
|
||
class ScheduleGenerator:
|
||
async def generate(
|
||
self,
|
||
db: AsyncSession,
|
||
user_id: str,
|
||
child_age: int,
|
||
preferences: List[str],
|
||
schedule_date: str,
|
||
description: Optional[str] = None
|
||
) -> Dict[str, Any]:
|
||
"""Генерация расписания через GigaChat"""
|
||
# Формируем промпт
|
||
prompt = SCHEDULE_GENERATION_PROMPT.format(
|
||
date=schedule_date,
|
||
age=child_age,
|
||
preferences=", ".join(preferences) if preferences else "нет особых предпочтений"
|
||
)
|
||
|
||
if description:
|
||
prompt += f"\n\nДополнительная информация: {description}"
|
||
|
||
# Генерируем через GigaChat
|
||
try:
|
||
response_text = await gigachat_service.generate_text(
|
||
prompt=prompt,
|
||
model=settings.GIGACHAT_MODEL_SCHEDULE
|
||
)
|
||
|
||
# Парсим JSON ответ
|
||
# Убираем markdown код блоки если есть
|
||
if "```json" in response_text:
|
||
response_text = response_text.split("```json")[1].split("```")[0].strip()
|
||
elif "```" in response_text:
|
||
response_text = response_text.split("```")[1].split("```")[0].strip()
|
||
|
||
schedule_data = json.loads(response_text)
|
||
|
||
# Создаем расписание в БД
|
||
schedule_create = ScheduleCreate(
|
||
title=schedule_data.get("title", f"Расписание на {schedule_date}"),
|
||
date=date.fromisoformat(schedule_date),
|
||
description=schedule_data.get("description") or description
|
||
)
|
||
|
||
db_schedule = await crud_schedule.create(
|
||
db,
|
||
{
|
||
**schedule_create.model_dump(),
|
||
"user_id": user_id
|
||
}
|
||
)
|
||
|
||
# Создаем задачи
|
||
tasks_data = schedule_data.get("tasks", [])
|
||
for task_data in tasks_data:
|
||
task_create = TaskCreate(
|
||
schedule_id=db_schedule.id,
|
||
title=task_data.get("title"),
|
||
description=task_data.get("description"),
|
||
duration_minutes=task_data.get("duration_minutes", 30),
|
||
category=task_data.get("category"),
|
||
order=task_data.get("order", 0)
|
||
)
|
||
|
||
await crud_task.create(db, task_create.model_dump())
|
||
|
||
await db.refresh(db_schedule)
|
||
|
||
return {
|
||
"schedule_id": db_schedule.id,
|
||
"title": db_schedule.title,
|
||
"tasks": [
|
||
{
|
||
"title": task.title,
|
||
"description": task.description,
|
||
"duration_minutes": task.duration_minutes,
|
||
"category": task.category,
|
||
"order": task.order
|
||
}
|
||
for task in db_schedule.tasks
|
||
]
|
||
}
|
||
except json.JSONDecodeError as e:
|
||
raise Exception(f"Failed to parse GigaChat response as JSON: {str(e)}")
|
||
except Exception as e:
|
||
raise Exception(f"Schedule generation error: {str(e)}")
|
||
|
||
|
||
schedule_generator = ScheduleGenerator()
|
||
|