Merge pull request #294 from Pseudo-Lab/feat/cert-watermark-verification

Feat/cert watermark verification

Author: yesinkim

cert/backend/pyproject.toml

@@ -8,9 +8,12 @@ dependencies = [
     "aiohttp>=3.12.15",
     "dotenv>=0.9.9",
     "fastapi>=0.109.1",
+    "invisible-watermark>=0.2.0",
+    "opencv-python>=4.11.0.86",
     "pillow>=10.1.0",
     "pydantic>=2.5.3",
     "pydantic-settings>=2.1.0",
+    "pypdf>=6.5.0",
     "python-dotenv>=1.0.1",
     "python-multipart>=0.0.6",
     "reportlab>=4.0.7",

cert/backend/src/routers/certificate.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, HTTPException, Response
+from fastapi import APIRouter, HTTPException, Response, File, UploadFile
 
 from ..models.project import Project, ProjectsBySeasonResponse
 from ..models.certificate import CertificateCreate, CertificateResponse, ErrorResponse
@@ -91,3 +91,19 @@ async def clear_cache():
     """캐시 삭제"""
     ProjectService.clear_cache()
     return {"message": "캐시 삭제 완료"}
+
+@certificate_router.post("/verify")
+async def verify_certificate(file: UploadFile = File(...)):
+    """수료증의 진위 여부를 확인합니다."""
+    # 파일 확장자 체크 (PDF만 허용)
+    if not file.filename.lower().endswith('.pdf'):
+        raise HTTPException(status_code=400, detail="PDF 파일만 업로드 가능합니다.")
+    
+    try:
+        file_bytes = await file.read()
+        result = await CertificateService.verify_certificate(file_bytes)
+        return result
+    except Exception as e:
+        import logging
+        logging.error(f"검증 중 오류 발생: {e}")
+        raise HTTPException(status_code=500, detail="파일 처리 중 오류가 발생했습니다.")

cert/backend/src/services/certificate_service.py

@@ -111,6 +111,82 @@ async def create_certificate(certificate_data: dict) -> CertificateResponse:
                 message="수료증 발급을 완료하지 못했습니다. 관리자에게 문의해주세요.",
                 data=None,
             )
+
+    @staticmethod
+    async def verify_certificate(file_bytes: bytes) -> dict:
+        """수료증 검증"""
+        try:
+            pdf_generator = PDFGenerator()
+            watermark_text = pdf_generator.extract_watermark_from_pdf(file_bytes)
+            
+            if not watermark_text:
+                return {
+                    "valid": False,
+                    "message": "수료증에서 워터마크를 찾을 수 없습니다."
+                }
+                
+            # 기본 검증 (PSEUDOLAB 접두사 확인)
+            if not watermark_text.startswith("PSEUDOLAB"):
+                return {
+                    "valid": False,
+                    "message": "유효하지 않은 수료증 워터마크입니다.",
+                    "debug_text": watermark_text
+                }
+                
+            # 수료증 번호 추출 (PSEUDOLAB_CERT-XXXX 포맷 기대)
+            cert_number = ""
+            if "_" in watermark_text:
+                cert_number = watermark_text.split("_")[1]
+            
+            # 번호가 없는 경우 (테스트용 등)
+            if not cert_number:
+                return {
+                    "valid": True,
+                    "message": "워터마크가 확인되었습니다 (테스트용/번호없음).",
+                    "watermark_text": watermark_text
+                }
+
+            # Notion 실데이터 조회
+            notion_client = NotionClient()
+            cert_page = await notion_client.get_certificate_by_number(cert_number)
+            
+            if not cert_page:
+                return {
+                    "valid": False,
+                    "message": f"수료증 번호({cert_number})에 해당하는 발급 기록을 찾을 수 없습니다."
+                }
+            
+            # Notion 결과 파싱
+            props = cert_page.get("properties", {})
+            
+            name = props.get("Name", {}).get("title", [{}])[0].get("plain_text", "알 수 없음")
+            course = props.get("Course Name", {}).get("rich_text", [{}])[0].get("plain_text", "알 수 없음")
+            season = props.get("Season", {}).get("select", {}).get("name", "알 수 없음")
+            issue_date = props.get("Issue Date", {}).get("date", {}).get("start", "알 수 없음")
+            status = props.get("Certificate Status", {}).get("status", {}).get("name", "알 수 없음")
+
+            return {
+                "valid": True,
+                "message": "수료증 진위 확인에 성공했습니다.",
+                "data": {
+                    "name": name,
+                    "course": course,
+                    "season": season,
+                    "issue_date": issue_date,
+                    "certificate_number": cert_number,
+                    "status": status
+                }
+            }
+            
+        except Exception as e:
+            if "워터마크를 찾을 수 없습니다" in str(e):
+                logger.info(f"수료증 검증 실패 (워터마크 없음): {str(e)}")
+            else:
+                logger.exception("수료증 검증 중 예상치 못한 오류")
+            return {
+                "valid": False,
+                "message": "수료증 검증 처리 중 오류가 발생했습니다."
+            }
 
     @staticmethod
     async def _reissue_certificate(
@@ -147,7 +223,26 @@ async def _reissue_certificate(
                     "기존 수료증 번호 없음. 새로 생성",
                     extra={"applicant_name": certificate_data["applicant_name"]},
                 )
-                existing_cert_number = f"CERT-{datetime.now().year}{participation_info['project_code']}{str(uuid.uuid4())[:2].upper()}"
+                # 기존 수료증 번호가 없으면 DB ID(Unique ID) 기반으로 생성
+                existing_data = existing_cert.get("existing_data", {})
+                properties = existing_data.get("properties", {})
+                
+                # 로그에서 확인된 구조: properties['ID']['unique_id']['number']
+                id_prop = properties.get("ID", {})
+                unique_identifier = None
+                
+                if id_prop.get("type") == "unique_id":
+                    unique_identifier = str(id_prop.get("unique_id", {}).get("number"))
+                
+                if not unique_identifier:
+                    # fallback: 혹시 ID 컬럼명이 다를 경우를 대비해 기존 방식 유지
+                    for prop_val in properties.values():
+                        if prop_val.get("type") == "unique_id":
+                            unique_identifier = str(prop_val.get("unique_id", {}).get("number"))
+                            break
+
+                
+                existing_cert_number = f"CERT-{datetime.now().year}{participation_info['project_code']}{unique_identifier.upper()}"
                 logger.info(
                     "새로운 수료증 번호 생성",
                     extra={"certificate_number": existing_cert_number},
@@ -170,14 +265,17 @@ async def _reissue_certificate(
 
             # 이메일 재발송
             email_sender = EmailSender()
-            await email_sender.send_certificate_email(
+            email_sent = 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
             )
+            
+            if not email_sent:
+                raise Exception("재발급 이메일 발송 실패")
 
             # 재발급 로그 기록
             reissue_log = await notion_client.log_certificate_reissue(
@@ -258,8 +356,27 @@ async def _create_new_certificate(
                 season=certificate_data["season"]
             )
 
-            # TODO: 임시 값, 추후 수정 필요
-            certificate_number = f"CERT-{datetime.now().year}{participation_info['project_code']}{str(uuid.uuid4())[:2].upper()}"
+            # DB ID(Unique ID)를 사용하여 수료증 번호 생성
+            properties = certificate_request.get("properties", {})
+            unique_identifier = None
+            
+            # 로그에서 확인된 구조: properties['ID']['unique_id']['number']
+            id_prop = properties.get("ID", {})
+            if id_prop.get("type") == "unique_id":
+                unique_identifier = str(id_prop.get("unique_id", {}).get("number"))
+
+            if not unique_identifier:
+                 # fallback: 혹시 ID 컬럼명이 다를 경우를 대비해 순회 검색
+                for prop_val in properties.values():
+                    if prop_val.get("type") == "unique_id":
+                        unique_identifier = str(prop_val.get("unique_id", {}).get("number"))
+                        break
+            
+            if not unique_identifier:
+                # fallback: Page ID의 마지막 5자리
+                unique_identifier = request_id.replace("-", "")[-5:]
+
+            certificate_number = f"CERT-{datetime.now().year}{participation_info['project_code']}{unique_identifier.upper()}"
             issue_date = datetime.now().strftime("%Y-%m-%d")
 
             # PDF 수료증 생성
@@ -275,14 +392,17 @@ async def _create_new_certificate(
             )
             # 이메일 발송
             email_sender = EmailSender()
-            await email_sender.send_certificate_email(
+            email_sent = 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
             )
+
+            if not email_sent:
+                raise Exception("이메일 발송 실패")
 
             # 수료증 상태 업데이트
             logger.info(

cert/backend/src/utils/email_sender.py

@@ -86,10 +86,13 @@ async def send_certificate_email(
                 )
 
             # 이메일 발송
+            # Port 465(SSL)는 use_tls=True, Port 587(STARTTLS)은 use_tls=False 필요
+            use_tls = (self.smtp_port == 465)
+            
             async with aiosmtplib.SMTP(
                 hostname=self.smtp_host,
                 port=self.smtp_port,
-                use_tls=False  # 포트 587은 STARTTLS 사용
+                use_tls=use_tls
             ) as smtp:
                 # STARTTLS는 aiosmtplib가 자동으로 처리
                 await smtp.login(self.smtp_username, self.smtp_password)

cert/backend/src/utils/notion_client.py

@@ -873,6 +873,40 @@ async def check_existing_certificate(
             logger.exception("기존 수료증 확인 중 오류")
             return None
 
+    async def get_certificate_by_number(self, certificate_number: str) -> Optional[Dict[str, Any]]:
+        """수료증 번호로 수료증 정보 조회"""
+        db_id = self.databases.get("certificate_requests")
+        if not db_id:
+            logger.error("수료증 데이터베이스 ID가 설정되지 않았습니다.")
+            return None
+            
+        url = f"{self.base_url}/databases/{db_id}/query"
+        payload = {
+            "filter": {
+                "property": "Certificate Number",
+                "rich_text": {
+                    "equals": certificate_number
+                }
+            }
+        }
+        
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(url, headers=self.headers, json=payload) as response:
+                    if response.status == 200:
+                        data = await response.json()
+                        results = data.get("results", [])
+                        if results:
+                            return results[0]
+                        return None
+                    else:
+                        error_text = await response.text()
+                        logger.error(f"Notion 수료증 조회 실패: {response.status}, {error_text}")
+                        return None
+        except Exception:
+            logger.exception("Notion 수료증 조회 중 예외 발생")
+            return None
+
     async def get_database_structure(self, database_type: str = "project_history") -> Optional[Dict[str, Any]]:
         """데이터베이스 구조 조회 (디버깅용)"""
         try: