import json import os import random from datetime import datetime, timedelta # passlib 1.7.x expects bcrypt.__about__.__version__, but bcrypt>=4 removed __about__. # Avoid noisy warnings by providing a compat shim. try: import bcrypt as _bcrypt # type: ignore if not hasattr(_bcrypt, "__about__"): class _About: # pragma: no cover __version__ = getattr(_bcrypt, "__version__", "0") _bcrypt.__about__ = _About() # type: ignore[attr-defined] except Exception: pass from passlib.context import CryptContext from fastapi import HTTPException, status from app.v1.dependencies.auth import create_access_token from app.v1.libraries.object import str_to_objectid from app.v1.services.saas.roles import get_role_service from app.v1.models.crudmodel import FeatureBase from app.v1.libraries.googleconnect import verify_google_token from app.v1.libraries.email_templates import send_welcome_email from bson import ObjectId from dotenv import load_dotenv import datetime load_dotenv() pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") access_token_expires = timedelta(minutes=14320) # Collection names (adjust as needed) COLLECTION_NAME = "users" ACCOUNT_COLLECTION_NAME = "accounts" SUBSCRIPTION_COLLECTION_NAME = "subscriptions" PROJECT_COLLECTION_NAME = "projects" class CustomJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.isoformat() if isinstance(obj, ObjectId): return str(obj) return super().default(obj) def serialize_data(obj): if isinstance(obj, dict): return {k: serialize_data(v) for k, v in obj.items()} elif isinstance(obj, list): return [serialize_data(item) for item in obj] elif isinstance(obj, datetime.datetime): return obj.isoformat() elif isinstance(obj, ObjectId): return str(obj) else: return obj async def enrich_user_data(user_data: dict, db) -> dict: """ Enriches user data with common details: - Account information (subscription, active agents, etc.) - Role rights from roles service. """ ALLOWED_FREE_AGENTS = int(os.getenv('ALLOWED_FREE_AGENTS', '5')) # Set default subscription values if not already set user_data.setdefault('account_type', 0) user_data.setdefault('subscription_id', '') user_data.setdefault('subscription_status', '') user_data.setdefault('subscription_agents', ALLOWED_FREE_AGENTS) user_data.setdefault('active_agents', 0) if user_data.get("account_id"): accounts_collection = db[ACCOUNT_COLLECTION_NAME] account_info = accounts_collection.find_one({"_id": str_to_objectid(user_data["account_id"])}) if account_info: user_data['account_type'] = account_info.get('account_type', 0) user_data['subscription_id'] = account_info.get('active_subscription_id', '') user_data['subscription_status'] = account_info.get('subscription_status', '') if account_info.get('active_subscription_id'): subscription_collection = db[SUBSCRIPTION_COLLECTION_NAME] subscription = subscription_collection.find_one({ "_id": str_to_objectid(account_info.get('active_subscription_id')) }) if subscription: user_data['active_agents'] = subscription.get('active_agents_count', 0) user_data['subscription_agents'] = subscription.get('subscription_agents_count', ALLOWED_FREE_AGENTS) else: project_collection = db[PROJECT_COLLECTION_NAME] active_projects_count = project_collection.count_documents({ "account_id": user_data["account_id"], "status": "ACTIVE" }) user_data['active_agents'] = active_projects_count # Enrich with role rights if a role_id exists. role_id = user_data.get("roles") if role_id: role = get_role_service(role_id, db) if role: user_data["role_rights"] = role.get("permissions", {}) # Directly fetch features from the features collection. features_collection = db["features"] # adjust the collection name if needed features_list = list(features_collection.find()) user_data["role_features"] = features_list #print("Enriched user data:", user_data) return user_data async def process_user_login(user_data: dict, db, token_expires_delta: timedelta) -> dict: """ Common login processing: - Enrich user data with account/role details. - Generate a JWT token. - Serialize user data for the response. """ # Enrich with common information user_data = await enrich_user_data(user_data, db) # Use email (or fallback to mobile) for token creation token = create_access_token( data={"sub": user_data.get("email", user_data.get("mobile", ""))}, expires_delta=token_expires_delta ) user_data = serialize_data(user_data) return {"message": "Login Successful", "user": user_data, "token": token} async def login_user(user_login, db) -> dict: """ Handles the email/password login. """ user_in_db = db[COLLECTION_NAME].find_one({"email": user_login.email}) if not user_in_db: raise HTTPException(status_code=404, detail="User not found") if not pwd_context.verify(user_login.password, user_in_db.get("hashed_password")): raise HTTPException(status_code=401, detail="Invalid credentials") return await process_user_login(user_in_db, db, access_token_expires) async def verify_phone_otp_service(payload: dict, db, otp_store: dict) -> dict: """ Verifies the OTP for a phone login. This function handles: - Validating the OTP (using an in-memory store for demo purposes) - Looking up (or creating) the user by phone - Processing the login (enriching user data and generating JWT token) """ phone = payload.get("phone") otp_submitted = payload.get("otp") if not phone or otp_submitted is None: raise HTTPException(status_code=400, detail="Phone number and OTP are required.") if phone not in otp_store: raise HTTPException(status_code=400, detail="OTP not sent for this phone number.") stored = otp_store[phone] if datetime.utcnow() > stored["expires"]: del otp_store[phone] raise HTTPException(status_code=400, detail="OTP expired.") if stored["otp"] != int(otp_submitted): raise HTTPException(status_code=400, detail="Invalid OTP.") # OTP verified; remove it. del otp_store[phone] # Look up or create a user by phone. users_collection = db[COLLECTION_NAME] user_in_db = users_collection.find_one({"mobile": phone}) if not user_in_db: # Create a new user record if not found. user_data = { "mobile": phone, "created_date": datetime.utcnow(), "last_login": datetime.utcnow(), "is_active": True, "is_verified": True, "mobile_verified": True, "account_id": "", "role_id": 3 # Default role id for phone login; adjust as needed. } inserted_id = users_collection.insert_one(user_data).inserted_id user_data["_id"] = inserted_id else: # Update last login timestamp for an existing user. users_collection.update_one( {"_id": user_in_db["_id"]}, {"$set": {"last_login": datetime.utcnow()}} ) user_data = user_in_db # Use common login processing (which enriches the data and creates the JWT token). return await process_user_login(user_data, db, access_token_expires) async def google_connect_user(google_token, background_tasks, db) -> dict: """ Handles Google Connect login. """ user_info = verify_google_token(google_token.token) if not user_info: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Google token.") users_collection = db[COLLECTION_NAME] user_in_db = users_collection.find_one({"email": user_info["email"]}) if not user_in_db: # Create a new user record user_data = { "email": user_info["email"], "name": user_info["name"], "picurl": user_info.get("picture", ""), "google_connect": user_info['refresh_token'], "created_date": datetime.utcnow(), "last_login": datetime.utcnow(), "date_of_birth": datetime.utcnow(), "is_active": True, "is_verified": True, "mobile_verified": False, } # Optionally, check for invitation details to set account/role info. emails_collection = db['emails'] invitation = emails_collection.find_one({"email": user_info["email"]}) if invitation: user_data['account_id'] = invitation.get("account_id", "") user_data['role_id'] = invitation.get("role", 9) #default role for any workforce inserted_id = users_collection.insert_one(user_data).inserted_id user_data["_id"] = inserted_id background_tasks.add_task(send_welcome_email, user_info["email"], user_info["name"]) return await process_user_login(user_data, db, access_token_expires) else: # Update last_login and google_connect token for an existing user. update_data = {"last_login": datetime.utcnow(), "google_connect": user_info['refresh_token']} users_collection.update_one({"_id": user_in_db["_id"]}, {"$set": update_data}) user_in_db.update(update_data) return await process_user_login(user_in_db, db, access_token_expires)