"""Ферма и кормораздатчик по recipe_id для analytics export.""" from __future__ import annotations from typing import Dict, Iterable, Set from sqlalchemy import select from app import db from app.models import FeedDispenser, FeedingPeriod, PeriodRecipe def build_recipe_location_maps( recipe_ids: Iterable[str], ) -> tuple[Dict[str, str], Dict[str, str]]: """recipe_id -> farm, recipe_id -> dispenser_name.""" ids = {str(rid) for rid in recipe_ids if rid} farm_by_recipe: Dict[str, str] = {} dispenser_by_recipe: Dict[str, str] = {} if not ids: return farm_by_recipe, dispenser_by_recipe rows = db.session.execute( select(PeriodRecipe.recipe_id, FeedDispenser.farm, FeedDispenser.name) .join(FeedingPeriod, FeedingPeriod.id == PeriodRecipe.period_id) .join(FeedDispenser, FeedDispenser.id == FeedingPeriod.dispenser_id) .where( PeriodRecipe.recipe_id.in_(ids), PeriodRecipe.is_deleted.is_(False), FeedingPeriod.is_deleted.is_(False), FeedDispenser.is_deleted.is_(False), ) ).all() for recipe_id, farm, disp_name in rows: rid = str(recipe_id) if rid not in farm_by_recipe and farm: farm_by_recipe[rid] = str(farm).strip() if rid not in dispenser_by_recipe and disp_name: dispenser_by_recipe[rid] = str(disp_name).strip() return farm_by_recipe, dispenser_by_recipe