# ============================================================ # app.py — 상상파인더 UI # ============================================================ import os, re, json, hashlib, warnings from datetime import datetime import html import streamlit as st from recommend import ( load_notices_from_supabase, two_tower_recommend, classify_job_type, get_supabase, ) from api.services.search_service import hybrid_search as pinecone_hybrid_search, _profile_summary try: from dotenv import load_dotenv load_dotenv() except ImportError: pass warnings.filterwarnings("ignore") os.environ["TRANSFORMERS_VERBOSITY"] = "error" # ── 경로 설정 ───────────────────────────────────────────────── _BASE_DIR = os.path.dirname(os.path.abspath(__file__)) GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") os.makedirs(os.path.join(_BASE_DIR, "data"), exist_ok=True) # ── 신버전 카테고리 ─────────────────────────────────────────── CATEGORIES = [ "취업/채용", "학사행정", "학생활동/비교과", "대외활동", "공모전/경진대회", "국제교류", "창업", "장학금", "기숙사", "ROTC" ] PINECONE_CATEGORY_MAP = { "취업/채용": ["취업/채용", "인턴십"], "학사행정": ["학사행정"], "학생활동/비교과": ["학생활동/비교과", "비교과", "교육/특강"], "대외활동": ["대외활동", "봉사/서포터즈"], "공모전/경진대회": ["공모전/경진대회"], "국제교류": ["국제교류"], "창업": ["창업"], "장학금": ["장학금", "학자금/근로장학"], "기숙사": ["기숙사", "기숙사/생활관"], "ROTC": ["ROTC"], } COLLEGE_MAP = { "크리에이티브인문예술대학": ["영미문화콘텐츠트랙", "영미언어정보트랙", "한국어교육트랙", "역사문화큐레이션트랙", "역사콘텐츠트랙", "지식정보문화트랙", "디지털인문정보학트랙", "동양화전공", "서양화전공", "한국무용전공", "현대무용전공", "발레전공"], "미래융합사회과학대학": ["국제무역트랙", "글로벌비지니스트랙", "기업ㆍ경제분석트랙", "경제금융투자트랙", "공공행정트랙", "법&정책트랙", "부동산트랙", "스마트도시ㆍ교통계획트랙", "기업경영트랙", "비지니스애널리틱스트랙", "회계ㆍ재무경영트랙"], "디자인대학": ["패션마케팅트랙", "패션디자인트랙", "패션크리에이티브디렉션트랙", "미디어디자인트랙", "시각디자인트랙", "영상ㆍ애니메이션디자인트랙", "UX/UI디자인트랙", "인테리어디자인트랙", "VMDㆍ전시디자인트랙", "게임그래픽디자인트랙", "뷰티디자인매니지먼트학과"], "IT공과대학": ["모바일소프트웨어트랙", "빅데이터트랙", "디지털콘텐츠ㆍ가상현실트랙", "웹공학트랙", "전자트랙", "시스템반도체트랙", "기계시스템디자인트랙", "AI로봇융합트랙", "산업공학트랙", "응용산업데이터공학트랙"], "창의융합대학": ["상상력인재학부", "문학문화콘텐츠학과", "AI응용학과", "융합보안학과", "미래모빌리티학과"], "글로벌인재대학": ["한국언어문화교육학과", "글로벌K비지니스학과", "영상엔터테인먼트학과", "패션뷰티크리에이션학과", "SW융합학과", "글로벌벤처창업학과"], "미래플러스대학": ["융합행정학과", "호텔외식경영학과", "뷰티디자인학과", "비지니스컨설팅학과", "ICT융합디자인학과", "AIㆍ소프트웨어학과", "뷰티매니지먼트학과", "디지털콘텐츠디자인학과", "인테리어디자인학과", "스마트제조혁신컨설팅학과"], } DEPT_URLS = { "한국언어문화교육학과":"https://www.hansung.ac.kr/global/1511/subview.do","글로벌K비지니스학과":"https://www.hansung.ac.kr/global/1516/subview.do","영상엔터테인먼트학과":"https://www.hansung.ac.kr/global/1521/subview.do","패션뷰티크리에이션학과":"https://www.hansung.ac.kr/global/1526/subview.do","SW융합학과":"https://www.hansung.ac.kr/global/1531/subview.do","글로벌벤처창업학과":"https://www.hansung.ac.kr/global/6807/subview.do","융합행정학과":"https://www.hansung.ac.kr/futureplus/731/subview.do","호텔외식경영학과":"https://www.hansung.ac.kr/futureplus/734/subview.do","뷰티디자인학과":"https://www.hansung.ac.kr/futureplus/737/subview.do","비지니스컨설팅학과":"https://www.hansung.ac.kr/futureplus/740/subview.do","ICT융합디자인학과":"https://www.hansung.ac.kr/futureplus/743/subview.do","AIㆍ소프트웨어학과":"https://www.hansung.ac.kr/futureplus/746/subview.do","뷰티매니지먼트학과":"https://www.hansung.ac.kr/futureplus/749/subview.do","디지털콘텐츠디자인학과":"https://www.hansung.ac.kr/futureplus/754/subview.do","인테리어디자인학과":"https://www.hansung.ac.kr/futureplus/759/subview.do","스마트제조혁신컨설팅학과":"https://www.hansung.ac.kr/futureplus/764/subview.do", "상상력인재학부":"https://www.hansung.ac.kr/CreCon/2761/subview.do","문학문화콘텐츠학과":"https://www.hansung.ac.kr/CreCon/2768/subview.do","AI응용학과":"https://www.hansung.ac.kr/CreCon/2777/subview.do","융합보안학과":"https://www.hansung.ac.kr/CreCon/2787/subview.do","미래모빌리티학과":"https://www.hansung.ac.kr/CreCon/2796/subview.do", "국제무역트랙":"https://www.hansung.ac.kr/SclScn/5260/subview.do","글로벌비지니스트랙":"https://www.hansung.ac.kr/SclScn/5267/subview.do","기업ㆍ경제분석트랙":"https://www.hansung.ac.kr/SclScn/5274/subview.do","경제금융투자트랙":"https://www.hansung.ac.kr/SclScn/5281/subview.do","공공행정트랙":"https://www.hansung.ac.kr/SclScn/5295/subview.do","법&정책트랙":"https://www.hansung.ac.kr/SclScn/5303/subview.do","부동산트랙":"https://www.hansung.ac.kr/SclScn/5313/subview.do","스마트도시ㆍ교통계획트랙":"https://www.hansung.ac.kr/SclScn/5321/subview.do","기업경영트랙":"https://www.hansung.ac.kr/SclScn/5328/subview.do","비지니스애널리틱스트랙":"https://www.hansung.ac.kr/SclScn/5336/subview.do","회계ㆍ재무경영트랙":"https://www.hansung.ac.kr/SclScn/5344/subview.do", "모바일소프트웨어트랙":"https://www.hansung.ac.kr/Engineering/4887/subview.do","빅데이터트랙":"https://www.hansung.ac.kr/Engineering/4894/subview.do","디지털콘텐츠ㆍ가상현실트랙":"https://www.hansung.ac.kr/Engineering/4901/subview.do","웹공학트랙":"https://www.hansung.ac.kr/Engineering/4908/subview.do","전자트랙":"https://www.hansung.ac.kr/Engineering/4915/subview.do","시스템반도체트랙":"https://www.hansung.ac.kr/Engineering/4922/subview.do","기계시스템디자인트랙":"https://www.hansung.ac.kr/Engineering/4929/subview.do","AI로봇융합트랙":"https://www.hansung.ac.kr/Engineering/4936/subview.do","산업공학트랙":"https://www.hansung.ac.kr/Engineering/4992/subview.do","응용산업데이터공학트랙":"https://www.hansung.ac.kr/Engineering/5020/subview.do", "패션마케팅트랙":"https://www.hansung.ac.kr/Design/5103/subview.do","패션디자인트랙":"https://www.hansung.ac.kr/Design/5110/subview.do","패션크리에이티브디렉션트랙":"https://www.hansung.ac.kr/Design/5117/subview.do","미디어디자인트랙":"https://www.hansung.ac.kr/Design/5124/subview.do","시각디자인트랙":"https://www.hansung.ac.kr/Design/5145/subview.do","영상ㆍ애니메이션디자인트랙":"https://www.hansung.ac.kr/Design/5131/subview.do","UX/UI디자인트랙":"https://www.hansung.ac.kr/Design/5173/subview.do","인테리어디자인트랙":"https://www.hansung.ac.kr/Design/5159/subview.do","VMDㆍ전시디자인트랙":"https://www.hansung.ac.kr/Design/5152/subview.do","게임그래픽디자인트랙":"https://www.hansung.ac.kr/Design/5166/subview.do","뷰티디자인매니지먼트학과":"https://www.hansung.ac.kr/Design/5180/subview.do", "영미문화콘텐츠트랙":"https://www.hansung.ac.kr/HmnArt/5641/subview.do","영미언어정보트랙":"https://www.hansung.ac.kr/HmnArt/5577/subview.do","한국어교육트랙":"https://www.hansung.ac.kr/HmnArt/5584/subview.do","역사문화큐레이션트랙":"https://www.hansung.ac.kr/HmnArt/5627/subview.do","역사콘텐츠트랙":"https://www.hansung.ac.kr/HmnArt/5634/subview.do","지식정보문화트랙":"https://www.hansung.ac.kr/HmnArt/5613/subview.do","디지털인문정보학트랙":"https://www.hansung.ac.kr/HmnArt/5620/subview.do","동양화전공":"https://www.hansung.ac.kr/HmnArt/5648/subview.do","서양화전공":"https://www.hansung.ac.kr/HmnArt/5655/subview.do","한국무용전공":"https://www.hansung.ac.kr/HmnArt/5662/subview.do","현대무용전공":"https://www.hansung.ac.kr/HmnArt/5669/subview.do","발레전공":"https://www.hansung.ac.kr/HmnArt/5676/subview.do", } # ============================================================ # 유틸 # ============================================================ def _load_image_b64(filename): import base64 try: with open(os.path.join(_BASE_DIR, filename), "rb") as f: return base64.b64encode(f.read()).decode() except: return "" def get_logo_base64(): return _load_image_b64("logo.png") def get_hsu_base64(): return _load_image_b64("hsu.png") # ============================================================ # 모델 로더 # ============================================================ def summarize_notice(title, body): import html as _html body = body or '' clean_body = re.sub(r'<[^>]+>', '', body) result = clean_body[:200] + '...' if len(clean_body) > 200 else clean_body return _html.escape(result) def get_gemini_model(api_key): try: import google.generativeai as genai genai.configure(api_key=api_key) return genai.GenerativeModel("gemini-2.5-flash") except Exception as e: st.error(f"[Gemini 오류] {e}"); return None def _result_content_for_llm(result: dict, body_map: dict[str, str] | None = None) -> str: content = str(result.get("content") or "").strip() if len(content) >= 80: return content if body_map: fallback = str(body_map.get(result.get("url"), "")).strip() if fallback: return fallback return content def generate_llm_reply(user_query, results, profile, is_first=False): model = get_gemini_model(GEMINI_API_KEY) if GEMINI_API_KEY else None if not model: return f"총 {len(results)}개의 관련 공지를 찾았습니다." if results else "관련 공지를 찾지 못했습니다." if not results: return "관련 공지를 찾지 못했습니다. 다른 키워드로 검색해보세요." top_results = results[:3] needs_body_fallback = any(len(str(r.get("content") or "").strip()) < 80 for r in top_results) body_map = {} if needs_body_fallback: notices_data = load_notices_from_supabase() body_map = {n["url"]: n.get("body", "") for n in notices_data} context_parts = [] for i, r in enumerate(top_results, start=1): body = _result_content_for_llm(r, body_map)[:800] context_parts.append( f"[공지 {i}]\n제목: {r.get('title', '')}\n날짜: {r.get('date', '')}\n내용: {body if body else '(본문 없음)'}" ) context = "\n\n".join(context_parts) greeting = f"{profile.get('name','')}님, 안녕하세요. " if is_first else "" route = (results[0].get("query_routing") or {}).get("route") if results else {} personalization = (route or {}).get("personalization") or {} profile_summary = personalization.get("profile_summary") or _profile_summary(profile) personalization_instruction = "" if personalization.get("enabled"): personalization_instruction = ( "\n- 사용자 프로필을 반영한 질문입니다. 각 추천 공지마다 사용자의 학과/학년/관심사/지역과 " "왜 맞는지 또는 지원 가능성과 관련된 근거를 한 줄로 덧붙이세요." ) if profile_summary: personalization_instruction += f"\n- 사용자 프로필: {profile_summary}" prompt = f"""당신은 한성대학교 공지사항 안내 도우미입니다. 아래 공지사항 본문을 바탕으로 사용자 질문에 직접적이고 구체적으로 답변하세요. - 날짜, 금액, 조건 등 구체적인 정보가 있으면 반드시 포함하세요. - "공지를 참고하세요" 같은 말은 절대 하지 마세요. - 2~3문장으로 간결하게 답변하세요. - 답변 시작: "{greeting}" {personalization_instruction} [공지 본문] {context} [질문] {user_query}""" try: return get_gemini_model(GEMINI_API_KEY).generate_content(prompt).text.strip() except Exception as e: return f"[Gemini 오류] {e}" # ============================================================ # CSS # ============================================================ GLOBAL_CSS = """ """ # ============================================================ # 알림 대상자 반환 # ============================================================ def _notification_user_id(profile: dict) -> str: phone = profile.get("phone") or "" seed = phone or profile.get("name") or json.dumps(profile, ensure_ascii=False, sort_keys=True) return hashlib.sha256(seed.encode("utf-8")).hexdigest()[:24] def sync_profile_to_supabase_user(profile: dict) -> None: """로컬 온보딩 프로필을 Supabase users 테이블에 업서트한다.""" if not profile.get("name"): return user_row = { "user_id": _notification_user_id(profile), "name": profile.get("name", ""), "phone": profile.get("phone"), "interests": profile.get("interests") or [], "track": profile.get("track", ""), "college": profile.get("college", ""), "grade": profile.get("grade", ""), "income_level": profile.get("income_level"), "gpa": profile.get("gpa"), "region": profile.get("region"), "loan": profile.get("loan"), "gender": profile.get("gender"), "dorm_interest": profile.get("dorm_interest") or [], "rotc_interest": profile.get("rotc_interest", False), } supabase_url = os.getenv("SUPABASE_URL") supabase_key = os.getenv("SUPABASE_SERVICE_ROLE_KEY") or os.getenv("SUPABASE_KEY") try: if supabase_url and supabase_key and not supabase_url.startswith("your_"): try: get_supabase().table("users").upsert(user_row, on_conflict="user_id").execute() except Exception as rest_error: if os.getenv("SUPABASE_DB_URL"): print(f"Supabase REST users 동기화 실패, DB 직접 연결로 재시도: {rest_error}") _sync_profile_to_supabase_user_via_db(user_row) else: raise else: _sync_profile_to_supabase_user_via_db(user_row) except Exception as e: print(f"Supabase users 동기화 실패: {e}") def _sync_profile_to_supabase_user_via_db(user_row: dict) -> None: """SUPABASE_DB_URL만 있는 로컬 환경에서 users 테이블 생성 후 업서트한다.""" from crawling.supabase_store import ensure_schema, _connect ensure_schema() with _connect() as conn: with conn.cursor() as cur: cur.execute( """ insert into users (user_id, name, phone, interests, track, updated_at) values (%s, %s, %s, %s, %s, now()) on conflict (user_id) do update set name = excluded.name, phone = excluded.phone, interests = excluded.interests, track = excluded.track, updated_at = now() """, ( user_row["user_id"], user_row["name"], user_row["phone"], user_row["interests"], user_row["track"], ), ) conn.commit() def get_notification_targets(notices: list[dict], users: list[dict] | None = None) -> list[dict]: """ 신규 공지 리스트를 받아 알림 대상자 정보 반환. 알림 담당자에게 전달할 데이터 구조: [ { "notice_id": "222025", "title": "공지 제목", "category": "장학금", "category_type": ["장학금"], "url": "https://...", "user_name": "홍길동", "phone": "010-1234-5678", "track": "영미문화콘텐츠트랙", "interests": ["장학금", "취업/채용"], }, ... ] """ if users is None: try: res = get_supabase().table("users").select("user_id,name,phone,interests,track").execute() users = res.data or [] except Exception as e: print(f"users 테이블 로드 오류: {e}") return [] targets = [] for notice in notices: category = notice.get('category', '') category_type = notice.get('category_type') or notice.get('job_types') or [] if isinstance(category_type, str): try: category_type = json.loads(category_type) except: category_type = [] for user in users: if not user or not user.get('phone'): continue interests = user.get('interests', []) if isinstance(interests, str): try: interests = json.loads(interests) except: interests = [interests] interest_set = set(interests) matched = category in interest_set or bool(interest_set.intersection(category_type)) if not matched: continue targets.append({ "notice_db_id": notice.get('id'), "notice_id": notice.get('notice_id', ''), "title": notice.get('title', ''), "category": category, "category_type": category_type, "url": notice.get('url', ''), "user_id": user.get('user_id') or user.get('id') or user.get('phone'), "user_name": user.get('name', ''), "phone": user.get('phone'), "track": user.get('track', ''), "interests": interests, }) return targets # ============================================================ # 필터링 함수 # ============================================================ def filter_scholarships(profile) -> tuple: """(신청 공지 리스트, 관련 공지 리스트) 반환""" try: today = datetime.now().strftime('%Y-%m-%d') income_level = profile.get('income_level') gpa = profile.get('gpa') region = profile.get('region') loan = profile.get('loan') grade = profile.get('grade', '') grade_num = None try: grade_num = int(grade.replace('학년', '').strip()) except: pass income_num = None if income_level and income_level != "모름/해당없음": try: income_num = int(income_level.replace("분위", "")) except: pass gpa_num = None if gpa and gpa != "모름/해당없음": try: gpa_num = float(gpa.split(" ")[0]) except: pass # 신청 공지 필터링 res = get_supabase().table("scholarships").select("*").eq("is_application", True).execute() filtered = [] for s in (res.data or []): target_grade = s.get('target_grade') if isinstance(target_grade, str): try: target_grade = json.loads(target_grade) except: target_grade = [] target_status = s.get('target_status') if isinstance(target_status, str): try: target_status = json.loads(target_status) except: target_status = [] if s.get('end_date_type') == '명시' and s.get('end_date'): if s['end_date'] < today: continue if grade_num and target_grade: if grade_num not in target_grade: continue if target_status and '재학' not in target_status: continue if income_num is not None and s.get('income_max') is not None: if income_num > s['income_max']: continue if gpa_num is not None and s.get('min_gpa') is not None: if gpa_num < s['min_gpa']: continue if s.get('region') is not None and region not in [None, "모름/해당없음"]: if s['region'] != region: continue if s.get('income_required') and loan != "대출 있음": continue filtered.append(s) application_notices = [] if filtered: notice_ids = [s['notice_id'] for s in filtered] notices_res = get_supabase().table("notices").select( "id,notice_id,title,url,posted_at,body,category" ).in_("notice_id", notice_ids).execute() notices_map = {n['notice_id']: n for n in (notices_res.data or [])} for s in filtered: notice = notices_map.get(s['notice_id'], {}) if notice: notice['scholarship_info'] = s application_notices.append(notice) # 관련 공지 최신순 3개 rel_res = get_supabase().table("scholarships").select("notice_id").eq("is_application", False).execute() related_notices = [] if rel_res.data: rel_ids = [r['notice_id'] for r in rel_res.data] rel_notices = get_supabase().table("notices").select( "id,notice_id,title,url,posted_at,body,category" ).in_("notice_id", rel_ids).order("posted_at", desc=True).limit(3).execute() related_notices = rel_notices.data or [] return application_notices, related_notices except Exception as e: print(f"장학금 필터링 오류: {e}") import traceback; traceback.print_exc() return [], [] def filter_dormitory(profile) -> list: try: today = datetime.now().strftime('%Y-%m-%d') gender = profile.get('gender') dorm_interest = profile.get('dorm_interest', []) res = get_supabase().table("dormitories").select("*").execute() filtered = [] for d in (res.data or []): if d.get('end_date_type') == '명시' and d.get('end_date'): if d['end_date'] < today: continue if dorm_interest: if not any(dorm in (d.get('name') or '') for dorm in dorm_interest): continue if gender == '남성' and d.get('male_quota') == 0: continue if gender == '여성' and d.get('female_quota') == 0: continue filtered.append(d) if not filtered: return [] notice_ids = [d['notice_id'] for d in filtered] notices_res = get_supabase().table("notices").select( "id,notice_id,title,url,posted_at,body,category" ).in_("notice_id", notice_ids).execute() notices_map = {n['notice_id']: n for n in (notices_res.data or [])} results = [] for d in filtered: notice = notices_map.get(d['notice_id'], {}) if notice: notice['dormitory_info'] = d results.append(notice) return results except Exception as e: print(f"기숙사 필터링 오류: {e}") import traceback; traceback.print_exc() return [] def filter_rotc() -> list: try: res = get_supabase().table("notices").select( "id,notice_id,title,url,posted_at,body,category" ).eq("category", "ROTC").order("posted_at", desc=True).limit(5).execute() return res.data or [] except Exception as e: print(f"ROTC 필터링 오류: {e}") return [] # ============================================================ # 온보딩 # ============================================================ def render_onboarding(): st.markdown(GLOBAL_CSS, unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) col_logo, col_form = st.columns([1, 2], gap="large") with col_logo: st.markdown("", unsafe_allow_html=True) logo_b64 = get_logo_base64() logo_img = f'