import json from datetime import datetime, timedelta from pathlib import Path from typing import Optional from uuid import uuid4 import bcrypt # Настройки JWT (читаются из config/settings) from config.settings import settings from jose import JWTError, jwt from models.user import Token, User, UserCreate SECRET_KEY = settings.JWT_SECRET_KEY ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 дней class AuthService: def __init__(self): from config.settings import settings self.data_dir = Path(settings.DATA_DIR) self.data_dir.mkdir(exist_ok=True) self.users_file = self.data_dir / "users.json" self.users = self._load_users() def _load_users(self) -> list[User]: if self.users_file.exists(): with open(self.users_file, "r", encoding="utf-8") as f: data = json.load(f) return [User(**u) for u in data] return [] def _save_users(self): with open(self.users_file, "w", encoding="utf-8") as f: json.dump( [u.model_dump() for u in self.users], f, ensure_ascii=False, indent=2 ) def verify_password(self, plain_password: str, hashed_password: str) -> bool: return bcrypt.checkpw( plain_password.encode("utf-8"), hashed_password.encode("utf-8") ) def get_password_hash(self, password: str) -> str: return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def get_user_by_username(self, username: str) -> Optional[User]: for user in self.users: if user.username == username: return user return None def get_user_by_id(self, user_id: str) -> Optional[User]: for user in self.users: if user.id == user_id: return user return None def create_user(self, user_create: UserCreate) -> User: if self.get_user_by_username(user_create.username): raise ValueError("Username already exists") new_user = User( id=str(uuid4()), username=user_create.username, email=user_create.email, hashed_password=self.get_password_hash(user_create.password), ) self.users.append(new_user) self._save_users() # Создаем папку для проектов пользователя user_projects_dir = self.data_dir / "users" / new_user.id / "projects" user_projects_dir.mkdir(parents=True, exist_ok=True) return new_user def authenticate_user(self, username: str, password: str) -> Optional[User]: user = self.get_user_by_username(username) if not user or not self.verify_password(password, user.hashed_password): return None return user def create_access_token(self, data: dict) -> str: to_encode = data.copy() expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) def get_token(self, username: str, password: str) -> Token: user = self.authenticate_user(username, password) if not user: raise ValueError("Invalid username or password") access_token = self.create_access_token( data={"sub": user.username, "user_id": user.id} ) return Token( access_token=access_token, token_type="bearer", user_id=user.id, username=user.username, ) # Singleton auth_service = AuthService()