import asyncio import json import os,sys BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../")) sys.path.insert(0, BASE_DIR) from function.file import extract_file as extract_file import function.filter.filter_role as filter_role_1 import function.filter.filter_sql_injection as filter_sql_injection_1 import function.filter.result as query_result_1 import function.chat as chat_sql import function.gemini_response.response_general as res_general import function.gemini_response.response_hello as res_hello import cloudinary import cloudinary.uploader from typing import List,Any import cloudinary from cloudinary.utils import cloudinary_url from cloudinary import uploader from typing import Optional import asyncio from models.Database_Entity import StopSignal async def check_should_stop(chat_id: str, stop_event:Optional[asyncio.Event]): # Trường hợp dừng qua RAM (in-memory) await asyncio.sleep(0.1) if stop_event and stop_event.is_set(): print("🛑 Dừng qua stop_event.") return {"status": "cancelled"} # Trường hợp dừng qua MongoDB await asyncio.sleep(0.1) if StopSignal.objects(chat_history= chat_id, is_stopped=True).first(): print("🛑 Dừng vì có StopSignal trong DB.") return {"status": "cancelled"} return None # Dán nguyên chuỗi URL này vào cloudinary.config( cloud_name = "dgj8n2ggn", api_key = "287331723817617", api_secret = "MAI7dpoikIy-ap7IjPCiwfRgMQ0", secure=True ) def upload_images_to_cloudinary_and_markdown(folder_path): markdown_images = "" image_files = [ f for f in os.listdir(folder_path) if f.lower().endswith(('.png', '.jpg', '.jpeg')) ] for img in image_files: img_path = os.path.join(folder_path, img) try: upload_result = cloudinary.uploader.upload(img_path) print(img_path) image_url = upload_result.get("secure_url") print(image_url) if image_url: markdown_images += f"![{img}]({image_url})\n\n" except Exception as e: print(f"❌ Upload lỗi với ảnh {img}: {e}") return markdown_images async def agent_information(task,user_id,languages, role, chat_id,token,stop_event: Optional[asyncio.Event]): result_check = await check_should_stop(chat_id,stop_event) if result_check: return result_check text_all= extract_file.extract_data2() result_check = await check_should_stop(chat_id,stop_event) if result_check: return result_check task = filter_sql_injection_1.normalize_query(task) print("text: ",text_all) result_check = await check_should_stop(chat_id,stop_event) if result_check: return result_check return extract_file.handle_query_upgrade_keyword_old(f"{task}",text_all,"demohmdrinks") async def agent_hello(task,user_id,languages, role,chat_id,token,stop_event: Optional[asyncio.Event]): result_check = await check_should_stop(chat_id,stop_event) if result_check: return result_check result = await res_hello.response_hello(task) result_check = await check_should_stop(chat_id,stop_event) if result_check: return result_check return result from mongoengine import connect from dotenv import load_dotenv load_dotenv() MONGO_URI = os.getenv("MONGO_URI", "") connect("chatbot_hmdrinks", host=MONGO_URI) from models.Database_Entity import User, ChatHistory, DetailChat def get_chat_history(chat_history_id): messages = DetailChat.objects(chat_history=chat_history_id).order_by('timestamp') history_text = "" for msg in messages: history_text += f"Người dùng hỏi: {msg.you_message}\n" history_text += f"Trợ lý: {msg.ai_message}\n\n" return history_text.strip() from mongoengine import connect from dotenv import load_dotenv load_dotenv() MONGO_URI = os.getenv("MONGO_URI", "") connect("chatbot_hmdrinks", host=MONGO_URI) import sys import os project_root = os.path.abspath(os.path.join(os.getcwd(), '..', '..', '..')) if project_root not in sys.path: sys.path.append(project_root) from models.Database_Entity import User, ChatHistory, DetailChat,ChatCart, CartProduct import base64 from advance_shopping.prompt import context_prompt, multitool_prompt, user_intent from advance_shopping.call_gemini import tool_call from advance_shopping.server_java import server_java from advance_shopping.function import test as test_func def get_chat_history(chat_history_id): messages = DetailChat.objects(chat_history=chat_history_id).order_by('timestamp') print(DetailChat._meta) # in ra cấu hình model history_text = "" for msg in messages: history_text += f"Người dùng: {msg.you_message}\n" history_text += f"Trợ lý: {msg.ai_message}\n\n" return history_text.strip() from bson import ObjectId def chat_history_has_unconfirmed_cart(chat_history_id: str) -> bool: """ Trả về True nếu trong đoạn chat có giỏ hàng chưa được xác nhận đặt hàng. Trả về False nếu không có giỏ hàng, hoặc tất cả giỏ hàng đã được xác nhận. """ return ChatCart.objects( chat_history=ObjectId(chat_history_id), status="pending" ).count() > 0 from bson import ObjectId def process_data(structured_data, data): list_data1 = structured_data[:] print("data",data) if data == "None": return structured_data if isinstance(data, list) and not data: print("✅ Đây là một list rỗng.") return structured_data if data: for item in data: action = item.get("action", "").lower() name = item.get("name", "").strip().lower() size = item.get("size", "").strip().lower() size_old = item.get("size_old", "").strip().lower() if "size_old" in item else None matched = False if action == "update" and size_old: for item_old in list_data1: name_old = item_old.get("name", "").strip().lower() size_old_in_list = item_old.get("size", "").strip().lower() if name == name_old and size_old == size_old_in_list: matched = True # Cập nhật các trường for key in ["quantity", "size"]: if key in item: item_old[key] = item[key] break else: for item_old in list_data1: name_old = item_old.get("name", "").strip().lower() size_old_in_list = item_old.get("size", "").strip().lower() if name == name_old and size == size_old_in_list: matched = True if action == "delete": list_data1.remove(item_old) break elif action == "update": for key in ["quantity", "size"]: if key in item: item_old[key] = item[key] break elif action == "add": if "quantity" in item: item_old["quantity"] = int(item_old.get("quantity", 0)) + int(item["quantity"]) break if not matched and action == "add": item_to_add = {k: v for k, v in item.items() if k != "action"} list_data1.append(item_to_add) filtered_list = [] for item in list_data1: if (item.get("name") not in [None, "", "None"]): filtered_list.append(item) return filtered_list def get_last_ai_message(chat_history_id: str) -> str: last_detail = DetailChat.objects( chat_history=ObjectId(chat_history_id), ai_message__ne=None ).order_by('-timestamp').first() return last_detail.ai_message if last_detail else "None" # async def agent_sql(task,user_id,languages, role, chat_id,token,stop_event: Optional[asyncio.Event] = None): # check_analyze = filter_sql_injection_1.filter_task_analyze(task) # check_restore_delete = filter_sql_injection_1.filter_question_remove_restore(task) # prompt_check_shopping = "" # print(check_restore_delete) # if check_restore_delete == True: # return "Bạn không được phép khôi phục hoặc xóa dữ liệu. Vui lòng kiểm tra lại câu hỏi của bạn." # print(check_analyze) # if check_analyze == False: # result_final = "" # cart_status = "" # if chat_history_has_unconfirmed_cart(chat_id) is True: # cart_status = "Yes (Đang Pending)" # else: # cart_status = "No (Không có)" # print(cart_status) # prompt_check_shopping = user_intent.build_detailed_user_intent_prompt(user_question=task, # chat_history=get_chat_history(chat_id), # cart_status=cart_status) # check_shopping = tool_call.generate(prompt_check_shopping) # print(check_shopping) # type = check_shopping["type"] # if str(type) == "shopping_question": # tools = ["confirm_order", "insert_cart", "update_cart", "delete_item_cart", "deleted all cart"] # last_ai_message = get_last_ai_message(chat_id) # context_prompt = multitool_prompt.build_multi_tool_prompt(question=last_ai_message,tools=tools,user_answer=task) # data_context = {} # try: # data_context = tool_call.generate(context_prompt) # if not data_context["question"]: # data_context = tool_call.generate(context_prompt) # except: # data_context = tool_call.generate(context_prompt) # if not data_context["question"]: # data_context = tool_call.generate(context_prompt) # print("data_context: ",data_context) # question_new = data_context["question"] # latest_unconfirmed_cart = ChatCart.objects( # chat_history=chat_id, # status="pending" # ).order_by('-created_at').first() # previous_data = [] # if latest_unconfirmed_cart: # for product in latest_unconfirmed_cart.cart_products: # product_dict = product.to_mongo().to_dict() # if "_id" in product_dict: # product_dict["_id"] = str(product_dict["_id"]) # previous_data.append(product_dict) # data = test_func.classify_data(input = f"""{question_new}""", structured_data=previous_data) # list_data = list() # print(data) # if data == "None" or data == '': # data = list() # else: # try: # list_data = json.loads(data) # except: # list_data = list() # print("List data:", list_data) # print("Preious_data", previous_data) # previous_data = list(previous_data) # list_data1 = previous_data # clean_list = list() # full_test = "" # result = [] # print(list_data) # try: # print("Nhay vo test2") # result = process_data(list_data1, list_data) # except: # if data is None or data == "None": # print("Nhay vo test1") # result = process_data(list_data1, []) # clean_cart_products = [] # chat_cart = ChatCart() # if list_data: # annotated_list, clean_list, full_test = test_func.validate_product_list(data_list=result,server_java=server_java) # print(result) # print(clean_list) # results = [] # if data_context["question_normal"]: # for key, q in data_context["question_normal"].items(): # task_1 = filter_sql_injection_1.normalize_query(q) # filtered_input = filter_sql_injection_1.filter_sql_injection(task_1) # filtered_role_input = filter_role_1.filter_role(filtered_input) # result = await chat_sql.execute_query_user(filtered_role_input, user_id, languages, role) # result_final = query_result_1.query_result(task, result) # results.append({"question": q, "result": result_final}) # result_final += str(results) # for item in clean_list: # # Đảm bảo quantity là số nguyên # product = CartProduct( # name=item.get('name'), # size=item.get('size'), # quantity=int(item.get('quantity', 1)) # ) # clean_cart_products.append(product) # print("Clean_cart", clean_cart_products) # print("Full test", full_test) # if latest_unconfirmed_cart: # print("🛒 Đã có giỏ hàng 'pending' trước đó. Đang cập nhật lại sản phẩm...") # latest_unconfirmed_cart.cart_products = clean_cart_products # latest_unconfirmed_cart.save() # chat_cart = latest_unconfirmed_cart # else: # print("Chưa có giỏ hàng 'pending'. Đang tạo mới...") # chat_cart = ChatCart( # chat_history=chat_id, # cart_products=clean_cart_products, # status="pending" # ) # chat_cart.save() # if full_test: # return full_test + f"Hãy tạo ra câu hỏi để người dùng thay đổi ý định mua sản phẩm. Ngoài ra luôn hiển thị lại tình trạng giỏ của người dùng: giỏ hàng{clean_list}" + str(results) # if check_shopping.get("is_delete_cart") == "Yes": # print("🗑️ Đang xóa toàn bộ sản phẩm trong giỏ hàng...") # chat_cart.cart_products = [] # chat_cart.status = "failed" # chat_cart.save() # result_final += "Đã xóa toàn bộ sản phẩm trong giỏ hàng..." # if check_shopping.get("is_view_cart") == "Yes": # print("👀 Đang xem giỏ hàng...") # chat_cart = ChatCart.objects(chat_history=chat_id, status="pending").first() # if not chat_cart or not chat_cart.cart_products: # print("🛒 Giỏ hàng hiện đang trống.") # result_final += "🛒 Giỏ hàng hiện đang trống.\n" # else: # message_lines = ["🛒 Danh sách sản phẩm trong giỏ:"] # for product in chat_cart.cart_products: # print(f"- Tên sản phẩm: {product.name} | Số Lượng: {product.quantity} | Size(Kích cỡ): {product.size}") # message_lines.append( # f"- Tên sản phẩm: {product.name} | Số Lượng: {product.quantity} | Size(Kích cỡ): {product.size}" # ) # data = "\n".join(message_lines) # result_final += data + "\n" # if check_shopping["is_confirm_order"] == "Yes": # try: # print("🔹 [STEP 1] Tạo giỏ hàng mới...") # data = server_java.create_cart(int(user_id), token) # cart_id = data.get("cartId") # print(f"✅ Giỏ hàng tạo thành công: cart_id = {cart_id}") # except Exception as e: # print(f"❌ Lỗi khi tạo giỏ hàng: {e}") # cart_id = None # if cart_id: # for idx, item in enumerate(clean_cart_products, 1): # try: # print(f"\n🔹 [STEP 2.{idx}] Tìm kiếm sản phẩm: {item['name']} (size: {item['size']})...") # pro_id, pro_name, size_check, stock_check, price_check = server_java.search_product( # keyword=item["name"], size=item["size"] # ) # print(f"✅ Tìm thấy sản phẩm '{pro_name}' (ID: {pro_id}, Size: {size_check})") # print(f"🔹 [STEP 3.{idx}] Thêm sản phẩm vào giỏ hàng...") # data1 = server_java.insert_cartItem(user_id, cart_id, pro_id, size_check, item["quantity"], token) # print(f"✅ Đã thêm sản phẩm '{pro_name}' vào giỏ hàng.") # except Exception as e: # print(f"❌ Lỗi khi xử lý sản phẩm thứ {idx}: {e}") # continue # vẫn tiếp tục vòng lặp cho sản phẩm tiếp theo # try: # print("\n🔹 [STEP 4] Tạo đơn hàng từ giỏ hàng...") # create_orders = server_java.create_orders(user_id=user_id, cartId=cart_id, token=token) # order_id = create_orders["body"]["orderId"] # print(f"✅ Đơn hàng được tạo: order_id = {order_id}") # except Exception as e: # print(f"❌ Lỗi khi tạo đơn hàng: {e}") # order_id = None # if order_id: # try: # print("🔹 [STEP 5] Xác nhận đơn hàng với AI...") # confirm_order = server_java.confirm_orders_AI(user_id=user_id, token=token, orderId=order_id) # print("✅ Phản hồi xác nhận đơn hàng từ AI:") # print(confirm_order["body"]) # if confirm_order["body"] == "Confirm success": # chat_cart.status = "confirmed" # chat_cart.confirmed_order = True # chat_cart.save() # from urllib.parse import urlencode # base_url = "https://vnexpress.net/tin-tuc-24h" # params = { # "order_id": order_id, # "token": token, # "user_id": user_id # } # link_payment = f"https://vnexpress.net/tin-tuc-24h?order_id={order_id}&token={token}&user_id={user_id}" # return "Yêu cầu: " + str(task) + f". Yêu cầu của bạn đã thành công. Tình trạng giỏ hiện tại (Luôn luôn hiển thị lại trạng thái này để người dùng biết giỏ của họ) {clean_list}. Bạn đã xác nhận thanh toán thành công sẽ tiến hành chuyển sang link thanh toán{link_payment}. Vui lòng thực hiện thanh toán trên trang này trong vòng 30 phút tối đa" + str(results) # else: # chat_cart.status = "failed" # chat_cart.confirmed_order = False # chat_cart.save() # except Exception as e: # print(f"❌ Lỗi khi xác nhận đơn hàng với AI: {e}") # chat_cart.status = "failed" # chat_cart.confirmed_order = False # chat_cart.save() # if check_shopping["is_confirm_order"] == "No" and len(clean_cart_products) != 0 : # result_final += "Yêu cầu: " + str(task) + f". Yêu cầu của bạn đã thành công. Tình trạng giỏ hiện tại(Luôn luôn hiển thị lại trạng thái này để người dùng biết giỏ của họ): {clean_list}" + "kết quả thực thi:" + str(results) # return result_final # else: # task = filter_sql_injection_1.normalize_query(task) # filtered_input = filter_sql_injection_1.filter_sql_injection(task) # filtered_role_input = filter_role_1.filter_role(filtered_input) # result = await chat_sql.execute_query_user(filtered_role_input, user_id, languages, role) # result_final = query_result_1.query_result(task, result) # return result_final # else: # result = "" # task = filter_sql_injection_1.normalize_query(task) # filtered_input = filter_sql_injection_1.filter_sql_injection(task) # filtered_role_input = filter_role_1.filter_role(filtered_input) # output = await chat_sql.generate_and_save_code(filtered_role_input, user_id, role, languages) # if isinstance(output, tuple): # result, path_image = output # else: # result = output # path_image = None # if path_image and os.path.exists(path_image): # images_md = upload_images_to_cloudinary_and_markdown(path_image) # if images_md: # result = f"### Tất cả Hình ảnh sinh ra:\n\n{images_md}\n\n{result}" # return result # # ----- PHẦN 2: HÀM AGENT_SQL ĐÃ TÁI CẤU TRÚC ----- # # --- Các hàm trợ giúp (Helper Functions) --- from typing import Optional async def _handle_shopping_intent(task: str, user_id: int, chat_id: Any, token: str, languages,role, stop_event: Optional[asyncio.Event]) -> str: """ Xử lý các tác vụ liên quan đến mua sắm: xem giỏ, thêm/sửa/xóa, xác nhận đơn hàng. """ result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check try: result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check cart_status = "Yes (Đang Pending)" if chat_history_has_unconfirmed_cart(chat_id) else "No (Không có)" prompt_context = multitool_prompt.build_multi_tool_prompt( question=get_last_ai_message(chat_id), tools=["confirm_order", "insert_cart", "update_cart", "delete_item_cart", "deleted all cart"], user_answer=task ) data_context = tool_call.generate(prompt_context) print(f"DEBUG: data_context: {data_context}") # 2. Lấy giỏ hàng đang chờ xử lý hoặc tạo mới result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check latest_unconfirmed_cart = ChatCart.objects(chat_history=chat_id, status="pending").order_by('-created_at').first() if latest_unconfirmed_cart: chat_cart = latest_unconfirmed_cart previous_cart_items = [p.to_mongo().to_dict() for p in chat_cart.cart_products] print("INFO: Đã tìm thấy giỏ hàng 'pending', sẽ cập nhật.") else: chat_cart = ChatCart(chat_history=chat_id, status="pending") previous_cart_items = [] print("INFO: Không tìm thấy giỏ hàng 'pending', sẽ tạo mới.") #Backup data chat_cart.backup_cart_products = chat_cart.cart_products.copy() chat_cart.backup_status = chat_cart.status chat_cart.backup_confirmed_order = chat_cart.confirmed_order chat_cart.save() # 3. Trích xuất và xử lý thông tin sản phẩm từ câu hỏi của người dùng result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check new_items_str = test_func.classify_data(input=f"{data_context.get('question', '')}", structured_data=previous_cart_items) new_items_list = json.loads(new_items_str) if new_items_str and new_items_str != "None" else [] updated_cart_list = process_data(previous_cart_items, new_items_list) # 4. Xác thực sản phẩm và cập nhật giỏ hàng trong DB result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check annotated_list, clean_list, validation_message = test_func.validate_product_list(data_list=updated_cart_list, server_java=server_java) # clean_cart_products = [CartProduct(**item) for item in clean_list] clean_cart_products = [] for item in clean_list: # Đảm bảo quantity là số nguyên product = CartProduct( name=item.get('name'), size=item.get('size'), quantity=int(item.get('quantity', 1)) ) clean_cart_products.append(product) print("Clean_cart", clean_cart_products) chat_cart.cart_products = clean_cart_products chat_cart.save() print(f"INFO: Giỏ hàng đã được cập nhật với {len(clean_cart_products)} sản phẩm.") # 5. Xử lý các câu hỏi phụ (nếu có) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check other_results = [] if data_context.get("question_normal"): for q_key, q_text in data_context["question_normal"].items(): result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check result = await _handle_simple_query(q_text, user_id, languages, role, stop_event, chat_id) other_results.append({"question": q_text, "result": result}) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check response_parts = [] if validation_message: response_parts.append(validation_message) final_cart_state_msg = f"Tình trạng giỏ hàng hiện tại của bạn: {clean_list}" result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check shopping_intent = tool_call.generate(user_intent.build_detailed_user_intent_prompt( user_question=task, chat_history=get_chat_history(chat_id), cart_status=cart_status )) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check if shopping_intent.get("is_delete_cart") == "Yes": chat_cart.cart_products = [] chat_cart.status = "failed" chat_cart.save() response_parts.append("Đã xóa toàn bộ sản phẩm trong giỏ hàng.") final_cart_state_msg = "Tình trạng giỏ hàng hiện tại của bạn: Giỏ hàng trống." elif shopping_intent.get("is_view_cart") == "Yes": result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check response_parts.append("Đây là giỏ hàng của bạn.") elif shopping_intent.get("is_confirm_order") == "Yes": result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check if not chat_cart.cart_products: return "Giỏ hàng của bạn đang trống. Vui lòng thêm sản phẩm trước khi xác nhận đơn hàng." result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check confirmation_result = await _execute_order_confirmation(user_id, token, chat_cart, stop_event,chat_id) confirmation_result_str = "" result_check = await check_should_stop(chat_id, stop_event) if result_check: chat_cart.status = chat_cart.backup_status chat_cart.confirmed_order = chat_cart.backup_confirmed_order chat_cart.save() if confirmation_result[1]: server_java.confirm_cancel_AI(int(user_id),token,confirmation_result[1]) server_java.remove_all_cartItem(int(user_id), confirmation_result[2], token) confirmation_result_str = confirmation_result[0] else: confirmation_result_str = confirmation_result return result_check response_parts.append(str(confirmation_result)) else: response_parts.append(f"Yêu cầu '{task}' của bạn đã được xử lý.") response_parts.append(final_cart_state_msg) if other_results: response_parts.append(f"Kết quả cho các câu hỏi khác: {other_results}") return " ".join(response_parts) except asyncio.CancelledError as e: print(f"STOP: {e}") if 'chat_cart' in locals(): if chat_cart.backup_cart_products: chat_cart.cart_products = chat_cart.backup_cart_products chat_cart.status = chat_cart.backup_status chat_cart.confirmed_order = chat_cart.backup_confirmed_order chat_cart.save() print("INFO: Đã khôi phục giỏ hàng từ bản backup do bị hủy.") return "Quá trình đã bị dừng theo yêu cầu. Giỏ hàng đã được khôi phục như ban đầu." # except Exception as e: # print(f"ERROR in _handle_shopping_intent: {e}") # return "Đã có lỗi xảy ra trong quá trình xử lý yêu cầu mua sắm của bạn. Vui lòng thử lại." # async def rollback_cart_and_order(chat_cart: ChatCart, user_id: int, token: str): # try: # if chat_cart.temp_order_id: # server_java.delete_order(chat_cart.temp_order_id, token) # print(f"INFO: Đã rollback đơn hàng {chat_cart.temp_order_id}") # if chat_cart.temp_cart_id: # server_java.delete_cart(chat_cart.temp_cart_id, token) # print(f"INFO: Đã rollback giỏ hàng {chat_cart.temp_cart_id}") # chat_cart.status = "cancelled" # chat_cart.save() # except Exception as e: # print(f"WARNING: Lỗi khi rollback: {e}") async def rollback_cart_and_order(chat_cart: ChatCart, user_id: int, token: str,order_id): try: if chat_cart.temp_order_id: server_java.delete_order(chat_cart.temp_order_id, token) print(f"INFO: Đã rollback đơn hàng {chat_cart.temp_order_id}") if chat_cart.temp_cart_id: server_java.delete_cart(chat_cart.temp_cart_id, token) print(f"INFO: Đã rollback giỏ hàng {chat_cart.temp_cart_id}") chat_cart.status = "cancelled" chat_cart.save() except Exception as e: print(f"WARNING: Lỗi khi rollback: {e}") async def _execute_order_confirmation(user_id: int, token: str, chat_cart: ChatCart, stop_event: Optional[asyncio.Event],chat_id:str= ""): """ Thực hiện quy trình 5 bước để xác nhận đơn hàng với server_java. """ try: #Backup data chat_cart.backup_status = chat_cart.status chat_cart.backup_confirmed_order = chat_cart.confirmed_order chat_cart.save() print("INFO: Bắt đầu quy trình xác nhận đơn hàng.") result_check = await check_should_stop(chat_id, stop_event) if result_check: chat_cart.status = chat_cart.backup_status chat_cart.confirmed_order = chat_cart.backup_confirmed_order chat_cart.save() return result_check cart_data = server_java.create_cart(int(user_id), token) cart_id = cart_data.get("cartId") if not cart_id: raise Exception("Không thể tạo giỏ hàng trên hệ thống.") print(f"SUCCESS: Tạo giỏ hàng tạm thành công: cart_id = {cart_id}") # Step 2 & 3: Add items to cart for item in chat_cart.cart_products: result_check = await check_should_stop(chat_id, stop_event) if result_check: chat_cart.status = chat_cart.backup_status chat_cart.confirmed_order = chat_cart.backup_confirmed_order chat_cart.save() rollback = server_java.remove_all_cartItem(int(user_id), cart_id, token) return result_check pro_id, _, size_check, _, _ = server_java.search_product(keyword=item.name, size=item.size) server_java.insert_cartItem(user_id, cart_id, pro_id, size_check, item.quantity, token) print(f"SUCCESS: Đã thêm '{item.name}' vào giỏ hàng tạm.") # Step 4: Create Order result_check = await check_should_stop(chat_id, stop_event) if result_check: chat_cart.status = "pending" chat_cart.confirmed_order = False chat_cart.save() server_java.remove_all_cartItem(int(user_id), cart_id, token) return result_check order_data = server_java.create_orders(user_id=user_id, cartId=cart_id, token=token) order_id = order_data["body"]["orderId"] if not order_id: raise Exception("Không thể tạo đơn hàng từ giỏ hàng.") print(f"SUCCESS: Tạo đơn hàng thành công: order_id = {order_id}") # Step 5: Confirm Order result_check = await check_should_stop(chat_id, stop_event) if result_check: chat_cart.status = chat_cart.backup_status chat_cart.confirmed_order = chat_cart.backup_confirmed_order chat_cart.save() server_java.confirm_cancel_AI(int(user_id),token,order_id) server_java.remove_all_cartItem(int(user_id), cart_id, token) return result_check confirm_data = server_java.confirm_orders_AI(user_id=user_id, token=token, orderId=order_id) if confirm_data.get("body") == "Confirm success": chat_cart.status = "confirmed" chat_cart.confirmed_order = True chat_cart.order_id = order_id chat_cart.cart_id = cart_id chat_cart.save() link_payment = f"https://hmdrinks-t8e3-giahans-projects.vercel.app/payment?order_id={order_id}&token={token}&user_id={user_id}" data = f"Đơn hàng của bạn đã được xác nhận thành công. Vui lòng truy cập link sau để thanh toán trong vòng 30 phút: {link_payment}(Vui lòng không bỏ sót link này)" return data, order_id, cart_id else: print(f"Hệ thống không thể xác nhận đơn hàng. Phản hồi: {confirm_data.get('body')}") result_check = await check_should_stop(chat_id, stop_event) if result_check: chat_cart.status = chat_cart.backup_status chat_cart.confirmed_order = chat_cart.backup_confirmed_order chat_cart.save() server_java.confirm_cancel_AI(int(user_id),token,order_id) server_java.remove_all_cartItem(int(user_id), cart_id, token) except Exception as e: print(f"ERROR in _execute_order_confirmation: {e}") chat_cart.status = "failed" chat_cart.save() return f"Đã xảy ra lỗi khi xác nhận đơn hàng: {e}. Giỏ hàng của bạn vẫn được lưu." async def _handle_simple_query(task: str, user_id: int, languages: List[str], role: str,stop_event: Optional[asyncio.Event],chat_id: str = "") -> str: """Xử lý một câu hỏi SQL đơn giản.""" result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check normalized_task = filter_sql_injection_1.normalize_query(task) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check filtered_input = filter_sql_injection_1.filter_sql_injection(normalized_task) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check filtered_role_input = filter_role_1.filter_role(filtered_input) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check query_result = await chat_sql.execute_query_user(filtered_role_input, user_id, languages, role) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check return query_result_1.query_result(task, query_result) async def _handle_analysis_intent(task: str, user_id: int, languages: List[str], role: str, stop_event: Optional[asyncio.Event],chat_id: str = "") -> str: """Xử lý các tác vụ yêu cầu phân tích, sinh code và vẽ biểu đồ.""" result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check normalized_task = filter_sql_injection_1.normalize_query(task) filtered_input = filter_sql_injection_1.filter_sql_injection(normalized_task) filtered_role_input = filter_role_1.filter_role(filtered_input) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check output = await chat_sql.generate_and_save_code( filtered_role_input, user_id, role, languages, stop_event, chat_id=chat_id ) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check result, path_image = output if isinstance(output, tuple) else (output, None) if path_image and os.path.exists(path_image): result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check # Trả về kết quả đã có images_md = upload_images_to_cloudinary_and_markdown(path_image) if images_md: result = f"### Biểu đồ được tạo:\n\n{images_md}\n\n{result}" return result # --- Hàm chính (Main Function) --- async def agent_sql(task: str, user_id: int, languages: str, role: str, chat_id: Any, token: str, stop_event: Optional[asyncio.Event] = None): """ Hàm chính điều phối các tác vụ, có khả năng dừng giữa chừng. """ # KIỂM TRA DỪNG NGAY LẬP TỨC result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check # 1. KIỂM TRA AN TOÀN BAN ĐẦU if filter_sql_injection_1.filter_question_remove_restore(task): return "Bạn không được phép khôi phục hoặc xóa dữ liệu. Vui lòng kiểm tra lại câu hỏi của bạn." # 2. PHÂN LOẠI Ý ĐỊNH CHÍNH: PHÂN TÍCH (cần sinh code) vs HỎI ĐÁP/MUA SẮM is_analysis_task = filter_sql_injection_1.filter_task_analyze(task) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check if is_analysis_task: print("INFO: Nhận diện ý định PHÂN TÍCH.") if result_check: return result_check data = await _handle_analysis_intent(task, user_id, languages, role, stop_event,chat_id) if result_check: return result_check return data, False # 3. PHÂN LOẠI Ý ĐỊNH PHỤ: MUA SẮM vs HỎI ĐÁP THÔNG THƯỜNG result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check cart_status = "Yes (Đang Pending)" if chat_history_has_unconfirmed_cart(chat_id) else "No (Không có)" prompt_check_shopping = user_intent.build_detailed_user_intent_prompt( user_question=task, chat_history=get_chat_history(chat_id), cart_status=cart_status ) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check intent_result = tool_call.generate(prompt_check_shopping) result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check if str(intent_result.get("type")) == "shopping_question": print("INFO: Nhận diện ý định MUA SẮM.") result = await _handle_shopping_intent(task, user_id, chat_id, token, languages, role, stop_event) result = query_result_1.query_result(task, result) return result, True else: print("INFO: Nhận diện ý định HỎI ĐÁP THÔNG THƯỜNG.") result = await _handle_simple_query(task, user_id, languages, role, stop_event,chat_id) result = query_result_1.query_result(task, result) return result, False async def agent_general(task, user_id, languages, role, chat_id, token, stop_event): task = filter_sql_injection_1.normalize_query(task) # Kiểm tra stop event trước khi gọi xử lý nặng result_check = await check_should_stop(chat_id, stop_event) if result_check: return result_check # Gọi xử lý chính result_task = asyncio.create_task(res_general.response_general(task)) # Chờ xử lý, nhưng cũng kiểm tra stop event while not result_task.done(): if stop_event.is_set(): print("Tín hiệu stop trong quá trình xử lý. Hủy task.") result_task.cancel() try: await result_task except asyncio.CancelledError: print("Task đã bị huỷ.") return None await asyncio.sleep(0.1) # Cho các task khác thời gian chạy function_mapping = { "question_hello": agent_hello, "question_information": agent_information, "question_sql": agent_sql, "question_general": agent_general } from typing import Optional async def process_toolcalls_with_order(toolcalls,token, stop_event: Optional[asyncio.Event] = None, chat_id:str = ""): task_results = {} ordered_results = {} filtered_toolcalls = [tc for tc in toolcalls if tc["name"] in function_mapping] for tc in filtered_toolcalls: orig_name = tc["name"] task_func = function_mapping.get(orig_name) if task_func is None: continue task_param = tc["args"].get("query", "") user_id = tc.get("user_id", "unknown_user") role = tc.get("role", "unknown_role") language = tc.get("language", "unknown_language") chat_id = tc.get("chat_id","unknow") if asyncio.iscoroutinefunction(task_func): result = await task_func(task_param, user_id, language,role,chat_id,token,stop_event) else: result = task_func(task_param, user_id, language,role,chat_id,token,stop_event) if isinstance(result, tuple) and len(result) == 2: result, is_shopping = result else: result = result is_shopping = False # Lưu kết quả theo id của tool_call task_results[tc["id"]] = { "result": result, "user_id": user_id, "role": role, "language": language, "is_shopping": is_shopping } ordered_results = {tc["id"]: task_results.get(tc["id"], {"result": "Không có kết quả"}) for tc in filtered_toolcalls} return ordered_results