From 8226db0e25f1f349d637131c556c7620eeab791e Mon Sep 17 00:00:00 2001 From: yesinkim Date: Thu, 18 Sep 2025 17:50:00 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=EC=88=98=EB=A3=8C=EC=A6=9D=20=EC=9E=AC?= =?UTF-8?q?=EB=B0=9C=EA=B8=89=20=EB=B0=8F=20=EC=8B=A0=EA=B7=9C=20=EB=B0=9C?= =?UTF-8?q?=EA=B8=89=20=EB=A1=9C=EC=A7=81=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 기존 수료증 확인 기능 추가: 이름, 코스명, 기수로 기존 수료증 검색 - 기존 수료증이 있을 경우 재발급 처리 로직 구현 - 신규 수료증 발급 로직 개선 및 오류 처리 강화 - NotionClient에 기존 수료증 확인 메서드 추가 --- .../src/services/certificate_service.py | 131 ++++++++++++++++-- cert/backend/src/utils/notion_client.py | 106 +++++++++++++- 2 files changed, 222 insertions(+), 15 deletions(-) diff --git a/cert/backend/src/services/certificate_service.py b/cert/backend/src/services/certificate_service.py index 53054a1..6ede880 100644 --- a/cert/backend/src/services/certificate_service.py +++ b/cert/backend/src/services/certificate_service.py @@ -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 @@ -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) @@ -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)}" - } - ) \ No newline at end of file + raise e \ No newline at end of file diff --git a/cert/backend/src/utils/notion_client.py b/cert/backend/src/utils/notion_client.py index 32999f5..4fa0363 100644 --- a/cert/backend/src/utils/notion_client.py +++ b/cert/backend/src/utils/notion_client.py @@ -1,4 +1,4 @@ -from datetime import datetime, timedelta +from datetime import datetime import os import aiohttp from typing import Optional, Dict, Any, List @@ -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: