146 lines
5.1 KiB
Python
146 lines
5.1 KiB
Python
from datetime import datetime, timedelta
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, and_
|
||
import pytz
|
||
|
||
from backend.database import Task, Event, WeeklyTaskException
|
||
from backend.models import ScheduleItem
|
||
|
||
TZ = pytz.timezone("Europe/London")
|
||
|
||
|
||
def get_weekday_from_date(date_str: str) -> int:
|
||
"""Получить день недели (0=Monday, 6=Sunday) из даты"""
|
||
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
||
return dt.weekday()
|
||
|
||
|
||
def get_week_range(date_str: str = None):
|
||
"""Получить диапазон дат недели (понедельник-воскресенье)"""
|
||
if date_str:
|
||
base_date = datetime.strptime(date_str, "%Y-%m-%d").date()
|
||
else:
|
||
base_date = datetime.now(TZ).date()
|
||
|
||
# Находим понедельник
|
||
days_since_monday = base_date.weekday()
|
||
monday = base_date - timedelta(days=days_since_monday)
|
||
sunday = monday + timedelta(days=6)
|
||
|
||
return monday.strftime("%Y-%m-%d"), sunday.strftime("%Y-%m-%d")
|
||
|
||
|
||
def format_date_russian(date_str: str) -> str:
|
||
"""Форматировать дату в русском формате: 'Вторник, 2 декабря'"""
|
||
dt = datetime.strptime(date_str, "%Y-%m-%d")
|
||
|
||
weekdays = ["Понедельник", "Вторник", "Среда", "Четверг", "Пятница", "Суббота", "Воскресенье"]
|
||
months = [
|
||
"января", "февраля", "марта", "апреля", "мая", "июня",
|
||
"июля", "августа", "сентября", "октября", "ноября", "декабря"
|
||
]
|
||
|
||
weekday = weekdays[dt.weekday()]
|
||
month = months[dt.month - 1]
|
||
|
||
return f"{weekday}, {dt.day} {month}"
|
||
|
||
|
||
async def check_event_overlap(
|
||
db: AsyncSession,
|
||
date: str,
|
||
start_time: str,
|
||
duration_min: int,
|
||
exclude_id: int = None
|
||
) -> bool:
|
||
"""Проверить пересечение по времени с другими events"""
|
||
# Парсим время начала
|
||
start_hour, start_min = map(int, start_time.split(":"))
|
||
start_minutes = start_hour * 60 + start_min
|
||
end_minutes = start_minutes + duration_min
|
||
|
||
# Получаем все events на эту дату
|
||
query = select(Event).where(Event.date == date)
|
||
if exclude_id:
|
||
query = query.where(Event.id != exclude_id)
|
||
|
||
result = await db.execute(query)
|
||
events = result.scalars().all()
|
||
|
||
for event in events:
|
||
ev_start_hour, ev_start_min = map(int, event.start_time.split(":"))
|
||
ev_start_minutes = ev_start_hour * 60 + ev_start_min
|
||
ev_end_minutes = ev_start_minutes + event.duration_min
|
||
|
||
# Проверка пересечения интервалов
|
||
if not (end_minutes <= ev_start_minutes or start_minutes >= ev_end_minutes):
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
async def materialize_weekly_tasks(
|
||
db: AsyncSession,
|
||
from_date: str,
|
||
to_date: str
|
||
) -> list[ScheduleItem]:
|
||
"""Материализовать weekly tasks в диапазоне дат"""
|
||
from backend.database import Task, WeeklyTaskException
|
||
|
||
items = []
|
||
|
||
# Получаем все weekly tasks
|
||
result = await db.execute(select(Task).where(Task.repeat_weekly == True))
|
||
weekly_tasks = result.scalars().all()
|
||
|
||
# Получаем все исключения
|
||
result = await db.execute(
|
||
select(WeeklyTaskException).where(
|
||
and_(
|
||
WeeklyTaskException.date >= from_date,
|
||
WeeklyTaskException.date <= to_date
|
||
)
|
||
)
|
||
)
|
||
exceptions = result.scalars().all()
|
||
exception_dict = {(ex.date, ex.weekday): ex for ex in exceptions}
|
||
|
||
from_dt = datetime.strptime(from_date, "%Y-%m-%d").date()
|
||
to_dt = datetime.strptime(to_date, "%Y-%m-%d").date()
|
||
|
||
current_date = from_dt
|
||
while current_date <= to_dt:
|
||
date_str = current_date.strftime("%Y-%m-%d")
|
||
weekday = current_date.weekday()
|
||
|
||
for task in weekly_tasks:
|
||
if task.weekday == weekday:
|
||
# Проверяем исключения
|
||
ex_key = (date_str, weekday)
|
||
if ex_key in exception_dict:
|
||
exception = exception_dict[ex_key]
|
||
if exception.action == "delete":
|
||
continue # Пропускаем эту дату
|
||
elif exception.action == "replace":
|
||
items.append(ScheduleItem(
|
||
kind="task",
|
||
id=task.id,
|
||
date=date_str,
|
||
title=exception.replacement_title,
|
||
repeat_weekly=True
|
||
))
|
||
continue
|
||
|
||
# Обычная weekly task
|
||
items.append(ScheduleItem(
|
||
kind="task",
|
||
id=task.id,
|
||
date=date_str,
|
||
title=task.title,
|
||
repeat_weekly=True
|
||
))
|
||
|
||
current_date += timedelta(days=1)
|
||
|
||
return items
|
||
|