"""Каскадное мягкое удаление связанных строк и постановка sync_queue для потомков.""" from __future__ import annotations import logging from datetime import datetime from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import select, update from app import db from app.models.component import Component, Ingredient from app.models.equipment import ( FeedDispenser, FeedMixer, FeedingLocation, FeedingPeriod, FeedingPoint, Trip, ) from app.models.recipe import PeriodRecipe, Recipe, UnloadingGroup from app.models.report import ( ComponentLoadingTime, LoadingReport, LoadingReportComponent, UnloadingReport, UnloadingReportGroup, ) from app.models.sklad import ComponentStock from app.timeutil import utc_now_naive logger = logging.getLogger(__name__) TABLE_MODEL_MAP: Dict[str, Any] = { "component": Component, "recipe": Recipe, "ingredient": Ingredient, "unloading_group": UnloadingGroup, "loading_report": LoadingReport, "loading_report_component": LoadingReportComponent, "component_loading_time": ComponentLoadingTime, "unloading_report": UnloadingReport, "unloading_report_group": UnloadingReportGroup, "feed_dispenser": FeedDispenser, "feeding_period": FeedingPeriod, "period_recipes": PeriodRecipe, "feed_mixer": FeedMixer, "feeding_location": FeedingLocation, "feeding_point": FeedingPoint, "trip": Trip, } CASCADE_CHILDREN: Dict[str, List[Tuple[str, str]]] = { "recipe": [ ("ingredient", "recipe_id"), ("unloading_group", "recipe_id"), ], "component": [ ("ingredient", "component_id"), ], "feed_dispenser": [ ("feeding_period", "dispenser_id"), ], "feeding_period": [ ("feeding_point", "period_id"), ("period_recipes", "period_id"), ], "feed_mixer": [ ("trip", "mixer_id"), ("feeding_location", "mixer_id"), ("feeding_point", "mixer_id"), ], "loading_report": [ ("loading_report_component", "report_id"), ("component_loading_time", "report_id"), ], "unloading_report": [ ("unloading_report_group", "report_id"), ], } def _active_child_record_ids(child_table: str, fk_column: str, parent_id: str) -> List[str]: model = TABLE_MODEL_MAP.get(child_table) if model is None: return [] col = getattr(model, fk_column, None) if col is None: return [] q = select(model).where(col == parent_id) if hasattr(model, "is_deleted"): q = q.where(model.is_deleted.is_(False)) rows = db.session.execute(q).unique().scalars().all() if child_table == "period_recipes": return [f"{r.period_id}:{r.recipe_id}" for r in rows] return [str(r.id) for r in rows if getattr(r, "id", None)] CASCADE_ENQUEUE_INFO_KEY = "_wesp_cascade_enqueue_pending" def _schedule_cascade_enqueues(rows: List[Tuple[str, str]]) -> None: if not rows: return db.session.info.setdefault(CASCADE_ENQUEUE_INFO_KEY, []).extend(rows) def cascade_soft_delete( table_name: str, record_id: str, *, deleted_by: str = "cascade", deleted_at: Optional[datetime] = None, enqueue: bool = True, defer_enqueue: bool = False, ) -> List[Tuple[str, str]]: """Помечает потомков is_deleted и ставит sync_queue update для каждой строки.""" when = deleted_at or utc_now_naive() affected: List[Tuple[str, str]] = [] seen: set[Tuple[str, str]] = set() queue: List[Tuple[str, str]] = [(table_name, record_id)] while queue: parent_table, parent_id = queue.pop(0) for child_table, fk_col in CASCADE_CHILDREN.get(parent_table, []): model = TABLE_MODEL_MAP.get(child_table) if model is None: continue child_ids = _active_child_record_ids(child_table, fk_col, parent_id) if not child_ids: continue tbl = model.__table__ db.session.execute( update(tbl) .where(tbl.c[fk_col] == parent_id) .where(tbl.c.is_deleted.is_(False)) .values( is_deleted=True, deleted_at=when, deleted_by=deleted_by, ) ) for cid in child_ids: key = (child_table, cid) if key in seen: continue seen.add(key) affected.append(key) queue.append(key) if table_name == "component": stock_tbl = ComponentStock.__table__ db.session.execute( update(stock_tbl) .where(stock_tbl.c.component_id == record_id) .where(stock_tbl.c.is_deleted.is_(False)) .values( is_deleted=True, deleted_at=when, deleted_by=deleted_by, ) ) if enqueue and affected: if defer_enqueue: _schedule_cascade_enqueues(affected) else: from app.services.sync_manager import enqueue_sync_queue_task with db.session.no_autoflush: for tname, rid in affected: enqueue_sync_queue_task(tname, rid, "update", priority=1) if affected: logger.info( "[SYNC-CASCADE] %s.%s → помечено потомков: %s", table_name, (record_id or "")[:36], len(affected), ) return affected