RAG &
VECTOR DB

┌─────────────────────────────────────────────────────────────┐
│                     데이터 수집 레이어                        │
│                                                             │
│  YouTube Data API v3                                        │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │  영상 정보    │  │   댓글 수집  │  │  대본 .txt 파일  │  │
│  │ title, desc  │  │ 3,000개      │  │  직접 보유분     │  │
│  │ view_count   │  │ 감정/요청 등 │  │                  │  │
│  └──────┬───────┘  └──────┬───────┘  └────────┬─────────┘  │
└─────────┼────────────────┼───────────────────┼─────────────┘
          └───────────────┬┘                   │
                          ↓                    ↓
┌─────────────────────────────────────────────────────────────┐
│                   전처리 & 임베딩 레이어                      │
│                                                             │
│  clean_text() → RecursiveCharacterTextSplitter(size=500)    │
│                          ↓                                  │
│         Gemini text-embedding-004 (무료 1500req/일)          │
│         또는 OpenAI text-embedding-3-small ($0.02/1M)        │
│                          ↓                                  │
│              float32[] 벡터 (768 or 1536차원)                │
└─────────────────────────┬───────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────────────┐
│                     저장 레이어                               │
│                                                             │
│   로컬 개발           │   Vercel 스택      │   무료 온라인   │
│   Chroma DB           │   Turso DB         │   Qdrant Cloud  │
│   (./chroma_db 폴더)  │   (SQL + 벡터 통합) │   (1GB 무료)    │
└─────────────────────────┬───────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────────────┐
│                    검색 & 생성 레이어                         │
│                                                             │
│  질문 입력 → 질문 임베딩 → cosine similarity 검색            │
│           → top-k=5 청크 반환 → LLM 프롬프트 조립           │
│           → GPT-4o-mini / Gemini Flash → 최종 답변          │
│                          ↓                                  │
│  출처: metadata의 video_id → youtube.com/watch?v=...&t=195  │
└─────────────────────────┬───────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────────────┐
│                      UI / 연동 레이어                         │
│                                                             │
│   로컬 테스트         │   기존 스택 통합   │   메신저 연동   │
│   Streamlit           │   Next.js + Vercel │   Slack Bot     │
│   (streamlit run)     │   (App Router API) │   Discord Bot   │
└─────────────────────────────────────────────────────────────┘

# 0_check_env.py — 환경 검증 스크립트
import sys

def check(label: str, fn):
    try:
        result = fn()
        print(f"  ✅ {label}: {result}")
        return True
    except Exception as e:
        print(f"  ❌ {label}: {e}")
        return False

print("\n🔍 환경 검증 시작...\n")

# 1. Python 버전
check("Python 버전", lambda: sys.version.split()[0])

# 2. 필수 패키지 임포트
import importlib
for pkg in ["langchain", "chromadb", "openai", "streamlit", "dotenv"]:
    check(f"패키지 {pkg}", lambda p=pkg: importlib.import_module(p) and "OK")

# 3. .env 로드 + API 키 존재 확인
from dotenv import load_dotenv
import os
load_dotenv()
check("OPENAI_API_KEY 존재",
      lambda: "설정됨 (" + os.getenv("OPENAI_API_KEY","")[:8] + "...)"
              if os.getenv("OPENAI_API_KEY") else (_ for _ in ()).throw(Exception("키 없음")))
check("YOUTUBE_API_KEY 존재",
      lambda: "설정됨 (" + os.getenv("YOUTUBE_API_KEY","")[:8] + "...)"
              if os.getenv("YOUTUBE_API_KEY") else (_ for _ in ()).throw(Exception("키 없음")))

# 4. OpenAI API 실제 연결 테스트 (소량 토큰 소모)
from openai import OpenAI
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
check("OpenAI API 연결",
      lambda: client.embeddings.create(
          model="text-embedding-3-small", input="테스트"
      ).data[0].embedding[:3])

print("\n검증 완료! 모두 ✅이면 다음 단계로 진행하세요.")

# 1_collect_data.py
import os, json, time
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from dotenv import load_dotenv

load_dotenv()
YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY")
CHANNEL_ID = "UCxxxxxxxxxxxxxxxxxxxxxxxx"  # 내 채널 ID
youtube = build("youtube", "v3", developerKey=YOUTUBE_API_KEY)


# ─── 1. 채널의 모든 영상 ID 수집 ───────────────────────────
def get_all_video_ids(channel_id: str) -> list[str]:
    """search.list API → 유닛 소모 100/호출. 50개씩 페이징."""
    video_ids = []
    next_page_token = None
    while True:
        try:
            res = youtube.search().list(
                channelId=channel_id,
                part="id",
                type="video",
                maxResults=50,
                pageToken=next_page_token,
            ).execute()
        except HttpError as e:
            print(f"[ERROR] search.list 실패: {e}")
            break

        for item in res.get("items", []):
            vid = item.get("id", {}).get("videoId")
            if vid:
                video_ids.append(vid)

        next_page_token = res.get("nextPageToken")
        if not next_page_token:
            break
        time.sleep(0.5)   # 초당 호출 제한 방지

    print(f"수집된 영상 ID: {len(video_ids)}개")
    return video_ids


# ─── 2. 영상 상세 정보 수집 (50개 배치) ─────────────────────
def get_video_details(video_ids: list[str]) -> list[dict]:
    """videos.list API → 유닛 소모 1/호출. 50개 배치 처리."""
    videos = []
    for i in range(0, len(video_ids), 50):
        batch = video_ids[i : i + 50]
        try:
            res = youtube.videos().list(
                part="snippet,statistics",
                id=",".join(batch),
            ).execute()
        except HttpError as e:
            print(f"[ERROR] videos.list 배치 {i//50+1} 실패: {e}")
            continue

        for item in res.get("items", []):
            snippet = item.get("snippet", {})
            stats   = item.get("statistics", {})
            videos.append({
                "video_id":     item["id"],
                "title":        snippet.get("title", ""),
                "description":  snippet.get("description", ""),
                "published_at": snippet.get("publishedAt", ""),
                "view_count":   int(stats.get("viewCount",   0)),
                "like_count":   int(stats.get("likeCount",   0)),
                "comment_count":int(stats.get("commentCount",0)),
                "type": "video",
            })
        time.sleep(0.2)

    print(f"수집된 영상 상세: {len(videos)}개")
    return videos


# ─── 3. 댓글 수집 ────────────────────────────────────────────
def get_comments(
    video_ids: list[str],
    max_per_video: int = 20,
) -> list[dict]:
    """
    commentThreads.list → 유닛 소모 1/호출.
    댓글 비활성화 영상은 HttpError 403으로 처리 후 스킵.
    """
    all_comments = []
    for idx, video_id in enumerate(video_ids):
        try:
            res = youtube.commentThreads().list(
                part="snippet",
                videoId=video_id,
                maxResults=max_per_video,
                order="relevance",   # 인기순 (최신순: "time")
            ).execute()
        except HttpError as e:
            # 댓글 사용 중지 영상 → 조용히 스킵
            if e.resp.status in (403, 404):
                continue
            print(f"[ERROR] 댓글 수집 실패 ({video_id}): {e}")
            continue

        for item in res.get("items", []):
            c = item["snippet"]["topLevelComment"]["snippet"]
            all_comments.append({
                "video_id":     video_id,
                "comment":      c.get("textDisplay", ""),
                "author":       c.get("authorDisplayName", ""),
                "like_count":   int(c.get("likeCount", 0)),
                "published_at": c.get("publishedAt", ""),
                "type": "comment",
            })

        if (idx + 1) % 50 == 0:
            print(f"  댓글 수집 진행 중... {idx+1}/{len(video_ids)}")
        time.sleep(0.3)

    print(f"수집된 댓글: {len(all_comments)}개")
    return all_comments


# ─── 4. 실행 및 저장 ─────────────────────────────────────────
if __name__ == "__main__":
    os.makedirs("data/raw", exist_ok=True)

    # (1) 영상 ID 수집
    video_ids = get_all_video_ids(CHANNEL_ID)

    # (2) 영상 상세 정보 저장
    videos = get_video_details(video_ids)
    with open("data/raw/videos.json", "w", encoding="utf-8") as f:
        json.dump(videos, f, ensure_ascii=False, indent=2)
    print("→ data/raw/videos.json 저장 완료")

    # (3) 댓글 저장
    comments = get_comments(video_ids)
    with open("data/raw/comments.json", "w", encoding="utf-8") as f:
        json.dump(comments, f, ensure_ascii=False, indent=2)
    print("→ data/raw/comments.json 저장 완료")

    print(f"\n✅ 수집 완료: 영상 {len(videos)}개 | 댓글 {len(comments)}개")

# 1b_collect_community.py
import os, json, time
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from dotenv import load_dotenv

load_dotenv()
youtube = build("youtube", "v3", developerKey=os.getenv("YOUTUBE_API_KEY"))
CHANNEL_ID = "UCxxxxxxxxxxxxxxxxxxxxxxxx"


def get_community_posts(channel_id: str) -> list[dict]:
    """
    activities.list API로 커뮤니티 게시글 수집.
    eventType="social"이 커뮤니티 포스트에 해당.
    """
    posts = []
    next_page_token = None

    while True:
        try:
            res = youtube.activities().list(
                part="snippet,contentDetails",
                channelId=channel_id,
                maxResults=50,
                pageToken=next_page_token,
            ).execute()
        except HttpError as e:
            print(f"[ERROR] activities.list 실패: {e}")
            break

        for item in res.get("items", []):
            snippet = item.get("snippet", {})
            # 커뮤니티 게시글만 필터
            if snippet.get("type") != "social":
                continue

            text = snippet.get("description", "").strip()
            if len(text) < 20:   # 너무 짧은 게시글 스킵
                continue

            posts.append({
                "post_id":      item["id"],
                "text":         text,
                "published_at": snippet.get("publishedAt", ""),
                "type": "community",
            })

        next_page_token = res.get("nextPageToken")
        if not next_page_token:
            break
        time.sleep(0.3)

    print(f"수집된 커뮤니티 게시글: {len(posts)}개")
    return posts


if __name__ == "__main__":
    os.makedirs("data/raw", exist_ok=True)
    posts = get_community_posts(CHANNEL_ID)
    with open("data/raw/community.json", "w", encoding="utf-8") as f:
        json.dump(posts, f, ensure_ascii=False, indent=2)
    print("→ data/raw/community.json 저장 완료")

# 먼저 설치
# pip install yt-dlp

# 1c_collect_subtitles.py
import os, json, subprocess
from pathlib import Path

SUBTITLES_DIR = Path("data/raw/subtitles")
SUBTITLES_DIR.mkdir(parents=True, exist_ok=True)


def download_subtitles_for_video(video_id: str, lang: str = "ko") -> bool:
    """
    단일 영상의 자막을 SRT로 다운로드.
    --write-subs      : 수동 업로드 자막
    --write-auto-subs : 자동 생성 자막 (없을 때 폴백)
    --skip-download   : 영상 파일은 받지 않음 (자막만)
    """
    url = f"https://www.youtube.com/watch?v={video_id}"
    output_template = str(SUBTITLES_DIR / f"{video_id}.%(ext)s")

    cmd = [
        "yt-dlp",
        "--write-subs",
        "--write-auto-subs",
        "--sub-langs", lang,
        "--sub-format", "srt",
        "--convert-subs", "srt",
        "--skip-download",
        "-o", output_template,
        url,
    ]

    result = subprocess.run(cmd, capture_output=True, text=True)
    # SRT 파일이 생성됐는지 확인
    srt_files = list(SUBTITLES_DIR.glob(f"{video_id}*.srt"))
    return len(srt_files) > 0


def batch_download_subtitles(video_ids: list[str], lang: str = "ko"):
    """영상 ID 목록 전체의 자막을 순차 다운로드."""
    success, fail = 0, 0
    for idx, video_id in enumerate(video_ids):
        ok = download_subtitles_for_video(video_id, lang)
        if ok:
            success += 1
        else:
            fail += 1  # 자막이 없는 영상은 건너뜀 (정상)

        if (idx + 1) % 20 == 0:
            print(f"  자막 다운로드 진행 중... {idx+1}/{len(video_ids)}")

        import time; time.sleep(1.0)  # yt-dlp 과호출 방지

    print(f"\n✅ 자막 다운로드 완료: 성공 {success}개 | 자막 없음 {fail}개")
    print(f"   저장 위치: {SUBTITLES_DIR}/")


if __name__ == "__main__":
    # videos.json에서 video_id 목록 로드
    with open("data/raw/videos.json", encoding="utf-8") as f:
        videos = json.load(f)
    video_ids = [v["video_id"] for v in videos]

    batch_download_subtitles(video_ids, lang="ko")
    # 영어 자막도 필요하면: batch_download_subtitles(video_ids, lang="en")

# 2_preprocess.py
import json, re, os, glob
from pathlib import Path


# ─── 공통 정제 함수 ─────────────────────────────────────────
def clean_text(text: str) -> str:
    if not text or not text.strip():
        return ""
    text = re.sub(r'<[^>]+>', ' ', text)                      # HTML 태그
    text = re.sub(r'https?://\S+|www\.\S+', '', text)       # URL
    text = re.sub(r'[\w.+-]+@[\w-]+\.[\w.-]+', '', text)   # 이메일
    text = re.sub(                                             # 이모지
        r'[\U0001F600-\U0001F64F\U0001F300-\U0001F5FF'
        r'\U0001F680-\U0001F6FF\U0001F1E0-\U0001F1FF'
        r'\u2600-\u26FF\u2700-\u27BF]+', '', text)
    text = re.sub(r'#\w+', '', text)                          # 해시태그
    text = re.sub(r'([!?.,ㅋㅎ])\1{2,}', r'\1', text)        # 반복 구두점
    text = re.sub(r'[^\w\s가-힣.,!?()\"\'%-]', ' ', text) # 허용 문자
    text = re.sub(r'\s+', ' ', text).strip()                  # 공백 정리
    return text


# ─── 1. 영상 설명란 처리 ───────────────────────────────────
def process_videos() -> list[dict]:
    path = Path("data/raw/videos.json")
    if not path.exists():
        print("⚠️  data/raw/videos.json 없음, 스킵")
        return []

    with open(path, encoding="utf-8") as f:
        videos = json.load(f)

    processed = []
    for v in videos:
        title       = clean_text(v.get("title", ""))
        description = clean_text(v.get("description", ""))

        if len(description) < 30:       # 설명이 너무 짧으면 스킵
            continue

        processed.append({
            "id": f"video_{v['video_id']}",
            "content": f"[영상 제목] {title}\n\n[내용]\n{description}",
            "metadata": {
                "type":         "video",
                "video_id":     v["video_id"],
                "title":        title,
                "view_count":   int(v.get("view_count", 0)),
                "like_count":   int(v.get("like_count", 0)),
                "published_at": v.get("published_at", ""),
            },
        })

    print(f"영상 처리 완료: {len(processed)}개")
    return processed


# ─── 2. 댓글 처리 ─────────────────────────────────────────
def process_comments(min_likes: int = 0) -> list[dict]:
    """
    min_likes: 이 값 이상의 좋아요를 받은 댓글만 포함.
    기본값 0 = 전부 포함. 스팸이 많으면 1~2로 올리세요.
    """
    path = Path("data/raw/comments.json")
    if not path.exists():
        print("⚠️  data/raw/comments.json 없음, 스킵")
        return []

    with open(path, encoding="utf-8") as f:
        comments = json.load(f)

    processed, skipped = [], 0
    for i, c in enumerate(comments):
        comment    = clean_text(c.get("comment", ""))
        like_count = int(c.get("like_count", 0))

        if len(comment) < 10:           # 너무 짧은 댓글
            skipped += 1
            continue
        if like_count < min_likes:      # 좋아요 필터
            skipped += 1
            continue

        processed.append({
            "id": f"comment_{i}",
            "content": f"[댓글] {comment}",
            "metadata": {
                "type":         "comment",
                "video_id":     c.get("video_id", ""),
                "like_count":   like_count,
                "published_at": c.get("published_at", ""),
            },
        })

    print(f"댓글 처리 완료: {len(processed)}개 (스킵 {skipped}개)")
    return processed


# ─── 3. 커뮤니티 게시글 처리 ───────────────────────────────
def process_community() -> list[dict]:
    path = Path("data/raw/community.json")
    if not path.exists():
        print("⚠️  data/raw/community.json 없음, 스킵")
        return []

    with open(path, encoding="utf-8") as f:
        posts = json.load(f)

    processed = []
    for p in posts:
        text = clean_text(p.get("text", ""))

        if len(text) < 20:
            continue

        processed.append({
            "id": f"community_{p.get('post_id', len(processed))}",
            "content": f"[커뮤니티] {text}",
            "metadata": {
                "type":         "community",
                "post_id":      p.get("post_id", ""),
                "published_at": p.get("published_at", ""),
            },
        })

    print(f"커뮤니티 처리 완료: {len(processed)}개")
    return processed


# ─── 4. 직접 보유 대본 .txt 처리 ──────────────────────────
def process_scripts() -> list[dict]:
    script_files = list(Path("data/raw/scripts").glob("*.txt"))
    if not script_files:
        print("⚠️  data/raw/scripts/*.txt 없음, 스킵")
        return []

    # 타임코드 패턴 제거 — [00:00], (00:00:00) 형식
    TIMECODE_RE = re.compile(r'\[?\d{1,2}:\d{2}(?::\d{2})?\]?')

    processed = []
    for filepath in script_files:
        filename = filepath.stem                      # 확장자 제외 파일명
        raw      = filepath.read_text(encoding="utf-8")
        raw      = TIMECODE_RE.sub('', raw)           # 타임코드 제거
        content  = clean_text(raw)

        if len(content) < 50:
            continue

        processed.append({
            "id": f"script_{filename}",
            "content": f"[대본: {filename}]\n\n{content}",
            "metadata": {
                "type":     "script",
                "filename": filename,
            },
        })

    print(f"대본 처리 완료: {len(processed)}개")
    return processed


# ─── 5. SRT 자막 처리 (타임스탬프 메타데이터 보존) ──────────
def process_subtitles() -> list[dict]:
    srt_files = list(Path("data/raw/subtitles").glob("*.srt"))
    if not srt_files:
        print("⚠️  data/raw/subtitles/*.srt 없음, 스킵")
        return []

    SRT_BLOCK_RE = re.compile(
        r'\d+\n'                                          # 인덱스 번호
        r'(\d{2}:\d{2}:\d{2},\d{3})'                    # 시작 시간
        r' --> '
        r'(\d{2}:\d{2}:\d{2},\d{3})\n'                 # 종료 시간
        r'((?:.+\n?)+)',                                    # 자막 텍스트
        re.MULTILINE
    )

    processed = []
    for srt_path in srt_files:
        video_id = srt_path.stem.split('.')[0]   # "dQw4w9WgXcQ.ko" → "dQw4w9WgXcQ"
        raw_srt  = srt_path.read_text(encoding="utf-8", errors="ignore")
        segments = SRT_BLOCK_RE.findall(raw_srt)

        # 10개 세그먼트씩 묶어서 하나의 청크로 만들기
        CHUNK_SIZE = 10
        for i in range(0, len(segments), CHUNK_SIZE):
            batch      = segments[i : i + CHUNK_SIZE]
            start_time = batch[0][0]             # 첫 세그먼트 시작 시간
            end_time   = batch[-1][1]            # 마지막 세그먼트 종료 시간
            text       = ' '.join(
                clean_text(seg[2].replace('\n', ' '))
                for seg in batch
                if clean_text(seg[2]) # 빈 세그먼트 스킵
            )

            if len(text) < 10:
                continue

            processed.append({
                "id": f"subtitle_{video_id}_{i // CHUNK_SIZE}",
                "content": f"[자막 {start_time}~{end_time}] {text}",
                "metadata": {
                    "type":        "subtitle",
                    "video_id":    video_id,
                    "start_time":  start_time,   # → 타임스탬프 링크용
                    "end_time":    end_time,
                    "chunk_index": i // CHUNK_SIZE,
                },
            })

    print(f"자막 처리 완료: {len(processed)}개 청크")
    return processed


# ─── 실행 ──────────────────────────────────────────────────
if __name__ == "__main__":
    Path("data/processed").mkdir(parents=True, exist_ok=True)

    all_data: list[dict] = []
    all_data.extend(process_videos())
    all_data.extend(process_comments(min_likes=0))   # 스팸 많으면 min_likes=1
    all_data.extend(process_community())
    all_data.extend(process_scripts())
    all_data.extend(process_subtitles())

    out_path = Path("data/processed/all_data.json")
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(all_data, f, ensure_ascii=False, indent=2)

    # 타입별 통계 출력
    from collections import Counter
    counts = Counter(d["metadata"]["type"] for d in all_data)
    print(f"\n✅ 전처리 완료!")
    print(f"   총 {len(all_data)}개 문서")
    for t, n in counts.items():
        print(f"   · {t}: {n}개")

# embed_utils.py — 배치 임베딩 유틸리티
import time
from openai import OpenAI
import os

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def embed_chunks_batch(
    texts: list[str],
    model: str = "text-embedding-3-small",
    batch_size: int = 100,    # OpenAI: 최대 2048, 안정적으로는 100
    sleep_sec: float = 0.1,   # 분당 요청 제한(RPM) 방지용 대기
) -> list[list[float]]:
    """
    대량 청크를 배치로 나눠 임베딩.
    실패한 배치는 재시도 후 빈 벡터로 채움.
    """
    all_embeddings: list[list[float]] = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]

        try:
            res = client.embeddings.create(
                model=model,
                input=batch,
            )
            # 인덱스 순서대로 정렬
            batch_vecs = [
                e.embedding
                for e in sorted(res.data, key=lambda x: x.index)
            ]
            all_embeddings.extend(batch_vecs)

        except Exception as e:
            print(f"[ERROR] 배치 {i//batch_size+1} 실패: {e}")
            # 실패한 배치는 빈 벡터로 채워 인덱스 유지
            all_embeddings.extend([[] for _ in batch])

        # 진행 상황 출력
        done = min(i + batch_size, len(texts))
        print(f"  임베딩 진행: {done}/{len(texts)}", end="\r")
        time.sleep(sleep_sec)

    print(f"\n✅ 임베딩 완료: {len(all_embeddings)}개")
    return all_embeddings


# 사용 예시
import json

with open("data/processed/all_data.json", encoding="utf-8") as f:
    all_data = json.load(f)

texts = [d["content"] for d in all_data]
vectors = embed_chunks_batch(texts, batch_size=100)

# 빈 벡터(실패) 개수 확인
failed = sum(1 for v in vectors if not v)
print(f"성공: {len(vectors)-failed}개 | 실패: {failed}개")

-- ① 벡터 저장 — vector32()에 JSON 배열 문자열 전달
INSERT INTO rag_documents
  (title, content, type, source_id, chunk_index, model, fingerprint, embedding)
VALUES (
  '유튜브 팁 영상 1',
  '썸네일 클릭률을 높이는 방법...',
  'video',
  'dQw4w9WgXcQ',
  0,
  'text-embedding-3-small',
  'a3f2c1d9...',
  vector32('[0.12, -0.34, 0.56, ...]')   -- JSON 배열 문자열 형식 필수!
);

-- ② 유사도 검색 — 반드시 같은 model로 필터!
SELECT title, content, type, source_id,
       vector_distance_cos(embedding, vector32('[0.10, -0.30, ...]')) AS distance
FROM rag_documents
WHERE model = 'text-embedding-3-small'   -- ← 이 줄 없으면 모델 혼재 시 오동작
ORDER BY distance
LIMIT 5;

-- ③ 메타데이터 필터 + 벡터 검색 동시에
SELECT title, content,
       vector_distance_cos(embedding, vector32('[...]')) AS distance
FROM rag_documents
WHERE model = 'text-embedding-3-small'
  AND type  = 'video'                    -- SQL 필터와 벡터 검색 동시 사용
ORDER BY distance
LIMIT 5;

-- ④ UPSERT — 같은 source_id + chunk_index면 업데이트
INSERT INTO rag_documents
  (title, content, type, source_id, chunk_index, model, fingerprint, embedding)
VALUES (?, ?, ?, ?, ?, ?, ?, vector32(?))
ON CONFLICT (source_id, chunk_index) DO UPDATE SET
  content     = excluded.content,
  model       = excluded.model,
  fingerprint = excluded.fingerprint,
  embedding   = excluded.embedding,
  updated_at  = CURRENT_TIMESTAMP;

// lib/init-db.ts
import { turso } from "./turso";

export async function initDB(): Promise<void> {
  // 트랜잭션으로 묶어 실행 — 일부 실패 시 전체 롤백
  await turso.batch([
    {
      sql: `CREATE TABLE IF NOT EXISTS rag_documents (
        id          INTEGER  PRIMARY KEY AUTOINCREMENT,
        title       TEXT     NOT NULL,
        content     TEXT     NOT NULL,
        type        TEXT     NOT NULL,
        source_id   TEXT,
        chunk_index INTEGER  DEFAULT 0,
        model       TEXT     NOT NULL,
        fingerprint TEXT,
        created_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
        updated_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
        embedding   BLOB
      )`,
      args: [],
    },
    {
      sql: `CREATE UNIQUE INDEX IF NOT EXISTS idx_source_chunk
            ON rag_documents(source_id, chunk_index)`,
      args: [],
    },
    {
      sql: `CREATE INDEX IF NOT EXISTS idx_type
            ON rag_documents(type)`,
      args: [],
    },
    {
      sql: `CREATE INDEX IF NOT EXISTS idx_model
            ON rag_documents(model)`,
      args: [],
    },
  ], "write");

  console.log("Turso DB 초기화 완료");
}

// 사용 예: npm run init-db 스크립트 또는 앱 최초 실행 시 1회 호출
// scripts/init-db.ts:
// import { initDB } from "../lib/init-db";
// initDB().then(() => process.exit(0));

# pip install libsql-experimental python-dotenv openai
import json, os, hashlib
import libsql_experimental as libsql
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

# ── 연결 ──────────────────────────────────────────────
conn = libsql.connect(
    database=os.getenv("TURSO_DATABASE_URL"),
    auth_token=os.getenv("TURSO_AUTH_TOKEN"),
)
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

EMBEDDING_MODEL = "text-embedding-3-small"

# ── 임베딩 함수 ────────────────────────────────────────
def embed(text: str) -> list[float]:
    res = openai_client.embeddings.create(
        model=EMBEDDING_MODEL,
        input=text.strip()[:8000],
    )
    return res.data[0].embedding

def fingerprint(text: str) -> str:
    return hashlib.sha256(text.encode()).hexdigest()

# ── 저장 (UPSERT) ──────────────────────────────────────
def upsert_chunk(
    title: str,
    content: str,
    content_type: str,
    source_id: str,
    chunk_index: int,
) -> None:
    fp = fingerprint(content)

    # 기존 fingerprint 확인 → 변경 없으면 스킵
    cursor = conn.execute(
        "SELECT fingerprint FROM rag_documents WHERE source_id = ? AND chunk_index = ?",
        (source_id, chunk_index)
    )
    row = cursor.fetchone()
    if row and row[0] == fp:
        print(f"  스킵 (변경 없음): {source_id}:{chunk_index}")
        return

    vec = embed(content)
    vec_str = json.dumps(vec)   # ← vector32()에 JSON 문자열 전달 필수

    conn.execute(
        """INSERT INTO rag_documents
             (title, content, type, source_id, chunk_index, model, fingerprint, embedding)
           VALUES (?, ?, ?, ?, ?, ?, ?, vector32(?))
           ON CONFLICT (source_id, chunk_index) DO UPDATE SET
             content     = excluded.content,
             model       = excluded.model,
             fingerprint = excluded.fingerprint,
             embedding   = excluded.embedding,
             updated_at  = CURRENT_TIMESTAMP""",
        (title, content, content_type, source_id, chunk_index,
         EMBEDDING_MODEL, fp, vec_str)
    )
    conn.commit()
    print(f"  저장 완료: {source_id}:{chunk_index}")

# ── 벡터 검색 ─────────────────────────────────────────
def search(query: str, top_k: int = 5, type_filter: str | None = None) -> list[dict]:
    query_vec = embed(query)
    vec_str = json.dumps(query_vec)

    if type_filter:
        cursor = conn.execute(
            """SELECT title, content, type, source_id,
                      vector_distance_cos(embedding, vector32(?)) AS distance
               FROM rag_documents
               WHERE model = ? AND type = ?
               ORDER BY distance
               LIMIT ?""",
            (vec_str, EMBEDDING_MODEL, type_filter, top_k)
        )
    else:
        cursor = conn.execute(
            """SELECT title, content, type, source_id,
                      vector_distance_cos(embedding, vector32(?)) AS distance
               FROM rag_documents
               WHERE model = ?
               ORDER BY distance
               LIMIT ?""",
            (vec_str, EMBEDDING_MODEL, top_k)
        )

    cols = [d[0] for d in cursor.description]
    return [dict(zip(cols, row)) for row in cursor.fetchall()]

# ── 사용 예시 ──────────────────────────────────────────
if __name__ == "__main__":
    # 저장
    upsert_chunk(
        title="썸네일 팁 영상",
        content="썸네일 클릭률을 높이는 방법...",
        content_type="video",
        source_id="dQw4w9WgXcQ",
        chunk_index=0,
    )

    # 검색
    results = search("썸네일 클릭률 높이는 방법", top_k=5)
    for r in results:
        print(f"  [{r['distance']:.3f}] {r['title']}: {r['content'][:60]}...")

# 4_rag_chain.py  (개념 이해용 — RetrievalQA)
import os
from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

load_dotenv()

# 1. 저장된 벡터 DB 로드
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectordb = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="youtube_rag"
)
print(f"벡터 DB 로드 완료. 총 {vectordb._collection.count()}개 청크")

# 2. 검색기 설정
retriever = vectordb.as_retriever(
    search_type="similarity",   # 코사인 유사도 기반
    search_kwargs={
        "k": 5,                 # 상위 5개 청크 반환
        # "filter": {"type": "video"}  # 메타데이터 필터 (선택)
    }
)

# 3. 맞춤 프롬프트
CUSTOM_PROMPT = PromptTemplate(
    template="""당신은 내 YouTube 채널(큐레이터 단비)의 전문 어시스턴트입니다.
아래 제공된 채널 데이터(영상 대본, 댓글 등)를 기반으로만 답변하세요.
데이터에 없는 내용은 솔직하게 "해당 데이터에서 찾을 수 없습니다"라고 말하세요.

채널 데이터:
{context}

질문: {question}
답변:""",
    input_variables=["context", "question"]
)

# 4. RAG 체인 생성
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": CUSTOM_PROMPT},
    return_source_documents=True
)

# 5. 질의 테스트
def ask(question: str):
    result = qa_chain.invoke({"query": question})
    print("\n💬 질문:", question)
    print("\n🤖 답변:", result["result"])
    print("\n📄 참고 소스:")
    for i, doc in enumerate(result["source_documents"]):
        print(
            f"  [{i+1}] {doc.metadata.get('title', '제목없음')}"
            f" | 유형: {doc.metadata.get('type', '?')}"
            f" | 조회수: {doc.metadata.get('view_count', 0):,}"
        )

if __name__ == "__main__":
    ask("썸네일 클릭률 높이는 방법이 뭐야?")
    ask("댓글에서 시청자들이 가장 많이 요청한 주제는?")

# 4_rag_chain_lcel.py  (권장 — LCEL 완전 구현)
import os
from operator import itemgetter
from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

# ── 벡터 DB & 검색기 ──────────────────────────────────
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectordb = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="youtube_rag"
)
retriever = vectordb.as_retriever(search_kwargs={"k": 5})

# ── 컨텍스트 포맷터 ────────────────────────────────────
def format_docs(docs) -> str:
    """검색된 문서 목록을 프롬프트용 문자열로 변환"""
    return "\n\n---\n\n".join(
        f"[{i+1}] ({doc.metadata.get('type','?')}) "
        f"{doc.metadata.get('title','?')}:\n{doc.page_content}"
        for i, doc in enumerate(docs)
    )

# ── 프롬프트 ───────────────────────────────────────────
prompt = ChatPromptTemplate.from_messages([
    ("system",
     """당신은 내 YouTube 채널(큐레이터 단비)의 전문 어시스턴트입니다.
반드시 아래 채널 데이터만 근거로 답변하세요.
근거 없는 내용은 '데이터에서 찾을 수 없습니다'라고 말하세요.

채널 데이터:
{context}"""),
    ("human", "{question}"),
])

# ── LLM ────────────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

# ── 체인 조합 (출처 문서 함께 반환) ───────────────────
rag_chain_with_sources = (
    {
        "context": retriever | RunnableLambda(format_docs),
        "question": RunnablePassthrough(),
        # 출처 문서를 별도로 보존
        "_docs": retriever,
    }
    | RunnablePassthrough.assign(
        answer=prompt | llm | StrOutputParser()
    )
)

def ask(question: str):
    result = rag_chain_with_sources.invoke(question)
    print("\n💬 질문:", question)
    print("\n🤖 답변:", result["answer"])
    print("\n📄 참고 소스:")
    for i, doc in enumerate(result["_docs"]):
        print(
            f"  [{i+1}] {doc.metadata.get('title','?')}"
            f" | 유형: {doc.metadata.get('type','?')}"
        )

if __name__ == "__main__":
    ask("썸네일 클릭률 높이는 방법이 뭐야?")

# 4_rag_chain_memory.py  (멀티턴 대화 — 히스토리 반영)
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

# ── 1. 히스토리를 반영한 검색 쿼리 재작성 프롬프트 ────
contextualize_q_prompt = ChatPromptTemplate.from_messages([
    ("system",
     """대화 히스토리와 최신 질문이 주어집니다.
히스토리 없이도 이해 가능한 독립적인 질문으로 재작성하세요.
질문을 답변하지 말고 필요하면 재작성, 아니면 그대로 반환하세요."""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])

history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_q_prompt
)

# ── 2. 답변 생성 프롬프트 (히스토리 포함) ────────────
qa_prompt = ChatPromptTemplate.from_messages([
    ("system",
     """당신은 내 YouTube 채널 전문 어시스턴트입니다.
아래 채널 데이터만 근거로 답변하세요.

{context}"""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])

question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)
rag_chain = create_retrieval_chain(
    history_aware_retriever, question_answer_chain
)

# ── 3. 대화 루프 ──────────────────────────────────────
chat_history = []

def chat(question: str) -> str:
    result = rag_chain.invoke({
        "input": question,
        "chat_history": chat_history,
    })
    # 히스토리 업데이트
    chat_history.extend([
        HumanMessage(content=question),
        AIMessage(content=result["answer"]),
    ])
    return result["answer"]

# 사용 예시
print(chat("썸네일 클릭률 높이는 방법이 뭐야?"))
print(chat("그 중에서 가장 효과적인 방법 하나만 뽑아줘"))  # 후속 질문
print(chat("그게 실제 내 채널 데이터에 있어?"))             # 히스토리 참조

import time
from openai import RateLimitError, APIError

def ask_with_retry(question: str, max_retries: int = 3) -> str:
    """API 실패 시 지수 백오프로 재시도"""
    for attempt in range(max_retries):
        try:
            result = rag_chain_with_sources.invoke(question)
            return result["answer"]

        except RateLimitError:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 1초 → 2초 → 4초
                print(f"Rate limit 초과. {wait}초 후 재시도 ({attempt+1}/{max_retries})")
                time.sleep(wait)
            else:
                return "⚠️ 요청 한도 초과입니다. 잠시 후 다시 시도해주세요."

        except APIError as e:
            print(f"API 오류: {e}")
            return "⚠️ AI 서비스 오류가 발생했습니다. 잠시 후 다시 시도해주세요."

        except Exception as e:
            print(f"예상치 못한 오류: {e}")
            return "⚠️ 오류가 발생했습니다."

    return "⚠️ 재시도 한도를 초과했습니다."

# 검색 결과가 없을 때 처리
def ask_safe(question: str) -> dict:
    docs = retriever.invoke(question)

    if not docs:
        return {
            "answer": "관련 데이터를 찾지 못했습니다. 다른 키워드로 질문해보세요.",
            "sources": []
        }

    # 유사도가 낮은 결과만 있을 때 경고
    # (with_score 사용 시: score < 임계값이면 신뢰도 낮음 표시)
    answer = ask_with_retry(question)
    return {"answer": answer, "sources": docs}

// lib/rag/search.ts
import { turso } from "@/lib/turso";
import { EMBEDDING_MODEL, TOP_K, MIN_SIMILARITY_THRESHOLD } from "./constants";

export type RagDoc = {
  id: number;
  title: string;
  content: string;
  type: string;
  source_id: string;
  distance: number;
};

export async function searchSimilarDocs(
  queryVector: number[],
  topK = TOP_K,
  typeFilter?: string   // 'video' | 'comment' | 'script' | undefined(전체)
): Promise<RagDoc[]> {
  const vecStr = JSON.stringify(queryVector);

  // WHERE model = ? 로 반드시 같은 모델의 벡터만 검색
  // 모델이 다른 벡터와 비교하면 유사도 수치가 무의미해짐
  const sql = typeFilter
    ? `SELECT id, title, content, type, source_id,
             vector_distance_cos(embedding, vector32(?)) AS distance
       FROM rag_documents
       WHERE model = ? AND type = ?
       ORDER BY distance
       LIMIT ?`
    : `SELECT id, title, content, type, source_id,
             vector_distance_cos(embedding, vector32(?)) AS distance
       FROM rag_documents
       WHERE model = ?
       ORDER BY distance
       LIMIT ?`;

  const args = typeFilter
    ? [vecStr, EMBEDDING_MODEL, typeFilter, topK]
    : [vecStr, EMBEDDING_MODEL, topK];

  const { rows } = await turso.execute({ sql, args });

  // 유사도 임계값 필터 — 관련 없는 결과 제거
  return (rows as RagDoc[]).filter(
    (row) => row.distance <= MIN_SIMILARITY_THRESHOLD
  );
}

// lib/rag/generate.ts
import OpenAI from "openai";
import { LLM_MODEL } from "./constants";
import type { RagDoc } from "./search";

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY! });

function buildContext(docs: RagDoc[]): string {
  if (docs.length === 0) return "관련 데이터 없음";
  return docs
    .map((d, i) => `[${i + 1}] (${d.type}) ${d.title}:\n${d.content}`)
    .join("\n\n---\n\n");
}

const SYSTEM_PROMPT = (context: string) =>
  `당신은 내 YouTube 채널 전문 어시스턴트입니다.
반드시 아래 채널 데이터만 근거로 답변하세요.
데이터에 없는 내용은 솔직하게 "해당 데이터에서 찾을 수 없습니다"라고 말하세요.

채널 데이터:
${context}`;

/** 일반 응답 (전체 텍스트 반환) */
export async function generateAnswer(
  query: string,
  docs: RagDoc[]
): Promise<string> {
  const completion = await openai.chat.completions.create({
    model: LLM_MODEL,
    temperature: 0.3,
    messages: [
      { role: "system", content: SYSTEM_PROMPT(buildContext(docs)) },
      { role: "user", content: query },
    ],
  });
  return completion.choices[0].message.content ?? "";
}

/** 스트리밍 응답 (ReadableStream 반환) */
export async function generateAnswerStream(
  query: string,
  docs: RagDoc[]
): Promise<ReadableStream<string>> {
  const stream = await openai.chat.completions.create({
    model: LLM_MODEL,
    temperature: 0.3,
    stream: true,
    messages: [
      { role: "system", content: SYSTEM_PROMPT(buildContext(docs)) },
      { role: "user", content: query },
    ],
  });

  return new ReadableStream({
    async start(controller) {
      for await (const chunk of stream) {
        const text = chunk.choices[0]?.delta?.content ?? "";
        if (text) controller.enqueue(text);
      }
      controller.close();
    },
  });
}

// app/api/rag/ask/route.ts
import { NextRequest } from "next/server";
import { embedQuery } from "@/lib/rag/embed";
import { searchSimilarDocs } from "@/lib/rag/search";
import { generateAnswerStream } from "@/lib/rag/generate";

// Edge Runtime — 스트리밍에 최적화, 타임아웃 제약 없음
// (단, Edge에서는 Node.js 전용 API 사용 불가)
export const runtime = "edge";

export async function POST(req: NextRequest) {
  try {
    const body = await req.json();
    const query: string = body?.query ?? "";
    const typeFilter: string | undefined = body?.typeFilter; // 선택적 타입 필터

    if (!query.trim()) {
      return new Response(
        JSON.stringify({ error: "query가 비어있습니다." }),
        { status: 400, headers: { "Content-Type": "application/json" } }
      );
    }

    // 1. 질문 임베딩
    const queryVec = await embedQuery(query);

    // 2. 벡터 검색
    const docs = await searchSimilarDocs(queryVec, 5, typeFilter);

    // 3. 검색 결과 없을 때 조기 반환
    if (docs.length === 0) {
      const fallback = "관련 채널 데이터를 찾지 못했습니다. 다른 키워드로 질문해 보세요.";
      return new Response(fallback, {
        headers: { "Content-Type": "text/plain; charset=utf-8" },
      });
    }

    // 4. 출처 메타데이터를 헤더에 포함 (스트리밍 중 body에 넣기 어려움)
    const sourceMeta = JSON.stringify(
      docs.map((d) => ({ title: d.title, type: d.type, source_id: d.source_id }))
    );

    // 5. LLM 스트리밍 응답 반환
    const stream = await generateAnswerStream(query, docs);
    return new Response(stream, {
      headers: {
        "Content-Type": "text/plain; charset=utf-8",
        "X-Rag-Sources": encodeURIComponent(sourceMeta), // 출처 헤더로 전달
        "X-Rag-Doc-Count": String(docs.length),
      },
    });

  } catch (err) {
    console.error("[RAG /ask] error:", err);
    return new Response(
      JSON.stringify({ error: "서버 오류가 발생했습니다. 잠시 후 다시 시도해 주세요." }),
      { status: 500, headers: { "Content-Type": "application/json" } }
    );
  }
}

// app/api/rag/sync/route.ts
import { NextRequest, NextResponse } from "next/server";
import crypto from "crypto";
import { turso } from "@/lib/turso";
import { embedBatch } from "@/lib/rag/embed";
import { EMBEDDING_MODEL } from "@/lib/rag/constants";

// SYNC_SECRET으로 무단 접근 차단 (Vercel 환경변수에 설정)
function isAuthorized(req: NextRequest): boolean {
  const secret = req.headers.get("x-sync-secret");
  return secret === process.env.SYNC_SECRET;
}

function fingerprint(text: string): string {
  return crypto.createHash("sha256").update(text).digest("hex");
}

type SyncItem = {
  title: string;
  content: string;
  type: "video" | "comment" | "script";
  source_id: string;
  chunk_index: number;
};

export async function POST(req: NextRequest) {
  if (!isAuthorized(req)) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

  const { items }: { items: SyncItem[] } = await req.json();
  if (!items?.length) {
    return NextResponse.json({ error: "items가 비어있습니다." }, { status: 400 });
  }

  // 1. 기존 fingerprint 조회 (변경 감지)
  const sourceIds = [...new Set(items.map((i) => i.source_id))];
  const { rows: existingRows } = await turso.execute({
    sql: `SELECT source_id, chunk_index, fingerprint
          FROM rag_documents
          WHERE source_id IN (${sourceIds.map(() => "?").join(",")})`,
    args: sourceIds,
  });

  type FpRow = { source_id: string; chunk_index: number; fingerprint: string };
  const existingFpMap = new Map(
    (existingRows as FpRow[]).map((r) => [`${r.source_id}:${r.chunk_index}`, r.fingerprint])
  );

  // 2. 변경된 항목만 필터링
  const toEmbed = items.filter((item) => {
    const key = `${item.source_id}:${item.chunk_index}`;
    const newFp = fingerprint(item.content);
    return existingFpMap.get(key) !== newFp;
  });

  if (toEmbed.length === 0) {
    return NextResponse.json({ synced: 0, skipped: items.length, message: "변경 없음" });
  }

  // 3. 배치 임베딩 (변경된 것만)
  const vectors = await embedBatch(toEmbed.map((i) => i.content));

  // 4. UPSERT — 중복 실행 안전
  let synced = 0;
  for (let i = 0; i < toEmbed.length; i++) {
    const item = toEmbed[i];
    const vec = vectors[i];
    const fp = fingerprint(item.content);

    await turso.execute({
      sql: `INSERT INTO rag_documents
              (title, content, type, source_id, chunk_index, model, fingerprint, embedding)
            VALUES (?, ?, ?, ?, ?, ?, ?, vector32(?))
            ON CONFLICT (source_id, chunk_index) DO UPDATE SET
              title       = excluded.title,
              content     = excluded.content,
              model       = excluded.model,
              fingerprint = excluded.fingerprint,
              embedding   = excluded.embedding,
              updated_at  = CURRENT_TIMESTAMP`,
      args: [
        item.title, item.content, item.type,
        item.source_id, item.chunk_index,
        EMBEDDING_MODEL, fp,
        JSON.stringify(vec),
      ],
    });
    synced++;
  }

  return NextResponse.json({
    synced,
    skipped: items.length - synced,
    model: EMBEDDING_MODEL,
  });
}

// Vercel Cron 설정 예시 (vercel.json)
// {
//   "crons": [{ "path": "/api/rag/sync", "schedule": "0 3 * * *" }]
// }
// → 매일 새벽 3시 자동 동기화 (POST + x-sync-secret 헤더 필요)

// app/my-rag-ui/page.tsx
"use client";
import { useState, useRef } from "react";

type Source = { title: string; type: string; source_id: string };

export default function MyRagPage() {
  const [query, setQuery] = useState("");
  const [answer, setAnswer] = useState("");
  const [sources, setSources] = useState<Source[]>([]);
  const [loading, setLoading] = useState(false);
  const [error, setError] = useState("");
  const abortRef = useRef<AbortController | null>(null);

  const handleAsk = async () => {
    if (!query.trim() || loading) return;

    // 이전 요청 취소
    abortRef.current?.abort();
    abortRef.current = new AbortController();

    setLoading(true);
    setAnswer("");
    setSources([]);
    setError("");

    try {
      const res = await fetch("/api/rag/ask", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ query }),
        signal: abortRef.current.signal,
      });

      if (!res.ok) {
        const err = await res.json();
        throw new Error(err.error ?? "서버 오류");
      }

      // 출처 헤더 파싱
      const sourceMeta = res.headers.get("X-Rag-Sources");
      if (sourceMeta) {
        setSources(JSON.parse(decodeURIComponent(sourceMeta)));
      }

      // 스트리밍 텍스트 읽기
      const reader = res.body?.getReader();
      const decoder = new TextDecoder();
      if (!reader) throw new Error("스트림을 읽을 수 없습니다.");

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        setAnswer((prev) => prev + decoder.decode(value, { stream: true }));
      }

    } catch (e: unknown) {
      if (e instanceof Error && e.name === "AbortError") return; // 취소는 무시
      setError(e instanceof Error ? e.message : "알 수 없는 오류가 발생했습니다.");
    } finally {
      setLoading(false);
    }
  };

  const handleKeyDown = (e: React.KeyboardEvent) => {
    if (e.key === "Enter" && (e.metaKey || e.ctrlKey)) handleAsk();
  };

  return (
    <div className="max-w-2xl mx-auto p-4 space-y-4">
      <h1 className="text-2xl font-bold">내 채널 AI 어시스턴트</h1>

      <textarea
        value={query}
        onChange={(e) => setQuery(e.target.value)}
        onKeyDown={handleKeyDown}
        placeholder="내 채널에 대해 뭐든 물어보세요... (Ctrl+Enter로 전송)"
        className="w-full border rounded-lg p-3 min-h-[100px] resize-none focus:outline-none focus:ring-2 focus:ring-violet-400"
      />

      <div className="flex gap-2">
        <button
          onClick={handleAsk}
          disabled={loading || !query.trim()}
          className="flex-1 px-6 py-2.5 bg-violet-600 hover:bg-violet-700 text-white rounded-lg disabled:opacity-50 transition-colors font-bold"
        >
          {loading ? "답변 생성 중..." : "질문하기"}
        </button>
        {loading && (
          <button
            onClick={() => abortRef.current?.abort()}
            className="px-4 py-2.5 bg-gray-200 hover:bg-gray-300 dark:bg-gray-700 dark:hover:bg-gray-600 rounded-lg text-sm"
          >
            중단
          </button>
        )}
      </div>

      {error && (
        <div className="p-3 bg-red-50 dark:bg-red-900/20 border border-red-200 dark:border-red-800 rounded-lg text-sm text-red-700 dark:text-red-300">
          ⚠️ {error}
        </div>
      )}

      {answer && (
        <div className="space-y-3">
          <div className="p-4 bg-gray-50 dark:bg-gray-900 rounded-lg border border-gray-200 dark:border-gray-700 whitespace-pre-wrap text-sm leading-relaxed">
            {answer}
            {loading && <span className="inline-block w-1 h-4 bg-violet-500 animate-pulse ml-1 align-middle" />}
          </div>

          {sources.length > 0 && (
            <details className="text-xs">
              <summary className="cursor-pointer text-gray-500 dark:text-gray-400 hover:text-gray-700 dark:hover:text-gray-200 select-none">
                📄 참고한 데이터 {sources.length}개 보기
              </summary>
              <ul className="mt-2 space-y-1 pl-2">
                {sources.map((s, i) => (
                  <li key={i} className="text-gray-500 dark:text-gray-400">
                    [{i + 1}] ({s.type}) {s.title}
                  </li>
                ))}
              </ul>
            </details>
          )}
        </div>
      )}
    </div>
  );
}

# OpenRouter 임베딩 API — openai 클라이언트 재사용 가능
pip install openai

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY"),
)

def embed_openrouter(
    text: str,
    model: str = "qwen/qwen3-embedding-8b"
) -> list[float]:
    """
    OpenRouter 임베딩 API 호출.
    model 문자열 형식: "provider/model-name"
    예시:
      "qwen/qwen3-embedding-8b"       — Qwen3 8B (다국어·범용)
      "qwen/qwen3-embedding-0.6b"     — Qwen3 0.6B (초경량)
      "openai/text-embedding-3-small" — OpenAI small (안정성)
      "openai/text-embedding-3-large" — OpenAI large (고품질)
      "mistralai/codestral-embed"     — 코드 특화
    """
    response = client.embeddings.create(
        model=model,
        input=text,
    )
    return response.data[0].embedding

# 배치 처리 (여러 텍스트 한 번에)
def embed_batch_openrouter(
    texts: list[str],
    model: str = "qwen/qwen3-embedding-8b"
) -> list[list[float]]:
    response = client.embeddings.create(
        model=model,
        input=texts,  # 리스트로 한 번에 전송 (API 호출 횟수 절감)
    )
    # 입력 순서대로 정렬
    return [item.embedding for item in sorted(response.data, key=lambda x: x.index)]

# 사용 예시
vec = embed_openrouter("썸네일 클릭률 높이는 방법")
print(f"벡터 차원: {len(vec)}")  # Qwen3-8B: 4096차원

# RAG용 저장 임베딩 vs 쿼리 임베딩
# ※ Qwen3은 task_type 파라미터가 없으므로 동일 모델로 저장·검색 모두 처리
store_vec = embed_openrouter("안녕하세요 오늘은 썸네일...")   # 문서 저장용
query_vec = embed_openrouter("썸네일 클릭률 높이는 방법?")   # 검색 쿼리용