""" ============================================================================== Goalskill 데이터 분류 알고리즘 (전역 유틸리티) ============================================================================== INPUT 데이터를 분석하여 Goalskill_DB의 48개 테이블 중 하나로 매칭·저장. 테이블명 패턴: {H|S}{A|B|C|D}_{T|P|R}{I|M} - H/S : 큰 목표(H) vs 작은 목표(S) ← 알고리즘이 판단 - A~D : 파트 ← 호출 시 전달 - T/P/R : 대화/조언/방향성 ← 알고리즘이 판단 - I/M : AI/User ← 호출 시 전달 사용법 (각 파트 라우터에서): from app.services.goalskill_classifier import classify_and_save await classify_and_save( session_id="sess_xxx", sender="M", # "M" = User, "I" = AI part="A", # "A" / "B" / "C" / "D" text="오늘 컨디션이 안 좋아서 쉬고 싶어" ) ============================================================================== """ import re from typing import Literal, Tuple from app.core.config import logger # ============================================================================ # 1. H/S 분류: 시간 범위 키워드 기반 # ============================================================================ # S(작은 목표): 어제~7일 이내의 시간 범위, 컨디션, 오늘/내일 관련 S_KEYWORDS = [ # 한국어 "오늘", "내일", "모레", "어제", "그제", "이번 주", "이번주", "며칠", "하루", "컨디션", "기분", "의지", "진행도", "진척", "몇 시간", "오전", "오후", "저녁", "아침", # 일본어 "今日", "明日", "明後日", "昨日", "一昨日", "今週", "数日", "1日", "コンディション", "気分", "やる気", "進捗", "何時間", "午前", "午後", "夕方", "朝", "体調", "調子", "今回", "次回", "直近", "最近", "この前", # 영어 "today", "tomorrow", "yesterday", "this week", "condition", "mood", "progress", "feeling", ] # H(큰 목표): 장기적 시간 범위, 방향성, 장기 계획 관련 H_KEYWORDS = [ # 한국어 "목표", "장기", "전체", "커리큘럼", "로드맵", "계획", "최종", "취업", "자격증", "경력", "개월", "년", "앞으로", "미래", "전략", "방향", "비전", "포부", "6개월", "1년", "반년", "한달", "다음 달", # 일본어 "目標", "長期", "全体", "カリキュラム", "ロードマップ", "計画", "最終", "就職", "資格", "キャリア", "ヶ月", "年", "将来", "未来", "戦略", "方向", "ビジョン", "抱負", "半年", "1年", "来月", "再来月", # 영어 "goal", "long-term", "career", "curriculum", "roadmap", "certificate", "future", "strategy", "vision", ] def classify_goal_scope(text: str) -> Tuple[Literal["H", "S"], int, int]: """ 텍스트를 분석하여 H(큰 목표) 또는 S(작은 목표) 분류. 판단 기준: - S: 어제~7일 이내 시간 범위, 컨디션, 진행도 관련 - H: 장기적 시간 범위, 큰 목표, 방향성, 커리어 관련 동일한 점수일 경우 S를 기본값으로 (일상적 대화가 더 빈번) Returns: (result, h_score, s_score) """ text_lower = text.lower() s_score = 0 h_score = 0 for kw in S_KEYWORDS: if kw.lower() in text_lower: s_score += 1 for kw in H_KEYWORDS: if kw.lower() in text_lower: h_score += 1 # H 키워드가 더 많으면 H, 그 외엔 S (일상 대화 기본값) if h_score > s_score: return "H", h_score, s_score return "S", h_score, s_score # ============================================================================ # 2. T/P/R 분류: 30개 예문 유사도 기반 # ============================================================================ # T(대화) 관련 30개 예문 T_EXAMPLES = [ # 한국어 (10) "안녕하세요", "오늘 기분이 좋아요", "잘 모르겠어요", "네 알겠습니다", "좀 더 설명해주세요", "그래요?", "어떻게 하면 돼요?", "오늘 뭐 했어요?", "고마워요", "재밌었어요", # 일본어 (10) "こんにちは", "今日は気分がいいです", "よくわかりません", "はい、わかりました", "もう少し説明してください", "そうですか？", "どうすればいいですか？", "今日は何をしましたか？", "ありがとうございます", "楽しかったです", # 영어 (10) "Hello", "I'm feeling good today", "I don't understand", "Yes, I got it", "Can you explain more?", "Really?", "How do I do this?", "What did you do today?", "Thank you", "That was fun", ] # P(조언) 관련 30개 예문 P_EXAMPLES = [ # 한국어 (10) "이렇게 하면 좋을 것 같아요", "먼저 기초부터 배우는 걸 추천해요", "파이썬부터 시작하세요", "매일 1시간씩 공부하면 효과적이에요", "이 부분은 다시 복습하세요", "유튜브 강의를 참고하세요", "실습을 많이 해보세요", "에러가 나면 구글링하세요", "이 자료를 읽어보면 도움이 돼요", "코드를 직접 쳐보면서 배우세요", # 일본어 (10) "こうするといいと思います", "まず基礎から学ぶことをお勧めします", "Pythonから始めてください", "毎日1時間ずつ勉強すると効果的です", "この部分はもう一度復習してください", "YouTube講義を参考にしてください", "実習をたくさんやってみてください", "エラーが出たらググってください", "この資料を読んでみると役に立ちます", "コードを直接書いてみながら学んでください", # 영어 (10) "I think you should try this approach", "I recommend starting with the basics", "Start with Python first", "Studying 1 hour daily is effective", "You should review this part again", "Check out YouTube tutorials", "Practice a lot with hands-on exercises", "Google the error when you get stuck", "Reading this material will help", "Learn by writing code yourself", ] # R(방향성) 관련 30개 예문 R_EXAMPLES = [ # 한국어 (10) "앞으로 백엔드 개발자를 목표로 해봐요", "다음 단계로 데이터베이스를 배워야 해요", "웹 개발 쪽으로 방향을 잡으면 좋겠어요", "이 커리큘럼으로 3개월 안에 완성할 수 있어요", "나중에 AWS도 배우면 좋아요", "풀스택 개발자가 되려면 이런 순서로", "자격증을 먼저 따고 취업 준비를 하세요", "프론트엔드와 백엔드 중 어디에 집중할 건가요", "장기적으로 이런 로드맵을 추천합니다", "최종 목표를 설정하고 역산해서 계획을 세워요", # 일본어 (10) "今後はバックエンド開発者を目指してみましょう", "次のステップとしてデータベースを学ぶ必要があります", "Web開発の方向に進むといいと思います", "このカリキュラムで3ヶ月以内に完成できます", "将来的にAWSも学ぶといいですよ", "フルスタック開発者になるにはこの順序で", "資格を先に取ってから就活準備をしてください", "フロントエンドとバックエンドのどちらに集中しますか", "長期的にこのロードマップをお勧めします", "最終目標を設定して逆算して計画を立てましょう", # 영어 (10) "Let's aim for a backend developer career", "The next step is to learn databases", "I suggest focusing on web development", "You can complete this curriculum in 3 months", "Learning AWS later would be beneficial", "To become a full-stack developer, follow this order", "Get certified first, then prepare for job hunting", "Which would you like to focus on, frontend or backend?", "I recommend this long-term roadmap", "Set your final goal and plan backwards", ] # 각 카테고리의 핵심 키워드 (예문에서 추출한 가중치 키워드) T_KEYWORDS = [ # 인사·감정·단순응답 "안녕", "네", "응", "아니", "모르", "그래", "뭐", "어떻게", "고마", "재미", "힘들", "기분", "좋아", "싫어", "궁금", "こんにちは", "はい", "いいえ", "わかり", "ありがとう", "楽し", "hello", "yes", "no", "thank", "fun", "don't know", "how", "what", "good", "bad", ] P_KEYWORDS = [ # 추천·조언·학습방법 "추천", "좋을", "하세요", "해보세요", "배우세요", "복습", "참고", "실습", "연습", "효과", "도움", "방법", "읽어", "お勧め", "してください", "復習", "参考", "実習", "練習", "効果", "役に立", "方法", "読んで", "recommend", "should", "try", "practice", "review", "effective", "helpful", "suggest", "tip", "advice", ] R_KEYWORDS = [ # 방향·목표·로드맵·장기계획 "목표", "방향", "로드맵", "단계", "커리큘럼", "계획", "앞으로", "장기", "다음", "최종", "취업", "자격증", "목指", "方向", "ロードマップ", "ステップ", "カリキュラム", "計画", "将来", "長期", "次", "最終", "就職", "資格", "goal", "direction", "roadmap", "step", "curriculum", "plan", "career", "next", "future", "long-term", "certificate", ] # ============================================================================ # A' PART 컨디션 점수 계산 알고리즘 # ============================================================================ # 키워드별 점수 변동값 (+: 긍정, -: 부정) CONDITION_SCORE_KEYWORDS = { # --- 긍정 키워드 (일본어) --- "元気": +3, "最高": +5, "爽やか": +3, "体調いい": +3, "よく眠れた": +3, "ぐっすり": +3, "快調": +4, "調子いい": +3, "調子がいい": +3, "万全": +4, "スッキリ": +3, "絶好調": +5, "いいね": +2, "いい感じ": +3, "良い": +2, "悪くない": +1, "まあまあ": +1, "普通": +0, # --- 긍정 키워드 (한국어) --- "좋아": +3, "최고": +5, "상쾌": +3, "컨디션 좋": +3, "잘 잤": +3, "푹 잤": +3, "쾌조": +4, "괜찮": +2, "멀쩡": +2, "건강": +3, # --- 긍정 키워드 (영어) --- "great": +4, "awesome": +5, "refreshed": +3, "well-rested": +3, "energetic": +4, "fine": +2, "good": +2, # --- 부정 키워드 (일본어) --- "疲れ": -3, "眠い": -2, "眠たい": -2, "頭痛": -3, "風邪": -3, "熱": -2, "体調悪い": -4, "調子悪い": -3, "だるい": -2, "しんどい": -3, "辛い": -3, "無気力": -3, "吐き気": -4, "めまい": -3, "倦怠感": -3, "痛い": -3, "痛む": -3, "具合悪い": -3, "気持ち悪い": -3, "寝不足": -3, "微熱": -2, "腹痛": -3, "腰痛": -3, "悪い": -2, "微妙": -1, "イマイチ": -1, "よくない": -2, # --- 부정 키워드 (한국어) --- "피곤": -3, "졸려": -2, "두통": -3, "감기": -3, "열이": -2, "컨디션 안": -3, "힘들": -3, "지쳐": -3, "아프": -3, "무기력": -3, "나른": -2, "구토": -4, "어지러": -3, "권태": -3, "아파": -3, "통증": -3, "속이 안": -3, "몸살": -4, # --- 부정 키워드 (영어) --- "tired": -3, "sleepy": -2, "headache": -3, "sick": -3, "exhausted": -4, "unwell": -3, "sluggish": -2, "nauseous": -4, "pain": -3, "hurt": -3, "sore": -2, "ill": -3, } # 강도 수식어 (배율) INTENSITY_AMPLIFIER = { # 강화 (일본어) "めっちゃ": 1.5, "すごく": 1.5, "とても": 1.3, "本当に": 1.5, "かなり": 1.5, "死ぬほど": 2.0, "超": 1.5, "マジで": 1.5, "非常に": 1.5, "結構": 1.2, "割と": 1.1, # 약화 (일본어) "ちょっと": 0.5, "少し": 0.5, "やや": 0.7, "まあまあ": 0.7, "なんとなく": 0.5, "微妙に": 0.5, # 강화 (한국어) "엄청": 1.5, "진짜": 1.5, "매우": 1.5, "너무": 1.5, "정말": 1.5, "죽을 만큼": 2.0, # 약화 (한국어) "조금": 0.5, "좀": 0.5, "약간": 0.7, "살짝": 0.5, # 영어 "very": 1.5, "extremely": 2.0, "really": 1.5, "super": 1.5, "a bit": 0.5, "slightly": 0.5, "somewhat": 0.7, } # 부정어 패턴 (극성 반전) NEGATION_PATTERNS = [ # 일본어 "ない", "ません", "じゃない", "ではない", "なくて", "なかった", # 한국어 "않", "안 ", "못 ", " 없", # 영어 "not ", "don't", "doesn't", "no ", ] def classify_content_type(text: str) -> Tuple[Literal["T", "P", "R"], float, float, float]: """ 텍스트를 분석하여 T(대화), P(조언), R(방향성) 분류. 2단계 판단: 1) 핵심 키워드 매칭 점수 2) 예문과의 단어 겹침 유사도 두 점수를 합산하여 가장 높은 카테고리를 선택. Returns: (result, t_total, p_total, r_total) """ text_lower = text.lower() # Step 1: 키워드 점수 t_kw_score = sum(1 for kw in T_KEYWORDS if kw.lower() in text_lower) p_kw_score = sum(1 for kw in P_KEYWORDS if kw.lower() in text_lower) r_kw_score = sum(1 for kw in R_KEYWORDS if kw.lower() in text_lower) # Step 2: 예문 유사도 점수 (단어 겹침 비율의 최대값) def word_overlap_score(text: str, examples: list) -> float: text_words = set(re.findall(r'\w+', text.lower())) if not text_words: return 0.0 max_score = 0.0 for ex in examples: ex_words = set(re.findall(r'\w+', ex.lower())) if not ex_words: continue overlap = len(text_words & ex_words) score = overlap / max(len(text_words), len(ex_words)) max_score = max(max_score, score) return max_score t_sim_score = word_overlap_score(text, T_EXAMPLES) p_sim_score = word_overlap_score(text, P_EXAMPLES) r_sim_score = word_overlap_score(text, R_EXAMPLES) # 합산 (키워드 점수 * 2 + 유사도 * 10 으로 가중치 조정) t_total = t_kw_score * 2 + t_sim_score * 10 p_total = p_kw_score * 2 + p_sim_score * 10 r_total = r_kw_score * 2 + r_sim_score * 10 # 가장 높은 점수의 카테고리 선택 (동점이면 T 기본값) scores = {"T": t_total, "P": p_total, "R": r_total} result = max(scores, key=scores.get) return result, t_total, p_total, r_total # ============================================================================ # 3. A/B 파트 전용: 커리큘럼 생성 관련성 판별 # ============================================================================ # ---------- progress: 어제까지의 커리큘럼 진행도 ---------- CURRICULUM_PROGRESS_KEYWORDS = [ # 한국어 "진행도", "진척도", "진척", "진행", "어디까지", "완료", "끝냈", "어제 배운", "어제 했", "어제 공부", "지난번", "저번", "몇 퍼센트", "몇 챕터", "몇 단원", "몇 강", "몇 과", "복습", "다시 보", "다시 했", "마무리", "이어서", "과제", "숙제", "제출", "미완료", "미흡", # 일본어 "進捗", "進行度", "どこまで", "完了", "終わった", "終えた", "昨日学んだ", "昨日やった", "昨日勉強", "前回", "先日", "何パーセント", "何章", "何課", "何講", "復習", "もう一度", "やり直し", "仕上げ", "続き", "課題", "宿題", "提出", "未完了", "不十分", # 영어 "progress", "how far", "completed", "finished", "yesterday learned", "yesterday studied", "last time", "previous", "percent", "chapter", "lesson", "lecture", "review", "redo", "continue", "incomplete", "unfinished", "assignment", "homework", "submitted", ] # ---------- condition: 오늘 유저의 컨디션 ---------- CURRICULUM_CONDITION_KEYWORDS = [ # 한국어 "컨디션", "몸 상태", "체력", "피곤", "졸려", "졸리", "아프", "아파", "두통", "감기", "열이", "배탈", "잠을 못", "잠이 부족", "수면", "밤새", "야근", "좋아요", "괜찮아", "안 좋", "힘들", "지쳐", "지치", "기분", "상쾌", "무기력", "나른", "멀쩡", "좋아", "최고", "쾌조", "괜찮", "건강", # 일본어 "コンディション", "体調", "体力", "疲れ", "眠い", "眠たい", "痛い", "頭痛", "風邪", "熱がある", "お腹", "眠れなかった", "睡眠不足", "睡眠", "徹夜", "残業", "元気", "大丈夫", "調子が悪い", "辛い", "疲れた", "気分", "爽やか", "だるい", "無気力", "しんどい", "最高", "体調いい", "よく眠れた", "ぐっすり", "快調", "調子いい", "万全", "スッキリ", "絶好調", "いいね", "いい感じ", "良い", "悪くない", "まあまあ", "普通", "悪い", "微妙", "イマイチ", "よくない", # 영어 "condition", "physical", "stamina", "tired", "sleepy", "sick", "headache", "cold", "fever", "stomachache", "couldn't sleep", "lack of sleep", "sleep", "all night", "overtime", "feeling good", "fine", "not well", "exhausted", "mood", "refreshed", "lethargic", "sluggish", "good", "great", "awesome", "well-rested", "energetic", "bad", "okay", "ok", "hurt", "pain" ] # ---------- passion: 오늘 유저의 열정/의지 ---------- CURRICULUM_PASSION_KEYWORDS = [ # 한국어 "의지", "열정", "동기", "의욕", "하고 싶", "열심히", "도전", "집중", "몰입", "흥미", "관심", "하기 싫", "귀찮", "포기", "그만", "쉬고 싶", "빨리 하고", "많이 하고", "조금만", "적당히", "자신감", "확신", "불안", "걱정", "두려", "해볼게", "해보겠", "할 수 있", "못 하겠", # 일본어 "やる気", "意欲", "情熱", "モチベーション", "やりたい", "頑張", "挑戦", "集中", "没頭", "興味", "関心", "やりたくない", "面倒", "諦め", "やめたい", "休みたい", "早くやりたい", "たくさん", "少しだけ", "ほどほど", "自信", "確信", "不安", "心配", "怖い", "やってみます", "できる", "できない", # 영어 "motivation", "passion", "willing", "eager", "want to", "hard", "challenge", "focus", "immerse", "interest", "curious", "don't want", "lazy", "give up", "quit", "want to rest", "confidence", "sure", "anxious", "worried", "scared", "I'll try", "I can", "I can't", ] # ---------- comprehension: 어제 내용에 대한 유저의 이해도 ---------- CURRICULUM_COMPREHENSION_KEYWORDS = [ # 한국어 "이해", "이해했", "이해 못", "이해가 안", "알겠", "모르겠", "기억", "기억 안", "기억나", "까먹", "잊어", "잊었", "어려웠", "어려워", "쉬웠", "쉬워", "헷갈", "혼란", "개념", "원리", "뜻", "의미", "왜 그런지", "다시 설명", "한번 더", "정리", "요약", "맞나요", "틀렸", "정답", "오답", "점수", "확실히", "확실하지 않", "애매", "불확실", # 일본어 "理解", "理解した", "理解できない", "わかった", "わからない", "覚えて", "覚えていない", "思い出せ", "忘れた", "忘れて", "難しかった", "難しい", "簡単だった", "簡単", "混乱", "混同", "概念", "原理", "意味", "なぜ", "もう一度説明", "もう一回", "整理", "まとめ", "合ってますか", "間違い", "正解", "不正解", "点数", "確実", "確実じゃない", "曖昧", "不確実", # 영어 "understand", "understood", "don't understand", "get it", "don't get", "remember", "don't remember", "forgot", "forget", "difficult", "hard", "easy", "confused", "confusing", "concept", "principle", "meaning", "why", "explain again", "one more time", "summarize", "summary", "correct", "wrong", "score", "grade", "sure about", "not sure", "ambiguous", "uncertain", ] # 카테고리 매핑 _CURRICULUM_CATEGORIES = { "progress": CURRICULUM_PROGRESS_KEYWORDS, "condition": CURRICULUM_CONDITION_KEYWORDS, "passion": CURRICULUM_PASSION_KEYWORDS, "comprehension": CURRICULUM_COMPREHENSION_KEYWORDS, } # 관련성 판정 임계값 (키워드 매칭 합계가 이 값 이상이면 관련 있음) _CURRICULUM_RELEVANCE_THRESHOLD = 1 def is_curriculum_relevant(text: str) -> dict: """ A/B 파트 전용: 오늘의 커리큘럼 생성에 필요한 문장인지 판별. 4개 카테고리로 판별: 1. progress - 어제까지의 커리큘럼 진행도 2. condition - 오늘 유저의 컨디션 3. passion - 오늘 유저의 열정/의지 4. comprehension - 어제 내용에 대한 유저의 이해도 키워드 매칭 합계가 임계값 이상이면 관련 있음으로 판정. 여러 카테고리에 매칭되면 가장 점수가 높은 카테고리를 대표로 선택. Returns: { "is_relevant": True/False, "category": "progress" | "condition" | "passion" | "comprehension" | None, "scores": {"progress": 0, "condition": 2, ...}, "total_score": 2 } """ text_lower = text.lower() scores = {} for category, keywords in _CURRICULUM_CATEGORIES.items(): score = sum(1 for kw in keywords if kw.lower() in text_lower) scores[category] = score total_score = sum(scores.values()) is_relevant = total_score >= _CURRICULUM_RELEVANCE_THRESHOLD # 가장 높은 점수의 카테고리 선택 best_category = None if is_relevant: best_category = max(scores, key=scores.get) logger.info( f"[Curriculum Relevance] " f"is_relevant={is_relevant}, " f"category={best_category}, " f"scores={scores}, total={total_score}" ) return { "is_relevant": is_relevant, "category": best_category, "scores": scores, "total_score": total_score, } # ============================================================================ # 4. 최종 분류 함수 (전역 유틸리티) # ============================================================================ def classify_table( sender: Literal["I", "M"], part: Literal["A", "B", "C", "D"], text: str ) -> dict: """ 4차원 분류를 수행하여 최종 테이블명과 각 점수를 반환. Args: sender: "I" (AI) 또는 "M" (User) part: "A" / "B" / "C" / "D" text: 저장할 텍스트 내용 Returns: { "table_name": "SA_TM", "goal_scope": "S", "content_type": "T", "h_score": 0, "s_score": 2, "t_score": 12.5, "p_score": 4.0, "r_score": 2.0, } """ goal_scope, h_score, s_score = classify_goal_scope(text) content_type, t_score, p_score, r_score = classify_content_type(text) table_name = f"{goal_scope}{part}_{content_type}{sender}" logger.info( f"[Goalskill Classifier] " f"sender={sender}, part={part}, " f"goal_scope={goal_scope}(H={h_score}/S={s_score}), " f"content_type={content_type}(T={t_score:.1f}/P={p_score:.1f}/R={r_score:.1f}) " f"→ table={table_name}" ) return { "table_name": table_name, "goal_scope": goal_scope, "content_type": content_type, "h_score": h_score, "s_score": s_score, "t_score": t_score, "p_score": p_score, "r_score": r_score, } def calculate_condition_score(text: str) -> int | None: """ 텍스트에서 컨디션 관련 키워드를 감지하고 1~10점 점수를 산출. 컨디션 키워드가 없으면 None 반환 (컨디션 문장이 아님). 알고리즘: 1. base_score = 5 (보통) 2. 각 키워드 매칭 → 변동값 적용 3. 강도 수식어가 키워드 앞에 있으면 배율 적용 4. 부정어가 키워드 뒤에 있으면 극성 반전 5. 최종 점수를 1~10으로 클램핑 Returns: int (1~10): 컨디션 점수 None: 컨디션 관련 키워드가 없는 경우 """ text_lower = text.lower() base_score = 5.0 total_delta = 0.0 matched_count = 0 for keyword, delta in CONDITION_SCORE_KEYWORDS.items(): kw_lower = keyword.lower() # 키워드가 텍스트에 포함되어 있는지 확인 if kw_lower not in text_lower: continue matched_count += 1 current_delta = float(delta) # --- Step 1: 강도 수식어 체크 (키워드 앞에 있는지) --- kw_pos = text_lower.find(kw_lower) before_text = text_lower[:kw_pos] # 키워드 앞부분 amplifier = 1.0 for amp_word, amp_value in INTENSITY_AMPLIFIER.items(): if amp_word.lower() in before_text: amplifier = amp_value break # 가장 먼저 매칭된 수식어 1개만 적용 # --- Step 2: 부정어 체크 (키워드 뒤 10글자 이내) --- after_start = kw_pos + len(kw_lower) after_near = text_lower[after_start:after_start + 10] is_negated = False for neg_pattern in NEGATION_PATTERNS: if neg_pattern.lower() in after_near: is_negated = True break # --- Step 3: 최종 변동값 계산 --- if is_negated: current_delta = -current_delta # 극성 반전 current_delta = current_delta * amplifier total_delta += current_delta logger.debug( f"[Condition Score] keyword='{keyword}', delta={delta}, " f"amplifier={amplifier}, negated={is_negated}, " f"final_delta={current_delta}" ) # コンディションキーワードが1つもなければ None if matched_count == 0: return None # 최종 점수 계산 (1~10 클램핑) raw_score = base_score + total_delta final_score = max(1, min(10, round(raw_score))) logger.info( f"[Condition Score Result] " f"matched={matched_count}, total_delta={total_delta:.1f}, " f"raw={raw_score:.1f}, final={final_score}" ) return final_score # ============================================================================ # 4. 분류 + 저장 통합 함수 (각 파트에서 호출) # ============================================================================ from app.models.goalskill_module import save_to_goalskill_table, save_admin_log, save_to_goalskill_table_score # ============================================================================ # Gemini 1~10 스코어링 (B파트 전용) # ============================================================================ def gemini_score_1to10(text: str, source_type: str, session_id: str = None) -> int: """ Gemini API로 텍스트를 1~10으로 수치화. source_type: 'learning' (어제 학습 자기평가) | 'mindset' (오늘 각오) learning의 경우, 어제 커리큘럼 내용을 기준으로 비교 판단. 실패 시 5(중립) 반환. """ from app.core.config import get_gemini_model import mysql.connector from app.core.config import get_db_config # ------------------------------------------------------------------ # learning: 어제 커리큘럼 내용을 가져와서 기준으로 사용 # ------------------------------------------------------------------ yesterday_curriculum = None if source_type == "learning" and session_id: try: # 1. result 테이블에서 오늘 일차(daily) 가져오기 gs_config = get_db_config("Goalskill_DB") conn = mysql.connector.connect(**gs_config) cursor = conn.cursor(dictionary=True) cursor.execute( "SELECT daily FROM result WHERE session_id = %s ORDER BY created_at DESC LIMIT 1", (session_id,) ) row = cursor.fetchone() conn.close() if row and row["daily"] and row["daily"] > 1: yesterday_day = row["daily"] - 1 # 2. C_DB에서 어제 커리큘럼 아이템 가져오기 from app.models.curriculum_module import get_today_item_info item_info = get_today_item_info(session_id, yesterday_day) if item_info and item_info.get("name"): yesterday_curriculum = item_info["name"] logger.info(f"[Gemini Score] 어제 커리큘럼 (Day {yesterday_day}): {yesterday_curriculum}") except Exception as e: logger.warning(f"[Gemini Score] 커리큘럼 조회 실패: {e}") # ------------------------------------------------------------------ # 프롬프트 생성 (구체적인 구분) # ------------------------------------------------------------------ if source_type == "learning" and yesterday_curriculum: context = ( f"昨日のカリキュラム内容: 「{yesterday_curriculum}」\n\n" "上記のカリキュラムに対するユーザーの理解度・学習達成度を1~10で判定してください。\n" "1~2=全く理解できていない(学習していない・サボった)\n" "3~4=少し理解した(表面的な理解・不十分)\n" "5~6=普通(基本的な概念は理解)\n" "7~8=よく理解した(応用もある程度可能)\n" "9~10=完全に理解(自主的な追加学習・深い理解)" ) elif source_type == "learning": context = ( "ユーザーが昨日の学習をどれだけ誠実に取り組んだかを1~10で細かく判定してください。\n" "1~2=全くやっていない(完全にサボった)\n" "3~4=少しだけやった(不十分)\n" "5~6=普通(最低限はやった)\n" "7~8=しっかりやった(真面目に取り組んだ)\n" "9~10=非常に頑張った(自主的に追加学習も行った)" ) else: context = ( "ユーザーの今日の学習に対する意気込み・覚悟を1~10で細かく判定してください。\n" "1~2=全くやる気がない(諦め・面倒)\n" "3~4=低い(消極的・不安)\n" "5~6=普通(特に強い感情なし)\n" "7~8=高い(積極的・意欲的)\n" "9~10=非常に高い(強い決意・情熱的)" ) prompt = f"""以下のユーザーメッセージを分析して、数値のみを返してください。 【評価基準】 {context} 【ユーザーメッセージ】 {text} 【重要な注意事項】 - 1から10の整数を1つだけ出力してください - 極端な値(1や10)は避け、細かい数値で判定してください - 文脈・ニュアンス・否定表現を慎重に考慮してください - 説明や文章は一切不要です。数字のみ出力してください""" try: client = get_gemini_model() response = client.models.generate_content( model="gemini-2.5-flash", contents=prompt, config={"temperature": 0.1} ) result_text = response.text.strip() digits = ''.join(c for c in result_text if c.isdigit()) if digits: score = int(digits) score = max(1, min(10, score)) logger.info(f"[Gemini Score] source={source_type}, score={score}") return score else: logger.warning(f"[Gemini Score] 숫자 추출 실패: '{result_text}' → 기본값 5") return 5 except Exception as e: logger.error(f"[Gemini Score] API 에러: {e} → 기본값 5") return 5 async def classify_and_save( session_id: str, sender: Literal["I", "M"], part: Literal["A", "B", "C", "D"], text: str, source_type: str = None, ) -> dict: """ INPUT 데이터를 분류하고 Goalskill_DB의 해당 테이블에 저장. T/P/R 모든 타입에서 실제 텍스트를 output에 저장. 동시에 admin 테이블에 분류 점수 로그를 기록. ★ A/B 파트: 커리큘럼 생성에 필요한 문장만 저장 (필터링) ★ C/D 파트: 기존대로 모든 INPUT 저장 Args: session_id: 유저 세션 ID sender: "I" (AI) 또는 "M" (User) part: "A" / "B" / "C" / "D" text: 저장할 텍스트 내용 Returns: 분류 결과 및 점수 정보를 포함한 dict """ result = classify_table(sender, part, text) table_name = result["table_name"] goal_scope = result["goal_scope"] content_type = result["content_type"] # ------------------------------------------------------------------ # ★ A/B 파트 전용: 커리큘럼 생성 관련성 필터링 # ------------------------------------------------------------------ relevance = None if part in ("A", "B"): relevance = is_curriculum_relevant(text) if not relevance["is_relevant"]: # 커리큘럼 생성과 무관한 문장 → 저장하지 않음 logger.info( f"[Goalskill Skip] part={part}, sender={sender}, " f"커리큘럼 관련성 없음 → 저장 스킵" ) # admin 로그에는 스킵된 것도 기록 (추적 용도) save_admin_log( session_id=session_id, sender=sender, part=part, input_text=text, h_scoring=result["h_score"], s_scoring=result["s_score"], t_scoring=result["t_score"], p_scoring=result["p_score"], r_scoring=result["r_score"], result_table=f"SKIP_{table_name}", ) return { "table_name": table_name, "goal_scope": goal_scope, "content_type": content_type, "sender": sender, "part": part, "session_id": session_id, "saved": False, "skip_reason": "not_curriculum_relevant", "relevance": relevance, "scores": { "H": result["h_score"], "S": result["s_score"], "T": result["t_score"], "P": result["p_score"], "R": result["r_score"], }, } # ------------------------------------------------------------------ # 컨디션 점수 계산 # ------------------------------------------------------------------ condition_score = calculate_condition_score(text) # ------------------------------------------------------------------ # 저장 실행 (A/B 관련 문장 또는 C/D 모든 문장) # ------------------------------------------------------------------ try: # ★ B파트 + source_type → Gemini 1~10 점수 저장 (동료 코드) status_score = None if part == "B" and source_type in ("learning", "mindset"): status_score = gemini_score_1to10(text, source_type, session_id=session_id) save_to_goalskill_table_score(table_name, session_id, text, source_type, status_score) else: # A 파트: 컨디션 점수를 status에 저장 / C·D 파트: status 없이 저장 save_to_goalskill_table(table_name, session_id, text, status=condition_score if part == "A" else None) # admin 테이블에 분류 점수 로그 저장 save_admin_log( session_id=session_id, sender=sender, part=part, input_text=text, h_scoring=result["h_score"], s_scoring=result["s_score"], t_scoring=result["t_score"], p_scoring=result["p_score"], r_scoring=result["r_score"], result_table=table_name, ) logger.info( f"[Goalskill Save] table={table_name}, " f"session={session_id}, sender={sender}" + (f", source={source_type}, status={status_score}" if status_score else "") ) response = { "table_name": table_name, "goal_scope": goal_scope, "content_type": content_type, "sender": sender, "part": part, "session_id": session_id, "saved": True, "condition_score": condition_score, "scores": { "H": result["h_score"], "S": result["s_score"], "T": result["t_score"], "P": result["p_score"], "R": result["r_score"], }, } if status_score is not None: response["source_type"] = source_type response["status"] = status_score # A/B 파트이면 relevance 정보도 추가 if relevance is not None: response["relevance"] = relevance return response except Exception as e: logger.error(f"[Goalskill Save Error] table={table_name}: {e}") return { "table_name": table_name, "goal_scope": goal_scope, "content_type": content_type, "sender": sender, "part": part, "session_id": session_id, "saved": False, "error": str(e), } def classify_only( sender: Literal["I", "M"], part: Literal["A", "B", "C", "D"], text: str ) -> dict: """ 분류만 수행 (저장하지 않음). 디버그/테스트용. 점수도 함께 반환합니다. """ result = classify_table(sender, part, text) return { "table_name": result["table_name"], "goal_scope": result["goal_scope"], "content_type": result["content_type"], "sender": sender, "part": part, "scores": { "H": result["h_score"], "S": result["s_score"], "T": result["t_score"], "P": result["p_score"], "R": result["r_score"], }, }