401 lines
15 KiB
Python
401 lines
15 KiB
Python
import os
|
||
import asyncio
|
||
from typing import Optional
|
||
from fastapi import Depends, APIRouter, HTTPException, Request, status
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, or_, and_, exists, func
|
||
from sqlalchemy.exc import IntegrityError
|
||
from google.oauth2.credentials import Credentials
|
||
from googleapiclient.discovery import build
|
||
|
||
from app.db import models
|
||
from app.core.security import get_current_user
|
||
from app.api import schemas
|
||
from app.core.config import config
|
||
from app.websocket import connection_manager
|
||
|
||
async def get_db():
|
||
async with models.AsyncSessionLocal() as db:
|
||
try:
|
||
yield db
|
||
finally:
|
||
await db.close()
|
||
|
||
def _get_drive_service_for_users():
|
||
credentials = Credentials(
|
||
token=None,
|
||
refresh_token=config.GOOGLE_REFRESH_TOKEN,
|
||
token_uri='https://oauth2.googleapis.com/token',
|
||
client_id=config.GOOGLE_CLIENT_ID,
|
||
client_secret=config.GOOGLE_CLIENT_SECRET,
|
||
scopes=['https://www.googleapis.com/auth/drive']
|
||
)
|
||
return build('drive', 'v3', credentials=credentials)
|
||
|
||
|
||
def _build_user_search_filter(query: str):
|
||
normalized = f"%{query.lower()}%"
|
||
return or_(
|
||
func.lower(models.User.username).like(normalized),
|
||
func.lower(models.User.first_name).like(normalized),
|
||
func.lower(models.User.last_name).like(normalized),
|
||
models.User.phone.like(normalized),
|
||
func.lower(models.User.first_name + " " + func.coalesce(models.User.last_name, '')).like(normalized),
|
||
func.lower(func.coalesce(models.User.last_name, '') + " " + models.User.first_name).like(normalized),
|
||
)
|
||
|
||
async def _delete_old_avatar_file(file_id: str, db: AsyncSession):
|
||
res = await db.execute(select(models.MediaItem).where(models.MediaItem.file_id == file_id))
|
||
cloud_items = res.scalars().all()
|
||
|
||
if not cloud_items:
|
||
return
|
||
|
||
try:
|
||
service = _get_drive_service_for_users()
|
||
except Exception as e:
|
||
print(f"Не удалось инициализировать Google Drive для удаления аватарки: {e}")
|
||
return
|
||
|
||
for item in cloud_items:
|
||
try:
|
||
await asyncio.to_thread(service.files().delete(fileId=item.storage_file_id).execute)
|
||
print(f"Старая аватарка {item.storage_file_id} успешно удалена из Google Drive")
|
||
except Exception as e:
|
||
print(f"Ошибка физического удаления аватарки из Google Drive: {e}")
|
||
|
||
await db.delete(item)
|
||
|
||
await db.commit()
|
||
print(f"Старая аватарка {file_id} успешно удалена из базы данных")
|
||
|
||
usersRouter = APIRouter(
|
||
prefix="/users",
|
||
tags=[],
|
||
)
|
||
|
||
@usersRouter.get("/me")
|
||
async def read_users_me(current_user: models.User = Depends(get_current_user)):
|
||
return {
|
||
"id": current_user.id,
|
||
"username": current_user.username,
|
||
"first_name": current_user.first_name,
|
||
"last_name": current_user.last_name,
|
||
"phone": current_user.phone,
|
||
"email": getattr(current_user, "email", None),
|
||
"about": current_user.about,
|
||
"public_key": current_user.public_key,
|
||
"encrypted_private_key": current_user.encrypted_private_key,
|
||
"avatar_file_id": current_user.avatar_file_id,
|
||
"totp_enabled": bool(current_user.totp_secret != None),
|
||
}
|
||
|
||
@usersRouter.put("/me")
|
||
async def update_users_me(
|
||
data: schemas.UpdateMe,
|
||
current_user: models.User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
user_to_update = await db.merge(current_user)
|
||
|
||
if data.username is not None:
|
||
user_to_update.username = data.username
|
||
if data.first_name is not None:
|
||
user_to_update.first_name = data.first_name
|
||
if data.last_name is not None:
|
||
user_to_update.last_name = data.last_name
|
||
if data.phone is not None:
|
||
user_to_update.phone = data.phone or None
|
||
if data.email is not None:
|
||
user_to_update.email = data.email or None
|
||
if data.about is not None:
|
||
user_to_update.about = data.about or None
|
||
|
||
try:
|
||
await db.commit()
|
||
except IntegrityError:
|
||
await db.rollback()
|
||
raise HTTPException(status_code=400, detail="phone/email already in use")
|
||
|
||
await db.refresh(user_to_update)
|
||
await connection_manager.manager.broadcast({'type': 'user_updated', 'user_id': current_user.id})
|
||
return {
|
||
"status": "ok",
|
||
"user": {
|
||
"id": user_to_update.id,
|
||
"username": user_to_update.username,
|
||
"first_name": user_to_update.first_name,
|
||
"last_name": user_to_update.last_name,
|
||
"phone": user_to_update.phone,
|
||
"email": getattr(user_to_update, "email", None),
|
||
"about": user_to_update.about,
|
||
},
|
||
}
|
||
|
||
@usersRouter.put("/me/encryption-key")
|
||
async def update_encrypted_private_key(
|
||
data: schemas.UpdateEncryptedPrivateKey,
|
||
current_user: models.User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
user_to_update = await db.merge(current_user)
|
||
user_to_update.encrypted_private_key = data.encrypted_private_key
|
||
|
||
try:
|
||
await db.commit()
|
||
except Exception:
|
||
await db.rollback()
|
||
raise HTTPException(status_code=500, detail="Не удалось сохранить ключ шифрования")
|
||
|
||
await db.refresh(user_to_update)
|
||
await connection_manager.manager.broadcast({'type': 'user_updated', 'user_id': current_user.id})
|
||
return {"status": "ok"}
|
||
|
||
@usersRouter.put("/me/password")
|
||
async def change_password(
|
||
data: schemas.ChangePassword,
|
||
current_user: models.User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
from app.core.security import verify_password, get_password_hash
|
||
|
||
if not verify_password(data.current_password, current_user.hashed_password):
|
||
raise HTTPException(status_code=400, detail="Неверный текущий пароль")
|
||
|
||
user_to_update = await db.merge(current_user)
|
||
user_to_update.hashed_password = get_password_hash(data.new_password)
|
||
|
||
try:
|
||
await db.commit()
|
||
except Exception:
|
||
await db.rollback()
|
||
raise HTTPException(status_code=500, detail="Не удалось изменить пароль")
|
||
|
||
await db.refresh(user_to_update)
|
||
return {"status": "ok"}
|
||
|
||
@usersRouter.put("/me/privacy")
|
||
async def update_privacy_settings(
|
||
data: schemas.UpdatePrivacySettings,
|
||
current_user: models.User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
user_to_update = await db.merge(current_user)
|
||
|
||
if data.show_email is not None:
|
||
user_to_update.show_email = 1 if data.show_email else 0
|
||
if data.show_phone is not None:
|
||
user_to_update.show_phone = 1 if data.show_phone else 0
|
||
if data.show_avatar is not None:
|
||
user_to_update.show_avatar = 1 if data.show_avatar else 0
|
||
if data.show_about is not None:
|
||
user_to_update.show_about = 1 if data.show_about else 0
|
||
user_to_update.show_username = 1
|
||
if data.show_last_online is not None:
|
||
user_to_update.show_last_online = 1 if data.show_last_online else 0
|
||
try:
|
||
await db.commit()
|
||
except Exception:
|
||
await db.rollback()
|
||
raise HTTPException(status_code=500, detail="Не удалось сохранить настройки конфиденциальности")
|
||
|
||
await db.refresh(user_to_update)
|
||
await connection_manager.manager.broadcast({'type': 'user_updated', 'user_id': current_user.id})
|
||
return {"status": "ok"}
|
||
|
||
@usersRouter.get("/me/privacy")
|
||
async def get_privacy_settings(current_user: models.User = Depends(get_current_user)):
|
||
return {
|
||
"show_email": bool(current_user.show_email),
|
||
"show_phone": bool(current_user.show_phone),
|
||
"show_avatar": bool(current_user.show_avatar),
|
||
"show_about": bool(current_user.show_about),
|
||
"show_username": True,
|
||
"show_last_online": bool(current_user.show_last_online),
|
||
}
|
||
|
||
@usersRouter.get("/all")
|
||
async def read_users_all(
|
||
current_user: models.User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
query: Optional[str] = None,
|
||
):
|
||
stmt = select(models.User)
|
||
if query and query.strip():
|
||
stmt = stmt.filter(_build_user_search_filter(query.strip()))
|
||
|
||
res = await db.execute(stmt)
|
||
users = res.scalars().all()
|
||
users_for_return = []
|
||
if current_user.id >= 100:
|
||
for user in users:
|
||
if not (1 < int(user.id) < 100 or int(user.id) == 0 or int(user.id) == current_user.id):
|
||
users_for_return.append(user)
|
||
else:
|
||
users_for_return = users
|
||
return [{"id": user.id, "username": user.username, "name": f"{user.first_name} {user.last_name or ''}".strip(), "public_key": user.public_key} for user in users_for_return]
|
||
|
||
@usersRouter.get("/chats")
|
||
async def read_users_chats(
|
||
request: Request,
|
||
current_user: models.User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
query: Optional[str] = None,
|
||
):
|
||
stmt = (
|
||
select(models.User)
|
||
.filter(models.User.id != current_user.id)
|
||
.filter(exists().where(
|
||
or_(
|
||
and_(models.Message.sender_id == current_user.id, models.Message.receiver_id == models.User.id),
|
||
and_(models.Message.sender_id == models.User.id, models.Message.receiver_id == current_user.id)
|
||
)
|
||
))
|
||
)
|
||
if query and query.strip():
|
||
stmt = stmt.filter(_build_user_search_filter(query.strip()))
|
||
|
||
users_res = await db.execute(stmt)
|
||
users = users_res.scalars().all()
|
||
|
||
result = []
|
||
for user in users:
|
||
last_msg_res = await db.execute(
|
||
select(models.Message)
|
||
.filter(
|
||
or_(
|
||
and_(models.Message.sender_id == current_user.id, models.Message.receiver_id == user.id),
|
||
and_(models.Message.sender_id == user.id, models.Message.receiver_id == current_user.id),
|
||
)
|
||
)
|
||
.order_by(models.Message.timestamp.desc())
|
||
)
|
||
last_msg = last_msg_res.scalars().first()
|
||
|
||
unread_res = await db.execute(
|
||
select(func.count(models.Message.id))
|
||
.filter(
|
||
models.Message.sender_id == user.id,
|
||
models.Message.receiver_id == current_user.id,
|
||
models.Message.read_at.is_(None),
|
||
)
|
||
)
|
||
unread_count = unread_res.scalar() or 0
|
||
|
||
result.append(
|
||
{
|
||
"id": user.id,
|
||
"username": user.username,
|
||
"name": f"{user.first_name} {user.last_name or ''}".strip(),
|
||
"public_key": user.public_key,
|
||
"avatar_file_id": user.avatar_file_id if (user.show_avatar or current_user.id == 1) else None,
|
||
"avatar_url": str(request.url_for("get_file", file_id=user.avatar_file_id)) if (user.show_avatar or current_user.id == 1) and user.avatar_file_id else None,
|
||
"last_message": last_msg.content if last_msg else None,
|
||
"last_message_time": (last_msg.timestamp.isoformat() if last_msg and last_msg.timestamp else None),
|
||
"unread_count": unread_count,
|
||
"online": str(user.id) in connection_manager.manager.online_users,
|
||
"last_message_id": last_msg.id if last_msg else None,
|
||
"last_message_type": last_msg.message_type if last_msg else None,
|
||
}
|
||
)
|
||
|
||
result.sort(key=lambda x: x['last_message_time'] or '', reverse=True)
|
||
return result
|
||
|
||
@usersRouter.get("/by-username/{username}", response_model=schemas.UserContactResponse)
|
||
async def get_user_by_username(username: str, request: Request, db: AsyncSession = Depends(get_db), current_user: models.User = Depends(get_current_user)):
|
||
res = await db.execute(select(models.User).where(models.User.username == username))
|
||
user = res.scalars().first()
|
||
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="Пользователь не найден")
|
||
|
||
profile_data = {
|
||
"id": user.id,
|
||
"public_key": user.public_key,
|
||
"first_name": user.first_name,
|
||
"last_name": user.last_name,
|
||
"username": user.username,
|
||
"show_avatar": bool(user.show_avatar),
|
||
"totp_enabled": bool(user.totp_secret),
|
||
"online": str(user.id) in connection_manager.manager.online_users
|
||
}
|
||
|
||
if user.show_avatar or current_user.id == 1:
|
||
profile_data["avatar_url"] = str(request.url_for("get_file", file_id=user.avatar_file_id)) if (user.show_avatar or current_user.id == 1) and user.avatar_file_id else None
|
||
|
||
if user.show_about or current_user.id == 1:
|
||
profile_data["about"] = user.about
|
||
|
||
if user.show_phone or current_user.id == 1:
|
||
profile_data["phone"] = user.phone
|
||
|
||
if user.show_email or current_user.id == 1:
|
||
profile_data["email"] = user.email
|
||
|
||
if user.show_last_online or current_user.id == 1:
|
||
profile_data["last_online"] = user.last_online.isoformat() if user.last_online else None
|
||
|
||
return profile_data
|
||
|
||
@usersRouter.get("/{user_id}", response_model=schemas.UserProfile)
|
||
async def get_user_by_id(
|
||
user_id: int,
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
current_user: models.User = Depends(get_current_user)
|
||
):
|
||
res = await db.execute(select(models.User).where(models.User.id == user_id))
|
||
user = res.scalars().first()
|
||
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="Пользователь не найден")
|
||
|
||
profile_data = {
|
||
"id": user.id,
|
||
"public_key": user.public_key,
|
||
"first_name": user.first_name,
|
||
"last_name": user.last_name,
|
||
"username": user.username,
|
||
"show_avatar": bool(user.show_avatar),
|
||
"totp_enabled": bool(user.totp_secret),
|
||
"online": str(user.id) in connection_manager.manager.online_users
|
||
}
|
||
|
||
if user.show_avatar or current_user.id == 1:
|
||
profile_data["avatar_url"] = str(request.url_for("get_file", file_id=user.avatar_file_id)) if (user.show_avatar or current_user.id == 1) and user.avatar_file_id else None
|
||
|
||
if user.show_about or current_user.id == 1:
|
||
profile_data["about"] = user.about
|
||
|
||
if user.show_phone or current_user.id == 1:
|
||
profile_data["phone"] = user.phone
|
||
|
||
if user.show_email or current_user.id == 1:
|
||
profile_data["email"] = user.email
|
||
|
||
if user.show_last_online or current_user.id == 1:
|
||
profile_data["last_online"] = user.last_online.isoformat() if user.last_online else None
|
||
|
||
return profile_data
|
||
|
||
@usersRouter.put("/me/avatar")
|
||
async def update_user_avatar(
|
||
data: dict,
|
||
current_user: models.User = Depends(get_current_user),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
user_to_update = await db.merge(current_user)
|
||
avatar_file_id = data.get("avatar_file_id")
|
||
if avatar_file_id:
|
||
old_avatar_file_id = user_to_update.avatar_file_id
|
||
if old_avatar_file_id and old_avatar_file_id != avatar_file_id:
|
||
await _delete_old_avatar_file(old_avatar_file_id, db)
|
||
|
||
user_to_update.avatar_file_id = avatar_file_id
|
||
await db.commit()
|
||
print(f"Пользователь {user_to_update.id} обновил аватар: {avatar_file_id}")
|
||
else:
|
||
raise HTTPException(status_code=400, detail="avatar_file_id is required")
|
||
await connection_manager.manager.broadcast({'type': 'user_updated', 'user_id': current_user.id})
|
||
return {"message": "Avatar updated"} |