import json from sqlalchemy import exists, false, func, select, text from sqlalchemy.ext.asyncio import AsyncSession from src.db.models.org_unit import OrgUnit from src.db.models.project import Project from src.db.models.rf_project_report import RfProjectReport from src.db.models.rf_project_report_line import RfProjectReportLine from src.db.models.rf_project_year import RfProjectYear from src.db.models.smeta import Smeta from src.domain.schemas import placement_type_from_db class ProjectRepository: def __init__(self, db: AsyncSession): self.db = db @staticmethod def _serialize_project(project: Project, org_unit_name: str | None, report_count: int, years: list[InterruptedError] | None = None) -> dict: return { "id": project.id, "name": project.name, "level": project.level, "parent_id": project.parent_id, "project_type": project.project_type, "vsp_format": project.vsp_format, "placement_type": placement_type_from_db(project.placement_type), "object_address": project.object_address, "staff_count": project.staff_count, "total_area": float(project.total_area) if project.total_area is not None else None, "org_unit_id": project.org_unit_id, "org_unit_name": org_unit_name, "created_at": project.created_at.isoformat() if project.created_at else None, "report_count": int(report_count), "years": years, } @staticmethod def _apply_project_filters(query, year: int | None, branch_id: int | None, org_unit_ids: list[int] | None): if year is not None: query = query.where( exists( select(1) .select_from(RfProjectReport) .join(RfProjectYear, RfProjectYear.id == RfProjectReport.rf_project_year_id) .where( RfProjectReport.project_id == Project.id, RfProjectYear.year == year, ) ) ) if branch_id is not None: query = query.where(Project.org_unit_id == branch_id) if org_unit_ids is not None: if len(org_unit_ids) == 0: query = query.where(false()) else: query = query.where(Project.org_unit_id.in_(org_unit_ids)) return query async def get_list( self, year: int | None = None, branch_id: int | None = None, offset: int | None = None, limit: int | None = None, with_count: bool = False, org_unit_ids: list[int] | None = None, ) -> list[dict] | tuple[int, list[dict]]: report_count_subq = ( select(func.count(RfProjectReport.id)) .where(RfProjectReport.project_id == Project.id) .scalar_subquery() ) report_years_subq = ( select(RfProjectYear.year) .select_from(RfProjectReport) .join(RfProjectYear, RfProjectYear.id == RfProjectReport.rf_project_year_id) .where(RfProjectReport.project_id == Project.id) .distinct() .scalar_subquery() ) query = ( select( Project, OrgUnit.title.label("org_unit_name"), report_count_subq.label("report_count"), func.array(report_years_subq).label("years"), ) .outerjoin(OrgUnit, OrgUnit.id == Project.org_unit_id) .order_by(Project.id) ) query = self._apply_project_filters(query, year, branch_id, org_unit_ids) if offset is not None: query = query.offset(offset) if limit is not None: query = query.limit(limit) rows = (await self.db.execute(query)).all() payload = [self._serialize_project(row[0], row[1], row[2], row[3]) for row in rows] if not with_count: return payload count_query = select(func.count(Project.id)) count_query = self._apply_project_filters(count_query, year, branch_id, org_unit_ids) total = int((await self.db.execute(count_query)).scalar() or 0) return total, payload @staticmethod def _serialize_project_with_reports(project: Project, years: list[tuple[RfProjectYear, Smeta | None]]) -> dict: return { "id": project.id, "name": project.name, "level": project.level, "parent_id": project.parent_id, "project_type": project.project_type, "vsp_format": project.vsp_format, "placement_type": placement_type_from_db(project.placement_type), "object_address": project.object_address, "staff_count": project.staff_count, "total_area": float(project.total_area) if project.total_area is not None else None, "org_unit_id": project.org_unit_id, "status": project.status, "technical_number": project.technical_number, "krf_decision_date": project.krf_decision_date.isoformat() if project.krf_decision_date else None, "fk_decision_date": project.fk_decision_date.isoformat() if project.fk_decision_date else None, "board_decision_date": project.board_decision_date.isoformat() if project.board_decision_date else None, "open_relocate_close_date": project.open_relocate_close_date.isoformat() if project.open_relocate_close_date else None, "funding_by_ko_decision": project.funding_by_ko_decision, "created_at": project.created_at.isoformat() if project.created_at else None, "sub_rows": [ { "id": rpy.id, "year": rpy.year, "project": f"Смета_{rpy.year}", "is_in_plan": smeta.is_in_plan if smeta else None, "is_in_plan_q2": smeta.is_in_plan_q2 if smeta else None, "is_in_plan_q3": smeta.is_in_plan_q3 if smeta else None, "is_in_plan_q4": smeta.is_in_plan_q4 if smeta else None, "development_block": smeta.development_block if smeta else None, "reserve_to_prrs_ahr": float(smeta.reserve_to_prrs_ahr) if smeta and smeta.reserve_to_prrs_ahr is not None else None, "reserve_to_prrs_kv": float(smeta.reserve_to_prrs_kv) if smeta and smeta.reserve_to_prrs_kv is not None else None, } for rpy, smeta in years ], } async def get_with_reports( self, year: int | None = None, branch_id: int | None = None, offset: int | None = None, limit: int | None = None, with_count: bool = False, org_unit_ids: list[int] | None = None, ) -> list[dict] | tuple[int, list[dict]]: project_query = select(Project).order_by(Project.id) project_query = self._apply_project_filters(project_query, year, branch_id, org_unit_ids) if offset is not None: project_query = project_query.offset(offset) if limit is not None: project_query = project_query.limit(limit) projects = (await self.db.execute(project_query)).scalars().all() years_by_project: dict[int, list[tuple[RfProjectYear, Smeta | None]]] = {} project_ids = [p.id for p in projects] if project_ids: years_query = ( select(RfProjectYear, Smeta) .outerjoin(Smeta, Smeta.rf_project_year_id == RfProjectYear.id) .where(RfProjectYear.project_id.in_(project_ids)) .order_by(RfProjectYear.year) ) for rpy, smeta in (await self.db.execute(years_query)).all(): years_by_project.setdefault(rpy.project_id, []).append((rpy, smeta)) payload = [ self._serialize_project_with_reports(p, years_by_project.get(p.id, [])) for p in projects ] if not with_count: return payload count_query = select(func.count(Project.id)) count_query = self._apply_project_filters(count_query, year, branch_id, org_unit_ids) total = int((await self.db.execute(count_query)).scalar() or 0) return total, payload async def get( self, project_id: int, org_unit_ids: list[int] | None = None, ) -> dict | None: report_count_subq = ( select(func.count(RfProjectReport.id)) .where(RfProjectReport.project_id == Project.id) .scalar_subquery() ) report_years_subq = ( select(RfProjectYear.year) .select_from(RfProjectReport) .join(RfProjectYear, RfProjectYear.id == RfProjectReport.rf_project_year_id) .where(RfProjectReport.project_id == Project.id) .distinct() .scalar_subquery() ) query = ( select( Project, OrgUnit.title.label("org_unit_name"), report_count_subq.label("report_count"), func.array(report_years_subq).label("years"), ) .outerjoin(OrgUnit, OrgUnit.id == Project.org_unit_id) .where(Project.id == project_id) ) if org_unit_ids is not None: if len(org_unit_ids) == 0: return None query = query.where(Project.org_unit_id.in_(org_unit_ids)) row = (await self.db.execute(query)).first() if not row: return None return self._serialize_project(row[0], row[1], row[2], row[3]) async def get_instance( self, project_id: int, org_unit_ids: list[int] | None = None, ) -> Project | None: query = ( select( Project, ) .where(Project.id == project_id) ) if org_unit_ids is not None: if len(org_unit_ids) == 0: return None query = query.where(Project.org_unit_id.in_(org_unit_ids)) return (await self.db.execute(query)).scalar_one_or_none() async def get_reports(self, project_id: int) -> list[dict]: line_count_subq = ( select(func.count(RfProjectReportLine.id)) .where(RfProjectReportLine.rf_project_report_id == RfProjectReport.id) .scalar_subquery() ) query = ( select( RfProjectReport.id, RfProjectYear.year, RfProjectReport.report_type, RfProjectReport.created_by, RfProjectReport.created_at, RfProjectReport.updated_by, RfProjectReport.updated_at, line_count_subq.label("line_count"), ) .join(RfProjectYear, RfProjectYear.id == RfProjectReport.rf_project_year_id) .where(RfProjectReport.project_id == project_id) .order_by(RfProjectYear.year, RfProjectReport.report_type) ) rows = (await self.db.execute(query)).all() reports: list[dict] = [] for row in rows: reports.append( { "id": row[0], "year": row[1], "report_type": row[2], "created_by": row[3], "created_at": row[4].isoformat() if row[4] is not None else None, "updated_by": row[5], "updated_at": row[6].isoformat() if row[6] is not None else None, "line_count": int(row[7]), } ) return reports async def resolve_report_id( self, project_id: int, year: int, report_type: str, org_unit_ids: list[int] | None = None, ) -> int | None: query = ( select(RfProjectReport.id) .join(Project, Project.id == RfProjectReport.project_id) .join(RfProjectYear, RfProjectYear.id == RfProjectReport.rf_project_year_id) .where( Project.id == project_id, RfProjectYear.year == year, RfProjectReport.report_type == report_type, ) ) if org_unit_ids is not None: if len(org_unit_ids) == 0: return None query = query.where(Project.org_unit_id.in_(org_unit_ids)) return (await self.db.execute(query)).scalar_one_or_none() async def get_report_rows(self, report_id: int, sections: list[str] | None = None) -> list[tuple]: query = text( """ SELECT row_type, depth, sort_order, data FROM v3.v_form3_report_jsonb(CAST(:report_id AS INT), CAST(:sections AS TEXT[])) """ ) return (await self.db.execute(query, {"report_id": report_id, "sections": sections})).all() async def get_rf_rollup_rows( self, branch_id: int, year: int, sections: list[str] | None = None, ) -> list[tuple]: query = text( """ SELECT row_type, depth, sort_order, data FROM v3.v_form3_rf_rollup_jsonb(CAST(:branch_id AS INT), CAST(:year AS INT), CAST(:sections AS TEXT[])) """ ) return ( await self.db.execute( query, {"branch_id": branch_id, "year": year, "sections": sections}, ) ).all() async def upd_form3_cell( self, report_id: int, line_id: int, column: str, value, ) -> list[tuple]: query = text( """ SELECT row_type, depth, sort_order, data FROM v3.upd_form3_cell( CAST(:report_id AS INT), CAST(:line_id AS INT), CAST(:column AS TEXT), CAST(:value AS JSONB) ) """ ) rows = ( await self.db.execute( query, { "report_id": report_id, "line_id": line_id, "column": column, "value": json.dumps(value), }, ) ).all() return [tuple(r) for r in rows] async def upd_form3_cells(self, report_id: int, changes: list[dict]) -> list[tuple]: query = text( """ SELECT row_type, depth, sort_order, data FROM v3.upd_form3_cells(CAST(:report_id AS INT), CAST(:changes AS JSONB)) """ ) rows = ( await self.db.execute( query, { "report_id": report_id, "changes": json.dumps(changes), }, ) ).all() return [tuple(r) for r in rows] async def add_form3_line(self, report_id: int, expense_item_id: int) -> list[tuple]: query = text( """ SELECT row_type, depth, sort_order, data FROM v3.add_form3_line(CAST(:report_id AS INT), CAST(:expense_item_id AS INT)) """ ) rows = ( await self.db.execute( query, { "report_id": report_id, "expense_item_id": expense_item_id, }, ) ).all() return [tuple(r) for r in rows] async def del_form3_line(self, line_id: int) -> list[tuple]: query = text( """ SELECT row_type, depth, sort_order, data FROM v3.del_form3_line(CAST(:line_id AS INT)) """ ) rows = (await self.db.execute(query, {"line_id": line_id})).all() return [tuple(r) for r in rows] async def del_project(self, project_id: int): query = text( """ SELECT v3.del_project(CAST(:project_id AS INT)) """ ) result = (await self.db.execute(query, {"project_id": project_id})).scalar_one() return result async def upd_project(self, project_id: int, data: dict) -> None: await self.db.execute( text( """ SELECT v3.upd_project( CAST(:project_id AS INT), CAST(:name AS VARCHAR), CAST(:status AS VARCHAR), CAST(:technical_number AS VARCHAR), CAST(:project_type AS VARCHAR), CAST(:vsp_format AS VARCHAR), CAST(:placement_type AS VARCHAR), CAST(:object_address AS VARCHAR), CAST(:staff_count AS INT), CAST(:total_area AS NUMERIC), CAST(:org_unit_id AS INT), CAST(:krf_decision_date AS DATE), CAST(:fk_decision_date AS DATE), CAST(:board_decision_date AS DATE), CAST(:open_relocate_close_date AS DATE), CAST(:funding_by_ko_decision AS VARCHAR) ) """ ), { "project_id": project_id, "name": data.get("name"), "status": data.get("status"), "technical_number": data.get("technical_number"), "project_type": data.get("project_type"), "vsp_format": data.get("vsp_format"), "placement_type": data.get("placement_type"), "object_address": data.get("object_address"), "staff_count": data.get("staff_count"), "total_area": data.get("total_area"), "org_unit_id": data.get("org_unit_id"), "krf_decision_date": data.get("krf_decision_date"), "fk_decision_date": data.get("fk_decision_date"), "board_decision_date": data.get("board_decision_date"), "open_relocate_close_date": data.get("open_relocate_close_date"), "funding_by_ko_decision": data.get("funding_by_ko_decision"), }, ) await self.db.flush() async def resolve_rf_project_year_id( self, project_id: int, year: int, org_unit_ids: list[int] | None = None, ) -> int | None: query = ( select(RfProjectYear.id) .join(Project, Project.id == RfProjectYear.project_id) .where(Project.id == project_id, RfProjectYear.year == year) ) if org_unit_ids is not None: if len(org_unit_ids) == 0: return None query = query.where(Project.org_unit_id.in_(org_unit_ids)) return (await self.db.execute(query)).scalar_one_or_none() async def upd_smeta(self, rf_project_year_id: int, data: dict) -> None: await self.db.execute( text( """ SELECT v3.upd_smeta( CAST(:rf_project_year_id AS INT), CAST(:is_in_plan AS BOOLEAN), CAST(:is_in_plan_q2 AS BOOLEAN), CAST(:is_in_plan_q3 AS BOOLEAN), CAST(:is_in_plan_q4 AS BOOLEAN), CAST(:development_block AS VARCHAR), CAST(:reserve_to_prrs_ahr AS NUMERIC), CAST(:reserve_to_prrs_kv AS NUMERIC) ) """ ), { "rf_project_year_id": rf_project_year_id, "is_in_plan": data.get("is_in_plan"), "is_in_plan_q2": data.get("is_in_plan_q2"), "is_in_plan_q3": data.get("is_in_plan_q3"), "is_in_plan_q4": data.get("is_in_plan_q4"), "development_block": data.get("development_block"), "reserve_to_prrs_ahr": data.get("reserve_to_prrs_ahr"), "reserve_to_prrs_kv": data.get("reserve_to_prrs_kv"), }, ) await self.db.flush() @staticmethod def _serialize_smeta(smeta: Smeta | None) -> dict | None: if smeta is None: return None return { "id": smeta.id, "rf_project_year_id": smeta.rf_project_year_id, "is_in_plan": smeta.is_in_plan, "is_in_plan_q2": smeta.is_in_plan_q2, "is_in_plan_q3": smeta.is_in_plan_q3, "is_in_plan_q4": smeta.is_in_plan_q4, "development_block": smeta.development_block, "reserve_to_prrs_ahr": float(smeta.reserve_to_prrs_ahr) if smeta.reserve_to_prrs_ahr is not None else None, "reserve_to_prrs_kv": float(smeta.reserve_to_prrs_kv) if smeta.reserve_to_prrs_kv is not None else None, } async def get_smeta(self, rf_project_year_id: int) -> dict | None: smeta = ( await self.db.execute( select(Smeta).where(Smeta.rf_project_year_id == rf_project_year_id) ) ).scalar_one_or_none() return self._serialize_smeta(smeta) async def add_project( self, name: str, year: int, org_unit_id: int, level: str = "project", parent_id: int | None = None, project_type: str | None = None, vsp_format: str | None = None, placement_type: str | None = None, object_address: str | None = None, staff_count: int | None = None, total_area: float | None = None, ) -> tuple[int, int, int]: query = text( """ SELECT project_id, limit_report_id, current_expenses_report_id FROM v3.add_project( CAST(:name AS VARCHAR), CAST(:year AS INT), CAST(:org_unit_id AS INT), CAST(:level AS VARCHAR), CAST(:parent_id AS INT), CAST(:project_type AS VARCHAR), CAST(:vsp_format AS VARCHAR), CAST(:placement_type AS VARCHAR), CAST(:object_address AS VARCHAR), CAST(:staff_count AS INT), CAST(:total_area AS NUMERIC) ) """ ) row = ( await self.db.execute( query, { "name": name, "year": year, "org_unit_id": org_unit_id, "level": level, "parent_id": parent_id, "project_type": project_type, "vsp_format": vsp_format, "placement_type": placement_type, "object_address": object_address, "staff_count": staff_count, "total_area": total_area, }, ) ).first() return int(row[0]), int(row[1]), int(row[2])