# ============================================================================ # AI 커리큘럼 생성 모듈 — Vertex AI (Gemini 3 Pro Preview) # 방법 1: AI는 우선순위만 결정, Python이 기계적 배치 # ============================================================================ import json import logging import mysql.connector from google.genai import types from app.core.config import get_vertex_client, A_DB_CONFIG, C_DB_CONFIG logger = logging.getLogger(__name__) VERTEX_MODEL = "gemini-3-pro-preview" # 블록·일수 설정: 6개월 = block 7~12, 각 block 20일 BLOCKS = list(range(7, 13)) # [7, 8, 9, 10, 11, 12] DAYS_PER_BLOCK = 20 # 각 블록당 일수 TOTAL_DAYS = len(BLOCKS) * DAYS_PER_BLOCK # 120일 # -------------------------------------------------------------------------- # 데이터 수집 헬퍼 # -------------------------------------------------------------------------- def _fetch_user_profile(session_id: str) -> dict: """A_DB.user_profile_summary에서 유저 프로필 조회""" conn = mysql.connector.connect(**A_DB_CONFIG) cursor = conn.cursor(dictionary=True) try: cursor.execute(""" SELECT session_id, username, final_goal, mbti, spi_language, spi_nonverbal, spi_personality, studyplan_duration, studyplan_status, studyplan_weekday_hours, studyplan_weekend_hours, studyplan_time_slot, it_level, ai_analysis FROM user_profile_summary WHERE session_id = %s """, (session_id,)) return cursor.fetchone() or {} finally: cursor.close() conn.close() def _fetch_master_items() -> list: """C_DB.master_items 전체 조회 (AI 참고용)""" conn = mysql.connector.connect(**C_DB_CONFIG) cursor = conn.cursor(dictionary=True) try: cursor.execute(""" SELECT item_id, block_id, name, category, item_type, sort_order FROM master_items ORDER BY block_id, sort_order, item_id """) return cursor.fetchall() finally: cursor.close() conn.close() # -------------------------------------------------------------------------- # 프롬프트 구성 (PJT 순서만 AI가 결정) # -------------------------------------------------------------------------- def _build_prompt(profile: dict, master_items: list) -> str: """AI에게 PJT 아이템의 우선순위만 결정하게 하는 프롬프트""" # --- 유저 프로필 요약 (1줄) --- parts = [] if profile.get("final_goal"): parts.append(f"目標:{profile['final_goal']}") if profile.get("mbti"): parts.append(f"MBTI:{profile['mbti']}") if profile.get("it_level"): parts.append(f"IT:{profile['it_level']}") if profile.get("studyplan_weekday_hours"): parts.append(f"平日:{profile['studyplan_weekday_hours']}h") if profile.get("ai_analysis"): parts.append(f"分析:{profile['ai_analysis'][:80]}") user_line = "/".join(parts) if parts else "情報なし" # --- PJT 아이템 ID만 추출 --- pjt_ids = [str(m['item_id']) for m in master_items if m['category'] == 'PJT'] prompt = f"""IT教育カリキュラムPJT教材の学習順序を最適化してください。 CERT（資格）教材は固定のため、PJTのみ並び替えます。 【ユーザー】{user_line} 【PJT ID({len(pjt_ids)}個)】{','.join(pjt_ids)} ルール: - 全{len(pjt_ids)}個のIDを必ず含めること。省略禁止 - 初心者=基礎→応用,中級者=基礎スキップし応用から - MBTIに基づき学習スタイルを最適化（例:ENTJ=効率重視で実践から） - ユーザーの目標に最も関連性の高い分野を前半に配置 出力:1行JSON。改行禁止。{{{{ "pjt":[全{len(pjt_ids)}個のid] }}}}""" return prompt # -------------------------------------------------------------------------- # JSON 파싱 헬퍼 (잘린 응답 복구) # -------------------------------------------------------------------------- def _try_parse_json(text: str) -> dict | None: """잘린 JSON도 최대한 복구하여 파싱""" # 1차: 그대로 파싱 try: result = json.loads(text) if isinstance(result, dict): return result except json.JSONDecodeError: pass # 2차: 닫히지 않은 괄호 닫아주기 for suffix in ["]}", "]}}", "]}}}", "]}}}}"]: try: result = json.loads(text + suffix) if isinstance(result, dict): return result except json.JSONDecodeError: continue # 3차: regex로 pjt 배열 추출 import re pjt_match = re.search(r'"pjt"\s*:\s*\[([0-9,\s]+)', text) if pjt_match: pjt_ids = [int(x.strip()) for x in pjt_match.group(1).split(",") if x.strip().isdigit()] if pjt_ids: logger.info(f"[AI Curriculum] JSON repaired: pjt={len(pjt_ids)}") return {"pjt": pjt_ids} return None # -------------------------------------------------------------------------- # Python 배치 로직: AI가 정한 PJT 순서 + CERT 고정 → 날짜별 배치 # -------------------------------------------------------------------------- def _place_items_to_days(pjt_order: list, all_pjt_ids: set, cert_items_by_day: dict) -> list: """ AI가 정해준 PJT 우선순위 + CERT 고정 순서를 120일에 배치. Args: pjt_order: AI가 정한 PJT ID 우선순위 리스트 all_pjt_ids: 유효한 PJT ID 집합 cert_items_by_day: {(block_id, sort_order): item_id} — CERT 고정 배치 Returns: [(item_id, sort_order), ...] """ # PJT: 유효한 ID만 필터링 (중복 제거) seen = set() pjt_clean = [] for i in pjt_order: if i in all_pjt_ids and i not in seen: pjt_clean.append(i) seen.add(i) # fallback: AI가 빠뜨린 PJT ID를 끝에 추가 missing_pjt = [i for i in sorted(all_pjt_ids) if i not in seen] pjt_clean.extend(missing_pjt) logger.info(f"[AI Curriculum] Placement: PJT={len(pjt_clean)}(+{len(missing_pjt)} fallback), " + f"CERT={len(cert_items_by_day)} items (fixed)") result = [] day_index = 0 for block_id in BLOCKS: for sort_order in range(1, DAYS_PER_BLOCK + 1): # PJT 배치 (AI 순서) if day_index < len(pjt_clean): result.append((pjt_clean[day_index], sort_order)) # CERT 배치 (master_items 고정) cert_id = cert_items_by_day.get((block_id, sort_order)) if cert_id: result.append((cert_id, sort_order)) day_index += 1 return result # -------------------------------------------------------------------------- # AI 호출 & DB 저장 # -------------------------------------------------------------------------- async def generate_ai_curriculum(session_id: str) -> dict: """ Vertex AI로 우선순위만 결정 → Python이 배치 → user_curriculum에 저장 Returns: {"success": bool, "message": str, "item_count": int} """ try: # 1. 데이터 수집 logger.info(f"[AI Curriculum] Fetching data for {session_id}") profile = _fetch_user_profile(session_id) if not profile: return {"success": False, "message": "ユーザープロフィールが見つかりません", "item_count": 0} master_items = _fetch_master_items() # 유효한 item_id 집합 + 카테고리 맵 valid_ids = {m["item_id"] for m in master_items} pjt_ids = {m["item_id"] for m in master_items if m["category"] == "PJT"} # CERT: master_items의 block_id + sort_order 그대로 고정 배치 cert_items_by_day = {} for m in master_items: if m["category"] in ("CERT", "REVIEW"): key = (m["block_id"], m["sort_order"]) cert_items_by_day[key] = m["item_id"] # 2. 프롬프트 구성 & AI 호출 (PJT만) prompt = _build_prompt(profile, master_items) logger.info(f"[AI Curriculum] Calling Vertex AI ({VERTEX_MODEL}) - PJT only mode") logger.info(f"[AI Curriculum] Prompt length: {len(prompt)} chars") logger.info(f"[AI Curriculum] Items: PJT={len(pjt_ids)}, CERT={len(cert_items_by_day)} (fixed)") client = get_vertex_client() response = client.models.generate_content( model=VERTEX_MODEL, contents=[types.Content(role="user", parts=[types.Part(text=prompt)])], config=types.GenerateContentConfig( temperature=0.3, max_output_tokens=4096, ), ) raw_text = response.text.strip() logger.info(f"[AI Curriculum] Raw response length: {len(raw_text)} chars") logger.info(f"[AI Curriculum] Raw text: {raw_text[:300]}") # 3. JSON 파싱 (정리 + 복구) # 마크다운 코드블록 제거 if raw_text.startswith("```"): lines = raw_text.split("\n") lines = [l for l in lines if not l.strip().startswith("```")] raw_text = "\n".join(lines) # 줄바꿈·공백 제거하여 1줄로 raw_text = raw_text.replace("\n", "").replace("\r", "").strip() # 잘린 JSON 복구 시도 ai_result = _try_parse_json(raw_text) if ai_result is None or "pjt" not in ai_result: return {"success": False, "message": "AI応答の形式が不正です（pjtキーが必要）", "item_count": 0} pjt_order = ai_result["pjt"] logger.info(f"[AI Curriculum] AI returned PJT:{len(pjt_order)} items") # 4. Python 배치 로직 (PJT=AI순서, CERT=고정) placements = _place_items_to_days(pjt_order, pjt_ids, cert_items_by_day) if len(placements) < 10: return {"success": False, "message": f"有効な項目が少なすぎます ({len(placements)}件)", "item_count": 0} logger.info(f"[AI Curriculum] Placed {len(placements)} items into schedule") # 5. DB 저장 (기존 커리큘럼 삭제 → 새로 INSERT) conn = mysql.connector.connect(**C_DB_CONFIG) cursor = conn.cursor() try: cursor.execute( "DELETE FROM user_curriculum WHERE session_id = %s", (session_id,), ) deleted = cursor.rowcount logger.info(f"[AI Curriculum] Deleted {deleted} old items") insert_sql = """ INSERT INTO user_curriculum (session_id, item_id, sort_order, status) VALUES (%s, %s, %s, 'NOT_STARTED') """ rows = [(session_id, item_id, sort_order) for item_id, sort_order in placements] cursor.executemany(insert_sql, rows) conn.commit() inserted = cursor.rowcount logger.info(f"[AI Curriculum] Inserted {inserted} new items for {session_id}") return { "success": True, "message": f"カリキュラムが作成されました（{inserted}項目）", "item_count": inserted, } except Exception as db_err: conn.rollback() logger.error(f"[AI Curriculum] DB Error: {db_err}") return {"success": False, "message": f"DB保存エラー: {str(db_err)}", "item_count": 0} finally: cursor.close() conn.close() except json.JSONDecodeError as je: logger.error(f"[AI Curriculum] JSON Parse Error: {je}") logger.error(f"[AI Curriculum] Raw text: {raw_text[:500]}") return {"success": False, "message": "AI応答のJSON解析に失敗しました", "item_count": 0} except Exception as e: logger.error(f"[AI Curriculum] Error: {e}") return {"success": False, "message": f"カリキュラム生成エラー: {str(e)}", "item_count": 0}