Implement password change functionality and initialize superuser

qwen3-tts-backend/api/auth.py

@@ -14,8 +14,8 @@ from core.security import (
     decode_access_token
 )
 from db.database import get_db
-from db.crud import get_user_by_username, get_user_by_email, create_user
-from schemas.user import User, UserCreate, Token
+from db.crud import get_user_by_username, get_user_by_email, create_user, change_user_password
+from schemas.user import User, UserCreate, Token, PasswordChange
 
 router = APIRouter(prefix="/auth", tags=["authentication"])
 
@@ -105,3 +105,33 @@ async def get_current_user_info(
     current_user: Annotated[User, Depends(get_current_user)]
 ):
     return current_user
+
+@router.post("/change-password", response_model=User)
+@limiter.limit("5/minute")
+async def change_password(
+    request: Request,
+    password_data: PasswordChange,
+    current_user: Annotated[User, Depends(get_current_user)],
+    db: Session = Depends(get_db)
+):
+    if not verify_password(password_data.current_password, current_user.hashed_password):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Current password is incorrect"
+        )
+
+    new_hashed_password = get_password_hash(password_data.new_password)
+
+    user = change_user_password(
+        db,
+        user_id=current_user.id,
+        new_hashed_password=new_hashed_password
+    )
+
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="User not found"
+        )
+
+    return user

qwen3-tts-backend/core/init_admin.py

@@ -0,0 +1,38 @@
+import logging
+from core.database import SessionLocal
+from core.security import get_password_hash
+from db.crud import count_users, create_user_by_admin
+
+logger = logging.getLogger(__name__)
+
+def init_superuser():
+    db = SessionLocal()
+    try:
+        user_count = count_users(db)
+
+        if user_count > 0:
+            logger.info(f"Database already has {user_count} user(s), skipping admin initialization")
+            return
+
+        logger.info("No users found in database, initializing default superuser")
+
+        hashed_password = get_password_hash("admin123456")
+        admin_user = create_user_by_admin(
+            db,
+            username="admin",
+            email="admin@example.com",
+            hashed_password=hashed_password,
+            is_superuser=True
+        )
+
+        logger.info(f"Default superuser created successfully: {admin_user.username}")
+        logger.warning("SECURITY WARNING: Default admin credentials are in use!")
+        logger.warning("  Username: admin")
+        logger.warning("  Password: admin123456")
+        logger.warning("  Please change the password immediately after first login!")
+
+    except Exception as e:
+        logger.error(f"Failed to initialize superuser: {e}")
+        raise
+    finally:
+        db.close()

qwen3-tts-backend/db/crud.py

@@ -11,6 +11,9 @@ def get_user_by_username(db: Session, username: str) -> Optional[User]:
 def get_user_by_email(db: Session, email: str) -> Optional[User]:
     return db.query(User).filter(User.email == email).first()
 
+def count_users(db: Session) -> int:
+    return db.query(User).count()
+
 def create_user(db: Session, username: str, email: str, hashed_password: str) -> User:
     user = User(
         username=username,
@@ -85,6 +88,21 @@ def delete_user(db: Session, user_id: int) -> bool:
     db.commit()
     return True
 
+def change_user_password(
+    db: Session,
+    user_id: int,
+    new_hashed_password: str
+) -> Optional[User]:
+    user = get_user_by_id(db, user_id)
+    if not user:
+        return None
+
+    user.hashed_password = new_hashed_password
+    user.updated_at = datetime.utcnow()
+    db.commit()
+    db.refresh(user)
+    return user
+
 def create_job(db: Session, user_id: int, job_type: str, input_data: Dict[str, Any]) -> Job:
     job = Job(
         user_id=user_id,

qwen3-tts-backend/main.py

@@ -70,6 +70,13 @@ async def lifespan(app: FastAPI):
         logger.error(f"Database initialization failed: {e}")
         raise
 
+    try:
+        from core.init_admin import init_superuser
+        init_superuser()
+    except Exception as e:
+        logger.error(f"Superuser initialization failed: {e}")
+        raise
+
     try:
         model_manager = await ModelManager.get_instance()
         await model_manager.load_model("custom-voice")

qwen3-tts-backend/schemas/user.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 from typing import Optional
-from pydantic import BaseModel, EmailStr, Field, field_validator, ConfigDict
+from pydantic import BaseModel, EmailStr, Field, field_validator, model_validator, ConfigDict
 import re
 
 class UserBase(BaseModel):
@@ -89,3 +89,25 @@ class Token(BaseModel):
 
 class TokenData(BaseModel):
     username: Optional[str] = None
+
+class PasswordChange(BaseModel):
+    current_password: str = Field(..., min_length=1)
+    new_password: str = Field(..., min_length=8, max_length=128)
+    confirm_password: str = Field(..., min_length=8, max_length=128)
+
+    @field_validator('new_password')
+    @classmethod
+    def validate_password_strength(cls, v: str) -> str:
+        if not re.search(r'[A-Z]', v):
+            raise ValueError('Password must contain at least one uppercase letter')
+        if not re.search(r'[a-z]', v):
+            raise ValueError('Password must contain at least one lowercase letter')
+        if not re.search(r'\d', v):
+            raise ValueError('Password must contain at least one digit')
+        return v
+
+    @model_validator(mode='after')
+    def passwords_match(self) -> 'PasswordChange':
+        if self.new_password != self.confirm_password:
+            raise ValueError('Passwords do not match')
+        return self