import asyncio import json import logging import os from pathlib import Path from typing import Any from uuid import uuid4 import pydantic try: from google.antigravity import Agent, LocalAgentConfig except ImportError: Agent = None LocalAgentConfig = None from app.core.postgres.erp_client import ErpDatabaseClient from app.core.settings import get_settings logger = logging.getLogger("dscons.antigravity_daemon") erp_client = ErpDatabaseClient() class TakeoffItemSchema(pydantic.BaseModel): item_name: str quantity: float unit: str category: str class TakeoffResultSchema(pydantic.BaseModel): items: list[TakeoffItemSchema] async def _call_agent_with_schema( agent_name: str, system_prompt: str, user_prompt: str, response_schema=None ) -> Any: if Agent is None or LocalAgentConfig is None: logger.warning(f"[AGY_DAEMON] google.antigravity chưa được cài đặt, bỏ qua agent {agent_name}") return None logger.info(f"[AGY_DAEMON] Invoking Agent {agent_name}...") settings = get_settings() config = LocalAgentConfig( model="gemini-3.7-flash", api_key=settings.gemini_api_key, system_instruction=system_prompt, response_schema=response_schema, ) async with Agent(config) as agent: resp = await agent.chat(user_prompt) if response_schema: return await resp.structured_output() return resp.text async def process_task(task_file: Path): logger.info(f"\n[AGY_DAEMON] Bắt đầu xử lý task: {task_file.name}") try: with open(task_file, "r", encoding="utf-8") as f: task = json.load(f) takeoff_id = task.get("takeoff_id") if not takeoff_id: if task_file.exists(): os.remove(task_file) return text_content = task.get("text_layer", "") if len(text_content) > 100000: text_content = text_content[:100000] quynh_sys = "Bạn là AI Quỳnh QS - Chuyên gia Bóc tách Khối lượng." quynh_usr = f"Dữ liệu bản vẽ/HSMT:\n{text_content}\nGợi ý: {task.get('drawing_hint', '')}\n\nHãy trích xuất TẤT CẢ các hạng mục công việc thành danh sách. Gán category phù hợp (concrete, rebar, formwork, earthwork, finishing, civil)." quynh_res = await _call_agent_with_schema( "Quỳnh", quynh_sys, quynh_usr, TakeoffResultSchema ) items = quynh_res.get("items", []) if quynh_res else [] lam_sys = "Bạn là AI Lâm - Kỹ sư Biện pháp thi công." lam_usr = f"Đây là các hạng mục Quỳnh đã bóc:\n{items}\n\nHãy đề xuất THÊM các hạng mục biện pháp thi công/phụ trợ (như bơm nước, lán trại, ván khuôn, cừ Larsen,...) còn thiếu. Nếu không thiếu gì, trả về mảng rỗng. Đừng lặp lại các mục Quỳnh đã bóc." lam_res = await _call_agent_with_schema( "Lâm", lam_sys, lam_usr, TakeoffResultSchema ) lam_items = lam_res.get("items", []) if lam_res else [] all_items = items + lam_items tung_sys = "Bạn là AI Tùng - Giám đốc Kỹ thuật." tung_usr = f"Quỳnh bóc được {len(items)} hạng mục. Lâm bổ sung {len(lam_items)} hạng mục. Hãy đưa ra 1 câu nhận xét chuyên nghiệp duy nhất để báo cáo với người dùng." ai_report_message = await _call_agent_with_schema("Tùng", tung_sys, tung_usr) erp_client.delete_drawing_takeoff_items(takeoff_id) total_cost = 0 total_concrete = 0.0 total_rebar = 0.0 total_formwork = 0.0 total_earthwork = 0.0 from app.modules.inventory.application.state_price_sync.pricing_resolver import ( PricingResolverService, ) resolver = PricingResolverService(erp_client=erp_client) for i, item in enumerate(all_items): item_id = str(uuid4()) qty = float(item.get("quantity", 0)) cat = item.get("category", "civil") name = item.get("item_name", "Không rõ") unit = item.get("unit", "") resolved = resolver.resolve_material_price_sync( keyword=name, context="Bóc tách tự động", material_group="Vật liệu", fallback_price=100000, ) u_price = float(resolved.get("price", 100000)) # Heuristic đơn giản cho các đơn vị 100m if unit in ["100m", "100m2", "100m3", "100 cây"]: u_price = u_price * 100 tot_vnd = round(qty * u_price) total_cost += tot_vnd nl = name.lower() if "bê tông" in nl: total_concrete += qty if "m3" in unit.lower() else 0.0 elif "thép" in nl: total_rebar += qty if "tấn" in unit.lower() else qty / 1000.0 elif "ván khuôn" in nl: total_formwork += qty * 100 if "100m2" in unit.lower() else qty elif "đào" in nl or "đắp" in nl: total_earthwork += qty * 100 if "100m3" in unit.lower() else qty erp_client.create_drawing_takeoff_item( { "id": item_id, "takeoff_id": takeoff_id, "item_order": i + 1, "wbs_code": f"1.{i + 1}", "norm_code": f"TT.{i + 1}", "item_name": name, "category": cat, "dimension_formula": f"AI Tự Động: {qty} {unit}", "unit": unit, "quantity": qty, "unit_price_vnd": u_price, "total_amount_vnd": tot_vnd, "price_source_url": f"/dashboard/material-price-comparison?q={name[:25].replace(' ', '+')}", } ) erp_client.update_drawing_takeoff( takeoff_id, { "takeoff_status": "completed", "confidence_score": 99, "confidence_level": "multi_agent_verified", "ai_report_message": ai_report_message, "ai_analysis_notes": f"Antigravity Daemon Multi-Agent chạy thành công. Quỳnh: {len(items)}, Lâm: {len(lam_items)}", "total_estimated_cost_vnd": total_cost, "total_concrete_volume_m3": round(total_concrete, 2), "total_rebar_weight_tons": round(total_rebar, 3), "total_formwork_area_m2": round(total_formwork, 2), "total_earthwork_volume_m3": round(total_earthwork, 2), }, ) if task_file.exists(): os.remove(task_file) logger.info( f"[AGY_DAEMON] Hoàn thành task {task_file.name}: {len(all_items)} hạng mục." ) except Exception as e: logger.error(f"[AGY_DAEMON] Lỗi xử lý {task_file.name}: {e}") if "takeoff_id" in locals() and takeoff_id: erp_client.update_drawing_takeoff( takeoff_id, { "takeoff_status": "errored", "ai_analysis_notes": f"Antigravity Daemon Error: {e}", }, ) if task_file.exists(): os.remove(task_file) async def start_antigravity_daemon_loop(): logger.info( "🚀 [DSCons] Antigravity Background Service started: listening to storage/antigravity_tasks/..." ) task_dir = Path("storage/antigravity_tasks") task_dir.mkdir(parents=True, exist_ok=True) while True: try: task_files = list(task_dir.glob("task_*.json")) for tf in task_files: await process_task(tf) except Exception as e: logger.error(f"[AGY_DAEMON] Lỗi trong loop: {e}") await asyncio.sleep(2)