from MongoDBConnection import db, users_collection, transactions_collection, products_collection import math import re import uuid import pytz import requests from urllib.parse import quote, urlparse, parse_qs from datetime import datetime, timedelta from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError from bson import ObjectId from model.user_model import UserModel from model.topup_model import TopupModel from utils.design_helper import ensure_design_local, is_local_design_value from utils.ai_visual_helper import ( extract_design_file_name, index_design_file, delete_design_file, ) orders_collection = db["orders"] VN_TZ = pytz.timezone("Asia/Ho_Chi_Minh") UPS_TRACK_BASE_URL = "https://www.ups.com/track" PLAYWRIGHT_TIMEOUT_MS = 45000 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): status = str(doc.get("status") or "").strip() canceled_by = str(doc.get("canceledBy") or doc.get("cancelledBy") or "").strip() return ( status in ["Canceled", "Cancelled"] and canceled_by == "user" and not OrderModel._is_user_processing_cancel(doc) ) @staticmethod def _is_canceled_status(status: str): return str(status or "").strip() in ["Canceled", "Cancelled"] @staticmethod def _is_user_processing_cancel(doc): canceled_by = str(doc.get("canceledBy") or doc.get("cancelledBy") or "").strip() if canceled_by != "user": return False from_status = str( doc.get("canceledFromStatus") or doc.get("cancelledFromStatus") or "" ).strip() if from_status == "Processing": return True refund_amount = OrderModel._safe_number(doc.get("refundAmount"), 0) refund_tx_id = str(doc.get("refundTxId") or "").strip() return refund_amount > 0 or bool(refund_tx_id) @staticmethod def _normalize_variant_text(value): return re.sub(r"\s+", " ", str(value or "")).strip().lower() @staticmethod def _get_variant_tier(variant: dict): return str((variant or {}).get("tier") or (variant or {}).get("material") or "").strip() @staticmethod def _get_item_applied_tier(source: dict): return str( (source or {}).get("appliedTier") or (source or {}).get("tier") or (source or {}).get("material") or "" ).strip() @staticmethod def _get_design_drive(source: dict): return str( (source or {}).get("designOriginal") or (source or {}).get("designDrive") or (source or {}).get("design") or "" ).strip() @staticmethod def _get_design_local(source: dict): return str((source or {}).get("designLocal") or (source or {}).get("design") or "").strip() @staticmethod def _extract_order_design_values(order_doc: dict): values = [] for item in (order_doc or {}).get("items", []) or []: if not isinstance(item, dict): continue for key in ["designLocal", "designFile", "designUrl"]: raw = str(item.get(key) or "").strip() if raw: values.append(raw) seen = set() result = [] for value in values: file_name = extract_design_file_name(value) if not file_name or file_name in seen: continue seen.add(file_name) result.append(file_name) return result @staticmethod def _cleanup_order_design_assets(order_doc: dict): file_names = OrderModel._extract_order_design_values(order_doc) for file_name in file_names: delete_design_file(file_name) @staticmethod def _normalize_order_item(item: dict): if not isinstance(item, dict): return item normalized = dict(item) applied_tier = OrderModel._get_item_applied_tier(item) if applied_tier: normalized["appliedTier"] = applied_tier design_drive = OrderModel._get_design_drive(item) design_local = str((item or {}).get("designLocal") or "").strip() if design_drive: normalized["design"] = design_drive normalized.setdefault("designOriginal", design_drive) normalized.setdefault("designDrive", design_drive) if design_local: normalized["designLocal"] = design_local return normalized @staticmethod def _normalize_issue_details(doc): raw = doc.get("issueDetails") if not isinstance(raw, list): return [] return [item for item in raw if isinstance(item, dict)] @staticmethod def _build_issue_state(errors: list = None): errors = [item for item in (errors or []) if isinstance(item, dict)] blocked = len(errors) > 0 return { "paymentBlocked": blocked, "issueType": "IMPORT_ERROR" if blocked else "", "issueStatus": "needs_fix" if blocked else "", "issueSummary": ( "Đơn này đang có sản phẩm hoặc ảnh thiết kế không hợp lệ. Vui lòng chỉnh sửa trước khi thanh toán." if blocked else "" ), "issueDetails": errors, "issueCount": len(errors), "issueUpdatedAt": datetime.utcnow().isoformat() if blocked else "", } @staticmethod def create_order(data: dict, actor: dict) -> dict: order_number = str(data.get("orderNumber") or "").strip() if not order_number: raise ValueError("orderNumber is required") channel_type = str(data.get("channelType") or "").strip().lower() if channel_type in ("epacket", "tiktok"): customer_data = data.get("customer") or {} required_customer_fields = { "name": "customer name", "address1": "address 1", "city": "city", "state": "state", "postalCode": "postal code", "country": "country", } for field, label in required_customer_fields.items(): if not str(customer_data.get(field) or "").strip(): raise ValueError(f"{label} is required") source_items = data.get("items") or [] if not source_items: raise ValueError("at least one item is required") for index, item in enumerate(source_items, start=1): if not str(item.get("skuNumber") or "").strip(): raise ValueError(f"item {index}: SKU number is required") if int(item.get("quantity") or 0) <= 0: raise ValueError(f"item {index}: quantity must be greater than zero") if not str(item.get("description") or "").strip(): raise ValueError(f"item {index}: description is required") if not str(item.get("saleUrl") or "").strip(): raise ValueError(f"item {index}: sale URL is required") now = datetime.now(VN_TZ) timestamp = now.strftime("%Y%m%d%H%M%S") + f"{now.microsecond // 1000:03d}" random_part = uuid.uuid4().hex[:6] order_id = f"{order_number}_{timestamp}_{random_part}" customer_data = data.get("customer") or {} customer = { "name": str(customer_data.get("name") or "").strip(), "phone": str(customer_data.get("phone") or "").strip(), "taxNumber": str(customer_data.get("taxNumber") or "").strip(), "address": { "street": str(customer_data.get("address1") or "").strip(), "address2": str(customer_data.get("address2") or "").strip(), "city": str(customer_data.get("city") or "").strip(), "state": str(customer_data.get("state") or "").strip(), "zipCode": str(customer_data.get("postalCode") or "").strip(), "country": str(customer_data.get("country") or "").strip(), }, } items = [] for item in (data.get("items") or []): items.append({ "skuNumber": str(item.get("skuNumber") or "").strip(), "quantity": int(item.get("quantity") or 1), "description": str(item.get("description") or "").strip(), "grossWeight": str(item.get("grossWeight") or "").strip(), "width": str(item.get("width") or "").strip(), "height": str(item.get("height") or "").strip(), "length": str(item.get("length") or "").strip(), "saleUrl": str(item.get("saleUrl") or "").strip(), }) now_iso = now.isoformat() doc = { "orderId": order_id, "orderNumber": order_number, "channelType": str(data.get("channelType") or "").strip(), "shippingPartner": str(data.get("shippingPartner") or "").strip(), "shipAddress": str(data.get("shipAddress") or "").strip(), "barcode": str(data.get("barcode") or "").strip(), "labelUrl": str(data.get("labelUrl") or "").strip(), "sellerAddress": str(data.get("sellerAddress") or "").strip(), "customer": customer, "weight": str(data.get("weight") or "").strip(), "length": str(data.get("length") or "").strip(), "width": str(data.get("width") or "").strip(), "height": str(data.get("height") or "").strip(), "chargeableWeight": str(data.get("chargeableWeight") or "").strip(), "items": items, "status": "Liên Hệ", "total": 0.0, "offerSetBy": "", "offerSetById": "", "offerSetAt": "", "offerApprovedAt": "", "offerRejectedAt": "", "tracking": "", "trackingNumber": "", "trackingLink": "", "sourceOrderIds": [], "uploadedByName": str(actor.get("name") or "").strip(), "uploadedByEmail": str(actor.get("email") or "").strip(), "uploadedById": str(actor.get("id") or "").strip(), "paymentBlocked": False, "issueType": "", "issueStatus": "", "issueSummary": "", "issueDetails": [], "issueCount": 0, "issueUpdatedAt": "", "createdAt": now_iso, "updatedAt": now_iso, } orders_collection.insert_one(doc) doc["_id"] = str(doc["_id"]) return doc @staticmethod def _map_db_status_to_fe(db_status: str) -> str: s = str(db_status or "").strip() if s in ("Processing", "Printing"): return "processing" if s == "Shipped": return "processed" if s == "Delivered": return "delivered" if s in ("Canceled", "Cancelled"): return "failed" # FBA&FBT statuses — pass through raw if s in ("Liên Hệ", "Offer", "Rejected", "Unpaid"): return s return "processing" @staticmethod def _map_channel_to_shipping_method(channel: str) -> str: mapping = { "epacket": "ePacket", "tiktok": "TikTok", "buylabel": "Buy Label", } return mapping.get(str(channel or "").lower(), channel) @staticmethod def _sanitize_my_order(doc: dict) -> dict: order_id = doc.get("orderId") or doc.get("id") or str(doc["_id"]) customer = doc.get("customer") or {} customer_name = customer.get("name", "") if isinstance(customer, dict) else str(customer or "") channel = str(doc.get("channelType") or "").strip() created_at = doc.get("createdAt", "") if created_at: created_at = str(created_at)[:10] return { "id": order_id, "docId": str(doc["_id"]), "orderNumber": doc.get("orderNumber") or order_id, "customer": customer_name, "channelType": channel, "shippingMethod": OrderModel._map_channel_to_shipping_method(channel), "status": OrderModel._map_db_status_to_fe(doc.get("status", "")), "weight": str(doc.get("weight") or doc.get("chargeableWeight") or ""), "createdAt": created_at, "importType": doc.get("importType", ""), "rawStatus": doc.get("status", ""), "total": round(float(doc.get("total") or 0), 2), "offerSetBy": doc.get("offerSetBy", ""), "offerSetAt": doc.get("offerSetAt", ""), } @staticmethod def get_my_orders( actor_id: str, keyword: str = None, status: str = None, channel: str = None, date_range: str = None, page: int = 1, page_size: int = 10, ): page = max(page, 1) page_size = max(page_size, 1) base = {"uploadedById": actor_id} # Build date range filter now = datetime.now(VN_TZ) date_from = None date_to = None if date_range == "today": date_from = now.strftime("%Y-%m-%d") date_to = now.strftime("%Y-%m-%d") elif date_range in ("7", "30"): days = int(date_range) date_from = (now - timedelta(days=days)).strftime("%Y-%m-%d") date_to = now.strftime("%Y-%m-%d") # Summary counts (always unfiltered, just by actor) summary_total = orders_collection.count_documents(base) summary_processing = orders_collection.count_documents({"$and": [base, {"status": {"$in": ["Processing", "Printing"]}}]}) summary_shipped = orders_collection.count_documents({"$and": [base, {"status": {"$in": ["Shipped", "Delivered"]}}]}) summary_failed = orders_collection.count_documents({"$and": [base, {"status": {"$in": ["Canceled", "Cancelled"]}}]}) # Filtered query conditions = [base] if keyword: escaped = re.escape(keyword.strip()) regex = {"$regex": escaped, "$options": "i"} conditions.append({"$or": [ {"orderId": regex}, {"orderNumber": regex}, {"customer.name": regex}, ]}) if status and status not in ("all", ""): db_statuses = { "processing": ["Processing", "Printing"], "processed": ["Shipped"], "delivered": ["Delivered"], "failed": ["Canceled", "Cancelled"], } mapped = db_statuses.get(status) if mapped: conditions.append({"status": {"$in": mapped}}) if channel and channel not in ("all", ""): conditions.append({"channelType": {"$regex": f"^{re.escape(channel)}$", "$options": "i"}}) date_filter = OrderModel._build_date_filter(date_from, date_to) if date_filter: conditions.append(date_filter) query = {"$and": conditions} if len(conditions) > 1 else conditions[0] total = orders_collection.count_documents(query) total_pages = math.ceil(total / page_size) if total > 0 else 0 skip = (page - 1) * page_size cursor = ( orders_collection.find(query) .sort([("createdAt", -1)]) .skip(skip) .limit(page_size) ) data = [OrderModel._sanitize_my_order(doc) for doc in cursor] return { "data": data, "total": total, "page": page, "pageSize": page_size, "totalPages": total_pages, "summary": { "total": summary_total, "processing": summary_processing, "shipped": summary_shipped, "failed": summary_failed, }, } @staticmethod def get_my_order_detail(order_id: str, actor_id: str) -> dict: conditions = [ {"uploadedById": actor_id}, {"$or": [{"orderId": order_id}, {"_id": ObjectId(order_id) if len(order_id) == 24 else None}]}, ] # Try orderId first doc = orders_collection.find_one({"orderId": order_id, "uploadedById": actor_id}) if not doc: try: doc = orders_collection.find_one({"_id": ObjectId(order_id), "uploadedById": actor_id}) except Exception: pass if not doc: raise ValueError("Order not found") order_id_val = doc.get("orderId") or doc.get("id") or str(doc["_id"]) customer_raw = doc.get("customer") or {} customer_name = "" customer_phone = "" customer_address = {} if isinstance(customer_raw, dict): customer_name = customer_raw.get("name", "") customer_phone = customer_raw.get("phone", "") addr = customer_raw.get("address") or {} if isinstance(addr, dict): customer_address = addr else: customer_name = str(customer_raw) channel = str(doc.get("channelType") or "").strip() created_at = str(doc.get("createdAt", ""))[:10] raw_items = doc.get("items") or [] items = [ { "skuNumber": item.get("skuNumber", ""), "quantity": item.get("quantity", 1), "description": item.get("description", ""), "grossWeight": item.get("grossWeight", ""), "length": item.get("length", ""), "width": item.get("width", ""), "height": item.get("height", ""), "saleUrl": item.get("saleUrl", ""), } for item in raw_items if isinstance(item, dict) ] return { "id": order_id_val, "docId": str(doc["_id"]), "orderNumber": doc.get("orderNumber") or order_id_val, "channelType": channel, "shippingMethod": OrderModel._map_channel_to_shipping_method(channel), "shippingPartner": doc.get("shippingPartner", ""), "shipAddress": doc.get("shipAddress", ""), "sellerAddress": doc.get("sellerAddress", ""), "barcode": doc.get("barcode", ""), "labelUrl": doc.get("labelUrl", ""), "status": OrderModel._map_db_status_to_fe(doc.get("status", "")), "rawStatus": doc.get("status", ""), "weight": doc.get("weight", ""), "length": doc.get("length", ""), "width": doc.get("width", ""), "height": doc.get("height", ""), "chargeableWeight": doc.get("chargeableWeight", ""), "tracking": doc.get("tracking") or doc.get("trackingNumber", ""), "trackingLink": doc.get("trackingLink", ""), "createdAt": created_at, "customer": { "name": customer_name, "phone": customer_phone, "address": customer_address, }, "items": items, } @staticmethod def _find_matching_variant( variants: list, requested_style: str, requested_size: str, preferred_tier: str = "", ): normalized_style = OrderModel._normalize_variant_text(requested_style) normalized_size = OrderModel._normalize_variant_text(requested_size) normalized_tier = OrderModel._normalize_variant_text(preferred_tier) exact_tier_matches = [] fallback_matches = [] for variant in variants or []: variant_style = OrderModel._normalize_variant_text(variant.get("style")) variant_size = OrderModel._normalize_variant_text(variant.get("size")) variant_tier = OrderModel._normalize_variant_text(OrderModel._get_variant_tier(variant)) if variant_style == normalized_style and variant_size == normalized_size: fallback_matches.append(variant) if normalized_tier and variant_tier == normalized_tier: exact_tier_matches.append(variant) if exact_tier_matches: return exact_tier_matches[0] if fallback_matches: return fallback_matches[0] return None @staticmethod def _build_design_error_detail(idx: int, item: dict, reason: str, message: str): return { "row": idx, "reason": reason, "message": message, "productId": str(item.get("productId") or "").strip(), "productName": str(item.get("productName") or "").strip(), "style": str(item.get("style") or "").strip(), "size": str(item.get("size") or "").strip(), "design": str(item.get("design") or "").strip(), } @staticmethod def _prepare_items_for_storage(items: list): normalized_items = [] design_errors = [] for idx, item in enumerate(items or [], start=1): if not isinstance(item, dict): normalized_items.append(item) continue next_item = dict(item) design_value = OrderModel._get_design_drive(next_item) result = ensure_design_local( design_value, filename_prefix=f"order_item_{idx}" ) if result["success"]: next_item["designOriginal"] = design_value next_item["designDrive"] = design_value next_item["designLocal"] = result["publicUrl"] next_item["design"] = design_value # Cho AI học ảnh vừa được lưu trong /pod-api/design index_design_file(result["publicUrl"]) else: next_item["design"] = design_value next_item.setdefault("designOriginal", design_value) next_item.setdefault("designDrive", design_value) design_errors.append( OrderModel._build_design_error_detail( idx=idx, item=next_item, reason=result["reason"], message=result["message"], ) ) normalized_items.append(next_item) return normalized_items, design_errors @staticmethod def _validate_customer_for_payment(customer: dict): errors = [] customer = customer or {} address = OrderModel._normalize_address(customer.get("address", {})) if not str(customer.get("name") or "").strip(): errors.append({ "row": 0, "reason": "MISSING_CUSTOMER", "message": "Thiếu tên khách hàng.", }) has_any_address = any( str(address.get(key) or "").strip() for key in ["street", "city", "state", "country", "zipCode"] ) if not has_any_address: errors.append({ "row": 0, "reason": "MISSING_ADDRESS", "message": "Thiếu địa chỉ.", }) return errors @staticmethod def _validate_items_for_payment(items: list): errors = [] if not isinstance(items, list) or len(items) == 0: return [{ "row": 0, "reason": "NO_ITEMS", "message": "Đơn hàng không có sản phẩm nào để thanh toán.", }] for idx, item in enumerate(items, start=1): if not isinstance(item, dict): errors.append({ "row": idx, "reason": "INVALID_ITEM", "message": f"Dòng {idx} không hợp lệ.", }) continue product_id = str(item.get("productId") or "").strip() requested_style = str(item.get("style") or "").strip() requested_size = str(item.get("size") or "").strip() requested_design = OrderModel._get_design_drive(item) requested_design_local = OrderModel._get_design_local(item) product_doc = products_collection.find_one({"productId": product_id}) if product_id else None product_name = str( (product_doc or {}).get("title") or item.get("productName") or product_id or "" ).strip() quantity = OrderModel._safe_int(item.get("quantity"), 0) if quantity <= 0: errors.append({ "row": idx, "reason": "INVALID_QUANTITY", "message": f'Sản phẩm "{product_name or product_id}" có số lượng không hợp lệ.', "productId": product_id, "productName": product_name, "style": requested_style, "size": requested_size, "quantity": item.get("quantity"), }) continue if not product_id: errors.append({ "row": idx, "reason": "MISSING_PRODUCT_ID", "message": "Thiếu mã sản phẩm.", "productId": "", "productName": product_name, "style": requested_style, "size": requested_size, }) continue if not product_doc: errors.append({ "row": idx, "reason": "PRODUCT_NOT_FOUND", "message": f'Không tìm thấy sản phẩm "{product_name or product_id}".', "productId": product_id, "productName": product_name or product_id, "style": requested_style, "size": requested_size, }) continue if not requested_style: errors.append({ "row": idx, "reason": "MISSING_STYLE", "message": f'Sản phẩm "{product_name}" đang thiếu style.', "productId": product_id, "productName": product_name, "style": requested_style, "size": requested_size, }) continue if not requested_size: errors.append({ "row": idx, "reason": "MISSING_SIZE", "message": f'Sản phẩm "{product_name}" đang thiếu size.', "productId": product_id, "productName": product_name, "style": requested_style, "size": requested_size, }) continue variants = product_doc.get("variants") or [] if not variants: errors.append({ "row": idx, "reason": "NO_VARIANTS", "message": f'Sản phẩm "{product_name}" hiện tại không còn mặt hàng này.', "productId": product_id, "productName": product_name, "style": requested_style, "size": requested_size, }) continue available_styles = sorted({ str(v.get("style") or "").strip() for v in variants if str(v.get("style") or "").strip() }) available_sizes = sorted({ str(v.get("size") or "").strip() for v in variants if str(v.get("size") or "").strip() }) normalized_requested_style = OrderModel._normalize_variant_text(requested_style) normalized_requested_size = OrderModel._normalize_variant_text(requested_size) normalized_available_styles = { OrderModel._normalize_variant_text(value) for value in available_styles } normalized_available_sizes = { OrderModel._normalize_variant_text(value) for value in available_sizes } if normalized_requested_style not in normalized_available_styles: errors.append({ "row": idx, "reason": "STYLE_NOT_FOUND", "message": f'Sản phẩm "{product_name}" hiện tại đã hết hoặc không có style "{requested_style}".', "productId": product_id, "productName": product_name, "style": requested_style, "size": requested_size, "availableStyles": available_styles, }) continue if normalized_requested_size not in normalized_available_sizes: errors.append({ "row": idx, "reason": "SIZE_NOT_FOUND", "message": f'Sản phẩm "{product_name}" hiện tại đã hết hoặc không có size "{requested_size}".', "productId": product_id, "productName": product_name, "style": requested_style, "size": requested_size, "availableSizes": available_sizes, }) continue matched_variant = None for variant in variants: variant_style = OrderModel._normalize_variant_text(variant.get("style")) variant_size = OrderModel._normalize_variant_text(variant.get("size")) if variant_style == normalized_requested_style and variant_size == normalized_requested_size: matched_variant = variant break if not matched_variant: errors.append({ "row": idx, "reason": "VARIANT_NOT_FOUND", "message": f'Sản phẩm "{product_name}" không có biến thể style "{requested_style}" - size "{requested_size}".', "productId": product_id, "productName": product_name, "style": requested_style, "size": requested_size, }) continue if not requested_design: errors.append({ "row": idx, "reason": "DESIGN_MISSING", "message": f'Sản phẩm "{product_name}" đang thiếu ảnh thiết kế.', "productId": product_id, "productName": product_name, "style": requested_style, "size": requested_size, "design": requested_design, }) continue if not requested_design_local or not is_local_design_value(requested_design_local): errors.append({ "row": idx, "reason": "DESIGN_NOT_READY", "message": ( f'Sản phẩm "{product_name}" chưa có ảnh thiết kế trong hệ thống ' 'hoặc Design URL hiện tại không tải được.' ), "productId": product_id, "productName": product_name, "style": requested_style, "size": requested_size, "design": requested_design, }) continue return errors @staticmethod def _validate_order_for_payment(doc: dict): customer = OrderModel._normalize_customer(doc or {}) items = (doc or {}).get("items", []) errors = [] errors.extend(OrderModel._validate_customer_for_payment(customer)) errors.extend(OrderModel._validate_items_for_payment(items)) return errors @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 = [OrderModel._normalize_order_item(item) for item in raw_items] if isinstance(raw_items, list) else [] item_count = len(items) total = OrderModel._resolve_order_total(doc) issue_details = OrderModel._normalize_issue_details(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), "canceledBy": doc.get("canceledBy") or doc.get("cancelledBy", ""), "canceledAt": doc.get("canceledAt") or doc.get("cancelledAt", ""), "cancelReason": doc.get("cancelReason", ""), "cancelledBy": doc.get("cancelledBy") or doc.get("canceledBy", ""), "itemCount": item_count, "total": total, "paymentBlocked": bool(doc.get("paymentBlocked")), "issueType": doc.get("issueType", ""), "issueStatus": doc.get("issueStatus", ""), "issueSummary": doc.get("issueSummary", ""), "issueCount": int(doc.get("issueCount", len(issue_details)) or 0), "previewImage": OrderModel._get_order_preview_image(doc), "matchedImage": OrderModel._get_order_preview_image(doc), "importType": doc.get("importType", ""), "offerSetBy": doc.get("offerSetBy", ""), "offerSetAt": doc.get("offerSetAt", ""), } @staticmethod def _get_order_preview_image(doc: dict): items = doc.get("items", []) if not isinstance(items, list): return "" for item in items: if not isinstance(item, dict): continue design_value = ( item.get("designLocal") or item.get("designUrl") or item.get("designFile") or item.get("design") or item.get("designOriginal") or item.get("designDrive") or item.get("image") or item.get("imageUrl") or "" ) design_value = str(design_value or "").strip() if design_value: return design_value return "" @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 = [OrderModel._normalize_order_item(item) for item in 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")) issue_details = OrderModel._normalize_issue_details(doc) 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), "canceledBy": doc.get("canceledBy") or doc.get("cancelledBy", ""), "canceledAt": doc.get("canceledAt") or doc.get("cancelledAt", ""), "cancelReason": doc.get("cancelReason", ""), "cancelledBy": doc.get("cancelledBy") or doc.get("canceledBy", ""), "refundAmount": doc.get("refundAmount", 0), "paymentBlocked": bool(doc.get("paymentBlocked")), "issueType": doc.get("issueType", ""), "issueStatus": doc.get("issueStatus", ""), "issueSummary": doc.get("issueSummary", ""), "issueCount": int(doc.get("issueCount", len(issue_details)) or 0), "issueDetails": issue_details, # ops-order fields "channelType": doc.get("channelType", ""), "shippingPartner": doc.get("shippingPartner", ""), "shipAddress": doc.get("shipAddress", ""), "sellerAddress": doc.get("sellerAddress", ""), "barcode": doc.get("barcode", ""), "labelUrl": doc.get("labelUrl", ""), "weight": doc.get("weight", ""), "length": doc.get("length", ""), "width": doc.get("width", ""), "height": doc.get("height", ""), "chargeableWeight": doc.get("chargeableWeight", ""), # offer-workflow fields "offerSetBy": doc.get("offerSetBy", ""), "offerSetAt": doc.get("offerSetAt", ""), } @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": if OrderModel._is_canceled_status(status): conditions.append({"status": {"$in": ["Canceled", "Cancelled"]}}) else: 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 = {} candidate_customer = OrderModel._normalize_customer(order) candidate_items = list(order.get("items", [])) if isinstance(order.get("items", []), list) else [] if "customer" in update_data: incoming_customer = update_data.get("customer") or {} merged_customer = { "name": incoming_customer.get("name", candidate_customer["name"]), "phone": incoming_customer.get("phone", candidate_customer["phone"]), "email": incoming_customer.get("email", candidate_customer["email"]), "address": { **candidate_customer["address"], **(incoming_customer.get("address") or {}), }, } candidate_customer = merged_customer payload["customer"] = merged_customer design_errors = [] if "items" in update_data: prepared_items, design_errors = OrderModel._prepare_items_for_storage(update_data["items"]) candidate_items = prepared_items payload["items"] = prepared_items payload["total"] = OrderModel._calculate_total(prepared_items) validation_errors = OrderModel._validate_order_for_payment({ **order, "customer": candidate_customer, "items": candidate_items, }) if design_errors: validation_errors = [ err for err in validation_errors if err.get("reason") not in {"DESIGN_MISSING", "DESIGN_NOT_READY"} ] validation_errors.extend(design_errors) payload.update(OrderModel._build_issue_state(validation_errors)) 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 _create_user_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} (user 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 cancel_order(order_id: str, id_khach_hang: str, cancel_reason: 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("This order can no longer be canceled.") canceled_at = datetime.utcnow().isoformat() cancel_reason = cancel_reason.strip() or "Canceled by customer" refund_amount = 0.0 refund_tx_id = None balance_refunded = False if current_status == "Unpaid": # Xoá cứng đơn chưa thanh toán ra khỏi database OrderModel._cleanup_order_design_assets(order) result = orders_collection.delete_one( { **OrderModel._order_match(order_id, id_khach_hang), "status": "Unpaid", } ) if result.deleted_count != 1: raise ValueError("Order could not be deleted") return { "canceled": True, "cancelled": True, "deleted": True, "hiddenFromUser": True, "refundAmount": 0, "refundTxId": "", "status": "Canceled", "canceledBy": "user", "canceledAt": canceled_at, "canceledFromStatus": "Unpaid", "cancelReason": cancel_reason, } # Processing: đơn đã thanh toán, phải hoàn tiền rồi mới hủy. 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") 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_user_refund_transaction(order, refund_amount) result = orders_collection.update_one( { **OrderModel._order_match(order_id, id_khach_hang), "status": "Processing", }, { "$set": { "status": "Canceled", "canceledBy": "user", "canceledByUserId": id_khach_hang, "canceledAt": canceled_at, "canceledFromStatus": "Processing", "cancelReason": cancel_reason, "cancelledBy": "user", "cancelledByUserId": id_khach_hang, "cancelledAt": canceled_at, "cancelledFromStatus": "Processing", "hiddenFromUser": False, "hiddenFromUserAt": "", "refundAmount": refund_amount, "refundTxId": refund_tx_id, "refundedAt": datetime.utcnow().isoformat(), } }, ) if result.modified_count != 1: raise ValueError("Order could not be cancelled") updated_order = orders_collection.find_one(OrderModel._order_match(order_id, id_khach_hang)) OrderModel._cleanup_order_design_assets(updated_order) return { "canceled": True, "cancelled": True, "hiddenFromUser": False, "refundAmount": refund_amount, "refundTxId": refund_tx_id, "status": "Canceled", "canceledBy": "user", "canceledAt": canceled_at, "canceledFromStatus": "Processing", "cancelReason": cancel_reason, } except Exception as e: # Nếu đã cộng tiền nhưng cập nhật order lỗi thì rollback balance. 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"Cancel refund failed: {str(e)}") # ========================= # TIER / DYNAMIC PRICING # ========================= @staticmethod def _get_tier_settings(): from model.tier_settings_model import TierSettingsModel return TierSettingsModel.get_settings() @staticmethod def _tier_rank_map(tiers: list) -> dict: return { str(item.get("code") or ""): int(item.get("order", 0)) for item in tiers or [] if str(item.get("code") or "").strip() } @staticmethod def _resolve_tier_from_paid_qty(paid_qty: int, tiers: list, default_tier: str) -> str: matched = default_tier sorted_tiers = sorted( tiers or [], key=lambda x: (int(x.get("order", 0)), int(x.get("minPaidQty", 0))) ) for item in sorted_tiers: if paid_qty >= int(item.get("minPaidQty", 0)): matched = str(item.get("code") or matched) return matched @staticmethod def _higher_tier(tier_a: str, tier_b: str, tiers: list) -> str: rank_map = OrderModel._tier_rank_map(tiers) return tier_a if rank_map.get(tier_a, 0) >= rank_map.get(tier_b, 0) else tier_b @staticmethod def _find_variant_price_for_tier(product_doc: dict, style: str, size: str, tier_code: str) -> float: variants = product_doc.get("variants") or [] exact_match = None fallback_match = None normalized_style = OrderModel._normalize_variant_text(style) normalized_size = OrderModel._normalize_variant_text(size) normalized_tier = OrderModel._normalize_variant_text(tier_code) for variant in variants: variant_style = OrderModel._normalize_variant_text(variant.get("style")) variant_size = OrderModel._normalize_variant_text(variant.get("size")) variant_tier = OrderModel._normalize_variant_text(OrderModel._get_variant_tier(variant)) if variant_style == normalized_style and variant_size == normalized_size: fallback_match = fallback_match or variant if normalized_tier and variant_tier == normalized_tier: exact_match = variant break chosen = exact_match or fallback_match if not chosen: raise ValueError( f"Không tìm thấy price variant cho product={product_doc.get('productId')} style={style}, size={size}, tier={tier_code}" ) return float(chosen.get("price", product_doc.get("price", 0)) or 0) @staticmethod def _expand_item_with_dynamic_tier( item: dict, running_paid_qty: int, baseline_tier: str, tiers: list, default_tier: str, ): product_id = str(item.get("productId") or "").strip() product_doc = products_collection.find_one({"productId": product_id}) if not product_doc: raise ValueError(f"Không tìm thấy sản phẩm {product_id}") qty = int(float(item.get("quantity", 0) or 0)) if qty <= 0: return [], running_paid_qty baseline_tier = str(baseline_tier or default_tier or "Silver").strip() segments = [] current_segment = None for _ in range(qty): qty_after_this_item = running_paid_qty + 1 # Tier đạt được theo số lượng đã mua trong tháng hiện tại, # tính cả sản phẩm đang xét. current_month_tier = OrderModel._resolve_tier_from_paid_qty( qty_after_this_item, tiers, default_tier, ) # Tier thực tế phải là tier cao hơn giữa: # 1. baseline_tier: tier được giữ từ tháng trước / tier hiện tại của user # 2. current_month_tier: tier đạt được trong tháng này effective_tier = OrderModel._higher_tier( baseline_tier, current_month_tier, tiers, ) unit_price = OrderModel._find_variant_price_for_tier( product_doc=product_doc, style=item.get("style", ""), size=item.get("size", ""), tier_code=effective_tier, ) if ( current_segment and current_segment["appliedTier"] == effective_tier and float(current_segment["price"]) == float(unit_price) ): current_segment["quantity"] += 1 else: if current_segment: segments.append(current_segment) current_segment = { **item, "quantity": 1, "price": unit_price, "appliedTier": effective_tier, "tier": effective_tier, } running_paid_qty += 1 if current_segment: segments.append(current_segment) return segments, running_paid_qty @staticmethod def _build_payment_quote(order_docs: list, user_id: str): settings = OrderModel._get_tier_settings() tiers = settings.get("tiers", []) default_tier = settings.get("defaultTier", "Silver") snapshot = UserModel.get_tier_snapshot(user_id) baseline_tier = str( snapshot.get("baselineTier") or snapshot.get("tierCurrent") or default_tier ).strip() tier_before_payment = str( snapshot.get("tierCurrent") or baseline_tier or default_tier ).strip() running_paid_qty = int(snapshot.get("currentMonthPaidQty") or 0) current_month_paid_qty_before_payment = running_paid_qty quoted_orders = [] grand_total = 0.0 for order_doc in order_docs: priced_items = [] raw_items = order_doc.get("items", []) or [] for item in raw_items: segments, running_paid_qty = OrderModel._expand_item_with_dynamic_tier( item=item, running_paid_qty=running_paid_qty, baseline_tier=baseline_tier, tiers=tiers, default_tier=default_tier, ) first_segment = True for segment in segments: cloned = dict(segment) if not first_segment: cloned["shipping"] = 0 cloned["tax"] = 0 first_segment = False priced_items.append(cloned) order_total = OrderModel._calculate_total(priced_items) grand_total += order_total quoted_orders.append({ "docId": str(order_doc["_id"]), "orderId": order_doc.get("orderId") or order_doc.get("id"), "items": priced_items, "total": round(order_total, 2), }) return { "orders": quoted_orders, "totalAmount": round(grand_total, 2), "baselineTier": baseline_tier, "tierBeforePayment": tier_before_payment, "currentMonthPaidQtyBeforePayment": current_month_paid_qty_before_payment, } @staticmethod def pay_orders(order_ids: list, id_khach_hang: str): if not id_khach_hang: raise ValueError("Missing id_khach_hang") base_query = { "uploadedById": id_khach_hang, "status": "Unpaid", "$or": [ {"orderId": {"$in": order_ids}}, {"id": {"$in": order_ids}}, ], } matched_orders = list(orders_collection.find(base_query)) matched_orders.sort( key=lambda doc: ( str(doc.get("createdAt") or ""), str(doc.get("orderId") or doc.get("id") or doc.get("_id")), ) ) if not matched_orders: raise ValueError("No unpaid orders found for payment") blocked_order_ids = [] for doc in matched_orders: # Đơn qua offer-workflow (admin set giá thủ công) → bỏ qua product validation is_offer_order = bool(doc.get("offerSetBy")) if is_offer_order: continue validation_errors = OrderModel._validate_order_for_payment(doc) issue_state = OrderModel._build_issue_state(validation_errors) should_update_issue = ( bool(doc.get("paymentBlocked")) != issue_state["paymentBlocked"] or doc.get("issueSummary", "") != issue_state["issueSummary"] or OrderModel._normalize_issue_details(doc) != issue_state["issueDetails"] ) if should_update_issue: orders_collection.update_one( {"_id": doc["_id"]}, {"$set": issue_state}, ) if issue_state["paymentBlocked"]: blocked_order_ids.append(str(doc.get("orderId") or doc.get("id") or doc.get("_id"))) if blocked_order_ids: raise ValueError( "Không thể thanh toán vì có đơn đang bị lỗi hoặc sản phẩm không còn hợp lệ: " + ", ".join(blocked_order_ids) + ". Vui lòng chỉnh sửa đơn rồi thử lại." ) user = UserModel.get_user_by_id(id_khach_hang) if not user: raise ValueError("User not found") # Tách offer-orders (giá admin set sẵn) và regular-orders (tính giá theo tier) offer_orders = [doc for doc in matched_orders if doc.get("offerSetBy")] regular_orders = [doc for doc in matched_orders if not doc.get("offerSetBy")] # Tính tổng tiền offer orders từ giá đã lưu offer_total = sum( round(float(doc.get("total") or 0), 2) for doc in offer_orders ) # Tính tổng tiền regular orders theo tier regular_quote = OrderModel._build_payment_quote(regular_orders, id_khach_hang) if regular_orders else {"totalAmount": 0.0, "orders": []} regular_total = regular_quote["totalAmount"] potential_total = round(offer_total + regular_total, 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: # Xử lý offer orders for doc in offer_orders: order_total = round(float(doc.get("total") or 0), 2) result = orders_collection.update_one( {"_id": doc["_id"], "status": "Unpaid"}, { "$set": { "status": "Processing", "paymentBlocked": False, "issueType": "", "issueStatus": "", "issueSummary": "", "issueDetails": [], "issueCount": 0, "issueUpdatedAt": "", } }, ) if result.modified_count == 1: paid_orders.append(doc.get("orderId") or doc.get("id")) paid_total += order_total # Xử lý regular orders quote_map = {item["docId"]: item for item in regular_quote["orders"]} for doc in regular_orders: doc_id = str(doc["_id"]) quote_doc = quote_map.get(doc_id) if not quote_doc: raise ValueError(f"Thiếu quote cho order {doc_id}") result = orders_collection.update_one( {"_id": doc["_id"], "status": "Unpaid"}, { "$set": { "items": quote_doc["items"], "total": quote_doc["total"], "status": "Processing", "paymentBlocked": False, "issueType": "", "issueStatus": "", "issueSummary": "", "issueDetails": [], "issueCount": 0, "issueUpdatedAt": "", } }, ) if result.modified_count == 1: paid_orders.append(doc.get("orderId") or doc.get("id")) paid_total += float(quote_doc["total"] or 0) 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) tier_snapshot = UserModel.sync_user_tier(id_khach_hang) 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, "tierCurrent": tier_snapshot["tierCurrent"], } @staticmethod def quote_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)) matched_orders.sort( key=lambda doc: ( str(doc.get("createdAt") or ""), str(doc.get("orderId") or doc.get("id") or doc.get("_id")), ) ) if not matched_orders: raise ValueError("No unpaid orders found for payment") blocked_order_ids = [] for doc in matched_orders: validation_errors = OrderModel._validate_order_for_payment(doc) issue_state = OrderModel._build_issue_state(validation_errors) if issue_state["paymentBlocked"]: blocked_order_ids.append( str(doc.get("orderId") or doc.get("id") or doc.get("_id")) ) if blocked_order_ids: raise ValueError( "Không thể thanh toán vì có đơn đang bị lỗi hoặc sản phẩm không còn hợp lệ: " + ", ".join(blocked_order_ids) + ". Vui lòng chỉnh sửa đơn rồi thử lại." ) user = UserModel.get_user_by_id(id_khach_hang) if not user: raise ValueError("User not found") quote = OrderModel._build_payment_quote(matched_orders, id_khach_hang) return { **quote, "orderCount": len(matched_orders), } @staticmethod def _admin_base_conditions(): return [ { "$or": [ {"status": {"$in": ["Processing", "Printing", "Shipped", "Delivered"]}}, {"status": {"$in": ["Liên Hệ", "Offer", "Rejected", "Unpaid"]}}, { "$and": [ {"status": {"$in": ["Canceled", "Cancelled"]}}, { "$or": [ {"canceledBy": {"$ne": "user"}}, {"cancelledBy": {"$ne": "user"}}, {"canceledFromStatus": "Processing"}, {"cancelledFromStatus": "Processing"}, {"refundAmount": {"$gt": 0}}, {"refundTxId": {"$nin": ["", None]}}, ] }, ] }, ] } ] @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", "Delivered", "Canceled", "Cancelled", "Liên Hệ", "Offer", "Rejected", "Unpaid", ] if status and status != "All": if status not in allowed_admin_statuses: return { "data": [], "total": 0, "page": page, "pageSize": page_size, "totalPages": 0, } if OrderModel._is_canceled_status(status): conditions.append({"status": {"$in": ["Canceled", "Cancelled"]}}) else: conditions.append({"status": status}) if OrderModel._is_canceled_status(status): conditions.append({ "$or": [ {"canceledBy": {"$ne": "user"}}, {"cancelledBy": {"$ne": "user"}}, {"canceledFromStatus": "Processing"}, {"cancelledFromStatus": "Processing"}, {"refundAmount": {"$gt": 0}}, {"refundTxId": {"$nin": ["", None]}}, ] }) 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_image_search_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() # Image Search KHÔNG hiển thị Delivered / Cancelled allowed_image_statuses = ["Processing", "Printing", "Shipped"] if status and status != "All": if status not in allowed_image_statuses: return { "data": [], "total": 0, "page": page, "pageSize": page_size, "totalPages": 0, } conditions.append({"status": status}) else: conditions.append({"status": {"$in": allowed_image_statuses}}) 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}) 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_orders_by_visual_matches( matches: list, keyword: str = None, status: str = None, dateFrom: str = None, dateTo: str = None, page: int = 1, page_size: int = 10, ): page = max(int(page or 1), 1) page_size = max(int(page_size or 10), 1) clean_matches = [] score_by_file = {} for item in matches or []: file_name = extract_design_file_name(item.get("file_name")) if not file_name: continue score = float(item.get("score") or 0) if file_name not in score_by_file or score > score_by_file[file_name]: score_by_file[file_name] = score clean_matches.append({ "file_name": file_name, "score": score, }) if not clean_matches: return { "data": [], "total": 0, "page": page, "pageSize": page_size, "totalPages": 0, } file_names = [item["file_name"] for item in clean_matches] conditions = OrderModel._admin_base_conditions() # Image Search chỉ hiển thị đơn còn đang xử lý, không hiện Delivered / Cancelled. allowed_image_statuses = ["Processing", "Printing", "Shipped"] if status and status != "All": if status not in allowed_image_statuses: return { "data": [], "total": 0, "page": page, "pageSize": page_size, "totalPages": 0, } conditions.append({"status": status}) else: conditions.append({"status": {"$in": allowed_image_statuses}}) file_conditions = [] for file_name in file_names: regex = {"$regex": re.escape(file_name), "$options": "i"} file_conditions.extend([ {"items.designLocal": regex}, {"items.designFile": regex}, {"items.designUrl": regex}, {"items.design": regex}, {"items.designOriginal": regex}, {"items.designDrive": regex}, {"items.image": regex}, {"items.imageUrl": regex}, ]) conditions.append({"$or": file_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}) 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} docs = list(orders_collection.find(query)) uploader_cache = {} rows = [] for doc in docs: matched_file = "" matched_score = 0.0 for item in doc.get("items", []) or []: if not isinstance(item, dict): continue for key in [ "designLocal", "designFile", "designUrl", "design", "designOriginal", "designDrive", "image", "imageUrl", ]: file_name = extract_design_file_name(item.get(key)) if file_name in score_by_file: if not matched_file or score_by_file[file_name] > matched_score: matched_file = file_name matched_score = score_by_file[file_name] if not matched_file: continue row = OrderModel._sanitize_order_list(doc, uploader_cache) row["matchedFileName"] = matched_file row["matchedImage"] = f"/pod-api/design/{matched_file}" row["matchScore"] = matched_score rows.append(row) rows.sort(key=lambda item: item.get("matchScore", 0), reverse=True) total_items = len(rows) total_pages = math.ceil(total_items / page_size) if total_items > 0 else 0 start = (page - 1) * page_size end = start + page_size return { "data": rows[start:end], "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", "Delivered"]}}] active_order_conditions = [{"status": {"$in": ["Processing", "Printing", "Shipped", "Delivered"]}}] 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) 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 == "Unpaid" or OrderModel._is_canceled_status(current_status): 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": payload.update( OrderModel._normalize_tracking_payload( tracking_number=tracking_number, tracking_link=tracking_link, ) ) 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 == "Unpaid" or OrderModel._is_canceled_status(current_status): 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": payload.update( OrderModel._normalize_tracking_payload( tracking_number=tracking_number, tracking_link=tracking_link, ) ) 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_mark_orders_as_delivered(order_ids: list): if not order_ids: raise ValueError("orderIds is required") unique_order_ids = [] seen = set() for raw_id in order_ids: order_id = str(raw_id or "").strip() if not order_id or order_id in seen: continue unique_order_ids.append(order_id) seen.add(order_id) if not unique_order_ids: raise ValueError("orderIds is required") updated = [] skipped = [] for order_id in unique_order_ids: order = orders_collection.find_one(OrderModel._admin_order_match(order_id)) if not order or OrderModel._is_hidden_from_admin(order): skipped.append({ "orderId": order_id, "reason": "Order not found", }) continue display_order_id = str(order.get("orderId") or order.get("id") or order_id) current_status = str(order.get("status") or "").strip() if current_status != "Shipped": skipped.append({ "orderId": display_order_id, "reason": "Order is not shipped", }) continue result = orders_collection.update_one( {"_id": order["_id"], "status": "Shipped"}, { "$set": { "status": "Delivered", "deliveredAt": datetime.utcnow().isoformat(), } }, ) if result.modified_count == 1: updated.append(display_order_id) updated_order = orders_collection.find_one({"_id": order["_id"]}) OrderModel._cleanup_order_design_assets(updated_order) else: skipped.append({ "orderId": display_order_id, "reason": "Order is not shipped", }) return { "updated": updated, "skipped": skipped, } @staticmethod def admin_cancel_order( identifier: str, admin_id: str = "", admin_name: str = "", cancel_reason: 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, "cancelReason": cancel_reason.strip() or "Cancelled by admin", } }, ) 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"]}) OrderModel._cleanup_order_design_assets(updated_order) 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 _build_ups_tracking_link(tracking_number: str): tracking_number = str(tracking_number or "").strip() if not tracking_number: return "" return f"{UPS_TRACK_BASE_URL}?tracknum={quote(tracking_number)}&loc=en_US" @staticmethod def _build_candidate_tracking_links(tracking_number: str, tracking_link: str = ""): candidates = [] seen = set() def add(url: str): url = str(url or "").strip() if not url or url in seen: return seen.add(url) candidates.append(url) tracking_number = str(tracking_number or "").strip().replace(" ", "").upper() tracking_link = str(tracking_link or "").strip() if tracking_number: add(f"{UPS_TRACK_BASE_URL}?tracknum={quote(tracking_number)}&loc=en_US") if tracking_link: add(tracking_link) if tracking_number: add(f"{UPS_TRACK_BASE_URL}?tracknum={quote(tracking_number)}&loc=en_US&requester=ST") add(f"{UPS_TRACK_BASE_URL}?tracknum={quote(tracking_number)}&loc=en_US&requester=ST/trackdetails") return candidates @staticmethod def _extract_tracking_number_from_link(tracking_link: str): tracking_link = str(tracking_link or "").strip() if not tracking_link: return "" try: parsed = urlparse(tracking_link) params = parse_qs(parsed.query) for key in ["tracknum", "trackNums", "trackingNumber"]: values = params.get(key) or [] if values and str(values[0]).strip(): return str(values[0]).strip().replace(" ", "").upper() except Exception: pass return "" @staticmethod def _normalize_tracking_payload(tracking_number: str = None, tracking_link: str = None): final_tracking_link = str(tracking_link or "").strip() final_tracking_number = str(tracking_number or "").strip().replace(" ", "").upper() if not final_tracking_number and final_tracking_link: final_tracking_number = OrderModel._extract_tracking_number_from_link(final_tracking_link) if not final_tracking_link and final_tracking_number: final_tracking_link = OrderModel._build_ups_tracking_link(final_tracking_number) if not final_tracking_number: raise ValueError("trackingNumber is required or must be extractable from trackingLink when confirming shipped") if not final_tracking_link: raise ValueError("trackingLink is required or could not be generated from trackingNumber") return { "trackingNumber": final_tracking_number, "tracking": final_tracking_number, "trackingLink": final_tracking_link, } @staticmethod def _build_ups_status_api_payload(tracking_number: str, tracking_link: str = ""): tracking_number = str(tracking_number or "").strip() tracking_link = str(tracking_link or "").strip() or OrderModel._build_ups_tracking_link(tracking_number) return { "Locale": "en_US", "TrackingNumber": [tracking_number.lower()], "isBarcodeScanned": False, "Requester": "st/trackdetails", "ClientUrl": tracking_link, "returnToValue": "", "AssociatedBcdnNumber": None, } @staticmethod def _extract_tracking_status_from_ups_json(data): track_details = [] if isinstance(data, dict): track_details = data.get("trackDetails") or [] detail = track_details[0] if isinstance(track_details, list) and track_details else {} if not isinstance(detail, dict): detail = {} package_status = str(detail.get("packageStatus") or "").strip() current_milestone = detail.get("currentMilestone") or {} current_milestone_name = "" if isinstance(current_milestone, dict): current_milestone_name = str(current_milestone.get("name") or "").strip() delivered_label_text = "" if str(detail.get("isDelivered") or "").lower() == "true" or detail.get("isDelivered") is True: delivered_label_text = package_status or current_milestone_name or "Delivered" elif package_status.lower() == "delivered": delivered_label_text = package_status elif current_milestone_name.lower() == "delivered": delivered_label_text = current_milestone_name else: milestones = detail.get("milestones") or [] if isinstance(milestones, list): for milestone in reversed(milestones): if not isinstance(milestone, dict): continue milestone_name = str(milestone.get("name") or "").strip() if milestone_name.lower() == "delivered": delivered_label_text = milestone_name break status_text = package_status or current_milestone_name or str(detail.get("progressBarType") or "").strip() preview = { "packageStatus": package_status, "isDelivered": detail.get("isDelivered"), "receivedBy": detail.get("receivedBy"), "leftAt": detail.get("leftAt"), "currentMilestone": current_milestone_name, } return { "isDelivered": bool(delivered_label_text), "deliveredLabelText": delivered_label_text, "statusText": status_text, "textPreview": str(preview)[:3000], "detail": detail, } @staticmethod def _fetch_ups_tracking_json(tracking_number: str, tracking_link: str = ""): tracking_number = str(tracking_number or "").strip() if not tracking_number: raise ValueError("Missing tracking number") client_url = str(tracking_link or "").strip() or OrderModel._build_ups_tracking_link(tracking_number) endpoint = "https://webapis.ups.com/track/api/Track/GetStatus?loc=en_US" payload = OrderModel._build_ups_status_api_payload( tracking_number=tracking_number, tracking_link=client_url, ) response = requests.post( endpoint, json=payload, headers={ "User-Agent": "Mozilla/5.0", "Accept": "application/json, text/plain, */*", "Content-Type": "application/json;charset=UTF-8", "Origin": "https://www.ups.com", "Referer": client_url, "Accept-Language": "en-US,en;q=0.9", }, timeout=30, ) response.raise_for_status() data = response.json() return { "json": data, "finalUrl": client_url, "httpStatus": response.status_code, "endpoint": endpoint, } @staticmethod def _extract_tracking_status_from_html(html: str): raw_html = str(html or "") if not raw_html: return { "isDelivered": False, "deliveredLabelText": "", "statusText": "", "textPreview": "", } collapsed_html = re.sub(r"\s+", " ", raw_html).strip() text_only = re.sub(r"<[^>]+>", " ", raw_html) collapsed_text = re.sub(r"\s+", " ", text_only).strip() searchable = f"{collapsed_html} {collapsed_text}" searchable_lower = searchable.lower() delivered_label_text = "" status_text = "" strong_delivered_checks = [ ("delivered on", "Delivered On"), ("proof of delivery", "Proof of Delivery"), ('"deliverystatus":"delivered"', 'deliveryStatus=Delivered'), ('"deliveryStatus":"Delivered"', 'deliveryStatus=Delivered'), ('"status":"Delivered"', 'status=Delivered'), ('"currentstatus":"Delivered"', 'currentStatus=Delivered'), ('"packageStatus":"Delivered"', 'packageStatus=Delivered'), ('"milestonestatus":"Delivered"', 'milestoneStatus=Delivered'), ] for needle, label in strong_delivered_checks: if needle.lower() in searchable_lower: delivered_label_text = label status_text = label break if not delivered_label_text: delivered_regexes = [ r"\bdelivered\s+on\b", r"\bproof\s+of\s+delivery\b", r"\breceived\s+by\s*:", r"\bdelivered\b.{0,200}\bcharlotte\b", r"Thursday,\s+March\s+26.{0,80}at\s+Dock", ] for pattern in delivered_regexes: match = re.search(pattern, searchable, re.IGNORECASE) if match: delivered_label_text = match.group(0) status_text = match.group(0) break if not status_text: in_transit_patterns = [ r"\bOn\s+the\s+Way\b", r"\bIn\s+Transit\b", r"\bOut\s+for\s+Delivery\b", r"\bLabel\s+Created\b", ] for pattern in in_transit_patterns: match = re.search(pattern, searchable, re.IGNORECASE) if match: status_text = match.group(0) break return { "isDelivered": bool(delivered_label_text), "deliveredLabelText": delivered_label_text, "statusText": status_text, "textPreview": searchable[:3000], } @staticmethod def _fetch_partial_tracking_html(candidate_link: str): response = requests.get( candidate_link, headers={ "User-Agent": "Mozilla/5.0", "Accept-Language": "en-US,en;q=0.9", }, timeout=30, allow_redirects=True, ) response.raise_for_status() response.encoding = response.encoding or response.apparent_encoding or "utf-8" html = response.text or "" return { "html": html, "finalUrl": response.url or candidate_link, "httpStatus": response.status_code, } @staticmethod def _check_delivered_on_with_requests(tracking_number: str, tracking_link: str = ""): tracking_number = str(tracking_number or "").strip().replace(" ", "").upper() tracking_link = str(tracking_link or "").strip() if not tracking_number and tracking_link: tracking_number = OrderModel._extract_tracking_number_from_link(tracking_link) if not tracking_number: raise ValueError("Missing tracking number") last_error = "" try: fetched_json = OrderModel._fetch_ups_tracking_json( tracking_number=tracking_number, tracking_link=tracking_link, ) parsed_json = OrderModel._extract_tracking_status_from_ups_json( fetched_json.get("json") or {} ) return { "isDelivered": parsed_json.get("isDelivered", False), "deliveredLabelText": parsed_json.get("deliveredLabelText", ""), "statusText": parsed_json.get("statusText", ""), "textPreview": parsed_json.get("textPreview", ""), "finalUrl": fetched_json.get("finalUrl", tracking_link or OrderModel._build_ups_tracking_link(tracking_number)), "httpStatus": fetched_json.get("httpStatus", 0), "endpoint": fetched_json.get("endpoint", ""), "source": "ups_json_endpoint", } except Exception as exc: last_error = str(exc) candidate_links = OrderModel._build_candidate_tracking_links( tracking_number=tracking_number, tracking_link=tracking_link, ) for candidate_link in candidate_links: try: fetched = OrderModel._fetch_partial_tracking_html(candidate_link) parsed = OrderModel._extract_tracking_status_from_html(fetched.get("html", "")) return { "isDelivered": parsed.get("isDelivered", False), "deliveredLabelText": parsed.get("deliveredLabelText", ""), "statusText": parsed.get("statusText", ""), "textPreview": parsed.get("textPreview", ""), "finalUrl": fetched.get("finalUrl", candidate_link), "httpStatus": fetched.get("httpStatus", 0), "endpoint": candidate_link, "source": "requests_html_fallback", } except Exception as exc: last_error = f"{candidate_link} :: {exc}" raise ValueError(last_error or "Failed to fetch UPS tracking page") @staticmethod def sync_one_shipped_order_to_delivered(order_doc: dict): if not isinstance(order_doc, dict): return {"updated": False, "reason": "invalid_order"} if str(order_doc.get("status") or "").strip() != "Shipped": return {"updated": False, "reason": "status_not_shipped"} tracking_link = str(order_doc.get("trackingLink") or "").strip() tracking_number = str(order_doc.get("trackingNumber") or order_doc.get("tracking") or "").strip() if not tracking_number and tracking_link: tracking_number = OrderModel._extract_tracking_number_from_link(tracking_link) if not tracking_link and tracking_number: tracking_link = OrderModel._build_ups_tracking_link(tracking_number) if not tracking_link: return {"updated": False, "reason": "missing_tracking_link"} if not tracking_number: return {"updated": False, "reason": "missing_tracking_number"} try: tracking_result = OrderModel._check_delivered_on_with_requests( tracking_number=tracking_number, tracking_link=tracking_link, ) base_payload = { "tracking": tracking_number, "trackingNumber": tracking_number, "trackingLink": tracking_link, "lastTrackingCheckedAt": datetime.utcnow().isoformat(), "lastTrackingSyncError": "", "lastTrackingSource": tracking_result.get("source", "requests_html"), "lastTrackingTextPreview": tracking_result.get("textPreview", ""), "lastTrackingDeliveredLabel": tracking_result.get("deliveredLabelText", ""), "lastTrackingFinalUrl": tracking_result.get("finalUrl", tracking_link), } if tracking_result.get("isDelivered"): base_payload["status"] = "Delivered" base_payload["deliveredAt"] = datetime.utcnow().isoformat() result = orders_collection.update_one( {"_id": order_doc["_id"], "status": "Shipped"}, {"$set": base_payload}, ) return { "updated": result.modified_count == 1, "reason": "ups_delivered_found", "deliveredLabelText": tracking_result.get("deliveredLabelText", ""), } result = orders_collection.update_one( {"_id": order_doc["_id"], "status": "Shipped"}, {"$set": base_payload}, ) updated_ok = result.modified_count == 1 if updated_ok: updated_order = orders_collection.find_one({"_id": order_doc["_id"]}) OrderModel._cleanup_order_design_assets(updated_order) return { "updated": updated_ok, "reason": "ups_delivered_found", "deliveredLabelText": tracking_result.get("deliveredLabelText", ""), } except Exception as e: orders_collection.update_one( {"_id": order_doc["_id"]}, { "$set": { "tracking": tracking_number, "trackingNumber": tracking_number, "trackingLink": tracking_link, "lastTrackingCheckedAt": datetime.utcnow().isoformat(), "lastTrackingSyncError": str(e), "lastTrackingSource": "requests_html_error", } }, ) return { "updated": False, "reason": str(e), } @staticmethod def sync_shipped_orders_to_delivered( limit: int = 100, doc_ids: list = None, min_age_seconds: int = 0, ): limit = max(int(limit or 0), 1) min_age_seconds = max(int(min_age_seconds or 0), 0) tracking_ready_condition = { "$or": [ {"trackingLink": {"$exists": True, "$ne": ""}}, {"trackingNumber": {"$exists": True, "$ne": ""}}, {"tracking": {"$exists": True, "$ne": ""}}, ] } query = { "status": "Shipped", "$and": [tracking_ready_condition], } if min_age_seconds > 0: stale_before = datetime.utcnow() - timedelta(seconds=min_age_seconds) query["$and"].append( { "$or": [ {"lastTrackingCheckedAt": {"$exists": False}}, {"lastTrackingCheckedAt": ""}, {"lastTrackingCheckedAt": {"$lt": stale_before.isoformat()}}, ] } ) if doc_ids: object_ids = [] for raw_id in doc_ids: try: object_ids.append(ObjectId(str(raw_id))) except Exception: continue if not object_ids: raise ValueError("No valid docIds provided") query["_id"] = {"$in": object_ids} cursor = ( orders_collection.find(query) .sort([("lastTrackingCheckedAt", 1), ("createdAt", 1)]) .limit(limit) ) checked = 0 updated = 0 results = [] for order_doc in cursor: checked += 1 sync_result = OrderModel.sync_one_shipped_order_to_delivered(order_doc) if sync_result.get("updated"): updated += 1 results.append({ "orderId": str(order_doc.get("orderId") or order_doc.get("id") or order_doc.get("_id")), "docId": str(order_doc.get("_id")), **sync_result, }) return {"checked": checked, "updated": updated, "results": results} @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] # ── FBA & FBT Flow ───────────────────────────────────────────────────────── @staticmethod def create_fba_fbt_order(data: dict, actor: dict) -> dict: order_number = str(data.get("orderNumber") or "").strip() if not order_number: raise ValueError("orderNumber is required") now = datetime.now(VN_TZ) timestamp = now.strftime("%Y%m%d%H%M%S") + f"{now.microsecond // 1000:03d}" random_part = uuid.uuid4().hex[:6] order_id = f"{order_number}_{timestamp}_{random_part}" customer_data = data.get("customer") or {} customer = { "name": str(customer_data.get("name") or "").strip(), "phone": str(customer_data.get("phone") or "").strip(), "taxNumber": str(customer_data.get("taxNumber") or "").strip(), "address": { "street": str(customer_data.get("address1") or "").strip(), "address2": str(customer_data.get("address2") or "").strip(), "city": str(customer_data.get("city") or "").strip(), "state": str(customer_data.get("state") or "").strip(), "zipCode": str(customer_data.get("postalCode") or "").strip(), "country": str(customer_data.get("country") or "").strip(), }, } items = [] for item in (data.get("items") or []): items.append({ "skuNumber": str(item.get("skuNumber") or "").strip(), "quantity": int(item.get("quantity") or 1), "description": str(item.get("description") or "").strip(), "price": 0.0, "shipping": 0.0, "tax": 0.0, }) now_iso = now.isoformat() doc = { "orderId": order_id, "orderNumber": order_number, "channelType": "fba_fbt", "importType": "FBA&FBT", "id_khach_hang": str(actor.get("id") or "").strip(), "customer": customer, "items": items, "total": 0.0, "status": "Liên Hệ", "tracking": "", "trackingNumber": "", "trackingLink": "", "sourceOrderIds": [], "uploadedByName": str(actor.get("name") or "").strip(), "uploadedByEmail": str(actor.get("email") or "").strip(), "uploadedById": str(actor.get("id") or "").strip(), "paymentBlocked": False, "issueType": "", "issueStatus": "", "issueSummary": "", "issueDetails": [], "issueCount": 0, "issueUpdatedAt": "", "createdAt": now_iso, "updatedAt": now_iso, } orders_collection.insert_one(doc) doc["_id"] = str(doc["_id"]) return doc @staticmethod def _sanitize_fba_fbt_order_list(doc: dict) -> dict: order_id = doc.get("orderId") or doc.get("id") or str(doc["_id"]) customer = doc.get("customer") or {} customer_name = customer.get("name", "") if isinstance(customer, dict) else str(customer or "") return { "id": order_id, "docId": str(doc["_id"]), "orderNumber": doc.get("orderNumber") or order_id, "customer": customer_name, "status": doc.get("status", "Liên Hệ"), "total": round(float(doc.get("total") or 0), 2), "createdAt": str(doc.get("createdAt", ""))[:10], "offerSetAt": doc.get("offerSetAt", ""), "offerSetByName": doc.get("offerSetBy", ""), "uploadedByName": doc.get("uploadedByName", ""), "uploadedByEmail": doc.get("uploadedByEmail", ""), } @staticmethod def _sanitize_fba_fbt_order_detail(doc: dict) -> dict: order_id = doc.get("orderId") or doc.get("id") or str(doc["_id"]) raw_customer = doc.get("customer") or {} if not isinstance(raw_customer, dict): raw_customer = {} customer = { "name": raw_customer.get("name", ""), "phone": raw_customer.get("phone", ""), "email": raw_customer.get("email", ""), "address1": raw_customer.get("address1", ""), "address2": raw_customer.get("address2", ""), "city": raw_customer.get("city", ""), "state": raw_customer.get("state", ""), "postalCode": raw_customer.get("postalCode", ""), "country": raw_customer.get("country", ""), } raw_items = doc.get("items", []) or [] items = [] for item in raw_items: if isinstance(item, dict): items.append({ "skuNumber": item.get("skuNumber") or item.get("productId") or (item.get("fbaRequest") or {}).get("sku", ""), "quantity": item.get("quantity", 1), "description": item.get("description") or item.get("productName") or (item.get("fbaRequest") or {}).get("productName", ""), "price": round(float(item.get("price") or 0), 2), "shipping": round(float(item.get("shipping") or 0), 2), "tax": round(float(item.get("tax") or 0), 2), }) return { "id": order_id, "docId": str(doc["_id"]), "orderNumber": doc.get("orderNumber") or order_id, "customer": customer, "items": items, "total": round(float(doc.get("total") or 0), 2), "status": doc.get("status", "Liên Hệ"), "offerSetBy": doc.get("offerSetBy", ""), "offerSetAt": doc.get("offerSetAt", ""), "offerApprovedAt": doc.get("offerApprovedAt", ""), "offerRejectedAt": doc.get("offerRejectedAt", ""), "tracking": doc.get("tracking") or doc.get("trackingNumber", ""), "trackingLink": doc.get("trackingLink", ""), "cancelReason": doc.get("cancelReason", ""), "createdAt": doc.get("createdAt", ""), "uploadedByName": doc.get("uploadedByName", ""), "uploadedByEmail": doc.get("uploadedByEmail", ""), } @staticmethod def get_my_fba_fbt_orders( actor_id: str, keyword: str = None, status: str = None, page: int = 1, page_size: int = 10, ): page = max(page, 1) page_size = max(page_size, 1) conditions = [ {"uploadedById": actor_id}, ] if keyword: escaped = re.escape(keyword.strip()) regex = {"$regex": escaped, "$options": "i"} conditions.append({"$or": [ {"orderId": regex}, {"orderNumber": regex}, {"customer.name": regex}, ]}) if status and status not in ("all", "All", ""): conditions.append({"status": status}) query = {"$and": conditions} if len(conditions) > 1 else conditions[0] total = orders_collection.count_documents(query) total_pages = math.ceil(total / page_size) if total > 0 else 0 skip = (page - 1) * page_size cursor = ( orders_collection.find(query) .sort([("createdAt", -1)]) .skip(skip) .limit(page_size) ) return { "data": [OrderModel._sanitize_fba_fbt_order_list(doc) for doc in cursor], "total": total, "page": page, "pageSize": page_size, "totalPages": total_pages, } @staticmethod def get_my_fba_fbt_order_detail(order_id: str, actor_id: str) -> dict: order = orders_collection.find_one({ "uploadedById": actor_id, "$or": [{"orderId": order_id}, {"id": order_id}], }) if not order: raise ValueError(f"Order '{order_id}' not found") return OrderModel._sanitize_fba_fbt_order_detail(order) @staticmethod def admin_get_fba_fbt_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 = [{"importType": "FBA&FBT"}] if keyword: escaped = re.escape(keyword.strip()) regex = {"$regex": escaped, "$options": "i"} conditions.append({"$or": [ {"orderId": regex}, {"orderNumber": regex}, {"customer.name": regex}, {"uploadedByName": regex}, {"uploadedByEmail": regex}, ]}) if status and status != "All": conditions.append({"status": status}) else: conditions.append({"status": {"$in": [ "Liên Hệ", "Offer", "Rejected", "Unpaid", "Processing", "Printing", "Shipped", "Delivered", ]}}) date_filter = OrderModel._build_date_filter(dateFrom, dateTo) if date_filter: conditions.append(date_filter) query = {"$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)]) .skip(skip) .limit(page_size) ) return { "data": [OrderModel._sanitize_fba_fbt_order_list(doc) for doc in cursor], "total": total_items, "page": page, "pageSize": page_size, "totalPages": total_pages, } @staticmethod def admin_get_fba_fbt_order_detail(identifier: str) -> dict: from bson.errors import InvalidId doc = None try: doc = orders_collection.find_one({"_id": ObjectId(identifier), "importType": "FBA&FBT"}) except (InvalidId, Exception): pass if not doc: doc = orders_collection.find_one({ "$and": [ {"importType": "FBA&FBT"}, {"$or": [{"orderId": identifier}, {"orderNumber": identifier}]}, ] }) if not doc: raise ValueError("Order not found") return OrderModel._sanitize_fba_fbt_order_detail(doc) @staticmethod def admin_set_offer(identifier: str, offer_items: list, admin_id: str, admin_name: str): try: obj_id = ObjectId(identifier) query = {"_id": obj_id} except Exception: query = {"$or": [{"orderId": identifier}, {"id": identifier}]} order = orders_collection.find_one(query) if not order: raise ValueError(f"Order '{identifier}' not found") if order.get("status") not in ["Liên Hệ", "Offer", "Rejected"]: raise ValueError( f"Chỉ đơn có trạng thái 'Liên Hệ', 'Offer' hoặc 'Rejected' mới có thể nhận giá offer. " f"Trạng thái hiện tại: {order.get('status')}" ) current_items = list(order.get("items", [])) updated_items = [] for offer_item in offer_items: item_idx = int(offer_item.get("index", 0)) if 0 <= item_idx < len(current_items): updated_item = dict(current_items[item_idx]) updated_item["price"] = float(offer_item.get("price", 0)) updated_item["shipping"] = float(offer_item.get("shipping", 0)) updated_item["tax"] = float(offer_item.get("tax", 0)) updated_items.append(updated_item) if not updated_items: raise ValueError("Danh sách offer_items không hợp lệ") total = OrderModel._calculate_total(updated_items) now = datetime.utcnow().isoformat() orders_collection.update_one( {"_id": order["_id"]}, { "$set": { "status": "Offer", "items": updated_items, "total": total, "offerSetBy": admin_name, "offerSetById": admin_id, "offerSetAt": now, "paymentBlocked": False, "issueType": "", "issueStatus": "", "issueSummary": "", "issueDetails": [], "issueCount": 0, } }, ) updated_order = orders_collection.find_one({"_id": order["_id"]}) return OrderModel._sanitize_fba_fbt_order_detail(updated_order) @staticmethod def user_respond_offer(order_id: str, id_khach_hang: str, action: str): order = orders_collection.find_one({ "uploadedById": id_khach_hang, "$or": [{"orderId": order_id}, {"id": order_id}], }) if not order: raise ValueError(f"Order '{order_id}' not found") if order.get("status") != "Offer": raise ValueError("Chỉ đơn có trạng thái 'Offer' mới có thể được chấp nhận hoặc từ chối") now = datetime.utcnow().isoformat() if action == "approve": orders_collection.update_one( {"_id": order["_id"]}, {"$set": {"status": "Unpaid", "offerApprovedAt": now}}, ) elif action == "reject": orders_collection.update_one( {"_id": order["_id"]}, {"$set": {"status": "Rejected", "offerRejectedAt": now}}, ) else: raise ValueError(f"Action không hợp lệ: '{action}'. Phải là 'approve' hoặc 'reject'.") updated_order = orders_collection.find_one({"_id": order["_id"]}) return OrderModel._sanitize_fba_fbt_order_detail(updated_order)