from __future__ import annotations import logging from typing import Any logger = logging.getLogger("dscons.state_price_sync.catalog") class StatePriceCatalogMixin: """Mixin for state price catalog querying and period management.""" def get_available_periods(self) -> List[Dict[str, Any]]: """Return available published periods with stats.""" with self.get_connection() as conn, conn.cursor() as cur: cur.execute(""" SELECT publish_period, region_code, count(*) as total_items, min(document_reference) as doc_ref, min(effective_date) as effective_date FROM erp_state_published_prices GROUP BY publish_period, region_code ORDER BY publish_period DESC, region_code ASC; """) return cur.fetchall() def get_full_state_price_catalog( self, publish_period: str = "2026-08", region_code: str = "HAI_PHONG_KHU_VUC_2", material_group: str | None = None, search_query: str | None = None, page: int = 1, limit: int = 200, ) -> Dict[str, Any]: """Truy xuất BẢNG GIÁ CÔNG BỐ SỞ XÂY DỰNG ĐẦY ĐỦ (Toàn bộ danh mục công bố chính thức) kèm phân trang, tìm kiếm và thống kê nhóm.""" with self.get_connection() as conn: with conn.cursor() as cur: where_clauses = ["publish_period = %s", "region_code = %s"] params: list[Any] = [publish_period, region_code] if material_group and material_group != "ALL": mg = material_group.upper().strip() if mg in ("LABOR", "LABOR_CONSTRUCTION"): where_clauses.append("material_group = 'LABOR'") elif mg in ("PIPES_M_E", "PIPES_DRAINAGE"): where_clauses.append("material_group = 'PIPES_M_E'") elif mg in ("TRANSPORT", "TRANSPORTATION_FREIGHT"): where_clauses.append("material_group = 'TRANSPORT'") elif mg in ("PRECAST_PILES", "PRECAST_PILE_CULVERT"): where_clauses.append("material_group = 'PRECAST_PILES'") elif mg in ("CONCRETE", "CONCRETE_READYMIX"): where_clauses.append("material_group = 'CONCRETE'") elif mg in ("STEEL", "STEEL_REBAR"): where_clauses.append("material_group = 'STEEL'") elif mg == "AGGREGATE_SAND": where_clauses.append( "(material_group = 'AGGREGATE' AND (material_code ILIKE '%%CAT%%' OR material_name ILIKE '%%Cát%%'))" ) elif mg == "AGGREGATE_STONE": where_clauses.append( "(material_group = 'AGGREGATE' AND (material_code ILIKE '%%DA%%' OR material_name ILIKE '%%Đá%%'))" ) else: where_clauses.append("material_group = %s") params.append(mg) if search_query and search_query.strip(): q = f"%{search_query.strip()}%" where_clauses.append( "(material_name ILIKE %s OR material_code ILIKE %s OR specifications ILIKE %s)" ) params.extend([q, q, q]) # Count total matching records count_sql = f"SELECT count(*) as total FROM erp_state_published_prices WHERE {' AND '.join(where_clauses)};" cur.execute(count_sql, params) total_count = cur.fetchone()["total"] # If no records in DB for this period, auto-seed and query again if total_count == 0 and not search_query and not material_group: self.seed_official_state_prices() return self.get_full_state_price_catalog( publish_period, region_code, material_group, search_query, page, limit, ) # Group summary stats cur.execute( """ SELECT material_group, count(*) as count, min(state_unit_price_vnd) as min_price, max(state_unit_price_vnd) as max_price FROM erp_state_published_prices WHERE publish_period = %s AND region_code = %s GROUP BY material_group ORDER BY material_group ASC; """, (publish_period, region_code), ) group_stats = cur.fetchall() # Pagination offset = (page - 1) * limit data_sql = f""" SELECT id, publish_period, region_code, material_group, material_code, material_name, specifications, unit, state_unit_price_vnd, document_reference, effective_date, notes FROM erp_state_published_prices WHERE {" AND ".join(where_clauses)} ORDER BY material_group ASC, material_code ASC LIMIT %s OFFSET %s; """ query_params = list(params) + [limit, offset] cur.execute(data_sql, query_params) items = cur.fetchall() # Format items with float prices for JSON formatted_items = [] for itm in items: formatted_items.append( { "id": str(itm["id"]), "publish_period": itm["publish_period"], "region_code": itm["region_code"], "material_group": itm["material_group"], "material_code": itm["material_code"], "material_name": itm["material_name"], "specifications": itm.get("specifications") or "", "unit": itm["unit"], "state_unit_price_vnd": float(itm["state_unit_price_vnd"]), "document_reference": itm["document_reference"], "effective_date": str(itm["effective_date"]), "notes": itm.get("notes") or "", } ) return { "period": publish_period, "region_code": region_code, "region_name": "Khu vực II: Kiến Thụy, Tiên Lãng, An Lão, Thủy Nguyên (Hải Phòng)" if "2" in region_code else "Khu vực I: Trung tâm TP Hải Phòng", "document_title": formatted_items[0]["document_reference"] if formatted_items else "Thông báo số 558/TB-SXD ngày 07/08/2026 của Sở Xây dựng TP. Hải Phòng", "total_items": total_count, "page": page, "limit": limit, "group_stats": group_stats, "items": formatted_items, } def add_or_update_custom_item( self, publish_period: str, region_code: str, material_group: str, material_code: str, material_name: str, specifications: str | None, unit: str, state_unit_price_vnd: Any, document_reference: str, effective_date: Any, notes: str | None = None, ) -> dict[str, Any]: """Thêm mới hoặc cập nhật một mục đơn giá công bố Sở Xây Dựng.""" from decimal import Decimal sql = """ INSERT INTO erp_state_published_prices ( id, publish_period, region_code, material_group, material_code, material_name, specifications, unit, state_unit_price_vnd, document_reference, effective_date, notes, created_at, updated_at ) VALUES ( gen_random_uuid(), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW() ) ON CONFLICT (publish_period, region_code, material_code) DO UPDATE SET material_name = EXCLUDED.material_name, specifications = EXCLUDED.specifications, unit = EXCLUDED.unit, state_unit_price_vnd = EXCLUDED.state_unit_price_vnd, document_reference = EXCLUDED.document_reference, effective_date = EXCLUDED.effective_date, notes = EXCLUDED.notes, updated_at = NOW() RETURNING *; """ with self.get_connection() as conn, conn.cursor() as cur: cur.execute( sql, ( publish_period, region_code, material_group, material_code, material_name, specifications, unit, Decimal(str(state_unit_price_vnd)), document_reference, effective_date, notes, ), ) saved = cur.fetchone() conn.commit() return dict(saved) if saved else {}