Данные из сателитныхх листов в АХР

api/alembic/versions/0031_form2_ahr_satellite_sync.py

@@ -0,0 +1,15 @@
+from sql_migration import run_sql_file_migration
+
+
+revision = "0031"
+down_revision = "0030"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    run_sql_file_migration("0031_form2_ahr_satellite_sync.sql")
+
+
+def downgrade() -> None:
+    pass

api/src/api/v1/websocket.py

@@ -33,6 +33,13 @@ from src.services.user_service import UserService
 
 logger = logging.getLogger()
 
+FORM_2_AHR_SATELLITE_SHEETS = {
+    "AHR_LIMIT",
+    "AHR_RENT",
+    "AHR_UTILITY",
+    "AHR_SECURITY",
+}
+
 
 @asynccontextmanager
 async def get_db_session(user_id: int | None = None):
@@ -215,6 +222,27 @@ class ConnectionManager:
             except:
                 pass  # Игнорируем недоступные соединения
 
+    async def broadcast_to_form_sheet(
+            self,
+            data: dict,
+            form_id: int,
+            sheet: str,
+    ):
+        data = data.copy()
+        data["is_self"] = None
+        message = dumps(data, default=_convert_error, ensure_ascii=False)
+
+        for key, connections in self.connections[ConnectionKeyEnum.FORM].items():
+            key_data = dict(key)
+            if key_data.get("form_id") != form_id or key_data.get("sheet") != sheet:
+                continue
+
+            for con_info in connections:
+                try:
+                    await con_info.ws.send_text(message)
+                except:
+                    pass  # Игнорируем недоступные соединения
+
     async def send_back(self, data: dict, websocket: WebSocket):
         data = data.copy()
         data["is_self"] = True
@@ -742,6 +770,93 @@ class ProjectEventProcess:
 manager = ConnectionManager()
 
 
+def get_satellite_source_row_id(data: dict) -> int | None:
+    event_name = data.get("event")
+
+    if event_name == "cell_updated":
+        return data.get("data", {}).get("line_id")
+
+    if event_name == "row_deleted":
+        return data.get("data", {}).get("row_id")
+
+    if event_name == "row_added":
+        result = data.get("result")
+        if not isinstance(result, dict):
+            return None
+
+        rows = result.get("data") or []
+        for row in rows:
+            if len(row) >= 4 and isinstance(row[3], dict) and row[3].get("id") is not None:
+                return row[3]["id"]
+
+    return None
+
+
+async def notify_ahr_about_satellite_change(
+        form_id: int,
+        source_sheet: str,
+        source_row_id: int,
+        source_event: str,
+        user_id: int,
+):
+    try:
+        async with get_db_session(user_id=user_id) as db:
+            ahr_delta = await SheetService(db).get_form2_ahr_satellite_delta(
+                form_id=form_id,
+                source_sheet=source_sheet,
+                source_row_id=source_row_id,
+            )
+
+        if not ahr_delta:
+            return
+
+        ahr_event = {
+            "event": "cell_updated",
+            "data": {
+                "source_sheet": source_sheet,
+            },
+            "result": ahr_delta,
+        }
+
+        if source_event == "row_added":
+            new_row = next(
+                (
+                    row for row in ahr_delta
+                    if row[0] == "INPUT"
+                ),
+                None,
+            )
+            if new_row is not None:
+                new_line_id = new_row[3].get("line_id")
+                ahr_event = {
+                    "event": "row_added",
+                    "data": {
+                        "source_sheet": source_sheet,
+                        "expense_item_id": new_row[3].get("header", {}).get("expense_item_id"),
+                    },
+                    "result": {
+                        "data": ahr_delta,
+                        "new_line_id": new_line_id,
+                    },
+                }
+
+        await manager.broadcast_to_form_sheet(
+            data=ahr_event,
+            form_id=form_id,
+            sheet="AHR",
+        )
+    except Exception:
+        logger.exception(
+            "Ошибка отправки на лист АХР",
+            extra={
+                "form_id": form_id,
+                "source_sheet": source_sheet,
+                "source_row_id": source_row_id,
+                "source_event": source_event,
+            },
+        )
+
+
 def _convert_error(o):
     try:
         return asdict(o)
@@ -910,6 +1025,16 @@ async def process_websocket(
                         user_id=user_id,
                         **kwargs_process
                     )
+                    should_refresh_ahr = (
+                        con_key == ConnectionKeyEnum.FORM
+                        and kwargs_process.get("sheet") in FORM_2_AHR_SATELLITE_SHEETS
+                        and data.get("event") in {"cell_updated", "row_added", "row_deleted"}
+                    )
+                    ahr_source_row_id = (
+                        get_satellite_source_row_id(data)
+                        if should_refresh_ahr
+                        else None
+                    )
                     await db.commit()
                     db_ms = (time.perf_counter() - t0) * 1000
 
@@ -956,6 +1081,17 @@ async def process_websocket(
                         **kwargs,
                     )
 
+                if should_refresh_ahr and ahr_source_row_id is not None:
+                    asyncio.create_task(
+                        notify_ahr_about_satellite_change(
+                            form_id=kwargs_process["form_id"],
+                            source_sheet=kwargs_process["sheet"],
+                            source_row_id=ahr_source_row_id,
+                            source_event=data["event"],
+                            user_id=user_id,
+                        )
+                    )
+
             except BasicAppException as e:
                 data["error"] = e.description or "Неизвестная ошибка"
                 await manager.send_back(

api/src/repository/sheet_repository.py

@@ -380,3 +380,31 @@ class SheetRepository:
                 )
             ).all()
         ]
+
+    async def get_form2_ahr_satellite_delta(
+            self,
+            form_id: int,
+            source_sheet: str,
+            source_row_id: int,
+    ) -> list[tuple]:
+        return [
+            tuple(r) for r in (
+                await self.db.execute(
+                    text(
+                        """
+                        SELECT row_type, depth, sort_order, data
+                          FROM v3.get_form2_ahr_satellite_delta(
+                              :form_id,
+                              :source_sheet,
+                              :source_row_id
+                          )
+                        """
+                    ),
+                    {
+                        "form_id": form_id,
+                        "source_sheet": source_sheet,
+                        "source_row_id": source_row_id,
+                    },
+                )
+            ).all()
+        ]

api/src/services/sheet_service.py

@@ -274,3 +274,15 @@ class SheetService:
             row_id=row_id,
             direction=direction,
         )
+
+    async def get_form2_ahr_satellite_delta(
+            self,
+            form_id: int,
+            source_sheet: str,
+            source_row_id: int,
+    ) -> list[tuple]:
+        return await self.sheet_repo.get_form2_ahr_satellite_delta(
+            form_id=form_id,
+            source_sheet=source_sheet,
+            source_row_id=source_row_id,
+        )