Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/scheduler/tasks/create_timeline_events.py: 20%
68 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 17:29 +0000
1from datetime import UTC, datetime, time, timedelta 1a
3from dateutil.tz import tzlocal 1a
4from pydantic import UUID4 1a
5from sqlalchemy.orm import Session 1a
7from mealie.db.db_setup import session_context 1a
8from mealie.repos.all_repositories import get_repositories 1a
9from mealie.schema.household.household import HouseholdRecipeUpdate 1a
10from mealie.schema.meal_plan.new_meal import PlanEntryType 1a
11from mealie.schema.recipe.recipe import RecipeSummary 1a
12from mealie.schema.recipe.recipe_timeline_events import RecipeTimelineEventCreate, TimelineEventType 1a
13from mealie.schema.response.pagination import PaginationQuery 1a
14from mealie.schema.user.user import DEFAULT_INTEGRATION_ID 1a
15from mealie.services.event_bus_service.event_bus_service import EventBusService 1a
16from mealie.services.event_bus_service.event_types import ( 1a
17 EventOperation,
18 EventRecipeData,
19 EventRecipeTimelineEventData,
20 EventTypes,
21)
22from mealie.services.household_services.household_service import HouseholdService 1a
25def _create_mealplan_timeline_events_for_household( 1a
26 event_time: datetime, session: Session, group_id: UUID4, household_id: UUID4
27) -> None:
28 repos = get_repositories(session, group_id=group_id, household_id=household_id)
29 household_service = HouseholdService(group_id, household_id, repos)
30 event_bus_service = EventBusService(session=session)
32 timeline_events_to_create: list[RecipeTimelineEventCreate] = []
33 recipes_to_update: dict[UUID4, RecipeSummary] = {}
34 recipe_id_to_slug_map: dict[UUID4, str] = {}
36 local_tz = tzlocal()
37 mealplans = repos.meals.get_today(tz=local_tz)
38 for mealplan in mealplans:
39 if not (mealplan.recipe and mealplan.user_id):
40 continue
42 user = repos.users.get_one(mealplan.user_id)
43 if not user:
44 continue
46 # TODO: make this translatable
47 if mealplan.entry_type == PlanEntryType.side:
48 event_subject = f"{user.full_name} made this as a side"
50 else:
51 event_subject = f"{user.full_name} made this for {mealplan.entry_type.value}"
53 query_start_time = datetime.combine(datetime.now(UTC).date(), time.min)
54 query_end_time = query_start_time + timedelta(days=1)
55 query = PaginationQuery(
56 query_filter=(
57 f'recipe_id = "{mealplan.recipe_id}" '
58 f'AND timestamp >= "{query_start_time.isoformat()}" '
59 f'AND timestamp < "{query_end_time.isoformat()}" '
60 f'AND subject = "{event_subject}"'
61 )
62 )
64 # if this event already exists, don't create it again
65 events = repos.recipe_timeline_events.page_all(pagination=query)
66 if events.items:
67 continue
69 # bump up the last made date
70 household_to_recipe = household_service.get_household_recipe(mealplan.recipe.slug)
71 last_made = household_to_recipe.last_made if household_to_recipe else None
72 if (not last_made or last_made.date() < event_time.date()) and mealplan.recipe_id not in recipes_to_update:
73 recipes_to_update[mealplan.recipe_id] = mealplan.recipe
75 timeline_events_to_create.append(
76 RecipeTimelineEventCreate(
77 user_id=user.id,
78 subject=event_subject,
79 event_type=TimelineEventType.info,
80 timestamp=event_time,
81 recipe_id=mealplan.recipe_id,
82 )
83 )
85 recipe_id_to_slug_map[mealplan.recipe_id] = mealplan.recipe.slug
87 if not timeline_events_to_create:
88 return
90 # TODO: use bulk operations
91 for event in timeline_events_to_create:
92 new_event = repos.recipe_timeline_events.create(event)
93 event_bus_service.dispatch(
94 integration_id=DEFAULT_INTEGRATION_ID,
95 group_id=group_id,
96 household_id=household_id,
97 event_type=EventTypes.recipe_updated,
98 document_data=EventRecipeTimelineEventData(
99 operation=EventOperation.create,
100 recipe_slug=recipe_id_to_slug_map[new_event.recipe_id],
101 recipe_timeline_event_id=new_event.id,
102 ),
103 )
105 for recipe in recipes_to_update.values():
106 household_service.set_household_recipe(recipe.slug, HouseholdRecipeUpdate(last_made=event_time))
107 repos.recipes.patch(recipe.slug, {"last_made": event_time})
108 event_bus_service.dispatch(
109 integration_id=DEFAULT_INTEGRATION_ID,
110 group_id=group_id,
111 household_id=household_id,
112 event_type=EventTypes.recipe_updated,
113 document_data=EventRecipeData(operation=EventOperation.update, recipe_slug=recipe.slug),
114 )
117def _create_mealplan_timeline_events_for_group(event_time: datetime, session: Session, group_id: UUID4) -> None: 1a
118 repos = get_repositories(session, group_id=group_id)
119 households_data = repos.households.page_all(PaginationQuery(page=1, per_page=-1))
120 household_ids = [household.id for household in households_data.items]
121 for household_id in household_ids:
122 _create_mealplan_timeline_events_for_household(event_time, session, group_id, household_id)
125def create_mealplan_timeline_events() -> None: 1a
126 event_time = datetime.now(UTC)
128 with session_context() as session:
129 repos = get_repositories(session)
130 groups_data = repos.groups.page_all(PaginationQuery(page=1, per_page=-1))
131 group_ids = [group.id for group in groups_data.items]
133 for group_id in group_ids:
134 _create_mealplan_timeline_events_for_group(event_time, session, group_id)