forked from hesabix/arc
280 lines
10 KiB
Python
Executable file
280 lines
10 KiB
Python
Executable file
from __future__ import annotations
|
|
|
|
from typing import Optional, List, Literal
|
|
from datetime import datetime, timedelta
|
|
|
|
from sqlalchemy import select, func, and_, or_, update
|
|
from sqlalchemy.orm import Session
|
|
|
|
from adapters.db.models.user import User
|
|
from adapters.db.repositories.base_repo import BaseRepository
|
|
from adapters.api.v1.schemas import QueryInfo
|
|
|
|
|
|
class UserRepository(BaseRepository[User]):
|
|
def __init__(self, db: Session) -> None:
|
|
super().__init__(db, User)
|
|
|
|
def get_by_email(self, email: str) -> Optional[User]:
|
|
stmt = select(User).where(User.email == email)
|
|
return self.db.execute(stmt).scalars().first()
|
|
|
|
def get_by_mobile(self, mobile: str) -> Optional[User]:
|
|
stmt = select(User).where(User.mobile == mobile)
|
|
return self.db.execute(stmt).scalars().first()
|
|
|
|
def get_signups_timeline_buckets(
|
|
self,
|
|
start_utc: datetime,
|
|
end_utc: datetime,
|
|
granularity: Literal["day", "week", "month"],
|
|
) -> List[tuple[datetime, int]]:
|
|
"""تعداد ثبت‌نام کاربران به‌ازای هر بازهٔ زمانی (PostgreSQL date_trunc)."""
|
|
if start_utc >= end_utc:
|
|
return []
|
|
if granularity == "day":
|
|
bucket = func.date_trunc("day", User.created_at)
|
|
elif granularity == "week":
|
|
bucket = func.date_trunc("week", User.created_at)
|
|
else:
|
|
bucket = func.date_trunc("month", User.created_at)
|
|
stmt = (
|
|
select(bucket.label("bucket"), func.count().label("cnt"))
|
|
.where(User.created_at >= start_utc, User.created_at < end_utc)
|
|
.group_by(bucket)
|
|
.order_by(bucket)
|
|
)
|
|
rows = self.db.execute(stmt).all()
|
|
out: List[tuple[datetime, int]] = []
|
|
for row in rows:
|
|
b = row.bucket
|
|
if b is None:
|
|
continue
|
|
out.append((b, int(row.cnt or 0)))
|
|
return out
|
|
|
|
def count_recently_active_users(self, within_minutes: int = 5) -> int:
|
|
"""تعداد کاربران فعالی که ضربان فعالیت در بازهٔ اخیر داشته‌اند (برآورد «کاربر آنلاین»)."""
|
|
cutoff = datetime.utcnow() - timedelta(minutes=max(1, min(within_minutes, 120)))
|
|
stmt = select(func.count()).select_from(User).where(
|
|
User.is_active == True,
|
|
User.last_activity_at.is_not(None),
|
|
User.last_activity_at >= cutoff,
|
|
)
|
|
return int(self.db.execute(stmt).scalar() or 0)
|
|
|
|
def touch_last_activity(self, user_id: int, throttle_seconds: int = 45) -> Optional[datetime]:
|
|
"""
|
|
به‌روزرسانی زمان آخرین فعالیت کاربر؛ اگر قبلاً در بازهٔ throttle به‌روز شده باشد، UPDATE زده نمی‌شود.
|
|
برمی‌گرداند timestamp فعالیت ثبت‌شده (جدید یا از DB در صورت throttle).
|
|
"""
|
|
now = datetime.utcnow()
|
|
threshold = now - timedelta(seconds=throttle_seconds)
|
|
stmt = (
|
|
update(User)
|
|
.where(User.id == user_id)
|
|
.where(or_(User.last_activity_at.is_(None), User.last_activity_at < threshold))
|
|
.values(last_activity_at=now)
|
|
)
|
|
result = self.db.execute(stmt)
|
|
self.db.commit()
|
|
if getattr(result, "rowcount", 0):
|
|
return now
|
|
u = self.get_by_id(user_id)
|
|
return u.last_activity_at if u else None
|
|
|
|
def get_by_referral_code(self, referral_code: str) -> Optional[User]:
|
|
stmt = select(User).where(User.referral_code == referral_code)
|
|
return self.db.execute(stmt).scalars().first()
|
|
|
|
def is_first_user(self) -> bool:
|
|
"""بررسی اینکه آیا این اولین کاربر سیستم است یا نه"""
|
|
stmt = select(func.count()).select_from(User)
|
|
count = self.db.execute(stmt).scalar() or 0
|
|
return count == 0
|
|
|
|
def create(self, *, email: str | None, mobile: str | None, password_hash: str, first_name: str | None, last_name: str | None, referral_code: str, referred_by_user_id: int | None = None, email_verified: bool = False) -> User:
|
|
# تعیین دسترسی‌های برنامه بر اساس اینکه آیا کاربر اول است یا نه
|
|
app_permissions = {"superadmin": True} if self.is_first_user() else {}
|
|
|
|
user = User(
|
|
email=email,
|
|
mobile=mobile,
|
|
password_hash=password_hash,
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
referral_code=referral_code,
|
|
referred_by_user_id=referred_by_user_id,
|
|
app_permissions=app_permissions,
|
|
email_verified=email_verified
|
|
)
|
|
self.db.add(user)
|
|
self.db.commit()
|
|
self.db.refresh(user)
|
|
return user
|
|
|
|
def count_referred(self, referrer_user_id: int, start: str | None = None, end: str | None = None) -> int:
|
|
stmt = select(func.count()).select_from(User).where(User.referred_by_user_id == referrer_user_id)
|
|
if start is not None:
|
|
stmt = stmt.where(User.created_at >= func.cast(start, User.created_at.type))
|
|
if end is not None:
|
|
stmt = stmt.where(User.created_at < func.cast(end, User.created_at.type))
|
|
return int(self.db.execute(stmt).scalar() or 0)
|
|
|
|
def count_referred_between(self, referrer_user_id: int, start_dt, end_dt) -> int:
|
|
stmt = select(func.count()).select_from(User).where(
|
|
and_(
|
|
User.referred_by_user_id == referrer_user_id,
|
|
User.created_at >= start_dt,
|
|
User.created_at < end_dt,
|
|
)
|
|
)
|
|
return int(self.db.execute(stmt).scalar() or 0)
|
|
|
|
def count_referred_filtered(self, referrer_user_id: int, start_dt=None, end_dt=None, search: str | None = None) -> int:
|
|
stmt = select(func.count()).select_from(User).where(User.referred_by_user_id == referrer_user_id)
|
|
if start_dt is not None:
|
|
stmt = stmt.where(User.created_at >= start_dt)
|
|
if end_dt is not None:
|
|
stmt = stmt.where(User.created_at < end_dt)
|
|
if search:
|
|
like = f"%{search}%"
|
|
stmt = stmt.where(or_(User.first_name.ilike(like), User.last_name.ilike(like), User.email.ilike(like)))
|
|
return int(self.db.execute(stmt).scalar() or 0)
|
|
|
|
def list_referred(self, referrer_user_id: int, start_dt=None, end_dt=None, search: str | None = None, offset: int = 0, limit: int = 20):
|
|
stmt = select(User).where(User.referred_by_user_id == referrer_user_id)
|
|
if start_dt is not None:
|
|
stmt = stmt.where(User.created_at >= start_dt)
|
|
if end_dt is not None:
|
|
stmt = stmt.where(User.created_at < end_dt)
|
|
if search:
|
|
like = f"%{search}%"
|
|
stmt = stmt.where(or_(User.first_name.ilike(like), User.last_name.ilike(like), User.email.ilike(like)))
|
|
stmt = stmt.order_by(User.created_at.desc()).offset(offset).limit(limit)
|
|
return self.db.execute(stmt).scalars().all()
|
|
|
|
def get_support_operators(self) -> List[User]:
|
|
"""دریافت لیست تمام اپراتورهای پشتیبانی فعال"""
|
|
from sqlalchemy import cast, or_
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
permissions = cast(User.app_permissions, JSONB)
|
|
stmt = select(User).where(
|
|
User.is_active == True,
|
|
or_(
|
|
permissions.contains({"support_operator": True}),
|
|
permissions.contains({"superadmin": True}),
|
|
),
|
|
)
|
|
return list(self.db.execute(stmt).scalars().all())
|
|
|
|
def is_support_operator(self, user_id: int) -> bool:
|
|
"""بررسی اینکه آیا کاربر یک اپراتور پشتیبانی است یا نه"""
|
|
user = self.get_by_id(user_id)
|
|
if not user or not user.is_active:
|
|
return False
|
|
if not user.app_permissions:
|
|
return False
|
|
# SuperAdmin هم می‌تواند اپراتور باشد
|
|
return bool(user.app_permissions.get("support_operator", False) or user.app_permissions.get("superadmin", False))
|
|
|
|
def to_dict(self, user: User, include_extended: bool = False) -> dict:
|
|
"""تبدیل User object به dictionary برای API response"""
|
|
# ساخت full_name
|
|
full_name = None
|
|
if user.first_name or user.last_name:
|
|
parts = [p for p in [user.first_name, user.last_name] if p]
|
|
full_name = " ".join(parts) if parts else None
|
|
|
|
# تعیین status از is_active و وضعیت تأیید
|
|
if not user.is_active:
|
|
status = "inactive"
|
|
elif not user.email_verified and not user.mobile_verified:
|
|
status = "pending"
|
|
else:
|
|
status = "active"
|
|
|
|
# تعیین role از app_permissions
|
|
role = "user"
|
|
if user.app_permissions:
|
|
if user.app_permissions.get("superadmin"):
|
|
role = "admin"
|
|
elif user.app_permissions.get("operator"):
|
|
role = "operator"
|
|
elif user.app_permissions.get("supervisor"):
|
|
role = "supervisor"
|
|
|
|
# شمارش کسب‌وکارها
|
|
businesses_count = 0
|
|
last_login_ip = None
|
|
last_login_at = None
|
|
|
|
if include_extended:
|
|
from adapters.db.repositories.business_permission_repo import BusinessPermissionRepository
|
|
from adapters.db.repositories.business_repo import BusinessRepository
|
|
from adapters.db.repositories.api_key_repo import ApiKeyRepository
|
|
|
|
# شمارش کسب‌وکارها (هم مالک و هم عضو)
|
|
business_repo = BusinessRepository(self.db)
|
|
bp_repo = BusinessPermissionRepository(self.db)
|
|
|
|
# دریافت کسب‌وکارهایی که کاربر مالک آن‌هاست
|
|
owned_businesses = business_repo.get_by_owner_id(user.id)
|
|
owned_business_ids = {b.id for b in owned_businesses}
|
|
|
|
# دریافت کسب‌وکارهایی که کاربر عضو آن‌هاست (از طریق BusinessPermission با join=True)
|
|
member_permissions = bp_repo.get_user_member_businesses(user.id)
|
|
member_business_ids = {perm.business_id for perm in member_permissions}
|
|
|
|
# ترکیب و حذف تکراری
|
|
all_business_ids = owned_business_ids | member_business_ids
|
|
businesses_count = len(all_business_ids)
|
|
|
|
# آخرین ورود
|
|
api_repo = ApiKeyRepository(self.db)
|
|
from sqlalchemy import select, desc
|
|
from adapters.db.models.api_key import ApiKey
|
|
stmt = select(ApiKey).where(
|
|
ApiKey.user_id == user.id,
|
|
ApiKey.revoked_at.is_(None)
|
|
).order_by(desc(ApiKey.last_used_at)).limit(1)
|
|
last_key = self.db.execute(stmt).scalars().first()
|
|
if last_key:
|
|
last_login_ip = last_key.ip
|
|
last_login_at = last_key.last_used_at or last_key.created_at
|
|
|
|
result = {
|
|
"id": user.id,
|
|
"email": user.email,
|
|
"mobile": user.mobile,
|
|
"first_name": user.first_name,
|
|
"last_name": user.last_name,
|
|
"full_name": full_name,
|
|
"is_active": user.is_active,
|
|
"status": status,
|
|
"role": role,
|
|
"referral_code": user.referral_code,
|
|
"referred_by_user_id": user.referred_by_user_id,
|
|
"app_permissions": user.app_permissions,
|
|
"created_at": user.created_at,
|
|
"updated_at": user.updated_at,
|
|
"signature_file_id": getattr(user, "signature_file_id", None),
|
|
}
|
|
|
|
if include_extended:
|
|
result.update({
|
|
"businesses_count": businesses_count,
|
|
"last_login_ip": last_login_ip,
|
|
"last_login_at": last_login_at,
|
|
"last_activity_at": user.last_activity_at,
|
|
})
|
|
|
|
return result
|
|
|
|
def query_admin_list(self, query_info: QueryInfo) -> tuple[list[User], int]:
|
|
"""لیست کاربران برای پنل مدیریت با فیلدهای مجازی UI."""
|
|
from app.services.user_list_query_service import query_users_admin
|
|
|
|
return query_users_admin(self.db, query_info)
|
|
|
|
|