c1aac7ecac
Co-authored-by: Cursor <cursoragent@cursor.com>
66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
from sqlalchemy import func
|
|
|
|
from app import db
|
|
from app.models import Photo, User, UserGroup
|
|
|
|
|
|
def get_default_group():
|
|
return UserGroup.query.filter_by(is_default=True).first()
|
|
|
|
|
|
def get_user_group(user):
|
|
if user.group_id and user.group:
|
|
return user.group
|
|
return get_default_group()
|
|
|
|
|
|
def get_user_storage_used(user_id):
|
|
result = db.session.query(func.coalesce(func.sum(Photo.file_size), 0)).filter(
|
|
Photo.user_id == user_id
|
|
).scalar()
|
|
return int(result or 0)
|
|
|
|
|
|
def get_group_quota_bytes(group):
|
|
if not group or group.disk_quota_mb == 0:
|
|
return None
|
|
return group.disk_quota_mb * 1024 * 1024
|
|
|
|
|
|
def check_upload_quota(user, new_file_size):
|
|
group = get_user_group(user)
|
|
quota_bytes = get_group_quota_bytes(group)
|
|
if quota_bytes is None:
|
|
return True, ""
|
|
|
|
used = get_user_storage_used(user.id)
|
|
if used + new_file_size > quota_bytes:
|
|
from app.models import Photo as _Photo
|
|
|
|
used_human = _Photo(file_size=used).size_human if used else "0 Б"
|
|
quota_human = f"{group.disk_quota_mb} МБ"
|
|
return False, f"Превышена квота группы «{group.name}»: {used_human} / {quota_human}"
|
|
return True, ""
|
|
|
|
|
|
def quota_status(user):
|
|
group = get_user_group(user)
|
|
used = get_user_storage_used(user.id)
|
|
quota_bytes = get_group_quota_bytes(group)
|
|
if quota_bytes is None:
|
|
return {
|
|
"group": group,
|
|
"used": used,
|
|
"quota_bytes": None,
|
|
"percent": 0,
|
|
"unlimited": True,
|
|
}
|
|
percent = min(100, int(used / quota_bytes * 100)) if quota_bytes else 0
|
|
return {
|
|
"group": group,
|
|
"used": used,
|
|
"quota_bytes": quota_bytes,
|
|
"percent": percent,
|
|
"unlimited": False,
|
|
}
|