c40a6569b4
All checks were successful
CI/CD Pipeline / build-and-deploy (push) Successful in 20s
Admins can now pick a color per event from a Catppuccin Mocha palette; existing events default to blue. Colors flow through to both the admin timeline and the new "Неделя" tab in the user view, which reuses the admin's weekly grid in a read-only mode. Also reworks event blocks to show a single-line "start–end title" so short events no longer overflow, tidies the admin toolbar, and expands mobile responsiveness across both frontends. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
242 lines
7.8 KiB
Python
242 lines
7.8 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, EventException
|
||
from backend.models import ScheduleItem
|
||
|
||
TZ = pytz.timezone("Europe/Moscow")
|
||
|
||
|
||
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}"
|
||
|
||
|
||
def _time_to_minutes(value: str) -> int:
|
||
hour, minute = map(int, value.split(":"))
|
||
return hour * 60 + minute
|
||
|
||
|
||
async def check_event_overlap(
|
||
db: AsyncSession,
|
||
date: str,
|
||
start_time: str,
|
||
duration_min: int,
|
||
exclude_id: int = None,
|
||
occurrence_date: str = None
|
||
) -> bool:
|
||
"""Проверить пересечение по времени с разовыми и weekly events."""
|
||
start_minutes = _time_to_minutes(start_time)
|
||
end_minutes = start_minutes + duration_min
|
||
|
||
events = await materialize_events(db, date, date)
|
||
|
||
for event in events:
|
||
if exclude_id and event.id == exclude_id and (occurrence_date is None or event.date == occurrence_date):
|
||
continue
|
||
|
||
ev_start_minutes = _time_to_minutes(event.start_time)
|
||
ev_end_minutes = ev_start_minutes + (event.duration_min or 0)
|
||
if not (end_minutes <= ev_start_minutes or start_minutes >= ev_end_minutes):
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
async def materialize_events(
|
||
db: AsyncSession,
|
||
from_date: str,
|
||
to_date: str
|
||
) -> list[ScheduleItem]:
|
||
items = []
|
||
|
||
result = await db.execute(
|
||
select(Event).where(
|
||
and_(
|
||
Event.date >= from_date,
|
||
Event.date <= to_date,
|
||
Event.repeat_weekly == False
|
||
)
|
||
)
|
||
)
|
||
one_off_events = result.scalars().all()
|
||
for event in one_off_events:
|
||
items.append(ScheduleItem(
|
||
kind="event",
|
||
id=event.id,
|
||
date=event.date,
|
||
source_date=event.date,
|
||
title=event.title,
|
||
start_time=event.start_time,
|
||
duration_min=event.duration_min,
|
||
repeat_weekly=False,
|
||
color=event.color or "blue"
|
||
))
|
||
|
||
result = await db.execute(select(Event).where(Event.repeat_weekly == True))
|
||
weekly_events = result.scalars().all()
|
||
if not weekly_events:
|
||
return items
|
||
|
||
result = await db.execute(
|
||
select(EventException).where(
|
||
and_(
|
||
EventException.date >= from_date,
|
||
EventException.date <= to_date
|
||
)
|
||
)
|
||
)
|
||
exceptions = result.scalars().all()
|
||
exception_dict = {(ex.event_id, ex.date): 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 event in weekly_events:
|
||
base_date = datetime.strptime(event.date, "%Y-%m-%d").date()
|
||
if current_date < base_date or event.weekday != weekday:
|
||
continue
|
||
|
||
exception = exception_dict.get((event.id, date_str))
|
||
if exception:
|
||
if exception.action == "delete":
|
||
continue
|
||
items.append(ScheduleItem(
|
||
kind="event",
|
||
id=event.id,
|
||
date=exception.replacement_date or date_str,
|
||
source_date=date_str,
|
||
title=exception.replacement_title or event.title,
|
||
start_time=exception.replacement_start_time or event.start_time,
|
||
duration_min=exception.replacement_duration_min or event.duration_min,
|
||
repeat_weekly=True,
|
||
color=exception.replacement_color or event.color or "blue"
|
||
))
|
||
continue
|
||
|
||
items.append(ScheduleItem(
|
||
kind="event",
|
||
id=event.id,
|
||
date=date_str,
|
||
source_date=date_str,
|
||
title=event.title,
|
||
start_time=event.start_time,
|
||
duration_min=event.duration_min,
|
||
repeat_weekly=True,
|
||
color=event.color or "blue"
|
||
))
|
||
|
||
current_date += timedelta(days=1)
|
||
|
||
return items
|
||
|
||
|
||
async def materialize_weekly_tasks(
|
||
db: AsyncSession,
|
||
from_date: str,
|
||
to_date: str
|
||
) -> list[ScheduleItem]:
|
||
"""Материализовать weekly tasks в диапазоне дат."""
|
||
items = []
|
||
|
||
result = await db.execute(select(Task).where(Task.repeat_weekly == True))
|
||
weekly_tasks = result.scalars().all()
|
||
if not weekly_tasks:
|
||
return items
|
||
|
||
result = await db.execute(
|
||
select(WeeklyTaskException).where(
|
||
and_(
|
||
WeeklyTaskException.date >= from_date,
|
||
WeeklyTaskException.date <= to_date
|
||
)
|
||
)
|
||
)
|
||
exceptions = result.scalars().all()
|
||
exception_dict = {}
|
||
for ex in exceptions:
|
||
if ex.task_id is not None:
|
||
exception_dict[(ex.task_id, ex.date)] = ex
|
||
else:
|
||
exception_dict[(None, ex.date, ex.weekday)] = ex
|
||
|
||
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:
|
||
base_date = datetime.strptime(task.date, "%Y-%m-%d").date()
|
||
if current_date < base_date or task.weekday != weekday:
|
||
continue
|
||
|
||
exception = exception_dict.get((task.id, date_str))
|
||
if not exception:
|
||
exception = exception_dict.get((None, date_str, weekday))
|
||
|
||
if exception:
|
||
if exception.action == "delete":
|
||
continue
|
||
items.append(ScheduleItem(
|
||
kind="task",
|
||
id=task.id,
|
||
date=exception.replacement_date or date_str,
|
||
source_date=date_str,
|
||
title=exception.replacement_title or task.title,
|
||
repeat_weekly=True
|
||
))
|
||
continue
|
||
|
||
items.append(ScheduleItem(
|
||
kind="task",
|
||
id=task.id,
|
||
date=date_str,
|
||
source_date=date_str,
|
||
title=task.title,
|
||
repeat_weekly=True
|
||
))
|
||
|
||
current_date += timedelta(days=1)
|
||
|
||
return items
|