IntelligentRailwayCosting/SourceCode/IntelligentRailwayCosting/app/stores/quota_input.py

from sqlalchemy import and_, or_
from datetime import datetime
from typing import Optional

import tools.db_helper as db_helper
from core.dtos import QuotaInputDto
from core.models import QuotaInputModel
from core.user_session import UserSession


class QuotaInputStore:
    def __init__(self):
        self._database = None
        self._current_user = None

    @property
    def current_user(self):
        if self._current_user is None:
            self._current_user = UserSession.get_current_user()
        return self._current_user

    def get_quotas_paginated(
        self,
        project_id: str,
        budget_id: int,
        item_id: int,
        page: int = 1,
        page_size: int = 10,
        keyword: Optional[str] = None,
    ):
        """分页查询定额输入列表

        Args:
            page: 页码，从1开始
            page_size: 每页数量
            project_id: 总概算序号
            budget_id: 总概算序号
            item_id: 条目序号
            keyword: 关键字

        Returns:
        """
        self._database = project_id
        with db_helper.sqlserver_query_session(self._database) as db_session:
            query = db_session.query(QuotaInputModel)

            # 构建查询条件
            conditions = [
                QuotaInputModel.budget_id == budget_id,
                QuotaInputModel.item_id == item_id,
            ]

            if keyword:
                conditions.append(
                    or_(
                        QuotaInputModel.quota_code.like(f"%{keyword}%"),
                        QuotaInputModel.project_name.like(f"%{keyword}%"),
                    )
                )

            query = query.filter(and_(*conditions))

            # 获取总数
            total_count = query.count()

            # 分页
            query = query.offset((page - 1) * page_size).limit(page_size)

            # 转换为DTO
            quotas = [QuotaInputDto.from_model(model) for model in query.all()]

            return {"total": total_count, "data": quotas}

    def get_quota(self, project_id: str, budget_id: int, item_id: int, quota_code: str):
        self._database = project_id
        with db_helper.sqlserver_query_session(self._database) as db_session:
            model = (
                db_session.query(QuotaInputModel)
                .filter(
                    QuotaInputModel.budget_id == budget_id,
                    QuotaInputModel.item_id == item_id,
                    QuotaInputModel.quota_code == quota_code,
                )
                .first()
            )
            if model is None:
                return None
            return QuotaInputDto.from_model(model)

    def create_quota(self, project_id: str, dto: QuotaInputDto) -> QuotaInputDto:
        """创建定额输入

        Args:
            project_id
            dto: 定额输入DTO

        Returns:
            QuotaInputDto
        """
        self._database = project_id
        with db_helper.sqlserver_session(self._database) as db_session:
            model = QuotaInputModel(
                budget_id=dto.budget_id,
                item_id=dto.item_id,
                quota_code=dto.quota_code,
                sequence_number=dto.sequence_number,
                project_name=dto.project_name,
                unit=dto.unit,
                project_quantity=dto.project_quantity,
                project_quantity_input=dto.project_quantity_input,
                quota_adjustment=dto.quota_adjustment,
                unit_price=dto.unit_price,
                # compilation_unit_price=dto.compilation_unit_price,
                # compilation_total_price=dto.compilation_total_price,
                unit_weight=dto.unit_weight,
                # total_weight=dto.total_weight,
                # labor_cost=dto.labor_cost,
                # compilation_labor_cost=dto.compilation_labor_cost,
                # material_cost=dto.material_cost,
                # compilation_material_cost=dto.compilation_material_cost,
                # deduct_material_cost=dto.deduct_material_cost,
                # compilation_deduct_material_cost=dto.compilation_deduct_material_cost,
                # mechanical_cost=dto.mechanical_cost,
                # compilation_mechanical_cost=dto.compilation_mechanical_cost,
                # equipment_cost=dto.equipment_cost,
                # compilation_equipment_cost=dto.compilation_equipment_cost,
                # transport_cost=dto.transport_cost,
                # compilation_transport_cost=dto.compilation_transport_cost,
                # quota_workday=dto.quota_workday,
                # total_workday=dto.total_workday,
                # workday_salary=dto.workday_salary,
                # compilation_workday_salary=dto.compilation_workday_salary,
                # quota_mechanical_workday=dto.quota_mechanical_workday,
                # total_mechanical_workday=dto.total_mechanical_workday,
                # mechanical_workday_salary=dto.mechanical_workday_salary,
                # compilation_mechanical_workday_salary=dto.compilation_mechanical_workday_salary,
                compiler=dto.compiler,
                modify_date=datetime.now().strftime("%Y/%m/%d %H:%M:%S"),
                # quota_consumption=dto.quota_consumption,0
                # basic_quota=dto.basic_quota,
                # quota_comprehensive_unit_price=dto.quota_comprehensive_unit_price,
                # quota_comprehensive_total_price=dto.quota_comprehensive_total_price
            )

            db_session.add(model)
            db_session.flush()

            return QuotaInputDto.from_model(model)

    def update_quota(
        self, project_id: str, dto: QuotaInputDto
    ) -> Optional[QuotaInputDto]:
        """更新定额输入

        Args:
            project_id:
            dto: 定额输入DTO

        Returns:
            QuotaInputDto or None
        """
        self._database = project_id
        with db_helper.sqlserver_session(self._database) as db_session:
            model = (
                db_session.query(QuotaInputModel)
                .filter(QuotaInputModel.quota_id == dto.quota_id)
                .first()
            )

            if model is None:
                return None
            model.budget_id = dto.budget_id
            model.item_id = dto.item_id
            model.quota_code = dto.quota_code
            model.sequence_number = dto.sequence_number
            model.project_name = dto.project_name
            model.unit = dto.unit
            model.project_quantity = dto.project_quantity
            model.project_quantity_input = str(dto.project_quantity)
            model.quota_adjustment = dto.quota_adjustment
            model.unit_price = dto.unit_price
            # model.compilation_unit_price = dto.compilation_unit_price
            # model.total_price = dto.total_price
            # model.compilation_total_price = dto.compilation_total_price
            model.unit_weight = dto.unit_weight
            # model.total_weight = dto.total_weight
            # model.labor_cost = dto.labor_cost
            # model.compilation_labor_cost = dto.compilation_labor_cost
            # model.material_cost = dto.material_cost
            # model.compilation_material_cost = dto.compilation_material_cost
            # model.deduct_material_cost = dto.deduct_material_cost
            # model.compilation_deduct_material_cost = dto.compilation_deduct_material_cost
            # model.mechanical_cost = dto.mechanical_cost
            # model.compilation_mechanical_cost = dto.compilation_mechanical_cost
            # model.equipment_cost = dto.equipment_cost
            # model.compilation_equipment_cost = dto.compilation_equipment_cost
            # model.transport_cost = dto.transport_cost
            # model.compilation_transport_cost = dto.compilation_transport_cost
            # model.quota_workday = dto.quota_workday
            # model.total_workday = dto.total_workday
            # model.workday_salary = dto.workday_salary
            # model.compilation_workday_salary = dto.compilation_workday_salary
            # model.quota_mechanical_workday = dto.quota_mechanical_workday
            # model.total_mechanical_workday = dto.total_mechanical_workday
            # model.mechanical_workday_salary = dto.mechanical_workday_salary
            # model.compilation_mechanical_workday_salary = dto.compilation_mechanical_workday_salary
            # model.compiler = dto.compiler
            model.modify_date = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
            # model.quota_consumption = dto.quota_consumption
            # model.basic_quota = dto.basic_quota
            # model.quota_comprehensive_unit_price = dto.quota_comprehensive_unit_price
            # model.quota_comprehensive_total_price = dto.quota_comprehensive_total_price

            db_session.merge(model)

            return QuotaInputDto.from_model(model)

    def delete_quota(self, quota_id: int) -> bool:
        """删除定额输入

        Args:
            quota_id: 定额序号

        Returns:
            bool
        """
        with db_helper.sqlserver_session(self._database) as db_session:
            model = (
                db_session.query(QuotaInputModel)
                .filter(QuotaInputModel.quota_id == quota_id)
                .first()
            )

            if model is None:
                return False

            db_session.delete(model)
            return True