76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from app.common.utils.id_gen import new_invite_code
|
|
from app.models.entities import InviteCode, InviteRelation
|
|
from app.modules.invites.repository import InviteRepository
|
|
|
|
|
|
class InviteService:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
self.repository = InviteRepository(db)
|
|
|
|
def get_invite_summary(self, user_id: int) -> dict:
|
|
code = self.repository.get_default_code(user_id)
|
|
if not code:
|
|
code = self._create_default_code(user_id)
|
|
relations = self.repository.inviter_relations(user_id)
|
|
rewarded = [item for item in relations if item.reward_status == "rewarded"]
|
|
return {
|
|
"inviteCode": code.invite_code,
|
|
"inviteLink": code.invite_link,
|
|
"invitedUsers": len(relations),
|
|
"rewardedUsers": len(rewarded),
|
|
"rewardedPoints": sum(item.reward_points for item in rewarded),
|
|
}
|
|
|
|
def create_invite_code(self, user_id: int) -> dict:
|
|
code = self.repository.get_default_code(user_id)
|
|
if code:
|
|
return {"inviteCode": code.invite_code, "inviteLink": code.invite_link}
|
|
code = self._create_default_code(user_id)
|
|
return {"inviteCode": code.invite_code, "inviteLink": code.invite_link}
|
|
|
|
def list_relations(self, user_id: int) -> list[dict]:
|
|
relations = self.repository.inviter_relations(user_id)
|
|
users = self.repository.users_by_ids([item.invitee_user_id for item in relations])
|
|
return [
|
|
{
|
|
"inviteeUserId": item.invitee_user_id,
|
|
"inviteeNickname": users.get(item.invitee_user_id).nickname if users.get(item.invitee_user_id) else "",
|
|
"rewardStatus": item.reward_status,
|
|
"rewardPoints": item.reward_points,
|
|
"createdAt": item.created_at.isoformat(),
|
|
"rewardedAt": item.rewarded_at.isoformat() if item.rewarded_at else None,
|
|
}
|
|
for item in relations
|
|
]
|
|
|
|
def list_rewards(self, user_id: int) -> list[dict]:
|
|
relations = self.repository.inviter_relations(user_id)
|
|
return [
|
|
{
|
|
"inviteeUserId": item.invitee_user_id,
|
|
"rewardStatus": item.reward_status,
|
|
"rewardPoints": item.reward_points,
|
|
"rewardedAt": item.rewarded_at.isoformat() if item.rewarded_at else None,
|
|
}
|
|
for item in relations
|
|
if item.reward_points > 0
|
|
]
|
|
|
|
def _create_default_code(self, user_id: int) -> InviteCode:
|
|
code_value = new_invite_code()
|
|
code = InviteCode(
|
|
user_id=user_id,
|
|
invite_code=code_value,
|
|
invite_link=f"http://localhost:3000/register?inviteCode={code_value}",
|
|
status=1,
|
|
is_default=True,
|
|
)
|
|
self.db.add(code)
|
|
self.db.commit()
|
|
self.db.refresh(code)
|
|
return code
|
|
|