from __future__ import annotations """Application Service for Flexible Warehouse Dispatch & Material Wastage Analytics. Adheres strictly to Hexagonal Architecture, Decimal(18, 4) financial precision, and Vietnam Construction Norms (Thông tư 12/2021/TT-BXD & Định mức 2026). """ from datetime import date, datetime from decimal import Decimal import logging from typing import Any, Dict, List, Optional, Tuple from uuid import UUID, uuid4 from app.core.postgres.base_pkg.base_client import BasePostgresClient from app.modules.inventory.domain.dispatch_entities import ( NORM_WASTAGE_PERCENTAGES, MaterialVarianceItem, VarianceAlertStatus, WarehouseDispatch, ) logger = logging.getLogger("dscons.inventory.dispatch_service") class WarehouseDispatchService: """Điều phối quy trình xuất kho vật tư linh hoạt và đối chiếu hao hụt 3 chiều: Dự Toán (BoQ) - Thực Xuất (Dispatches) - Quyết Toán (Settlement). """ def __init__(self, postgres_client: Optional[BasePostgresClient] = None) -> None: self._db = postgres_client or BasePostgresClient() def _ensure_schema(self, cur) -> None: """Đảm bảo bảng erp_warehouse_dispatches tồn tại tự động nếu chưa chạy migration thủ công.""" cur.execute(""" CREATE TABLE IF NOT EXISTS erp_warehouse_dispatches ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), inventory_item_id UUID NOT NULL REFERENCES erp_warehouse_inventory(id) ON DELETE CASCADE, dispatch_code VARCHAR(64) NOT NULL UNIQUE, project_id UUID REFERENCES projects(id) ON DELETE SET NULL, wbs_id UUID REFERENCES erp_project_wbs(id) ON DELETE SET NULL, takeoff_item_id UUID REFERENCES erp_drawing_takeoff_items(id) ON DELETE SET NULL, dispatch_date DATE NOT NULL DEFAULT CURRENT_DATE, quantity_dispatched NUMERIC(18, 4) NOT NULL CHECK (quantity_dispatched > 0), unit VARCHAR(50) NOT NULL, unit_price_vnd NUMERIC(18, 4) NOT NULL DEFAULT 0.0000, total_value_vnd NUMERIC(18, 4) NOT NULL DEFAULT 0.0000, recipient_name VARCHAR(255), transport_vehicle VARCHAR(100), purpose VARCHAR(255), notes TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_erp_wh_disp_inv_item ON erp_warehouse_dispatches(inventory_item_id); CREATE INDEX IF NOT EXISTS idx_erp_wh_disp_project ON erp_warehouse_dispatches(project_id); CREATE INDEX IF NOT EXISTS idx_erp_wh_disp_date ON erp_warehouse_dispatches(dispatch_date); """) def create_dispatch( self, inventory_item_id: str, quantity_dispatched: Decimal | float, project_id: Optional[str] = None, wbs_id: Optional[str] = None, takeoff_item_id: Optional[str] = None, dispatch_date: Optional[date | str] = None, recipient_name: Optional[str] = None, transport_vehicle: Optional[str] = None, purpose: Optional[str] = None, notes: Optional[str] = None, ) -> Dict[str, Any]: """Tạo phiếu xuất kho linh hoạt. - Kiểm tra tồn kho khả dụng (chống xuất âm) bằng SELECT ... FOR UPDATE. - Trừ quantity_remaining và cập nhật trạng thái kho. - Cho phép liên kết linh hoạt: Theo dự án, WBS, hoặc xuất dùng chung (không bắt buộc). """ qty = Decimal(str(quantity_dispatched)) if qty <= Decimal("0"): raise ValueError("Số lượng xuất kho phải lớn hơn 0.") d_date = dispatch_date or date.today() if isinstance(d_date, str): d_date = date.fromisoformat(d_date) with self._db.get_connection() as conn: with conn.cursor() as cur: self._ensure_schema(cur) # 1. Khóa bản ghi lô vật tư để bảo vệ Race Condition cur.execute(""" SELECT id, item_code, item_name, material_group, unit, quantity_remaining, unit_price_vnd, allocated_project_id FROM erp_warehouse_inventory WHERE id = %s FOR UPDATE; """, (inventory_item_id,)) item = cur.fetchone() if not item: raise ValueError(f"Không tìm thấy lô vật tư có ID {inventory_item_id} trong kho.") remaining = Decimal(str(item["quantity_remaining"])) if remaining < qty: raise ValueError( f"Tồn kho khả dụng ({remaining} {item['unit']}) không đủ để xuất ({qty} {item['unit']})." ) new_remaining = remaining - qty new_status = "ĐÃ_XUẤT_HẾT" if new_remaining <= Decimal("0") else "ĐÃ_XUẤT_MỘT_PHẦN" # 2. Cập nhật tồn kho lô vật tư update_proj_sql = ", allocated_project_id = %s" if project_id and not item["allocated_project_id"] else "" update_params = [new_remaining, new_status] if update_proj_sql: update_params.append(project_id) update_params.append(inventory_item_id) cur.execute(f""" UPDATE erp_warehouse_inventory SET quantity_remaining = %s, status = %s, updated_at = CURRENT_TIMESTAMP {update_proj_sql} WHERE id = %s; """, update_params) # 3. Tạo mã phiếu xuất chuẩn ISO XK-YYYY-XXXX cur.execute("SELECT COUNT(*) as cnt FROM erp_warehouse_dispatches;") seq = (cur.fetchone()["cnt"] or 0) + 1 dispatch_code = f"XK-{d_date.year}-{seq:04d}" unit_price = Decimal(str(item["unit_price_vnd"])) total_val = qty * unit_price # 4. Lưu phiếu xuất kho cur.execute(""" INSERT INTO erp_warehouse_dispatches ( inventory_item_id, dispatch_code, project_id, wbs_id, takeoff_item_id, dispatch_date, quantity_dispatched, unit, unit_price_vnd, total_value_vnd, recipient_name, transport_vehicle, purpose, notes ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) RETURNING id, created_at; """, ( inventory_item_id, dispatch_code, project_id, wbs_id, takeoff_item_id, d_date, qty, item["unit"], unit_price, total_val, recipient_name, transport_vehicle, purpose, notes )) res = cur.fetchone() conn.commit() logger.info(f"[WarehouseDispatch] Đã xuất {qty} {item['unit']} mã {item['item_code']} theo phiếu {dispatch_code}") return { "id": str(res["id"]), "dispatch_code": dispatch_code, "inventory_item_id": inventory_item_id, "item_name": item["item_name"], "quantity_dispatched": float(qty), "quantity_remaining_after": float(new_remaining), "unit": item["unit"], "unit_price_vnd": float(unit_price), "total_value_vnd": float(total_val), "dispatch_date": d_date.isoformat(), "project_id": project_id, "recipient_name": recipient_name, "transport_vehicle": transport_vehicle, "purpose": purpose, "notes": notes, "status": new_status, } def list_dispatches( self, project_id: Optional[str] = None, inventory_item_id: Optional[str] = None, search: Optional[str] = None, limit: int = 50, offset: int = 0, ) -> Tuple[List[Dict[str, Any]], int]: """Truy vấn danh sách lịch sử phiếu xuất kho với đầy đủ liên kết dự án và lô vật tư gốc.""" with self._db.get_connection() as conn: with conn.cursor() as cur: self._ensure_schema(cur) where_clauses = ["1=1"] params: List[Any] = [] if project_id: where_clauses.append("d.project_id = %s") params.append(project_id) if inventory_item_id: where_clauses.append("d.inventory_item_id = %s") params.append(inventory_item_id) if search: where_clauses.append("(d.dispatch_code ILIKE %s OR w.item_name ILIKE %s OR d.recipient_name ILIKE %s OR d.purpose ILIKE %s)") term = f"%{search.strip()}%" params.extend([term, term, term, term]) where_sql = " AND ".join(where_clauses) # Total count cur.execute(f""" SELECT COUNT(*) as total FROM erp_warehouse_dispatches d JOIN erp_warehouse_inventory w ON d.inventory_item_id = w.id LEFT JOIN projects p ON d.project_id = p.id WHERE {where_sql}; """, params) total_cnt = cur.fetchone()["total"] # Rows query query = f""" SELECT d.id, d.dispatch_code, d.dispatch_date, d.quantity_dispatched, d.unit, d.unit_price_vnd, d.total_value_vnd, d.recipient_name, d.transport_vehicle, d.purpose, d.notes, d.created_at, w.id as inventory_item_id, w.item_code, w.item_name, w.material_group, w.invoice_series_number, w.supplier_name, p.id as project_id, p.project_code, p.project_name, t.item_name as takeoff_work_name FROM erp_warehouse_dispatches d JOIN erp_warehouse_inventory w ON d.inventory_item_id = w.id LEFT JOIN projects p ON d.project_id = p.id LEFT JOIN erp_drawing_takeoff_items t ON d.takeoff_item_id = t.id WHERE {where_sql} ORDER BY d.dispatch_date DESC, d.created_at DESC LIMIT %s OFFSET %s; """ params.extend([limit, offset]) cur.execute(query, params) rows = cur.fetchall() return [dict(r) for r in rows], total_cnt def get_project_takeoff_options(self, project_id: str) -> List[Dict[str, Any]]: """Lấy danh sách các công tác bóc tách dự toán của dự án để gợi ý chọn khi xuất kho.""" with self._db.get_connection() as conn: with conn.cursor() as cur: cur.execute(""" SELECT t.id, t.item_order, t.wbs_code, t.norm_code, t.item_name, t.category, t.unit, t.quantity as estimated_quantity, t.hsmt_quantity, t.actual_quantity as settled_quantity, t.unit_price_vnd, t.total_amount_vnd, dt.drawing_code, dt.drawing_title FROM erp_drawing_takeoff_items t JOIN erp_drawing_takeoffs dt ON t.takeoff_id = dt.id WHERE dt.project_id = %s ORDER BY t.item_order ASC; """, (project_id,)) rows = cur.fetchall() return [dict(r) for r in rows] def calculate_material_variance_report( self, project_id: Optional[str] = None ) -> Dict[str, Any]: """Tính toán ma trận đối chiếu 3 chiều: Dự toán (BoQ) - Thực xuất (Dispatches) - Quyết toán (Settlement) và phân tích tỷ lệ hao hụt vật tư thực tế so với định mức nhà nước (Thông tư 12/2021 & Định mức 2026). """ with self._db.get_connection() as conn: with conn.cursor() as cur: self._ensure_schema(cur) # ------------------------------------------------------------------ # 1. ĐỌC DỰ TOÁN ($Q_{\text{dt}}$) & QUYẾT TOÁN ($Q_{\text{qt}}$) TỪ HỒ SƠ BÓC TÁCH CAD # ------------------------------------------------------------------ takeoff_where = "1=1" takeoff_params: List[Any] = [] if project_id: takeoff_where = "dt.project_id = %s" takeoff_params.append(project_id) cur.execute(f""" SELECT t.id, t.item_name, t.category, t.unit, t.quantity as estimated_quantity, t.actual_quantity as settled_quantity, t.material_breakdown, t.material_cost_vnd FROM erp_drawing_takeoff_items t JOIN erp_drawing_takeoffs dt ON t.takeoff_id = dt.id WHERE {takeoff_where}; """, takeoff_params) takeoff_rows = cur.fetchall() # Phân rã khối lượng định mức theo từng nhóm vật tư estimated_by_group: Dict[str, Dict[str, Any]] = {} for row in takeoff_rows: cat = (row.get("category") or "other").upper() mat_group = self._map_category_to_material_group(cat, str(row.get("item_name") or "")) # Khối lượng công tác dự toán ($Q_{\text{dt}}$) và quyết toán ($Q_{\text{qt}}$) q_dt = Decimal(str(row.get("estimated_quantity") or 0)) # Nếu actual_quantity chưa có thì mặc định bằng q_dt cho công trình đang thi công q_act = Decimal(str(row["settled_quantity"])) if row.get("settled_quantity") is not None else q_dt # Phân tích material_breakdown nếu có breakdowns = row.get("material_breakdown") if isinstance(breakdowns, list) and breakdowns: for m in breakdowns: code = m.get("material_code", "") m_name = m.get("material_name", row["item_name"]) norm_qty = Decimal(str(m.get("norm_quantity") or 1.0)) unit = m.get("unit", row["unit"]) sub_group = self._map_material_name_to_group(m_name, cat) key = f"{sub_group}::{unit}" if key not in estimated_by_group: estimated_by_group[key] = { "material_group": sub_group, "material_name": m_name, "unit": unit, "estimated_quantity": Decimal("0"), "settled_quantity": Decimal("0"), } estimated_by_group[key]["estimated_quantity"] += q_dt * norm_qty estimated_by_group[key]["settled_quantity"] += q_act * norm_qty else: # Gộp trực tiếp theo Category key = f"{mat_group}::{row['unit']}" if key not in estimated_by_group: estimated_by_group[key] = { "material_group": mat_group, "material_name": row["item_name"], "unit": row["unit"], "estimated_quantity": Decimal("0"), "settled_quantity": Decimal("0"), } estimated_by_group[key]["estimated_quantity"] += q_dt estimated_by_group[key]["settled_quantity"] += q_act # ------------------------------------------------------------------ # 2. ĐỌC THỰC TẾ XUẤT KHO ($Q_{\text{tx}}$) TỪ BẢNG PHIẾU XUẤT KHO # ------------------------------------------------------------------ disp_where = "1=1" disp_params: List[Any] = [] if project_id: disp_where = "d.project_id = %s" disp_params.append(project_id) cur.execute(f""" SELECT w.material_group, w.item_name, d.unit, SUM(d.quantity_dispatched) as total_dispatched, AVG(d.unit_price_vnd) as avg_unit_price, SUM(d.total_value_vnd) as total_dispatched_value FROM erp_warehouse_dispatches d JOIN erp_warehouse_inventory w ON d.inventory_item_id = w.id WHERE {disp_where} GROUP BY w.material_group, w.item_name, d.unit; """, disp_params) dispatch_rows = cur.fetchall() dispatched_by_group: Dict[str, Dict[str, Any]] = {} for r in dispatch_rows: grp = r["material_group"] unit = r["unit"] key = f"{grp}::{unit}" dispatched_by_group[key] = { "material_group": grp, "material_name": r["item_name"], "unit": unit, "dispatched_quantity": Decimal(str(r["total_dispatched"] or 0)), "unit_price_vnd": Decimal(str(r["avg_unit_price"] or 0)), "total_dispatched_value_vnd": Decimal(str(r["total_dispatched_value"] or 0)), } # ------------------------------------------------------------------ # 3. KẾT HỢP MA TRẬN 3 CỘT & TÍNH TỶ LỆ HAO HỤT SO VỚI ĐỊNH MỨC 2026 # ------------------------------------------------------------------ all_keys = set(estimated_by_group.keys()) | set(dispatched_by_group.keys()) variance_items: List[Dict[str, Any]] = [] total_dispatched_val = Decimal("0") total_wastage_cost = Decimal("0") exceeded_count = 0 for key in sorted(all_keys): est_data = estimated_by_group.get(key, {}) disp_data = dispatched_by_group.get(key, {}) grp = est_data.get("material_group") or disp_data.get("material_group") or "OTHER_SUPPLIES" unit = est_data.get("unit") or disp_data.get("unit") or "" mat_name = disp_data.get("material_name") or est_data.get("material_name") or grp q_dt = est_data.get("estimated_quantity", Decimal("0")) q_qt = est_data.get("settled_quantity", Decimal("0")) q_tx = disp_data.get("dispatched_quantity", Decimal("0")) unit_price = disp_data.get("unit_price_vnd", Decimal("0")) val_tx = disp_data.get("total_dispatched_value_vnd", Decimal("0")) total_dispatched_val += val_tx norm_pct = NORM_WASTAGE_PERCENTAGES.get(grp, Decimal("2.5")) # Tính toán hao hụt thực tế # Chuẩn đối chiếu: Ưu tiên Quyết toán (q_qt), nếu chưa có thì so với Dự toán (q_dt) baseline_q = q_qt if q_qt > Decimal("0") else q_dt if baseline_q > Decimal("0"): # Hao hụt khối lượng = Thực xuất - Chuẩn nghiệm thu actual_wastage_qty = q_tx - baseline_q if baseline_q > 0: actual_wastage_pct = (actual_wastage_qty / baseline_q) * Decimal("100") else: actual_wastage_pct = Decimal("0") # Cảnh báo dựa trên tỷ lệ hao hụt định mức nhà nước (TT 12/2021) if actual_wastage_pct <= norm_pct: status = VarianceAlertStatus.SAFE eval_msg = f"Trong định mức cho phép (≤ {norm_pct}%). Quản lý hao hụt tốt." elif actual_wastage_pct <= norm_pct * Decimal("1.2"): status = VarianceAlertStatus.WARNING eval_msg = f"Tiệm cận ngưỡng định mức ({actual_wastage_pct:.2f}% vs {norm_pct}%). Cần chú ý bãi kho." else: status = VarianceAlertStatus.EXCEEDED exceeded_count += 1 eval_msg = f"VƯỢT ĐỊNH MỨC ({actual_wastage_pct:.2f}% > {norm_pct}%). Cảnh báo thất thoát hoặc hao phí thi công!" else: # Xuất độc lập không có dự toán bóc tách actual_wastage_qty = Decimal("0") actual_wastage_pct = Decimal("0") status = VarianceAlertStatus.NO_ESTIMATE eval_msg = "Xuất kho độc lập (Chưa gắn dự toán bóc tách CAD)." # Giá trị hao hụt thành tiền (nếu xuất nhiều hơn nghiệm thu) wastage_cost = (max(Decimal("0"), actual_wastage_qty)) * unit_price total_wastage_cost += wastage_cost variance_items.append({ "material_group": grp, "material_name": mat_name, "unit": unit, "estimated_quantity": float(round(q_dt, 4)), "settled_quantity": float(round(q_qt, 4)), "dispatched_quantity": float(round(q_tx, 4)), "norm_wastage_pct": float(round(norm_pct, 2)), "actual_wastage_quantity": float(round(actual_wastage_qty, 4)), "actual_wastage_pct": float(round(actual_wastage_pct, 2)), "variance_status": status.value, "unit_price_vnd": float(round(unit_price, 2)), "wastage_cost_vnd": float(round(wastage_cost, 2)), "evaluation_message": eval_msg, }) # Tính tỷ lệ hao hụt trung bình của các mục có dự toán valid_pcts = [item["actual_wastage_pct"] for item in variance_items if item["variance_status"] != VarianceAlertStatus.NO_ESTIMATE.value] avg_pct = sum(valid_pcts) / len(valid_pcts) if valid_pcts else 0.0 return { "project_id": project_id, "total_materials_monitored": len(variance_items), "total_dispatched_value_vnd": float(round(total_dispatched_val, 2)), "total_wastage_cost_vnd": float(round(total_wastage_cost, 2)), "average_wastage_pct": round(avg_pct, 2), "exceeded_items_count": exceeded_count, "items": variance_items, } @staticmethod def _map_category_to_material_group(category: str, name: str) -> str: """Ánh xạ danh mục công tác sang nhóm vật tư chính.""" c = category.lower() n = name.lower() if "rebar" in c or "thép" in n or "cốt thép" in n: return "STEEL" if "concrete" in c or "bê tông" in n: return "CONCRETE" if "cement" in c or "xi măng" in n: return "CEMENT" if "masonry" in c or "gạch" in n: return "BRICK" if "earthwork" in c or "đất" in n or "cát" in n or "đá" in n: return "AGGREGATE" if "paint" in c or "sơn" in n or "chống thấm" in n: return "FINISHING_PAINT" if "pipe" in c or "ống" in n: return "PIPES_FITTINGS" return "OTHER_SUPPLIES" @staticmethod def _map_material_name_to_group(name: str, fallback_cat: str) -> str: """Nhận diện nhóm vật tư từ tên chi tiết trong bảng phân rã định mức.""" nl = name.lower() if any(k in nl for k in ["thép", "rebar", "buộc"]): return "STEEL" if any(k in nl for k in ["xi măng", "pcb40", "chinfon"]): return "CEMENT" if any(k in nl for k in ["cát", "đá 1x2", "đá 4x6", "đá dăm"]): return "AGGREGATE" if any(k in nl for k in ["bê tông", "concrete"]): return "CONCRETE" if any(k in nl for k in ["gạch", "ngói"]): return "BRICK" if any(k in nl for k in ["sơn", "sika", "chống thấm"]): return "FINISHING_PAINT" if any(k in nl for k in ["ống", "nhựa", "cút"]): return "PIPES_FITTINGS" if any(k in nl for k in ["dầu", "xăng", "diesel"]): return "FUEL_OIL" return WarehouseDispatchService._map_category_to_material_group(fallback_cat, name)