Files
colio/codes/services.py

151 lines
4.5 KiB
Python

import random
import string
import requests
from django.db import IntegrityError
from django.conf import settings
from django.utils.timezone import now
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from .models import *
from common.models.choiceModels import CertificateCodeUseType, InviteCodeUseType
from common.utils.codeManger import set_expire
from solapi import SolapiMessageService
from solapi.model import RequestMessage
from users.models import User
from projects.models import Project, ProjectTeamList
from datetime import timedelta
from rest_framework_simplejwt.tokens import AccessToken
# from .schemas import send_sms_post_schema # Swagger나 drf-spectacular 등에 사용되는 데코레이터
INVITE_CHOICE_USE_TYPE ={
'p': InviteCodeUseType.PROJECT,
'h': InviteCodeUseType.HACKATHON
}
PASSWORD_RESET_TOKEN_TTL_MINUTES = 5
class CertificateService:
@staticmethod
def send(code, identifier):
pass
# 유효시간 5분 설정해서 저장, 이전 같은 번호 sms 요청 값들 is_used=True로 바꾸기
@staticmethod
def save_certificate_info(use_type: CertificateCodeUseType, code, identifier):
try:
befores = CertificationCode.objects.filter(
use_type=use_type,
identifier=identifier,
is_used=False
).update(is_used=True)
instance = CertificationCode.objects.create(
use_type = use_type,
code = code,
expire_at = set_expire(5),
identifier = identifier
)
return True
except IntegrityError as e:
return False
# code 체크 후 is_used=True로 설정
@staticmethod
def check_code(use_type: CertificateCodeUseType, code, identifier):
if check := CertificationCode.objects.filter(
use_type=use_type,
identifier=identifier,
is_used=False,
expire_at__gte=now()
).order_by('-created_at').first():
if check.code == code:
check.is_used=True
check.save()
return True
return False
class SmsService(CertificateService):
# 같은 전번에 대해 요청 텀 설정은 views 단에서 하자.
@staticmethod
def send(code, identifier):
message_service = SolapiMessageService(
api_key = settings.SOLAPI_API_KEY, api_secret = settings.SOLAPI_API_SECRET
)
message = RequestMessage(
from_ = settings.FROM_PHONE_NUMBER,
to = identifier,
text = "colio 서비스 회원가입 인증번호는 " + code +" 입니다."
)
try:
res = message_service.send(message)
return True
except Exception as e:
# print(f"메시지 발송 실패: {str(e)}")
return False
class InviteService:
# 종류마다 사용하는 테이블이 다르므로 오버라이딩
@staticmethod
def add_member(invitee, work):
pass
# url 생성
@staticmethod
def create_invite_object_and_url(use_type, identifier, code) -> str:
InviteCode.objects.create(
use_type = INVITE_CHOICE_USE_TYPE[use_type],
code = code,
expire_at = set_expire(10080),
identifier = identifier,
)
return f"https://{settings.DOMAIN_NAME}/invite?t={use_type}&i={identifier}&c={code}"
# 코드 유효성 검사
@staticmethod
def check_code(use_type, identifier, code) -> bool:
if InviteCode.objects.filter(
use_type = INVITE_CHOICE_USE_TYPE[use_type],
identifier = identifier,
code = code,
expire_at__gte=now()
).exists():
return True
else:
return False
class ProjectInviteService(InviteService):
# 사용자 초대
@staticmethod
def add_member(invitee: User, work: Project):
return ProjectTeamList.objects.create(user=invitee, project=work)
class PasswordResetTokenService:
@staticmethod
def issue_temp_access_token(*, user_id: str, identifier: str, use_type: str) -> str:
token = AccessToken()
token.set_exp(lifetime=timedelta(minutes=PASSWORD_RESET_TOKEN_TTL_MINUTES))
token["purpose"] = "password_reset"
token["user_id"] = str(user_id)
token["identifier"] = identifier
token["use_type"] = use_type
return str(token)