from MongoDBConnection import db, users_collection, transactions_collection, products_collection import math import re import uuid import pytz from datetime import datetime, timedelta from model.user_model import UserModel from model.topup_model import TopupModel from bson import ObjectId orders_collection = db["orders"] VN_TZ = pytz.timezone("Asia/Ho_Chi_Minh") class OrderModel: @staticmethod def _order_match(order_id: str, id_khach_hang: str): return { "id_khach_hang": id_khach_hang, "$or": [ {"orderId": order_id}, {"id": order_id}, ], } @staticmethod def _blank_address(): return { "street": "", "city": "", "state": "", "country": "", "zipCode": "", } @staticmethod def _normalize_address(address): if isinstance(address, dict): return { "street": address.get("street", ""), "city": address.get("city", ""), "state": address.get("state", ""), "country": address.get("country", ""), "zipCode": address.get("zipCode", ""), } if isinstance(address, str): return { "street": address, "city": "", "state": "", "country": "", "zipCode": "", } return OrderModel._blank_address() @staticmethod def _normalize_customer(doc): customer = doc.get("customer") if isinstance(customer, dict): return { "name": customer.get("name", ""), "phone": customer.get("phone", doc.get("phone", "")), "email": customer.get("email", doc.get("email", "")), "address": OrderModel._normalize_address( customer.get("address", doc.get("address", {})) ), } return { "name": customer or doc.get("customerName", ""), "phone": doc.get("phone", ""), "email": doc.get("email", ""), "address": OrderModel._normalize_address(doc.get("address", {})), } @staticmethod def _calculate_total(items): if not isinstance(items, list): return 0.0 total = 0.0 for item in items: if not isinstance(item, dict): continue quantity = float(item.get("quantity", 0) or 0) price = float(item.get("price", 0) or 0) shipping = float(item.get("shipping", 0) or 0) tax = float(item.get("tax", 0) or 0) total += quantity * price + shipping + tax return round(total, 2) @staticmethod def _resolve_order_total(doc): total = doc.get("total") try: return round(float(total), 2) except (TypeError, ValueError): return OrderModel._calculate_total(doc.get("items", [])) @staticmethod def _build_uploader_info_from_user(user_doc): if not user_doc: return { "uploadedByName": "", "uploadedByEmail": "", } full_name = f"{user_doc.get('firstName', '')} {user_doc.get('lastName', '')}".strip() return { "uploadedByName": full_name or user_doc.get("email", ""), "uploadedByEmail": user_doc.get("email", ""), } @staticmethod def _get_uploader_info(user_id: str, cache: dict = None): user_id = str(user_id or "").strip() if not user_id: return { "uploadedByName": "", "uploadedByEmail": "", } if cache is not None and user_id in cache: return cache[user_id] user = UserModel.get_user_by_id(user_id) info = OrderModel._build_uploader_info_from_user(user) if cache is not None: cache[user_id] = info return info @staticmethod def _find_uploader_ids_by_keyword(keyword: str): cleaned_keyword = (keyword or "").strip() if not cleaned_keyword: return [] matched_ids = set() regex = {"$regex": re.escape(cleaned_keyword), "$options": "i"} direct_matches = users_collection.find( { "role": "user", "$or": [ {"email": regex}, {"firstName": regex}, {"lastName": regex}, ], }, {"_id": 1}, ) for user in direct_matches: matched_ids.add(str(user["_id"])) lowered_keyword = cleaned_keyword.lower() if " " in lowered_keyword: name_matches = users_collection.find( {"role": "user"}, {"_id": 1, "firstName": 1, "lastName": 1, "email": 1}, ) for user in name_matches: full_name = f"{user.get('firstName', '')} {user.get('lastName', '')}".strip().lower() email = str(user.get("email", "")).lower() if lowered_keyword in full_name or lowered_keyword in email: matched_ids.add(str(user["_id"])) return list(matched_ids) @staticmethod def _normalize_source_order_ids(doc): raw_ids = doc.get("sourceOrderIds") if isinstance(raw_ids, list): result = [] seen = set() for raw in raw_ids: value = str(raw or "").strip() if not value or value in seen: continue seen.add(value) result.append(value) if result: return result fallback_order_id = str(doc.get("orderId") or doc.get("id") or "").strip() if fallback_order_id: return [fallback_order_id] return [] @staticmethod def _is_hidden_from_admin(doc): return doc.get("status") == "Cancelled" and doc.get("cancelledBy") == "user" @staticmethod def _sanitize_order_list(doc, uploader_cache: dict = None): order_id = doc.get("orderId") or doc.get("id") or str(doc["_id"]) customer = OrderModel._normalize_customer(doc) uploaded_by_name = doc.get("uploadedByName", "") uploaded_by_email = doc.get("uploadedByEmail", "") if not uploaded_by_name and not uploaded_by_email: uploader_info = OrderModel._get_uploader_info(doc.get("id_khach_hang"), uploader_cache) uploaded_by_name = uploader_info["uploadedByName"] uploaded_by_email = uploader_info["uploadedByEmail"] raw_items = doc.get("items", []) items = raw_items if isinstance(raw_items, list) else [] item_count = len(items) total = OrderModel._resolve_order_total(doc) return { "id": order_id, "docId": str(doc["_id"]), "date": doc.get("date") or (doc.get("createdAt", "")[:10] if doc.get("createdAt") else ""), "customer": customer.get("name", ""), "tracking": doc.get("tracking") or doc.get("trackingNumber", ""), "trackingLink": doc.get("trackingLink", ""), "status": doc.get("status", "Unpaid"), "uploadedByName": uploaded_by_name, "uploadedByEmail": uploaded_by_email, "sourceOrderIds": OrderModel._normalize_source_order_ids(doc), "cancelledBy": doc.get("cancelledBy", ""), "itemCount": item_count, "total": total, } @staticmethod def _sanitize_order_detail(doc): order_id = doc.get("orderId") or doc.get("id") or str(doc["_id"]) customer = OrderModel._normalize_customer(doc) raw_items = doc.get("items", []) items = raw_items if isinstance(raw_items, list) else [] total = doc.get("total") try: total = float(total) except (TypeError, ValueError): total = OrderModel._calculate_total(items) uploader_info = { "uploadedByName": doc.get("uploadedByName", ""), "uploadedByEmail": doc.get("uploadedByEmail", ""), } if not uploader_info["uploadedByName"] and not uploader_info["uploadedByEmail"]: uploader_info = OrderModel._get_uploader_info(doc.get("id_khach_hang")) return { "orderId": order_id, "docId": str(doc["_id"]), "customer": customer, "items": items, "status": doc.get("status", "Unpaid"), "total": round(total, 2), "tracking": doc.get("tracking") or doc.get("trackingNumber", ""), "trackingLink": doc.get("trackingLink", ""), "uploadedByName": uploader_info["uploadedByName"], "uploadedByEmail": uploader_info["uploadedByEmail"], "sourceOrderIds": OrderModel._normalize_source_order_ids(doc), "cancelledBy": doc.get("cancelledBy", ""), "refundAmount": doc.get("refundAmount", 0), } @staticmethod def _build_date_filter(date_from=None, date_to=None): if not date_from and not date_to: return None date_condition = {} created_at_condition = {} if date_from: date_condition["$gte"] = date_from created_at_condition["$gte"] = f"{date_from}T00:00:00" if date_to: date_condition["$lte"] = date_to created_at_condition["$lte"] = f"{date_to}T23:59:59.999999" or_conditions = [] if date_condition: or_conditions.append({"date": date_condition}) if created_at_condition: or_conditions.append({"createdAt": created_at_condition}) if not or_conditions: return None return {"$or": or_conditions} @staticmethod def get_all_orders( id_khach_hang: str, keyword: str = None, status: str = None, dateFrom: str = None, dateTo: str = None, page: int = 1, page_size: int = 5, ): if not id_khach_hang: raise ValueError("Missing id_khach_hang") page = max(page, 1) page_size = max(page_size, 1) conditions = [{"id_khach_hang": id_khach_hang}] if keyword: regex = {"$regex": keyword, "$options": "i"} conditions.append( { "$or": [ {"orderId": regex}, {"id": regex}, {"tracking": regex}, {"customer.name": regex}, {"customer": regex}, ] } ) if status and status != "All": conditions.append({"status": status}) date_filter = OrderModel._build_date_filter(dateFrom, dateTo) if date_filter: conditions.append(date_filter) query = conditions[0] if len(conditions) == 1 else {"$and": conditions} total_items = orders_collection.count_documents(query) total_pages = math.ceil(total_items / page_size) if total_items > 0 else 0 skip = (page - 1) * page_size cursor = ( orders_collection.find(query) .sort([("createdAt", -1), ("date", -1)]) .skip(skip) .limit(page_size) ) orders = [OrderModel._sanitize_order_list(doc) for doc in cursor] return { "data": orders, "total": total_items, "page": page, "pageSize": page_size, "totalPages": total_pages, } @staticmethod def get_order_detail(order_id: str, id_khach_hang: str): order = orders_collection.find_one(OrderModel._order_match(order_id, id_khach_hang)) if not order: raise ValueError("Order not found") return OrderModel._sanitize_order_detail(order) @staticmethod def update_order(order_id: str, id_khach_hang: str, update_data: dict): order = orders_collection.find_one(OrderModel._order_match(order_id, id_khach_hang)) if not order: raise ValueError("Order not found") if order.get("status") not in ["Unpaid", "Processing"]: raise ValueError("Only Unpaid or Processing orders can be updated") payload = {} if "customer" in update_data: current_customer = OrderModel._normalize_customer(order) incoming_customer = update_data.get("customer") or {} merged_customer = { "name": incoming_customer.get("name", current_customer["name"]), "phone": incoming_customer.get("phone", current_customer["phone"]), "email": incoming_customer.get("email", current_customer["email"]), "address": { **current_customer["address"], **(incoming_customer.get("address") or {}), }, } payload["customer"] = merged_customer if "items" in update_data: payload["items"] = update_data["items"] payload["total"] = OrderModel._calculate_total(update_data["items"]) if payload: orders_collection.update_one( OrderModel._order_match(order_id, id_khach_hang), {"$set": payload}, ) updated_order = orders_collection.find_one(OrderModel._order_match(order_id, id_khach_hang)) return OrderModel._sanitize_order_detail(updated_order) @staticmethod def cancel_order(order_id: str, id_khach_hang: str): order = orders_collection.find_one(OrderModel._order_match(order_id, id_khach_hang)) if not order: raise ValueError("Order not found") current_status = order.get("status") if current_status not in ["Unpaid", "Processing"]: raise ValueError("Only Unpaid or Processing orders can be cancelled") orders_collection.update_one( OrderModel._order_match(order_id, id_khach_hang), { "$set": { "status": "Cancelled", "cancelledBy": "user", "cancelledByUserId": id_khach_hang, "cancelledAt": datetime.utcnow().isoformat(), } }, ) return True @staticmethod def pay_orders(order_ids: list, id_khach_hang: str): if not id_khach_hang: raise ValueError("Missing id_khach_hang") base_query = { "id_khach_hang": id_khach_hang, "status": "Unpaid", "$or": [ {"orderId": {"$in": order_ids}}, {"id": {"$in": order_ids}}, ], } matched_orders = list(orders_collection.find(base_query)) if not matched_orders: raise ValueError("No unpaid orders found for payment") user = UserModel.get_user_by_id(id_khach_hang) if not user: raise ValueError("User not found") potential_total = round( sum(OrderModel._calculate_total(doc.get("items", [])) for doc in matched_orders), 2, ) if potential_total <= 0: raise ValueError("Không tìm thấy tổng tiền hợp lệ để thanh toán") UserModel.deduct_balance(id_khach_hang, potential_total) paid_orders = [] paid_total = 0.0 try: for doc in matched_orders: order_identifier = doc.get("orderId") or doc.get("id") order_total = OrderModel._calculate_total(doc.get("items", [])) result = orders_collection.update_one( {"_id": doc["_id"], "status": "Unpaid"}, {"$set": {"status": "Processing"}}, ) if result.modified_count == 1: paid_orders.append(order_identifier) paid_total += order_total except Exception as e: refund_amount = round(potential_total - paid_total, 2) if refund_amount > 0: UserModel.add_balance(id_khach_hang, refund_amount) raise ValueError(f"Thanh toán thất bại: {str(e)}") paid_total = round(paid_total, 2) if not paid_orders: UserModel.add_balance(id_khach_hang, potential_total) raise ValueError("Thanh toán thất bại. Số dư đã được hoàn lại.") refund_amount = round(potential_total - paid_total, 2) if refund_amount > 0: UserModel.add_balance(id_khach_hang, refund_amount) ten_khach_hang = ( f"{user.get('firstName', '')} {user.get('lastName', '')}".strip() or user.get("email", "") or "Unknown User" ) try: TopupModel.create_deduction_transaction( id_khach_hang=id_khach_hang, ten_khach_hang=ten_khach_hang, amount=paid_total, order_ids=paid_orders, ) except Exception as e: print(f"Lỗi ghi transaction deduction: {e}") return { "paidCount": len(paid_orders), "totalAmount": paid_total, } @staticmethod def _admin_base_conditions(): return [ { "$or": [ {"status": {"$in": ["Processing", "Printing", "Shipped"]}}, { "$and": [ {"status": "Cancelled"}, {"cancelledBy": {"$ne": "user"}}, ] }, ] } ] @staticmethod def _allowed_next_status(current_status: str): flow = { "Processing": "Printing", "Printing": "Shipped", } return flow.get(current_status) @staticmethod def _create_admin_refund_transaction(order_doc: dict, amount: float): tx_id = f"TX-{uuid.uuid4().hex[:6].upper()}" now_vn = datetime.now(VN_TZ) order_ref = str(order_doc.get("orderId") or order_doc.get("id") or order_doc.get("_id")) customer_doc = UserModel.get_user_by_id(order_doc.get("id_khach_hang")) ten_khach_hang = "Unknown User" if customer_doc: ten_khach_hang = ( f"{customer_doc.get('firstName', '')} {customer_doc.get('lastName', '')}".strip() or customer_doc.get("email", "") or "Unknown User" ) new_doc = { "id": tx_id, "date": now_vn, "type": "Deposit", "amount": f"+${amount:,.2f}", "numericAmount": amount, "status": "Completed", "reasonKey": f"Hoàn tiền đơn hàng {order_ref} (admin huỷ)", "referenceId": order_ref, "paymentMethod": "Balance Refund", "receiptUrl": None, "id_khach_hang": order_doc.get("id_khach_hang", ""), "ten_khach_hang": ten_khach_hang, } transactions_collection.insert_one(new_doc) return tx_id @staticmethod def admin_get_all_orders( keyword: str = None, status: str = None, dateFrom: str = None, dateTo: str = None, page: int = 1, page_size: int = 10, ): page = max(page, 1) page_size = max(page_size, 1) conditions = OrderModel._admin_base_conditions() if keyword: escaped_keyword = re.escape(keyword.strip()) regex = {"$regex": escaped_keyword, "$options": "i"} matched_uploader_ids = OrderModel._find_uploader_ids_by_keyword(keyword) keyword_conditions = [ {"orderId": regex}, {"id": regex}, {"tracking": regex}, {"trackingNumber": regex}, {"customer.name": regex}, {"customer": regex}, {"uploadedByName": regex}, {"uploadedByEmail": regex}, ] if matched_uploader_ids: keyword_conditions.append({"id_khach_hang": {"$in": matched_uploader_ids}}) conditions.append({"$or": keyword_conditions}) allowed_admin_statuses = ["Processing", "Printing", "Shipped", "Cancelled"] if status and status != "All": if status not in allowed_admin_statuses: return { "data": [], "total": 0, "page": page, "pageSize": page_size, "totalPages": 0, } conditions.append({"status": status}) if status == "Cancelled": conditions.append({"cancelledBy": {"$ne": "user"}}) date_filter = OrderModel._build_date_filter(dateFrom, dateTo) if date_filter: conditions.append(date_filter) query = conditions[0] if len(conditions) == 1 else {"$and": conditions} total_items = orders_collection.count_documents(query) total_pages = math.ceil(total_items / page_size) if total_items > 0 else 0 skip = (page - 1) * page_size cursor = ( orders_collection.find(query) .sort([("createdAt", -1), ("date", -1)]) .skip(skip) .limit(page_size) ) uploader_cache = {} orders = [OrderModel._sanitize_order_list(doc, uploader_cache) for doc in cursor] return { "data": orders, "total": total_items, "page": page, "pageSize": page_size, "totalPages": total_pages, } @staticmethod def admin_get_dashboard_summary( dateFrom: str = None, dateTo: str = None, salesPage: int = 1, salesPageSize: int = 15, salesSortBy: str = "revenue", salesSortOrder: str = "desc", ): salesPage = max(1, int(salesPage)) salesPageSize = max(1, min(100, int(salesPageSize))) salesSortBy = str(salesSortBy or "revenue").strip() salesSortOrder = str(salesSortOrder or "desc").strip().lower() if salesSortBy not in ["revenue", "qtySold"]: salesSortBy = "revenue" if salesSortOrder not in ["asc", "desc"]: salesSortOrder = "desc" order_date_filter = OrderModel._build_date_filter(dateFrom, dateTo) paid_conditions = [{"status": {"$in": ["Processing", "Printing", "Shipped"]}}] active_order_conditions = [{"status": {"$in": ["Processing", "Printing", "Shipped"]}}] if order_date_filter: paid_conditions.append(order_date_filter) active_order_conditions.append(order_date_filter) paid_query = paid_conditions[0] if len(paid_conditions) == 1 else {"$and": paid_conditions} active_order_query = ( active_order_conditions[0] if len(active_order_conditions) == 1 else {"$and": active_order_conditions} ) total_revenue = 0.0 sales_map = {} product_name_cache = {} cursor = orders_collection.find( paid_query, { "items": 1, "total": 1, } ) for doc in cursor: total_revenue += OrderModel._resolve_order_total(doc) items = doc.get("items", []) if not isinstance(items, list): continue for item in items: if not isinstance(item, dict): continue quantity = OrderModel._safe_int(item.get("quantity"), 0) if quantity <= 0: continue price = OrderModel._safe_number(item.get("price"), 0) shipping = OrderModel._safe_number(item.get("shipping"), 0) tax = OrderModel._safe_number(item.get("tax"), 0) # Theo đúng yêu cầu dashboard của bạn: line_revenue = round((price + shipping + tax) * quantity, 2) product_name = OrderModel._resolve_sales_product_name(item, product_name_cache) if product_name not in sales_map: sales_map[product_name] = { "name": product_name, "qtySold": 0, "revenue": 0.0, } sales_map[product_name]["qtySold"] += quantity sales_map[product_name]["revenue"] += line_revenue sales_rows = [] for row in sales_map.values(): sales_rows.append({ "name": row["name"], "qtySold": int(row["qtySold"]), "revenue": round(row["revenue"], 2), }) if salesSortBy == "qtySold": sales_rows.sort( key=lambda x: (x["qtySold"], x["revenue"], x["name"].lower()), reverse=(salesSortOrder == "desc"), ) else: sales_rows.sort( key=lambda x: (x["revenue"], x["qtySold"], x["name"].lower()), reverse=(salesSortOrder == "desc"), ) sales_total = len(sales_rows) sales_total_pages = math.ceil(sales_total / salesPageSize) if sales_total > 0 else 1 start = (salesPage - 1) * salesPageSize end = start + salesPageSize paged_sales = sales_rows[start:end] sales_grand_qty = sum(row["qtySold"] for row in sales_rows) sales_grand_revenue = round(sum(row["revenue"] for row in sales_rows), 2) total_orders = orders_collection.count_documents(active_order_query) topup_query = { "type": "Deposit", "status": "Approved", "paymentMethod": {"$ne": "Balance Refund"}, } transaction_date_filter = OrderModel._build_transaction_date_filter(dateFrom, dateTo) if transaction_date_filter: topup_query["date"] = transaction_date_filter total_topup = 0.0 tx_cursor = transactions_collection.find(topup_query, {"numericAmount": 1}) for tx in tx_cursor: total_topup += OrderModel._safe_number(tx.get("numericAmount"), 0) return { "totalRevenue": round(total_revenue, 2), "totalOrders": int(total_orders), "totalTopup": round(total_topup, 2), "salesByProduct": paged_sales, "salesPage": salesPage, "salesPageSize": salesPageSize, "salesTotal": sales_total, "salesTotalPages": sales_total_pages, "salesGrandQty": sales_grand_qty, "salesGrandRevenue": sales_grand_revenue, "salesSortBy": salesSortBy, "salesSortOrder": salesSortOrder, } @staticmethod def admin_batch_update_order_status( doc_ids: list, new_status: str, tracking_number: str = None, tracking_link: str = None, ): if not doc_ids: raise ValueError("No orders selected") unique_doc_ids = [] seen = set() for raw_id in doc_ids: doc_id = str(raw_id or "").strip() if not doc_id or doc_id in seen: continue try: unique_doc_ids.append(ObjectId(doc_id)) seen.add(doc_id) except Exception: raise ValueError(f"Invalid docId: {doc_id}") if not unique_doc_ids: raise ValueError("No valid orders selected") orders = list(orders_collection.find({"_id": {"$in": unique_doc_ids}})) if len(orders) != len(unique_doc_ids): raise ValueError("One or more orders not found") if new_status == "Shipped" and len(unique_doc_ids) != 1: raise ValueError("Shipped can only update 1 order at a time") for order in orders: current_status = order.get("status", "") if current_status in ["Unpaid", "Cancelled"]: raise ValueError("One or more selected orders are not available for admin operations") allowed_next = OrderModel._allowed_next_status(current_status) if not allowed_next: raise ValueError("One or more selected orders can no longer be updated") if new_status != allowed_next: raise ValueError(f"Invalid status transition: {current_status} -> {new_status}") payload = {"status": new_status} if new_status == "Shipped": if not tracking_number or not tracking_number.strip(): raise ValueError("trackingNumber is required when confirming shipped") if not tracking_link or not tracking_link.strip(): raise ValueError("trackingLink is required when confirming shipped") payload["tracking"] = tracking_number.strip() payload["trackingNumber"] = tracking_number.strip() payload["trackingLink"] = tracking_link.strip() result = orders_collection.update_many( {"_id": {"$in": unique_doc_ids}}, {"$set": payload}, ) return { "updatedCount": result.modified_count, "status": new_status, "docIds": [str(x) for x in unique_doc_ids], } @staticmethod def admin_update_order_status( identifier: str, new_status: str, tracking_number: str = None, tracking_link: str = None, ): order = orders_collection.find_one(OrderModel._admin_order_match(identifier)) if not order or OrderModel._is_hidden_from_admin(order): raise ValueError("Order not found") current_status = order.get("status", "") if current_status in ["Unpaid", "Cancelled"]: raise ValueError("This order is not available for admin operations") allowed_next = OrderModel._allowed_next_status(current_status) if not allowed_next: raise ValueError("This order can no longer be updated") if new_status != allowed_next: raise ValueError(f"Invalid status transition: {current_status} -> {new_status}") payload = {"status": new_status} if new_status == "Shipped": if not tracking_number or not tracking_number.strip(): raise ValueError("trackingNumber is required when confirming shipped") if not tracking_link or not tracking_link.strip(): raise ValueError("trackingLink is required when confirming shipped") payload["tracking"] = tracking_number.strip() payload["trackingNumber"] = tracking_number.strip() payload["trackingLink"] = tracking_link.strip() orders_collection.update_one( {"_id": order["_id"]}, {"$set": payload}, ) updated_order = orders_collection.find_one({"_id": order["_id"]}) return OrderModel._sanitize_order_detail(updated_order) @staticmethod def admin_cancel_order( identifier: str, admin_id: str = "", admin_name: str = "", ): order = orders_collection.find_one(OrderModel._admin_order_match(identifier)) if not order or OrderModel._is_hidden_from_admin(order): raise ValueError("Order not found") current_status = order.get("status", "") if current_status not in ["Processing", "Printing"]: raise ValueError("Only Processing or Printing orders can be cancelled by admin") user_id = str(order.get("id_khach_hang") or "").strip() if not user_id: raise ValueError("Missing customer id for refund") refund_amount = OrderModel._resolve_order_total(order) if refund_amount <= 0: raise ValueError("Invalid refund amount") cancelled_at = datetime.utcnow().isoformat() update_result = orders_collection.update_one( {"_id": order["_id"], "status": {"$in": ["Processing", "Printing"]}}, { "$set": { "status": "Cancelled", "cancelledBy": "admin", "cancelledByAdminId": admin_id or "", "cancelledByAdminName": admin_name or "", "cancelledAt": cancelled_at, } }, ) if update_result.modified_count != 1: raise ValueError("Order could not be cancelled") balance_refunded = False refund_tx_id = None try: refund_ok = UserModel.add_balance(user_id, refund_amount) if not refund_ok: raise ValueError("Refund failed while updating user balance") balance_refunded = True refund_tx_id = OrderModel._create_admin_refund_transaction(order, refund_amount) orders_collection.update_one( {"_id": order["_id"]}, { "$set": { "refundAmount": refund_amount, "refundTxId": refund_tx_id, "refundedAt": datetime.utcnow().isoformat(), } }, ) updated_order = orders_collection.find_one({"_id": order["_id"]}) return { "order": OrderModel._sanitize_order_detail(updated_order), "refundAmount": refund_amount, "refundTxId": refund_tx_id, } except Exception as e: rollback_update = { "$set": { "status": current_status, }, "$unset": { "cancelledBy": "", "cancelledByAdminId": "", "cancelledByAdminName": "", "cancelledAt": "", "refundAmount": "", "refundTxId": "", "refundedAt": "", }, } orders_collection.update_one({"_id": order["_id"]}, rollback_update) if balance_refunded: try: UserModel.deduct_balance(user_id, refund_amount) except Exception: pass if refund_tx_id: try: transactions_collection.delete_one({"id": refund_tx_id}) except Exception: pass raise ValueError(f"Admin cancel refund failed: {str(e)}") @staticmethod def admin_get_order_detail(identifier: str): order = orders_collection.find_one(OrderModel._admin_order_match(identifier)) if not order or OrderModel._is_hidden_from_admin(order): raise ValueError("Order not found") return OrderModel._sanitize_order_detail(order) @staticmethod def _admin_order_match(identifier: str): conditions = [ {"orderId": identifier}, {"id": identifier}, ] try: conditions.insert(0, {"_id": ObjectId(identifier)}) except Exception: pass return {"$or": conditions} @staticmethod def _safe_number(value, default=0.0): try: return float(value or 0) except (TypeError, ValueError): return float(default) @staticmethod def _safe_int(value, default=0): try: return int(float(value or 0)) except (TypeError, ValueError): return int(default) @staticmethod def _build_transaction_date_filter(date_from=None, date_to=None): if not date_from and not date_to: return None date_query = {} if date_from: date_query["$gte"] = VN_TZ.localize(datetime.strptime(date_from, "%Y-%m-%d")) if date_to: date_query["$lte"] = VN_TZ.localize( datetime.strptime(date_to, "%Y-%m-%d") + timedelta(days=1, seconds=-1) ) return date_query or None @staticmethod def _resolve_sales_product_name(item: dict, cache: dict): direct_name = str( item.get("productName") or item.get("title") or item.get("name") or "" ).strip() if direct_name: return direct_name product_id = str(item.get("productId") or "").strip() if not product_id: return "Unknown Product" if product_id not in cache: doc = products_collection.find_one({"productId": product_id}, {"title": 1}) cache[product_id] = str((doc or {}).get("title") or product_id).strip() return cache[product_id]