import json import os import re from datetime import datetime, time from bson import ObjectId from pymongo.errors import DuplicateKeyError from fastapi import HTTPException, status from dotenv import load_dotenv from typing import Optional,List, Dict, Any from pymongo import ASCENDING, DESCENDING from app.v1.libraries.object import str_to_objectid from ...models.platform.workforcemodel import ( Workforce, WorkforceBase, WorkforceUpdate, WorkforceResponseList ) from app.v1.services.sequence import get_next_sequence_value_int load_dotenv() # Collection name COLLECTION_NAME = "workforce" TAG_COLLECTION = "tag_name" ACCOUNTS_COLLECTION = "accounts" def create_workforce_service(workforce: WorkforceBase, db) -> dict: workforce_collection = db[COLLECTION_NAME] # ✅ Step 1: Get account document accounts_collection = db[ACCOUNTS_COLLECTION] account = accounts_collection.find_one({"_id": ObjectId(workforce.account_id)}) if not account: raise HTTPException(status_code=404, detail="Account not found") # ✅ Step 2: Extract max allowed workforce count from account workforce_range = str(account.get("workforce_count", "0-10")).strip() if "-" in workforce_range: # Example: "0-10" → take "10" try: max_allowed = int(workforce_range.split("-")[-1].strip()) except ValueError: max_allowed = 10 elif workforce_range.endswith("+"): # Example: "1000+" → treat as unlimited (very high cap) try: base = int(workforce_range.replace("+", "").strip()) max_allowed = 10**9 # effectively unlimited except ValueError: max_allowed = 10**9 else: # Example: "250" try: max_allowed = int(workforce_range) except ValueError: max_allowed = 10 # fallback # ✅ Step 3: Count existing workforce for this account existing_count = workforce_collection.count_documents({"account_id": workforce.account_id}) if existing_count >= max_allowed: raise HTTPException( status_code=400, detail=f"Workforce limit reached. Maximum allowed for this account is {max_allowed}." ) # Check for existing mobile/email if workforce_collection.find_one({"$or": [ {"mobile_number": workforce.mobile_number}, {"email_id": workforce.email_id} ]}): raise HTTPException(status_code=400, detail="Workforce with this contact information already exists") # ✅ Handle tag_name before inserting workforce tag_collection = db[TAG_COLLECTION] if workforce.tag_name: existing_tag = tag_collection.find_one({"tag_name": workforce.tag_name}) if not existing_tag: # only insert if new tag_collection.insert_one({ "tag_name": workforce.tag_name, "created_date": datetime.utcnow() }) # Create base data from model workforce_data = workforce.dict() # Changed variable name from new_workforce_data to workforce_data workforce_data.update({ "created_date": datetime.utcnow(), "last_updated": datetime.utcnow(), "status": "active" # Added default status }) # Explicitly include photo even if None if "photo" not in workforce_data: workforce_data["photo"] = None workforce_data["is_fleet_assigned"] = False try: # Insert the complete document workforce_data["w_id"] = get_next_sequence_value_int("w_id", db) result = workforce_collection.insert_one(workforce_data) # Get the inserted document inserted_data = workforce_collection.find_one({"_id": result.inserted_id}) # Convert for response inserted_data["_id"] = str(inserted_data["_id"]) if inserted_data.get("photo"): inserted_data["photo"] = f"{BASE_URL}/public/workforce/{inserted_data['photo']}" return inserted_data except DuplicateKeyError: raise HTTPException(status_code=400, detail="Duplicate workforce entry") def get_workforce_service( skip: int, limit: int, q: Optional[str], status: Optional[str], created_date_from: Optional[str], created_date_to: Optional[str], sort_by: Optional[str], sort_order: Optional[str], account_id: Optional[str], db, current_user ) -> dict: workforce_collection = db[COLLECTION_NAME] query = {} if q: regex_query = {"$regex": q, "$options": "i"} query["$or"] = [ {"first_name": regex_query}, {"last_name": regex_query}, {"email_id": regex_query}, {"mobile_number": regex_query} ] if status: query["status"] = status # 🔹 Account ID filter if account_id: query["account_id"] = account_id # Add date range filter if created_date_from or created_date_to: date_filter = {} if created_date_from: date_filter["$gte"] = datetime.strptime(created_date_from, "%Y-%m-%d") if created_date_to: date_filter["$lte"] = datetime.strptime(created_date_to, "%Y-%m-%d") query["created_date"] = date_filter # Sorting sort_fields = { "first_name": "first_name", "last_name": "last_name", "mobile_number": "mobile_number", "email_id": "email_id", "status": "status", "created_date": "created_date" } sort_field = sort_fields.get(sort_by, "created_date") sort_direction = ASCENDING if sort_order == "asc" else DESCENDING cursor = workforce_collection.find(query).sort(sort_field, sort_direction).skip(skip).limit(limit) workforce = list(cursor) #workforce = list(workforce_collection.find(query).skip(skip).limit(limit)) for entry in workforce: entry["workforce_id"] = str(entry["_id"]) for field in ['created_date', 'last_updated', 'date_of_birth', 'join_date']: if field in entry and isinstance(entry[field], datetime): entry[field] = entry[field].isoformat() total_count = workforce_collection.count_documents(query) return {"total_count": total_count, "workforce": workforce} def read_workforce_service(workforce_id: str, db) -> dict: workforce = db[COLLECTION_NAME].find_one({"_id": str_to_objectid(workforce_id)}) if workforce: workforce["id"] = str(workforce["_id"]) return workforce def update_workforce_service(workforce_id: str, workforce_data: WorkforceUpdate, db) -> dict: workforce_collection = db[COLLECTION_NAME] existing = workforce_collection.find_one({"_id": str_to_objectid(workforce_id)}) if not existing: raise HTTPException(status_code=404, detail="Workforce not found") update_data = {k: v for k, v in workforce_data.dict().items() if v is not None} result = workforce_collection.update_one( {"_id": str_to_objectid(workforce_id)}, {"$set": update_data} ) if result.matched_count == 0: raise HTTPException(status_code=404, detail="Workforce not found") updated = workforce_collection.find_one({"_id": str_to_objectid(workforce_id)}) if updated: updated["id"] = str(updated.pop("_id")) return updated raise HTTPException(status_code=404, detail="Workforce not found after update") def delete_workforce_service(workforce_id: str, db) -> dict: workforce_collection = db[COLLECTION_NAME] workforce = workforce_collection.find_one({"_id": str_to_objectid(workforce_id)}) if not workforce: raise HTTPException(status_code=404, detail="Workforce not found") workforce["id"] = str(workforce["_id"]) del workforce["_id"] workforce_collection.delete_one({"_id": str_to_objectid(workforce_id)}) return workforce def get_workforce_dropdown_list(db, account_id: str, query: str = "") -> List[Dict[str, str]]: collection = db[COLLECTION_NAME] filters = {"account_id": account_id} if query: filters["first_name"] = {"$regex": re.escape(query), "$options": "i"} results = collection.find(filters, {"first_name": 1}) return [{"id": str(f["_id"]), "first_name": f.get("first_name", "")} for f in results] def get_freeworkforce_with_counts(db, account_id: str, query: str = "") -> List[Dict[str, object]]: collection = db[COLLECTION_NAME] schedule_collection = db["schedules"] task_collection = db["tasks"] filters = {"account_id": account_id} if query: filters["first_name"] = {"$regex": re.escape(query), "$options": "i"} workforce_cursor = collection.find(filters, {"first_name": 1}) response = [] for f in workforce_cursor: workforce_id = str(f["_id"]) # Count schedules for this workforce schedule_count = schedule_collection.count_documents({ "account_id": account_id, "workforce_id": workforce_id }) # Count tasks for this workforce task_count = task_collection.count_documents({ "account_id": account_id, "workforce_id": workforce_id }) response.append({ "id": workforce_id, "first_name": f.get("first_name", ""), "total_schedules": schedule_count, "total_tasks": task_count }) return response # async def get_workforce_details_with_tasks_and_schedules(db, account_id: str) -> List[Dict[str, Any]]: # workforce_collection = db["workforce"] # schedule_collection = db["schedules"] # task_collection = db["tasks"] # # Fetch all workforces for the account # workforces = workforce_collection.find({"account_id": account_id}) # response = [] # for workforce in workforces: # workforce_id = str(workforce["_id"]) # # Get schedules # schedules = list(schedule_collection.find({ # "account_id": account_id, # "workforce_id": workforce_id # })) # # Get tasks # tasks = list(task_collection.find({ # "account_id": account_id, # "workforce_info.workforce_id": workforce_id # })) # # Prepare response # response.append({ # "workforce_id": workforce_id, # "first_name": workforce.get("first_name", ""), # "last_name": workforce.get("last_name", ""), # "email_id": workforce.get("email_id", ""), # "total_schedules": len(schedules), # "total_tasks": len(tasks), # "schedules": [ # { # "schedule_id": str(sch["_id"]), # "schedule_name": sch.get("schedule_name", ""), # "status": sch.get("status", ""), # "start_time": sch.get("start_time"), # "end_time": sch.get("end_time"), # "created_date": sch.get("created_date") # } # for sch in schedules # ], # "tasks": [ # { # "task_id": str(task["_id"]), # "status": task.get("status"), # "scheduled_start": task.get("scheduled_start"), # "scheduled_end": task.get("scheduled_end"), # "priority": task.get("priority"), # "task_mode": task.get("task_mode"), # "customer_info": task.get("customer_info", {}), # "fleet_info": task.get("fleet_info", {}) # } # for task in tasks # ] # }) # return response # async def get_workforce_details_with_tasks_and_schedules( # db, # account_id: str, # workforce_id: Optional[str] = None # ) -> List[Dict[str, Any]]: # workforce_collection = db["workforce"] # schedule_collection = db["schedules"] # task_collection = db["tasks"] # filters = {"account_id": account_id} # if workforce_id: # filters["_id"] = workforce_id # If you're storing _id as ObjectId, wrap with ObjectId(workforce_id) # workforces = workforce_collection.find(filters) # response = [] # for workforce in workforces: # wf_id_str = str(workforce["_id"]) # schedules = list(schedule_collection.find({ # "account_id": account_id, # "workforce_id": wf_id_str # })) # tasks = list(task_collection.find({ # "account_id": account_id, # "workforce_info.workforce_id": wf_id_str # })) # response.append({ # "workforce_id": wf_id_str, # "first_name": workforce.get("first_name", ""), # "last_name": workforce.get("last_name", ""), # "email_id": workforce.get("email_id", ""), # "total_schedules": len(schedules), # "total_tasks": len(tasks), # "schedules": [ # { # "schedule_id": str(sch["_id"]), # "schedule_name": sch.get("schedule_name", ""), # "status": sch.get("status", ""), # "start_time": sch.get("start_time"), # "end_time": sch.get("end_time"), # "created_date": sch.get("created_date") # } # for sch in schedules # ], # "tasks": [ # { # "task_id": str(task["_id"]), # "status": task.get("status"), # "scheduled_start": task.get("scheduled_start"), # "scheduled_end": task.get("scheduled_end"), # "priority": task.get("priority"), # "task_mode": task.get("task_mode"), # "customer_info": task.get("customer_info", {}), # "fleet_info": task.get("fleet_info", {}) # } # for task in tasks # ] # }) # return response async def get_workforce_details_with_tasks_and_schedules( db, account_id: str, workforce_id: Optional[str] = None ) -> List[Dict[str, Any]]: workforce_collection = db["workforce"] schedule_collection = db["schedules"] task_collection = db["tasks"] filters = {"account_id": account_id} if workforce_id: filters["_id"] = ObjectId(workforce_id) workforces = workforce_collection.find(filters) response = [] for workforce in workforces: wf_id_str = str(workforce["_id"]) # 1️⃣ Get schedules for this workforce schedules = list(schedule_collection.find({ "account_id": account_id, "workforce_id": wf_id_str })) # Extract schedule_ids as strings schedule_ids = [str(sch["_id"]) for sch in schedules] # 2️⃣ Get tasks matching by schedule_id tasks = [] if schedule_ids: tasks = list(task_collection.find({ #"account_id": account_id, "schedule_id": {"$in": schedule_ids} })) # 3️⃣ Append response entry response.append({ "workforce_id": wf_id_str, "first_name": workforce.get("first_name", ""), "last_name": workforce.get("last_name", ""), "email_id": workforce.get("email_id", ""), "fleet_info": workforce.get("fleet_info", []), "total_schedules": len(schedules), "total_tasks": len(tasks), "schedules": [ { "schedule_id": str(sch["_id"]), "schedule_name": sch.get("schedule_name", ""), "status": sch.get("status", ""), "start_time": sch.get("start_time"), "end_time": sch.get("end_time"), "created_date": sch.get("created_date") } for sch in schedules ], "tasks": [ { "task_id": str(task["_id"]), "status": task.get("status"), "scheduled_start": task.get("scheduled_start"), "scheduled_end": task.get("scheduled_end"), "timing": task.get("timing"), "location": task.get("location"), "priority": task.get("priority"), "task_mode": task.get("task_mode"), "customer_info": task.get("customer_info", {}), "fleet_info": task.get("fleet_info", {}) } for task in tasks ] }) return response def get_shift_period(shift_timing: str): """Return (start_time, end_time) tuple for a shift_timing label.""" shift_map = { "Morning": (time(6, 0), time(12, 0)), "Afternoon": (time(12, 0), time(18, 0)), "Evening": (time(18, 0), time(23, 0)), "Night": (time(23, 0), time(6, 0)), # spans midnight "Flexible": (time(0, 0), time(23, 59)), } return shift_map.get(shift_timing, (time(0, 0), time(23, 59))) def get_workforce_list_by_type(db, account_id: str, workforce_type: str, query: str = "", skills: Optional[List[str]] = None, vehicle_required: Optional[bool] = None, can_fly: Optional[bool] = None, addons_name: Optional[str] = None,vehicle_model_id: Optional[str] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, customer_id: Optional[str] = None) -> List[Dict[str, object]]: collection = db[COLLECTION_NAME] schedule_collection = db["schedules"] task_collection = db["tasks"] filters = {"account_id": account_id} if query: filters["first_name"] = {"$regex": re.escape(query), "$options": "i"} if skills and len(skills) > 0: filters["skills"] = {"$in": skills} # ✅ Added vehicle_required filter if vehicle_required is not None: if vehicle_required is False: filters["$or"] = [ {"is_fleet_assigned": False}, {"is_fleet_assigned": {"$exists": False}} ] else: # vehicle_required == True filters["is_fleet_assigned"] = True # 🔹 Filter by can_fly (check both top-level and nested fleet_info.can_fly) if can_fly is not None: if can_fly is False: # False = include missing fields also filters["$or"] = [ {"can_fly": False}, {"can_fly": {"$exists": False}}, {"fleet_info.can_fly": False}, {"fleet_info.can_fly": {"$exists": False}} ] else: # True = only match explicit True filters["$or"] = [ {"can_fly": True}, {"fleet_info.can_fly": True} ] # 🔹 Filter by addons_name (exact match) if addons_name is not None: filters["addons_name"] = addons_name # 🔹 Filter by model_name (exact match) if vehicle_model_id is not None: filters["vehicle_model_id"] = vehicle_model_id # --------------------------------------------------------- # ⭐ NEW CUSTOMER-specific favourite & restricted logic # --------------------------------------------------------- restricted_ids = [] favourite_ids = [] if customer_id: # Only if customer_id is passed customer = db["customers"].find_one({"_id": str_to_objectid(customer_id)}) if customer: restricted_ids = customer.get("restricted_workforce", []) or [] favourite_ids = customer.get("favourite_workforce", []) or [] # --------------------------------------------------------- workforce_cursor = collection.find(filters, {"first_name": 1,"skills": 1, "addons_name": 1, "vehicle_model_id": 1, "can_fly": 1, "fleet_info.can_fly": 1}) result_list = [] for f in workforce_cursor: workforce_id = str(f["_id"]) # --------------------------------------------------------- # ⭐ SKIP restricted ONLY if customer_id provided # --------------------------------------------------------- if customer_id and workforce_id in restricted_ids: continue # ------------------------------- # ❌ EXCLUDE maintenance fleets # ------------------------------- fleet_info = f.get("fleet_info", {}) if isinstance(fleet_info, dict): fleet_id = fleet_info.get("fleet_id") if fleet_id: fleet = db["fleets"].find_one( {"_id": str_to_objectid(fleet_id)}, {"maintenance_status": 1} ) if fleet and fleet.get("maintenance_status") == "maintenance": continue # skip this workforce shift_label = f.get("shift_timing", "Flexible") # ✅ If start_time and end_time given, filter by matching shift_timing if start_time: task_time = start_time.time() shift_start, shift_end = get_shift_period(shift_label) # Handle Night shift spanning midnight if shift_label == "Night": in_shift = ( task_time >= shift_start or task_time <= shift_end ) else: in_shift = shift_start <= task_time <= shift_end # Skip non-matching (non-flexible) shifts if not in_shift and shift_label != "Flexible": continue schedule_count = schedule_collection.count_documents({ "account_id": account_id, "workforce_id": workforce_id }) task_count = task_collection.count_documents({ #"account_id": account_id, "workforce_id": workforce_id }) # Prefer nested fleet_info.can_fly if exists can_fly_value = ( f.get("fleet_info", {}).get("can_fly") if isinstance(f.get("fleet_info"), dict) else f.get("can_fly", False) ) workforce_data = { "id": workforce_id, "first_name": f.get("first_name", ""), "skills": f.get("skills", []), "addons_name": f.get("addons_name", ""), "vehicle_model_id": f.get("vehicle_model_id", ""), "can_fly": can_fly_value, "total_schedules": schedule_count, "total_tasks": task_count } # if workforce_type == "free" and schedule_count == 0: # result_list.append(workforce_data) # elif workforce_type == "scheduled" and schedule_count > 0: # result_list.append(workforce_data) # --------------------------------------------------------- # ⭐ PRIORITY: favourite workforce first # --------------------------------------------------------- # For FREE if workforce_type == "free" and schedule_count == 0: if customer_id and workforce_id in favourite_ids: result_list.insert(0, workforce_data) # favourite on top else: result_list.append(workforce_data) # For SCHEDULED elif workforce_type == "scheduled" and schedule_count > 0: if customer_id and workforce_id in favourite_ids: result_list.insert(0, workforce_data) else: result_list.append(workforce_data) return result_list def get_workforce_list_with_status(db, account_id: str, query: str = "") -> List[dict]: collection = db[COLLECTION_NAME] schedule_collection = db["schedules"] task_collection = db["tasks"] filters = {"account_id": account_id} if query: filters["first_name"] = {"$regex": re.escape(query), "$options": "i"} workforce_cursor = collection.find(filters, {"first_name": 1, "last_name": 1, "status": 1}) result_list = [] for f in workforce_cursor: workforce_id = str(f["_id"]) # Count schedules schedule_count = schedule_collection.count_documents({ "account_id": account_id, "workforce_id": workforce_id }) # Check task statuses inprogress_task = task_collection.find_one({ "workforce_id": workforce_id, "status": "inprogress" }) completed_task = task_collection.find_one({ "workforce_id": workforce_id, "status": "completed" }) # Default status workforce_status = 4 # active if schedule_count == 0: workforce_status = 0 # free elif inprogress_task: workforce_status = 2 # inprogress elif completed_task and not inprogress_task: workforce_status = 3 # completed elif schedule_count > 0: workforce_status = 1 # busy workforce_data = { "workforce_id": workforce_id, "account_id": f.get("account_id", account_id), "user_id": f.get("user_id", ""), "first_name": f.get("first_name", ""), "last_name": f.get("last_name", ""), "mobile_number": f.get("mobile_number", ""), "email_id": f.get("email_id", ""), "gender": f.get("gender", ""), "date_of_birth": f.get("date_of_birth"), "license_id": f.get("license_id", ""), "national_id": f.get("national_id", ""), "emergency_contact": f.get("emergency_contact", ""), "shift_timing": f.get("shift_timing", ""), "address": f.get("address", ""), "blood_group": f.get("blood_group", ""), "is_human": f.get("is_human", True), "created_date": f.get("created_date"), "last_updated": f.get("last_updated"), "status": f.get("status"), "cense_number": f.get("cense_number"), "join_date": f.get("join_date"), "is_fleet_assigned": f.get("is_fleet_assigned", False), "skills": f.get("skills", []), "w_id": f.get("w_id"), "workforce_category_name": f.get("workforce_category_name", ""), "skills_name": f.get("skills_name", ""), "tag_name": f.get("tag_name"), "workforce_category_id": f.get("workforce_category_id"), "workforce_status": workforce_status, "status_label": {0: "free", 1: "busy", 2: "inprogress", 3: "completed", 4: "active"}[workforce_status], } result_list.append(workforce_data) return result_list def get_tag_name_list(db, query: str = "") -> List[dict]: tag_collection = db[TAG_COLLECTION] filters = {} if query: filters["tag_name"] = {"$regex": re.escape(query), "$options": "i"} tags = list(tag_collection.find(filters, {"_id": 0, "tag_name": 1})) return [{"tag_name": t["tag_name"]} for t in tags]