# ============================================================ # notice_processor.py — 신규 공지 Gemini 분류 + 임베딩 계산 # 크롤러 파이프라인: auto_crawler.py 실행 후 자동 실행 # ============================================================ import os import re import json import math import time import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from datetime import datetime from supabase import create_client import google.generativeai as genai from sentence_transformers import SentenceTransformer SUPABASE_URL = os.getenv("SUPABASE_URL") SUPABASE_KEY = os.getenv("SUPABASE_KEY") GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") TWO_TOWER_MODEL_PATH = os.getenv("TWO_TOWER_MODEL_PATH", "models/two_tower_model_v4.pt") BASE_MODEL_EMBED = "jhgan/ko-sroberta-multitask" BASE_DATE = datetime(2026, 4, 30) supabase = create_client(SUPABASE_URL, SUPABASE_KEY) genai.configure(api_key=GEMINI_API_KEY) gemini = genai.GenerativeModel( "gemini-2.5-flash", generation_config={"response_mime_type": "application/json"} ) def clean_for_postgres(value): """Postgres text/jsonb values cannot contain literal NUL bytes.""" if isinstance(value, str): return value.replace("\x00", "") if isinstance(value, list): return [clean_for_postgres(item) for item in value] if isinstance(value, dict): return {key: clean_for_postgres(item) for key, item in value.items()} return value # ── 카테고리 정의 ───────────────────────────────────────────── VALID_CATEGORIES = [ "취업/채용", "학사행정", "학생활동/비교과", "대외활동", "공모전/경진대회", "국제교류", "창업", "장학금", "기숙사", "ROTC" ] CATEGORY_TYPE_MAP = { "취업/채용": ["경영/금융/사무", "IT/정보통신", "디자인/예술/방송", "공학/기술", "교육/법률/공공", "교내채용"], "학생활동/비교과": ["IT/AI/SW", "진로/취업/현장실습", "디자인/콘텐츠", "심리/상담/자기계발", "인문/어학", "행사/생활"], "공모전/경진대회": ["IT/AI/SW", "창업/아이디어", "디자인/콘텐츠", "글쓰기/발표/어학", "정책/사회/ESG"], "창업": ["창업"], "국제교류": ["교환학생/파견", "해외인턴/현장실습", "해외연수", "외국인학생/글로벌교류", "해외봉사"], "학사행정": ["수업/수강", "전공/트랙/학적", "성적/졸업", "시험/학사일정", "시설/시스템/행정"], "장학금": ["장학금"], "대외활동": ["봉사활동", "멘토링", "서포터즈/홍보대사", "기획/미디어"], "ROTC": ["ROTC"], "기숙사": ["기숙사"], } HALF_LIFE_MAP = { "취업/채용": 7, "공모전/경진대회": 7, "장학금": 7, "학생활동/비교과": 14, "대외활동": 14, "국제교류": 14, "창업": 14, "ROTC": 30, "기숙사": 30, "학사행정": 30, } # ── notice_score 계산 ───────────────────────────────────────── def calc_notice_score(notice: dict) -> float: try: date_str = str(notice.get('posted_at') or notice.get('date', ''))[:10].replace('.', '-') posted = datetime.strptime(date_str, '%Y-%m-%d') days = max((datetime.now() - posted).days, 1) except: days = 999 views = notice.get('views', 0) or 0 category = notice.get('category', '기타') h = HALF_LIFE_MAP.get(category, 14) alpha = math.log(2) / h return round(math.log(views + 1) * math.exp(-alpha * days), 4) # ── Gemini 카테고리 분류 ────────────────────────────────────── def classify_with_gemini(title: str, body: str, max_retry: int = 3) -> dict: prompt = f""" 한성대학교 공지사항을 아래 카테고리 중 하나로 분류하고, 세부 유형도 분류해. 반드시 아래 '카테고리' 목록에 있는 정확한 단어만 사용해. 카테고리: {json.dumps(VALID_CATEGORIES, ensure_ascii=False)} 카테고리별 세부 유형: {json.dumps(CATEGORY_TYPE_MAP, ensure_ascii=False)} 공지 제목: {title} 공지 본문: {(body or '')[:300]} 아래 JSON 형식으로만 출력해: {{"category": "카테고리명", "category_type": ["세부유형1"]}} """ for attempt in range(max_retry): try: res = gemini.generate_content(prompt) raw_text = res.text.strip() # 마크다운 찌꺼기 제거 if raw_text.startswith("```"): raw_text = re.sub(r"^```(?:json)?\s*", "", raw_text) raw_text = re.sub(r"\s*```$", "", raw_text).strip() result = json.loads(raw_text) category = result.get('category', '').strip() if category in VALID_CATEGORIES: return result else: print(f" [디버깅] 목록에 없는 카테고리 반환됨: '{category}' (시도 {attempt+1}/{max_retry})") except json.JSONDecodeError as e: print(f" [디버깅] JSON 파싱 실패 (시도 {attempt+1}/{max_retry}): {e} | 원본: {res.text[:100]}") except Exception as e: print(f" Gemini 분류 실패 (시도 {attempt+1}/{max_retry}): {e}") time.sleep(3) return {"category": "기타", "category_type": []} # ── 장학금 파싱 ─────────────────────────────────────────────── def parse_scholarship(title: str, body: str, max_retry: int = 3) -> dict: SEOUL_GU = ["종로구","중구","용산구","성동구","광진구","동대문구","중랑구","성북구","강북구","도봉구","노원구","은평구","서대문구","마포구","양천구","강서구","구로구","금천구","영등포구","동작구","관악구","서초구","강남구","송파구","강동구"] GYEONGGI_SI = ["수원시","용인시","고양시","화성시","성남시","부천시","남양주시","안산시","평택시","안양시","시흥시","파주시","김포시","의정부시","광주시","하남시","양주시","광명시","군포시","오산시","이천시","안성시","구리시","포천시","의왕시","양평군","여주시","동두천시","과천시","가평군","연천군"] OTHER_REGIONS = ["부산광역시","대구광역시","인천광역시","광주광역시","대전광역시","울산광역시","세종특별자치시","강원특별자치도","충청북도","충청남도","전북특별자치도","전라남도","경상북도","경상남도","제주특별자치도"] seoul_list = ", ".join([f'"서울특별시 {g}"' for g in SEOUL_GU]) gyeonggi_list = ", ".join([f'"경기도 {s}"' for s in GYEONGGI_SI]) other_list = ", ".join([f'"{r}"' for r in OTHER_REGIONS]) prompt = f""" 아래 한성대학교 장학금 공지를 분석해서 JSON으로 반환해. 공지 제목: {title} 공지 본문: {(body or '')[:500]} 아래 JSON 형식으로만 출력해: {{ "is_application": true 또는 false, "type": "교내장학금 또는 국가장학금 또는 교외장학금 또는 근로장학금 중 하나", "income_max": null 또는 숫자 (소득분위 제한 있으면 숫자, 없으면 null), "income_required": true 또는 false, "income_priority_max": null 또는 숫자, "end_date": null 또는 "MM-DD" 형식, "end_date_type": "명시" 또는 "상시" 또는 "미정", "target_status": ["재학"] 또는 ["재학", "휴학"] 등, "target_grade": [1, 2, 3, 4] 또는 특정 학년만, "min_gpa": null 또는 숫자, "region": 지역 조건 없으면 null, 있으면 아래 목록 중 정확히 하나만 선택: 서울 전체: "서울특별시" 서울 특정 구: {seoul_list} 경기 전체: "경기도" 경기 특정 시군: {gyeonggi_list} 기타 광역시도: {other_list}, "extra_info": "기타 중요 정보 한 줄 요약" }} is_application 판단 기준: - true: 실제 장학금 신청을 받는 공고 (신청기간, 신청방법, 지원자격 등이 명시된 경우) - false: 장학금 관련 안내공지 (수혜자 발표, 지급 안내, 제도 소개 등) region 규칙: - 지역 조건이 없으면 반드시 null - 있으면 반드시 위 목록 중 하나만 선택 (임의 형식 사용 금지) """ for attempt in range(max_retry): try: res = gemini.generate_content(prompt) result = json.loads(res.text) return result except Exception as e: print(f" 장학금 파싱 실패 (시도 {attempt+1}/{max_retry}): {e}") time.sleep(3) return None # ── 기숙사 파싱 ─────────────────────────────────────────────── def parse_dormitory(title: str, body: str, max_retry: int = 3) -> dict: prompt = f""" 아래 한성대학교 기숙사 공지를 분석해서 JSON으로 반환해. 공지 제목: {title} 공지 본문: {(body or '')[:500]} 아래 JSON 형식으로만 출력해: {{ "name": "기숙사 이름 (상상빌리지/우촌학사/동소문행복기숙사/에피소드/임대기숙사 중 하나)", "start_date": null 또는 "YYYY-MM-DD", "end_date": null 또는 "YYYY-MM-DD", "end_date_type": "명시" 또는 "미정", "male_quota": null 또는 숫자, "female_quota": null 또는 숫자, "target_grade": [1, 2, 3, 4] 또는 특정 학년만, "extra_info": "기타 중요 정보 한 줄 요약" }} """ for attempt in range(max_retry): try: res = gemini.generate_content(prompt) result = json.loads(res.text) return result except Exception as e: print(f" 기숙사 파싱 실패 (시도 {attempt+1}/{max_retry}): {e}") time.sleep(3) return None # ── Two-Tower 모델 로드 ─────────────────────────────────────── def load_model(): device = torch.device('cpu') sbert = SentenceTransformer(BASE_MODEL_EMBED, device="cpu") HIDDEN_DIM = 256 OUTPUT_DIM = 128 SCORE_DIM = 16 class TwoTowerModel(nn.Module): def __init__(self): super().__init__() self.user_tower = nn.Sequential( nn.Linear(768, HIDDEN_DIM), nn.ReLU(), nn.Dropout(0.2), nn.Linear(HIDDEN_DIM, OUTPUT_DIM) ) self.item_text_layer = nn.Sequential(nn.Linear(768, HIDDEN_DIM), nn.ReLU()) self.item_score_layer = nn.Sequential(nn.Linear(1, SCORE_DIM), nn.ReLU()) self.item_final_layer = nn.Sequential( nn.Dropout(0.2), nn.Linear(HIDDEN_DIM + SCORE_DIM, OUTPUT_DIM) ) def forward_item(self, x): text_emb = x[:, :-1] score_emb = x[:, -1:].float() text_vec = self.item_text_layer(text_emb) score_vec = self.item_score_layer(score_emb) return F.normalize(self.item_final_layer( torch.cat([text_vec, score_vec], dim=-1)), dim=-1) model = TwoTowerModel().to(device) if os.path.exists(TWO_TOWER_MODEL_PATH): model.load_state_dict(torch.load(TWO_TOWER_MODEL_PATH, map_location=device)) model.eval() return sbert, model, device # ── 임베딩 계산 ─────────────────────────────────────────────── def calc_embedding(notice: dict, sbert, model, device, notice_score: float) -> list: notice_text = f"{notice.get('category','')} {notice.get('title','')} {(notice.get('body') or '')[:100]}" text_emb = sbert.encode([notice_text], convert_to_numpy=True) score_norm = min(notice_score / 5.0, 1.0) item_emb = np.concatenate([text_emb[0], [score_norm]]) item_tensor = torch.tensor(item_emb, dtype=torch.float).unsqueeze(0).to(device) with torch.no_grad(): item_vec = model.forward_item(item_tensor).cpu().numpy()[0] return item_vec.tolist() # ── 장학금 테이블 저장 ──────────────────────────────────────── def save_scholarship(notice_id: str, title: str, body: str, posted_at: str): print(f" → 장학금 파싱 중...") result = parse_scholarship(title, body) if not result: print(f" → 장학금 파싱 실패") return end_date = result.get('end_date') if end_date: year = (posted_at or '2026')[:4] end_date = f"{year}-{end_date}" try: supabase.table("scholarships").insert({ "notice_id": notice_id, "type": result.get('type'), "income_max": result.get('income_max'), "income_required": result.get('income_required', False), "income_priority_max": result.get('income_priority_max'), "end_date": end_date, "end_date_type": result.get('end_date_type', '미정'), "target_status": json.dumps(result.get('target_status', ['재학']), ensure_ascii=False), "target_grade": json.dumps(result.get('target_grade', [1,2,3,4]), ensure_ascii=False), "min_gpa": result.get('min_gpa'), "region": result.get('region'), "extra_info": result.get('extra_info'), "is_application": result.get('is_application', False), }).execute() is_app = result.get('is_application', False) print(f" → 장학금 저장 완료 (is_application={is_app})") except Exception as e: print(f" → 장학금 저장 실패: {e}") # ── 기숙사 테이블 저장 ──────────────────────────────────────── def save_dormitory(notice_id: str, title: str, body: str): print(f" → 기숙사 파싱 중...") result = parse_dormitory(title, body) if not result: print(f" → 기숙사 파싱 실패") return try: supabase.table("dormitories").insert({ "notice_id": notice_id, "name": result.get('name'), "start_date": result.get('start_date'), "end_date": result.get('end_date'), "end_date_type": result.get('end_date_type', '미정'), "male_quota": result.get('male_quota'), "female_quota": result.get('female_quota'), "target_grade": json.dumps(result.get('target_grade', [1,2,3,4]), ensure_ascii=False), "extra_info": result.get('extra_info'), }).execute() print(f" → 기숙사 저장 완료 (관리자 확인 필요)") except Exception as e: print(f" → 기숙사 저장 실패: {e}") # ── 메인 처리 함수 ──────────────────────────────────────────── def process_notices(notices: list[dict]) -> int: print(f"공지 처리 시작: {len(notices)}건") sbert, model, device = load_model() print("모델 로드 완료!") success = 0 failed = 0 for i, notice in enumerate(notices): try: notice = clean_for_postgres(dict(notice)) title = notice.get('title', '') body = notice.get('body', '') or '' url = notice.get('url', '') posted_at = str(notice.get('posted_at') or '') print(f"\n[{i+1}/{len(notices)}] {title[:45]}") # 1. Gemini 분류 classification = classify_with_gemini(title, body) category = classification.get('category', '기타') category_type = clean_for_postgres(classification.get('category_type', [])) notice['category'] = category # 2. notice_score 계산 notice_score = calc_notice_score(notice) # 3. 임베딩 계산 embedding = calc_embedding(notice, sbert, model, device, notice_score) # 4. notice_id 추출 notice_id_match = re.search(r'/(\d+)/artclView\.do', url) notice_id = notice_id_match.group(1) if notice_id_match else notice.get('notice_id') # 5. Supabase notices 업데이트 payload = clean_for_postgres({ "source": "hansung", "notice_id": notice_id, "title": title, "url": url, "posted_at": posted_at[:10].replace('.', '-') if posted_at else None, "posted_date_text": notice.get('date') or notice.get('posted_date_text'), "category": category, "category_type": category_type, "job_types": category_type, "body": body, "views": notice.get('views', 0), "notice_score": notice_score, "embedding": json.dumps(embedding), "raw": notice, }) supabase.table("notices").upsert(payload, on_conflict="url").execute() print(f" → {category} {category_type} | score:{notice_score}") # 6. 장학금/기숙사 추가 파싱 if category == '장학금' and notice_id: save_scholarship(notice_id, title, body, posted_at) elif category == '기숙사' and notice_id: save_dormitory(notice_id, title, body) success += 1 except Exception as e: print(f" 실패: {e}") import traceback; traceback.print_exc() failed += 1 time.sleep(2) # Gemini API rate limit 방지 print(f"\n처리 완료! 성공: {success}개 | 실패: {failed}개") return success # ── 단독 실행 ───────────────────────────────────────────────── if __name__ == "__main__": # category_type이 없는 신규 공지만 처리 res = supabase.table("notices").select( "id,notice_id,title,url,posted_at,body,views,category" ).is_("category_type", "null").order("posted_at", desc=True).limit(20).execute() # category_type이 빈 배열인 것도 처리 res2 = supabase.table("notices").select( "id,notice_id,title,url,posted_at,body,views,category,category_type" ).order("posted_at", desc=True).limit(100).execute() unprocessed = [ n for n in (res2.data or []) if not n.get('category_type') ] all_notices = list({n['id']: n for n in (res.data or []) + unprocessed}.values()) if all_notices: print(f"미처리 공지 {len(all_notices)}건 발견") process_notices(all_notices) else: print("미처리 공지 없음")