#/opt/docker/dev/service_finder/backend/app/api/v1/endpoints/users.py import copy from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.exc import SQLAlchemyError from sqlalchemy import select, or_ from sqlalchemy.orm import joinedload from sqlalchemy.orm.attributes import flag_modified from typing import Dict, Any, List, Optional from datetime import datetime from app.api.deps import get_db, get_current_user from app.schemas.user import UserResponse, UserUpdate, ActiveOrganizationUpdate, UserWithTokenResponse, PersonUpdate, ChangePasswordRequest from app.models.identity import User, Person from app.models.identity.address import Address from app.services.trust_engine import TrustEngine from app.services.geo_service import GeoService from app.services.auth_service import AuthService from app.core.security import create_tokens, DEFAULT_RANK_MAP, verify_password, get_password_hash from app.core.config import settings from pydantic import BaseModel router = APIRouter() trust_engine = TrustEngine() # MLM Hálózat válasz sémák class NetworkMemberL1(BaseModel): email: str referral_code: str folder_slug: Optional[str] joined_at: datetime class NetworkMemberL2L3(BaseModel): referral_code: str joined_at: datetime class NetworkResponse(BaseModel): level1: List[NetworkMemberL1] level2: List[NetworkMemberL2L3] level3: List[NetworkMemberL2L3] def _build_user_response(user: User, active_org_id: Optional[int] = None) -> dict: """ Segédfüggvény a UserResponse dict előállításához. Beágyazza a Person és Address adatokat a 'person' mezőbe. """ from app.models.marketplace.organization import Organization, OrganizationMember from app.models.marketplace.organization import OrgUserRole # Determine active organization ID if active_org_id is None: if user.scope_id is not None: try: active_org_id = int(user.scope_id) except (ValueError, TypeError): active_org_id = None if active_org_id is None: # 1. Check if user is a member of any organization with ADMIN/OWNER role # (This is a helper - in real usage the caller should pass active_org_id) pass person = user.person person_data = None if person: address_data = None if person.address: address_data = { "zip": getattr(person.address, 'zip', None) or getattr(getattr(person.address, 'postal_code', None), 'zip_code', None), "city": getattr(person.address, 'city', None) or getattr(getattr(person.address, 'postal_code', None), 'city', None), "street_name": person.address.street_name, "street_type": person.address.street_type, "house_number": person.address.house_number, "stairwell": person.address.stairwell, "floor": person.address.floor, "door": person.address.door, "parcel_id": person.address.parcel_id, "full_address_text": person.address.full_address_text, "latitude": person.address.latitude, "longitude": person.address.longitude, } person_data = { "id": person.id, "id_uuid": str(person.id_uuid) if person.id_uuid else None, "first_name": person.first_name, "last_name": person.last_name, "phone": person.phone, "mothers_last_name": person.mothers_last_name, "mothers_first_name": person.mothers_first_name, "birth_place": person.birth_place, "birth_date": person.birth_date, "identity_docs": person.identity_docs, "ice_contact": person.ice_contact, "is_active": person.is_active, "address": address_data, } # Get first_name/last_name from person for backward compatibility first_name = person.first_name if person else "" last_name = person.last_name if person else "" return { "id": user.id, "email": user.email, "first_name": first_name, "last_name": last_name, "is_active": user.is_active, "region_code": user.region_code, "person_id": user.person_id, "role": user.role.value if hasattr(user.role, 'value') else str(user.role), "subscription_plan": user.subscription_plan, "scope_level": user.scope_level or "individual", "scope_id": str(active_org_id) if active_org_id else None, "ui_mode": user.ui_mode or "personal", "active_organization_id": active_org_id, "preferred_language": user.preferred_language or "hu", "person": person_data, } @router.get("/me", response_model=UserResponse) async def read_users_me( db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """Visszaadja a bejelentkezett felhasználó profilját beágyazott Person és Address adatokkal.""" from app.models.marketplace.organization import Organization, OrganizationMember from app.models.marketplace.organization import OrgUserRole # Determine active organization ID active_org_id = None # If user already has a scope_id, use it if current_user.scope_id is not None: try: active_org_id = int(current_user.scope_id) except (ValueError, TypeError): active_org_id = None # If still no active org ID, try to find user's primary organization if active_org_id is None: # 1. Check if user is a member of any organization with ADMIN/OWNER role stmt = select(OrganizationMember.organization_id).where( OrganizationMember.user_id == current_user.id, or_( OrganizationMember.role == OrgUserRole.ADMIN, OrganizationMember.role == OrgUserRole.OWNER ) ).limit(1) result = await db.execute(stmt) org_member_row = result.first() if org_member_row: active_org_id = org_member_row[0] else: # 2. Check if user owns any organization (owner_id matches user.id) stmt = select(Organization.id).where( Organization.owner_id == current_user.id ).limit(1) result = await db.execute(stmt) org_owner_row = result.first() if org_owner_row: active_org_id = org_owner_row[0] else: # 3. Fallback: get first organization they're a member of stmt = select(OrganizationMember.organization_id).where( OrganizationMember.user_id == current_user.id ).limit(1) result = await db.execute(stmt) org_row = result.first() active_org_id = org_row[0] if org_row else None response_data = _build_user_response(current_user, active_org_id) return UserResponse.model_validate(response_data) @router.put("/me/person", response_model=UserResponse) async def update_my_person( update_data: PersonUpdate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """ Frissíti a bejelentkezett felhasználó Person (és Address) rekordját. Elfogadja a PersonUpdate sémát, amely tartalmazza a személyes, okmány és cím adatokat. Visszaadja a frissített UserResponse objektumot. """ # Reload user with eager loaded person + address stmt = ( select(User) .where(User.id == current_user.id) .options(joinedload(User.person).joinedload(Person.address)) ) result = await db.execute(stmt) user = result.unique().scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="User not found") person = user.person if not person: raise HTTPException(status_code=400, detail="No Person record found. Complete KYC first.") # ── Update Person fields ── update_dict = update_data.dict(exclude_unset=True) person_fields = ["first_name", "last_name", "phone", "mothers_last_name", "mothers_first_name", "birth_place", "birth_date"] address_fields = ["address_zip", "address_city", "address_street_name", "address_street_type", "address_house_number", "address_stairwell", "address_floor", "address_door", "address_hrsz"] for field in person_fields: if field in update_dict and update_dict[field] is not None: setattr(person, field, update_dict[field]) # ── Update identity_docs (if provided and not None) ── if "identity_docs" in update_dict and update_dict["identity_docs"] is not None: # Use deep copy + flag_modified to ensure SQLAlchemy detects the JSON mutation existing_docs = copy.deepcopy(person.identity_docs) if person.identity_docs else {} if isinstance(existing_docs, dict) and isinstance(update_dict["identity_docs"], dict): existing_docs.update(update_dict["identity_docs"]) person.identity_docs = existing_docs flag_modified(person, "identity_docs") else: person.identity_docs = update_dict["identity_docs"] flag_modified(person, "identity_docs") # ── Update Address fields ── has_address_update = any(f in update_dict for f in address_fields) if has_address_update: # Use GeoService to get or create address addr_id = await GeoService.get_or_create_full_address( db, zip_code=update_dict.get("address_zip"), city=update_dict.get("address_city"), street_name=update_dict.get("address_street_name"), street_type=update_dict.get("address_street_type"), house_number=update_dict.get("address_house_number"), parcel_id=update_dict.get("address_hrsz"), ) if addr_id: person.address_id = addr_id try: await db.commit() # Refresh to get the latest state await db.refresh(person) if person.address: await db.refresh(person.address) except SQLAlchemyError as e: await db.rollback() raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") # Build and return the full response response_data = _build_user_response(user) return UserResponse.model_validate(response_data) @router.put("/me/password", status_code=200) async def change_my_password( change_data: ChangePasswordRequest, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """ Jelszó módosítás a bejelentkezett felhasználó számára. Ellenőrzi a jelenlegi jelszót, validálja az új jelszó komplexitását, majd elmenti az új hash-t. Válaszként egy {'status': 'ok', 'message': ...} dict-et ad. """ # 1. Ellenőrizzük a jelenlegi jelszót if not verify_password(change_data.current_password, current_user.hashed_password): raise HTTPException( status_code=400, detail="A jelenlegi jelszó helytelen." ) # 2. Ellenőrizzük, hogy az új jelszó nem egyezik-e a régivel if verify_password(change_data.new_password, current_user.hashed_password): raise HTTPException( status_code=400, detail="Az új jelszó nem egyezhet meg a jelenlegi jelszóval." ) # 3. Jelszó komplexitás ellenőrzése (dinamikus admin beállítások alapján) await AuthService._validate_password_complexity(db, change_data.new_password, current_user.region_code) # 4. Új jelszó hash-elése és mentése current_user.hashed_password = get_password_hash(change_data.new_password) try: await db.commit() await db.refresh(current_user) except SQLAlchemyError as e: await db.rollback() raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") return {"status": "ok", "message": "Jelszó sikeresen megváltoztatva."} @router.get("/me/trust") async def get_user_trust( db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), force_recalculate: bool = False, ) -> Dict[str, Any]: """ Visszaadja a felhasználó Gondos Gazda Index (Trust Score) értékét. A számítás dinamikusan betölti a paramétereket a SystemParameter rendszerből (Global/Country/Region/User hierarchia). Paraméterek: - force_recalculate: Ha True, akkor újraszámolja a trust score-t (alapértelmezetten cache-elt értéket ad vissza, ha kevesebb mint 24 órája számoltuk) """ trust_data = await trust_engine.calculate_user_trust( db=db, user_id=current_user.id, force_recalculate=force_recalculate ) return trust_data @router.patch("/me/preferences", response_model=UserResponse) async def update_user_preferences( update_data: UserUpdate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """ Update user preferences (ui_mode, preferred_language, etc.) """ # Filter out None values update_dict = update_data.dict(exclude_unset=True) if not update_dict: raise HTTPException(status_code=400, detail="No fields to update") # Validate ui_mode if present if "ui_mode" in update_dict: if update_dict["ui_mode"] not in ["personal", "fleet"]: raise HTTPException(status_code=422, detail="ui_mode must be 'personal' or 'fleet'") # Update user fields for field, value in update_dict.items(): if hasattr(current_user, field): setattr(current_user, field, value) else: raise HTTPException(status_code=400, detail=f"Invalid field: {field}") try: await db.commit() await db.refresh(current_user) except SQLAlchemyError as e: await db.rollback() raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") # Return the Pydantic model instead of raw SQLAlchemy object response_data = _build_user_response(current_user) return UserResponse.model_validate(response_data) @router.patch("/me/active-organization", response_model=UserWithTokenResponse) async def update_active_organization( update_data: ActiveOrganizationUpdate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """ Update the user's active organization (scope_id). Accepts an organization_id (UUID/string) or None to revert to personal mode. Returns a new JWT token with updated scope_id in the payload. """ # Extract organization_id from request org_id = update_data.organization_id # Validate that the user has access to this organization if org_id is provided if org_id is not None: from sqlalchemy import select from app.models.marketplace.organization import OrganizationMember # Check if user is a member of the organization stmt = select(OrganizationMember).where( OrganizationMember.organization_id == org_id, OrganizationMember.user_id == current_user.id ) result = await db.execute(stmt) member = result.scalar_one_or_none() if not member: raise HTTPException( status_code=403, detail="You are not a member of this organization" ) # Update user's scope_id current_user.scope_id = org_id try: await db.commit() await db.refresh(current_user) except SQLAlchemyError as e: await db.rollback() raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") # Generate new JWT token with updated scope_id role_key = current_user.role.value.upper() token_payload = { "sub": str(current_user.id), "role": role_key, "rank": DEFAULT_RANK_MAP.get(role_key, "user"), "scope_level": "organization" if org_id else "personal", "scope_id": org_id, "person_id": str(current_user.person_id) if current_user.person_id else None, } access_token, _ = create_tokens(data=token_payload) # Return user data with new token response_data = _build_user_response(current_user) return UserWithTokenResponse( user=UserResponse.model_validate(response_data), access_token=access_token, token_type="bearer" ) @router.get("/me/network", response_model=NetworkResponse) async def get_my_network( db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user), ): """ Visszaadja a felhasználó MLM hálózatát 3 szinten: - L1: Közvetlen meghívottak (email + kód) - L2: L1 meghívottai (csak kód) - L3: L2 meghívottai (csak kód) """ # L1: Közvetlen meghívottak l1_stmt = select(User).where(User.referred_by_id == current_user.id) l1_result = await db.execute(l1_stmt) l1_users = l1_result.scalars().all() level1 = [ NetworkMemberL1( email=u.email, referral_code=u.referral_code or "", folder_slug=u.folder_slug, joined_at=u.created_at ) for u in l1_users ] # L2: L1 userek által meghívottak level2 = [] if l1_users: l1_ids = [u.id for u in l1_users] l2_stmt = select(User).where(User.referred_by_id.in_(l1_ids)) l2_result = await db.execute(l2_stmt) l2_users = l2_result.scalars().all() level2 = [ NetworkMemberL2L3( referral_code=u.referral_code or "", joined_at=u.created_at ) for u in l2_users ] # L3: L2 userek által meghívottak level3 = [] if l2_users: l2_ids = [u.id for u in l2_users] l3_stmt = select(User).where(User.referred_by_id.in_(l2_ids)) l3_result = await db.execute(l3_stmt) l3_users = l3_result.scalars().all() level3 = [ NetworkMemberL2L3( referral_code=u.referral_code or "", joined_at=u.created_at ) for u in l3_users ] return NetworkResponse( level1=level1, level2=level2, level3=level3 )