Skip to content

feat: 수료증 재발급 및 신규 발급 로직 추가 #139

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: deploy/cert-develop
Choose a base branch
from
Open

feat: 수료증 재발급 및 신규 발급 로직 추가 #139

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 117 additions & 14 deletions cert/backend/src/services/certificate_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,8 @@
import uuid
from datetime import datetime
from typing import Optional, List
from fastapi import HTTPException



from ..models.project import Project, ProjectsBySeasonResponse
from ..models.certificate import CertificateResponse, CertificateData, CertificateStatus, Role
from ..constants.error_codes import ErrorCodes, ResponseStatus
from ..utils.notion_client import NotionClient
from ..utils.pdf_generator import PDFGenerator
from ..utils.email_sender import EmailSender
Expand Down Expand Up @@ -49,7 +44,122 @@ async def create_certificate(certificate_data: dict) -> CertificateResponse:
notion_client = NotionClient()

try:
# 1. 기존 수료증 확인 (재발급 여부 판단)
existing_cert = await notion_client.check_existing_certificate(
applicant_name=certificate_data["applicant_name"],
course_name=certificate_data["course_name"],
season=certificate_data["season"],
recipient_email=certificate_data.get("recipient_email")
)

# 기존 수료증 확인이 성공하고 기존 수료증이 있는 경우 재발급 처리
if existing_cert and existing_cert.get("found"):
print(f"기존 수료증 발견: {existing_cert.get('certificate_number')}")
return await CertificateService._reissue_certificate(
certificate_data, existing_cert, notion_client
)

# 2. 신규 수료증 발급 처리
return await CertificateService._create_new_certificate(
certificate_data, notion_client
)

except Exception as e:
print(f"수료증 발급 중 오류: {e}")
return CertificateResponse(
status="500",
message=f"수료증 발급 중 오류가 발생했습니다: {str(e)}",
data=None
)

@staticmethod
async def _reissue_certificate(
certificate_data: dict,
existing_cert: dict,
notion_client: NotionClient
) -> CertificateResponse:
"""기존 수료증 재발급"""
try:
# 기존 수료증 정보 사용
existing_page_id = existing_cert.get("page_id")
existing_cert_number = existing_cert.get("certificate_number")

print("🔄 기존 수료증 재발급 시작 (이름, 코스, 기수 일치):")
print(f" - 기존 수료증 번호: '{existing_cert_number}'")
print(f" - 요청 이메일: '{certificate_data.get('recipient_email', '')}'")

# 사용자 참여 이력 재확인 (역할 정보 가져오기)
participation_info = await notion_client.verify_user_participation(
user_name=certificate_data["applicant_name"],
course_name=certificate_data["course_name"],
season=certificate_data["season"]
)

# 수료증 번호가 없는 경우 새로 생성
if not existing_cert_number:
print("⚠️ 기존 수료증에 번호가 없어 새로 생성합니다.")
existing_cert_number = f"CERT-{datetime.now().year}{participation_info['project_code']}{str(uuid.uuid4())[:2].upper()}"
print(f"🆕 새로 생성된 수료증 번호: {existing_cert_number}")

# PDF 수료증 재생성
pdf_generator = PDFGenerator()
pdf_bytes = pdf_generator.create_certificate(
name=certificate_data["applicant_name"],
season=certificate_data['season'],
course_name=certificate_data["course_name"],
role=participation_info["user_role"],
period=participation_info["period"],
)

# 이메일 재발송
email_sender = EmailSender()
await email_sender.send_certificate_email(
recipient_email=certificate_data["recipient_email"],
recipient_name=certificate_data["applicant_name"],
course_name=certificate_data["course_name"],
season=certificate_data["season"],
role=participation_info["user_role"],
certificate_bytes=pdf_bytes
)

# 기존 수료증 상태를 재발급으로 업데이트
await notion_client.update_certificate_status(
page_id=existing_page_id,
status=CertificateStatus.ISSUED,
certificate_number=existing_cert_number,
role=participation_info["user_role"]
)

print(f"수료증 재발급 완료: {existing_cert_number}")

return CertificateResponse(
status="200",
message="기존 수료증이 재발급되었습니다.\n제출하신 이메일로 수료증이 발송되었습니다.",
data=CertificateData(
id=existing_page_id,
name=certificate_data["applicant_name"],
recipient_email=certificate_data["recipient_email"],
certificate_number=existing_cert_number,
issue_date=datetime.now().strftime("%Y-%m-%d"),
certificate_status=CertificateStatus.ISSUED,
season=certificate_data["season"],
course_name=certificate_data["course_name"],
role=Role.BUILDER if participation_info["user_role"] == "BUILDER" else Role.RUNNER
)
)

except Exception as e:
print(f"수료증 재발급 중 오류: {e}")
raise e

@staticmethod
async def _create_new_certificate(
certificate_data: dict,
notion_client: NotionClient
) -> CertificateResponse:
"""신규 수료증 발급"""
request_id = None
try:
# 수료증 요청 내역 생성
certificate_request = await notion_client.create_certificate_request(certificate_data)

Expand Down Expand Up @@ -121,17 +231,10 @@ async def create_certificate(certificate_data: dict) -> CertificateResponse:

except Exception as e:
# 시스템 오류
print(f"시스템 오류: {e}")
print(f"신규 수료증 발급 중 시스템 오류: {e}")
if request_id: # request_id가 존재하는 경우에만 상태 업데이트
await notion_client.update_certificate_status(
page_id=request_id,
status=CertificateStatus.SYSTEM_ERROR
)
raise HTTPException(
status_code=500,
detail={
"status": ResponseStatus.FAIL,
"error_code": ErrorCodes.PIPELINE_ERROR,
"message": f"{str(e)}"
}
)
raise e
106 changes: 105 additions & 1 deletion cert/backend/src/utils/notion_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from datetime import datetime, timedelta
from datetime import datetime
import os
import aiohttp
from typing import Optional, Dict, Any, List
Expand Down Expand Up @@ -448,6 +448,110 @@ def sort_key(season_group: SeasonGroup) -> int:
print(f"기수별 프로젝트 조회 중 오류: {e}")
return None

async def check_existing_certificate(
self,
applicant_name: str,
course_name: str,
season: int,
recipient_email: str = None
) -> Optional[Dict[str, Any]]:
"""기존 수료증 확인 (재발급용) - 이름, 코스명, 기수로 검색 (이메일 무관)"""
try:
async with aiohttp.ClientSession() as session:
url = f"{self.base_url}/databases/{self.databases['certificate_requests']}/query"

# 필터 조건 구성 (이름, 코스명, 기수로만 검색 - 이메일은 무관)
filters = [
{
"property": "Name",
"title": {
"equals": applicant_name
}
},
{
"property": "Course Name",
"rich_text": {
"equals": course_name
}
},
{
"property": "Season",
"select": {
"equals": f"{season}기"
}
}
]

payload = {
"filter": {
"and": filters
}
}

async with session.post(url, headers=self.headers, json=payload) as response:
if response.status == 200:
data = await response.json()
if data["results"]:
# 가장 최근 수료증 반환 (첫 번째 결과)
existing_cert = data["results"][0]
properties = existing_cert.get("properties", {})

# 기존 수료증 정보 추출
certificate_number = ""
if "Certificate Number" in properties:
cert_num_prop = properties["Certificate Number"].get("rich_text", [])
if cert_num_prop:
certificate_number = cert_num_prop[0].get("plain_text", "")

role = ""
if "Role" in properties:
role_prop = properties["Role"].get("select", {})
if role_prop:
role = role_prop.get("name", "")

status = ""
if "Certificate Status" in properties:
status_prop = properties["Certificate Status"].get("status", {})
if status_prop:
status = status_prop.get("name", "")

# 이메일 정보 추출
existing_email = ""
if "Recipient Email" in properties:
email_prop = properties["Recipient Email"].get("email", "")
if email_prop:
existing_email = email_prop

print("🔍 기존 수료증 발견 (이름, 코스, 기수 일치):")
print(f" - 이름: {applicant_name}")
print(f" - 코스: {course_name}")
print(f" - 기수: {season}기")
print(f" - 수료증 번호: '{certificate_number}'")
print(f" - 역할: '{role}'")
print(f" - 상태: '{status}'")

return {
"found": True,
"page_id": existing_cert.get("id"),
"certificate_number": certificate_number,
"role": role,
"status": status,
"issue_date": properties.get("Issue Date", {}).get("date", {}).get("start", ""),
"existing_email": existing_email,
"existing_data": existing_cert
}
else:
print(f"🔍 기존 수료증 없음: {applicant_name}, {course_name}, {season}기")
return {"found": False}
else:
error_text = await response.text()
print(f"기존 수료증 확인 오류: {response.status} - {error_text}")
return None

except Exception as e:
print(f"기존 수료증 확인 중 오류: {e}")
return None

async def get_database_structure(self, database_type: str = "project_history") -> Optional[Dict[str, Any]]:
"""데이터베이스 구조 조회 (디버깅용)"""
try:
Expand Down