recommend.py

# ============================================================
# recommend.py — Two-Tower 추천 시스템 로직 (v4)
# ============================================================

import os
import re
import json
import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F
from datetime import datetime
from sentence_transformers import SentenceTransformer
from supabase import create_client

import streamlit as st

# ── 경로 설정 ─────────────────────────────────────────────────
_BASE_DIR            = os.path.dirname(os.path.abspath(__file__))
TWO_TOWER_MODEL_PATH = os.path.join(_BASE_DIR, "models", "two_tower_model_v4.pt")
BASE_MODEL_EMBED     = "jhgan/ko-sroberta-multitask"
SUPABASE_URL         = os.getenv("SUPABASE_URL", "your_supabase_url")
SUPABASE_KEY         = os.getenv("SUPABASE_KEY", "your_supabase_key")

# ── 추천 가중치 ───────────────────────────────────────────────
MODEL_WEIGHT    = 0.4
CATEGORY_WEIGHT = 0.4
JOB_TYPE_WEIGHT = 0.1
SCORE_WEIGHT    = 0.1

PENALTY_CATEGORIES     = ['국제교류', '대외활동', '창업']
NO_JOB_TYPE_CATEGORIES = ['국제교류', '학사행정', '대외활동', 'ROTC', '기숙사']
FILTER_CATEGORIES      = ['장학금', '기숙사', 'ROTC']
OLD_CATEGORIES = ['비교과', '교육/특강', '봉사/서포터즈', '인턴십', '인턴쉽', '기타', '학자금/근로장학']

# ── 신버전 카테고리 ───────────────────────────────────────────
VALID_CATEGORIES = [
    '취업/채용', '학사행정', '학생활동/비교과',
    '대외활동', '공모전/경진대회', '국제교류', '창업',
    '장학금', '기숙사', 'ROTC'
]

TRACK_DOMAIN = {
    "IT": [
        "모바일소프트웨어트랙", "빅데이터트랙", "디지털콘텐츠ㆍ가상현실트랙",
        "웹공학트랙", "전자트랙", "시스템반도체트랙", "기계시스템디자인트랙",
        "AI로봇융합트랙", "산업공학트랙", "응용산업데이터공학트랙",
        "AI응용학과", "융합보안학과", "미래모빌리티학과",
        "SW융합학과", "글로벌벤처창업학과",
        "AIㆍ소프트웨어학과", "ICT융합디자인학과", "스마트제조혁신컨설팅학과",
    ],
    "경영": [
        "기업경영트랙", "회계ㆍ재무경영트랙", "경제금융투자트랙",
        "기업ㆍ경제분석트랙", "비지니스애널리틱스트랙",
        "국제무역트랙", "글로벌비지니스트랙",
        "글로벌K비지니스학과", "비지니스컨설팅학과", "호텔외식경영학과",
    ],
    "행정/공공": [
        "공공행정트랙", "법&정책트랙", "부동산트랙",
        "스마트도시ㆍ교통계획트랙", "융합행정학과",
    ],
    "디자인": [
        "패션마케팅트랙", "패션디자인트랙", "패션크리에이티브디렉션트랙",
        "미디어디자인트랙", "시각디자인트랙", "영상ㆍ애니메이션디자인트랙",
        "UX/UI디자인트랙", "인테리어디자인트랙", "VMDㆍ전시디자인트랙",
        "게임그래픽디자인트랙", "뷰티디자인매니지먼트학과",
        "패션뷰티크리에이션학과", "영상엔터테인먼트학과",
        "뷰티디자인학과", "뷰티매니지먼트학과",
        "디지털콘텐츠디자인학과", "인테리어디자인학과",
    ],
    "인문": [
        "영미문화콘텐츠트랙", "영미언어정보트랙", "한국어교육트랙",
        "역사문화큐레이션트랙", "역사콘텐츠트랙", "지식정보문화트랙",
        "디지털인문정보학트랙", "문학문화콘텐츠학과", "한국언어문화교육학과",
    ],
    "예술": [
        "동양화전공", "서양화전공",
        "한국무용전공", "현대무용전공", "발레전공",
    ],
    "융합": ["상상력인재학부"],
}

track_to_domain = {}
for _domain, _tracks in TRACK_DOMAIN.items():
    for _track in _tracks:
        track_to_domain[_track] = _domain
track_to_domain["트랙 미정"] = "융합"
track_to_domain["상상력인재학부 (트랙 미정)"] = "융합"

DOMAIN_TO_JOB_TYPE = {
    "취업/채용": {
        "IT":        "IT/정보통신",
        "경영":      "경영/금융/사무",
        "행정/공공": "교육/법률/공공",
        "디자인":    "디자인/예술/방송",
        "인문":      "교육/법률/공공",
        "예술":      "디자인/예술/방송",
        "융합":      None,
    },
    "학생활동/비교과": {
        "IT":        "IT/AI/SW",
        "경영":      "진로/취업/현장실습",
        "행정/공공": "진로/취업/현장실습",
        "디자인":    "디자인/콘텐츠",
        "인문":      "인문/어학",
        "예술":      "디자인/콘텐츠",
        "융합":      None,
    },
    "공모전/경진대회": {
        "IT":        "IT/AI/SW",
        "경영":      "창업/아이디어",
        "행정/공공": "정책/사회/ESG",
        "디자인":    "디자인/콘텐츠",
        "인문":      "글쓰기/발표/어학",
        "예술":      "디자인/콘텐츠",
        "융합":      None,
    },
    "창업": {
        "IT":        "창업",
        "경영":      "창업",
        "행정/공공": "창업",
        "디자인":    "창업",
        "인문":      "창업",
        "예술":      "창업",
        "융합":      None,
    },
    "국제교류": {
        "IT":        "교환학생/파견",
        "경영":      "교환학생/파견",
        "행정/공공": "교환학생/파견",
        "디자인":    "교환학생/파견",
        "인문":      "외국인학생/글로벌교류",
        "예술":      "교환학생/파견",
        "융합":      "교환학생/파견",
    },
    "대외활동": {
        "IT":        "멘토링",
        "경영":      "서포터즈/홍보대사",
        "행정/공공": "봉사활동",
        "디자인":    "기획/미디어",
        "인문":      "봉사활동",
        "예술":      "기획/미디어",
        "융합":      None,
    },
}

# ============================================================
# Supabase
# ============================================================

@st.cache_resource
def get_supabase():
    return create_client(SUPABASE_URL, SUPABASE_KEY)

@st.cache_data(ttl=300, show_spinner=False)
def load_notices_from_supabase() -> list:
    try:
        supabase  = get_supabase()
        all_data  = []
        page_size = 1000
        offset    = 0
        while True:
            res = supabase.table("notices").select(
                "id,notice_id,title,url,posted_at,posted_date_text,category,body,views,job_types,category_type,notice_score"
            ).order("posted_at", desc=True).range(offset, offset + page_size - 1).execute()
            batch = res.data or []
            if not batch: break
            all_data.extend(batch)
            if len(batch) < page_size: break
            offset += page_size
        for n in all_data:
            if not n.get('notice_score'):
                n['notice_score'] = 0.0
            raw = str(n.get('posted_date_text') or n.get('posted_at', ''))
            raw = re.sub(r'<[^>]+>', '', raw).strip()
            n['date'] = raw[:10].replace('.', '-')
        print(f"Supabase 공지 {len(all_data)}건 로드 완료")
        return all_data
    except Exception as e:
        print(f"Supabase 로드 오류: {e}")
        import traceback; traceback.print_exc()
        return []

@st.cache_data(ttl=300, show_spinner=False)
def load_embeddings_from_supabase() -> dict:
    try:
        supabase  = get_supabase()
        all_embs  = []
        page_size = 1000
        offset    = 0
        while True:
            res = supabase.table("notices").select(
                "id,embedding"
            ).order("posted_at", desc=True).range(offset, offset + page_size - 1).execute()
            batch = res.data or []
            if not batch: break
            all_embs.extend(batch)
            if len(batch) < page_size: break
            offset += page_size

        def parse_embedding(emb):
            if emb is None: return None
            if isinstance(emb, str): emb = json.loads(emb)
            return np.array(emb, dtype=np.float32)

        result = {r['id']: parse_embedding(r.get('embedding')) for r in all_embs}
        valid  = {k: v for k, v in result.items() if v is not None}
        print(f"임베딩 {len(valid)}건 로드 완료")
        return valid
    except Exception as e:
        print(f"임베딩 로드 오류: {e}")
        return {}

# ============================================================
# Two-Tower 모델 (v4)
# ============================================================

@st.cache_resource
def load_two_tower_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_user(self, x):
            return F.normalize(self.user_tower(x), dim=-1)

        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)

        def forward(self, u, i):
            return self.forward_user(u), self.forward_item(i)

    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()
    print("Two-Tower 모델 v4 로드 완료!")
    return sbert, model, device

# ============================================================
# job_type / 도메인
# ============================================================

def get_user_domain(track: str) -> str:
    if "상상력인재학부" in track:
        for t, d in track_to_domain.items():
            if t in track and t != "상상력인재학부":
                return d
        return "융합"
    return track_to_domain.get(track, "융합")

def classify_job_type(notice: dict, threshold: float = 0.35, top_k: int = 2) -> list:
    """Supabase category_type 컬럼 직접 사용"""
    job_types = notice.get('job_types') or notice.get('category_type') or []
    if isinstance(job_types, str):
        try:
            job_types = json.loads(job_types)
        except:
            job_types = []
    return [{'job_type': jt, 'score': 1.0} for jt in job_types[:top_k]]

def get_job_score(track: str, notice: dict) -> float:
    cat = notice.get('category', '')
    if cat in NO_JOB_TYPE_CATEGORIES:
        return 0.0
    notice_types = [t['job_type'] for t in classify_job_type(notice)]
    if not notice_types:
        return 0.0
    domain  = get_user_domain(track)
    user_jt = DOMAIN_TO_JOB_TYPE.get(cat, {}).get(domain)
    if user_jt is None:
        return 0.0
    return 1.0 if user_jt in notice_types else 0.0

# ============================================================
# Two-Tower 추천
# ============================================================

def two_tower_recommend(college, track, year, interests, top_k=10):
    try:
        sbert, model, device = load_two_tower_model()
        notices              = load_notices_from_supabase()
        emb_dict             = load_embeddings_from_supabase()

        if not notices or not emb_dict:
            return []

        # 필터링 카테고리 제외한 관심사만
        rec_interests = [i for i in interests if i not in FILTER_CATEGORIES]

        user_text   = f"{college} {track} {year} 관심사: {', '.join(rec_interests)}"
        user_emb    = sbert.encode([user_text], convert_to_numpy=True)
        user_tensor = torch.tensor(user_emb, dtype=torch.float).to(device)
        with torch.no_grad():
            user_vec = model.forward_user(user_tensor).cpu().numpy()

        scores      = np.array([n.get('notice_score', 0) for n in notices])
        max_score   = scores.max() if scores.max() > 0 else 1
        scores_norm = scores / max_score

        results = []
        for n_idx, notice in enumerate(notices):
            nid      = notice.get('id')
            item_emb = emb_dict.get(nid)
            if item_emb is None:
                continue

            category = notice.get('category', '')

            # 구버전 + 필터링 카테고리 제외
            if category in FILTER_CATEGORIES or category in OLD_CATEGORIES:
                continue
            # notice_score 0인 공지 제외
            if notice.get('notice_score', 0) <= 0.01:
                continue

            n_score   = float(scores_norm[n_idx])
            sim_score = float(np.dot(item_emb, user_vec.T).flatten()[0])
            sim_norm  = (sim_score + 1.0) / 2.0

            if category in rec_interests:
                cat_score = 1.0
            elif category in PENALTY_CATEGORIES:
                cat_score = -1.0
            else:
                cat_score = 0.0

            job_score   = get_job_score(track, notice)
            final_score = (
                MODEL_WEIGHT    * sim_norm  +
                CATEGORY_WEIGHT * cat_score +
                JOB_TYPE_WEIGHT * job_score +
                SCORE_WEIGHT    * n_score
            )
            results.append({
                'notice':      notice,
                'final_score': final_score,
                'sim_score':   sim_norm,
                'cat_score':   cat_score,
                'job_score':   job_score,
                'n_score':     n_score,
            })

        results.sort(key=lambda x: x['final_score'], reverse=True)

        # 관심 카테고리 2개, 비관심 1개 제한
        interest_cat_count     = {}
        non_interest_cat_count = {}
        filtered = []

        for res in results:
            if res['final_score'] < 0.5:
                break
            cat = res['notice'].get('category', '')
            if cat in rec_interests:
                if interest_cat_count.get(cat, 0) < 2:
                    filtered.append(res)
                    interest_cat_count[cat] = interest_cat_count.get(cat, 0) + 1
            else:
                if non_interest_cat_count.get(cat, 0) < 1:
                    filtered.append(res)
                    non_interest_cat_count[cat] = non_interest_cat_count.get(cat, 0) + 1
            if len(filtered) == top_k:
                break

        # 터미널 출력
        print(f"\n{'='*60}")
        print(f"유저: {college} {track} {year}")
        print(f"관심사: {rec_interests}")
        print(f"{'='*60}")
        for i, res in enumerate(filtered):
            n = res['notice']
            print(f"{i+1:2d}. [{n.get('category')}] {n.get('title','')[:40]}")
            print(f"     최종:{res['final_score']:.3f} 모델:{res['sim_score']:.3f} "
                  f"카테고리:{res['cat_score']:.1f} 직무:{res['job_score']:.1f} "
                  f"공지:{res['n_score']:.3f}")

        return filtered

    except Exception as e:
        print(f"Two-Tower 추천 오류: {e}")
        import traceback; traceback.print_exc()
        return []