import time

import torch
from PIL import Image
from sentence_transformers import SentenceTransformer, util


def calculate_similarity(img_emb, text_embs):
    """ cosine similarity between two vectors """
    cosine_scores = util.cos_sim(img_emb, text_embs)
    return cosine_scores

class CLIPInferenceService:
    # https://huggingface.co/models 에서 모델 찾기
    def __init__(self, model_name='clip-ViT-L-14'):
        print(f"Loading model '{model_name}' into memory...")
        start_time = time.time()

        # NVIDIA GPU 가속이 가능하면 cuda, 아니면 cpu 사용
        self.device = 'cuda' if torch.cuda.is_available() else 'cpu'
        self.model = SentenceTransformer(model_name, device=self.device)

        print(f"Model loaded in {time.time() - start_time:.2f} seconds on {self.device}")

    def encode_image(self, image_path):
        """ image to vector """
        img = Image.open(image_path)
        start_time = time.time()
        embeddings = self.model.encode(img)
        print(f"Image encoding latency: {(time.time() - start_time)*1000:.2f} ms")
        return embeddings

    def encode_text(self, text_list):
        # text(or list) to vector
        start_time = time.time()
        embeddings = self.model.encode(text_list)
        print(f"Text encoding latency: {(time.time() - start_time)*1000:.2f} ms")
        return embeddings


if __name__ == "__main__":
    clip_service = CLIPInferenceService()
    sample_image_path = "./anime.jpg"

    queries = [
        "비가 내리는 우울한 장면",
        "주인공이 검을 휘두르는 화려한 액션",
        "등장인물들이 웃으며 식사하는 일상",
        "울며 점프하는 엘프를 보는 주인공 일행",
        "침대 위에서 울고 있는 장면",
        "어느 여관 안에 있는 모습"
    ]

    print("\n--- Extracting Embeddings ---")
    img_vec = clip_service.encode_image(sample_image_path)
    text_vec = clip_service.encode_text(queries)

    print(f"\nImage Vector Shape: {img_vec.shape}")
    # Float32, 차원은 모델마다 다름 (예: ViT-L-14는 768, ViT-B-32는 512)

    print("\n--- Cosine Similarity Results ---")
    scores = calculate_similarity(img_vec, text_vec)

    # sort
    results = list(zip(queries, scores[0]))
    results.sort(key=lambda x: x[1].item(), reverse=True)
    for i, (query, score) in enumerate(results):
        print(f"{i + 1}. Score: {score.item():.4f} -> Query: '{query}'")

Loading model 'clip-ViT-L-14' into memory...
Model loaded in 7.83 seconds on cpu

--- Extracting Embeddings ---
Image encoding latency: 586.39 ms
Text encoding latency: 155.15 ms

Image Vector Shape: (768,)

--- Cosine Similarity Results ---
1. Score: 0.1655 -> Query: '어느 여관 안에 있는 모습'
2. Score: 0.1540 -> Query: '침대 위에서 울고 있는 장면'
3. Score: 0.1535 -> Query: '울며 점프하는 엘프를 보는 주인공 일행'
4. Score: 0.1477 -> Query: '주인공이 검을 휘두르는 화려한 액션'
5. Score: 0.1404 -> Query: '비가 내리는 우울한 장면'
6. Score: 0.1347 -> Query: '등장인물들이 웃으며 식사하는 일상'

Loading model 'clip-ViT-B-32' into memory...
Model loaded in 7.77 seconds on cpu

--- Extracting Embeddings ---
Image encoding latency: 96.44 ms
Text encoding latency: 76.44 ms

Image Vector Shape: (512,)

--- Cosine Similarity Results ---
1. Score: 0.2219 -> Query: '주인공이 검을 휘두르는 화려한 액션'
2. Score: 0.2109 -> Query: '침대 위에서 울고 있는 장면'
3. Score: 0.2097 -> Query: '어느 여관 안에 있는 모습'
4. Score: 0.1976 -> Query: '비가 내리는 우울한 장면'
5. Score: 0.1974 -> Query: '울며 점프하는 엘프를 보는 주인공 일행'
6. Score: 0.1949 -> Query: '등장인물들이 웃으며 식사하는 일상'

Process finished with exit code 0

queries = [
    "A gloomy scene with falling rain", # 비가 내리는 우울한 장면
    "Flashy action of the protagonist swinging a sword", # 주인공이 검을 휘두르는 화려한 액션
    "Characters having a meal and laughing together", # 등장인물들이 웃으며 식사하는 일상
    "The protagonist's party watching an elf jumping while crying", # 울며 점프하는 엘프를 보는 주인공 일행
    "A scene of someone crying on the bed", # 침대 위에서 울고 있는 장면
    "A scene inside an inn", # 어느 여관 안에 있는 모습
    "There are four people inside the inn", # 여관 안에 사람이 4명이 있다
    "There are three people inside the inn", # 여관 안에 사람이 3명이 있다
    "There are two people inside the inn", # 여관 안에 사람이 2명이 있다
    "There is a wardrobe and a bed in the room", # 방 안에 장롱과 침대가 있다
    "There is a bed and a wardrobe in the room" # 방 안에 침대와 장롱이 있다
]

Loading model 'clip-ViT-L-14' into memory...
Model loaded in 7.57 seconds on cpu

--- Extracting Embeddings ---
Image encoding latency: 552.03 ms
Text encoding latency: 441.05 ms

Image Vector Shape: (768,)

--- Cosine Similarity Results ---
1. Score: 0.2711 -> Query: 'A scene of someone crying on the bed'
2. Score: 0.2227 -> Query: 'A scene inside an inn'
3. Score: 0.2223 -> Query: 'There is a bed and a wardrobe in the room'
4. Score: 0.2205 -> Query: 'There are four people inside the inn'
5. Score: 0.2205 -> Query: 'There are three people inside the inn'
6. Score: 0.2161 -> Query: 'The protagonist's party watching an elf jumping while crying'
7. Score: 0.2159 -> Query: 'There are two people inside the inn'
8. Score: 0.2094 -> Query: 'There is a wardrobe and a bed in the room'
9. Score: 0.2020 -> Query: 'Characters having a meal and laughing together'
10. Score: 0.1944 -> Query: 'Flashy action of the protagonist swinging a sword'
11. Score: 0.1459 -> Query: 'A gloomy scene with falling rain'

이미지 파일 로드 → CLIP 전처리 → 배치 Tensor 생성 → GPU 이동 → 이미지 벡터 추출
텍스트 쿼리 → CLIP 토큰화 → GPU 이동 → 텍스트 벡터 추출
이미지 벡터 N개 × 텍스트 벡터 M개 → N:M 유사도 행렬 → 히트맵 시각화

!pip install yt-dlp
!pip install git+https://github.com/openai/CLIP.git
# yt-dlp로 프레임 추출에 사용할 영상 다운로드
!yt-dlp -f "bestvideo[ext=mp4]/best[ext=mp4]/best" "https://www.youtube.com/watch?v=qDFeurdWDNE" -o sample_anime.mp4

# ffmpeg로 1차 리사이징 및 프레임 추출 (I/O 병목 방지)
# ViT-L/14 모델의 기본 입력은 224px이므로 scale=-1:224로 맞춤
!mkdir -p frames
!ffmpeg -i sample_anime.mp4 -vf "fps=2.5,scale=-1:224" -q:v 2 frames/frame_%04d.jpg

import os
import time
import torch # 딥러닝 프레임워크
import clip # 신경망
from PIL import Image # 이미지 가공
import matplotlib.pyplot as plt # 시각화
import seaborn as sns # 데이터 통계 시각화
import matplotlib.gridspec as gridspec # grid 세팅

# CLIP 모델과 이미지 전처리 함수 로드
device = "cuda" if torch.cuda.is_available() else "cpu"
print("Loading ViT-L/14 model...")
# 이 모델은 224px 이미지를 14px 패치로 나누므로 16 x 16 = 256개 패치를 사용함
# 이미지 크기를 224에서 336으로 늘리면 패치 수가 증가해 GPU 비용도 함께 증가함
# 224px이어도 CLIP은 색감, 구도, 객체, 분위기 같은 정보를 의미적 특징 벡터로 인코딩함
model, preprocess = clip.load("ViT-L/14", device=device)
model.eval() # 모델을 추론 모드로 설정

# CUDA 연산은 비동기로 실행되므로 시간 측정 전후에 사용할 동기화 함수
def sync_if_cuda():
  if device == "cuda":
    torch.cuda.synchronize()

# 전체 프레임 중 시각화할 이미지 샘플링
image_count = 20
all_frame_files = sorted([os.path.join("frames", f) for f in os.listdir("frames") if f.endswith(".jpg")])

if len(all_frame_files) > image_count and image_count > 1:
  sampled_indices = [
      round(i * (len(all_frame_files) - 1) / (image_count - 1))
      for i in range(image_count)
  ]
  frame_files = [all_frame_files[i] for i in sampled_indices]
else:
  frame_files = all_frame_files[:image_count]
print(f"Extracting features for {len(frame_files)} frames...")

# 샘플링한 이미지들을 전처리한 뒤 하나의 배치 Tensor로 묶어 GPU로 이동
image_tensors = torch.stack([preprocess(Image.open(f)) for f in frame_files]).to(device)
# 텍스트 토큰화, sentence-transformers를 쓰지 않으므로 CLIP 토크나이저를 직접 사용
queries = [
    "Netflix",
    "Sad scene",
    "Neon-lit Night City",
    "Giant Hologram Advertisement",
    "Rainy Dystopian Street",
    "Cybernetic Body Enhancements",
    "Flying Vehicle in the Sky",
    "Digital Glitch Effect",
    "High-tech Hacking Interface",
    "Wires and Cables Underground",
    "Cyberpunk Techwear Fashion",
    "Melancholic Loneliness"
]
text_tokens = clip.tokenize(queries).to(device)

# 특징 추출 (Feature Extraction & Normalization)
with torch.no_grad(): # 그래디언트 추적을 꺼서 추론 속도와 VRAM 사용량을 개선
  # 이미지 배치와 텍스트 배치를 각각 한 번에 인코딩
  sync_if_cuda()
  encode_start = time.perf_counter()
  image_features = model.encode_image(image_tensors)
  text_features = model.encode_text(text_tokens)
  sync_if_cuda()
  encode_time = time.perf_counter() - encode_start

  # L2 정규화: 벡터의 길이를 1로 맞춰 방향성(Cosine Similarity)만 비교할 수 있게 함, 비교하기 위해 정규화를 적용했다고 이해하면 될 것 같습니다.
  image_features /= image_features.norm(dim=-1, keepdim=True)
  text_features /= text_features.norm(dim=-1, keepdim=True)

  # 1:1 방식: 각 이미지-텍스트 쌍을 하나씩 비교
  sync_if_cuda()
  loop_start = time.perf_counter()
  loop_scores = []
  for image_feature in image_features:
    row = []
    for text_feature in text_features:
      row.append(image_feature @ text_feature)
    loop_scores.append(torch.stack(row))
  loop_scores = torch.stack(loop_scores)
  sync_if_cuda()
  loop_time = time.perf_counter() - loop_start

  # N:M 방식: 행렬곱 한 번으로 전체 유사도 행렬 계산
  # image_features (프레임 수, 768) @ text_features.T (768, 쿼리 수) = cos_sim (프레임 수, 쿼리 수)
  sync_if_cuda()
  matrix_start = time.perf_counter()
  cos_sim_tensor = image_features @ text_features.t()
  sync_if_cuda()
  matrix_time = time.perf_counter() - matrix_start

  # 유사도 점수를 CLIP의 logits와 softmax 상대 점수로 변환
  logit_scale = model.logit_scale.exp()
  # logits은 코사인 유사도에 CLIP이 학습한 스케일 값을 곱한 점수
  logits = logit_scale * cos_sim_tensor

  # softmax는 전체 정답 확률이 아니라 현재 쿼리 목록 안에서의 상대 점수
  probs = logits.softmax(dim=-1).cpu().numpy()
  # 순수 코사인 유사도
  # cpu().numpy(): GPU 텐서를 CPU 메모리로 옮긴 뒤 NumPy 배열로 변환
  cos_sim = cos_sim_tensor.cpu().numpy()

# 이미지/텍스트를 CLIP 벡터로 바꾸는 모델 추론 시간이 어느 정도인지 확인
print(f"CLIP feature encoding time: {encode_time:.6f}s")
# 이미지-텍스트 쌍을 하나씩 비교하면 얼마나 걸리는지 확인
print(f"1:1 loop similarity time: {loop_time:.6f}s")
# 같은 비교를 N:M 행렬곱 한 번으로 처리하면 얼마나 걸리는지 확인
print(f"N:M matrix similarity time: {matrix_time:.6f}s")
# 전체 비교 기준으로 행렬곱 방식이 1:1 반복 방식보다 몇 배 빠른지 확인
print(f"Matrix speedup: {loop_time / max(matrix_time, 1e-12):.2f}x")

# 두 방식이 같은 코사인 유사도 행렬을 만드는지 최대 오차로 체크
max_diff = (loop_scores - cos_sim_tensor).abs().max().item()
print(f"Max difference between two methods: {max_diff:.8f}")

# 시각화 설정
num_frames = len(frame_files)
fig = plt.figure(figsize=(16, num_frames * 1.5))
# grid 설정
gs = gridspec.GridSpec(1, 2, width_ratios=[3, 1])

# 코사인 유사도와 softmax 상대 점수를 함께 표시
heatmap_labels = [
    [f"{cos_sim[i, j]:.3f} ({probs[i, j]:.2f})" for j in range(len(queries))]
    for i in range(num_frames)
]
# 히트맵 색상은 cos_sim 기준, 셀 텍스트에는 cos_sim과 probs를 함께 표시
ax_cos = plt.subplot(gs[0])
sns.heatmap(cos_sim, annot=heatmap_labels, cmap='Blues', fmt="",
            xticklabels=queries,
            yticklabels=[f"Frame {i+1}" for i in range(num_frames)],
            ax=ax_cos, cbar_kws={"shrink": 0.5})
ax_cos.set_title(
    "CLIP Cosine Similarity (Softmax Relative Score)", 
    fontsize=15, 
    pad=20
)

# 오른쪽에 이미지 그리기 
gs_img = gridspec.GridSpecFromSubplotSpec(num_frames, 1, subplot_spec=gs[1])
for i, img_path in enumerate(frame_files):
  ax_img = plt.subplot(gs_img[i])
  img = Image.open(img_path)
  ax_img.imshow(img)
  ax_img.axis('off') # 축 숨기기
  ax_img.set_title(f"Frame {i+1}", fontsize=9)

plt.tight_layout()
plt.show()

    image_features = model.encode_image(image_tensors)
    text_features = model.encode_text(text_tokens)

dataloader = DataLoader(
    dataset,                   # 배치마다 샘플을 꺼내는 Dataset
    batch_size=batch_size,     # 한 스텝에서 묶어서 모델에 넘길 이미지 개수
    shuffle=False,             # 순서 섞기 여부
    num_workers=num_workers,   # 디스크 로딩·전처리용 서브프로세스 수 (0이면 메인만)
    pin_memory=pin,            # pinned memory 사용 (CUDA→GPU 복사 최적화 등, MPS에선 효과 제한적일 수 있음)
)

docker pull qdrant/qdrant
docker run -p 6333:6333 -p 6334:6334 \
    -v "$(pwd)/qdrant_storage:/qdrant/storage:z" \
    qdrant/qdrant

curl -X PUT http://localhost:6333/collections/test_collection \
  -H 'Content-Type: application/json' \
  -d '{
    "vectors": { "size": 4, "distance": "Cosine" }
  }'

curl -X PUT http://localhost:6333/collections/test_collection/points \
  -H 'Content-Type: application/json' \
  -d '{
    "points": [
      { "id": 1, "vector": [0.05, 0.61, 0.76, 0.74], "payload": {"city": "Seoul", "age": 26} },
      { "id": 2, "vector": [0.19, 0.81, 0.75, 0.11], "payload": {"city": "Tokyo", "age": 30} }
    ]
  }'

curl -X POST http://localhost:6333/collections/test_collection/points/search \
  -H 'Content-Type: application/json' \
  -d '{
    "vector": [0.2, 0.1, 0.9, 0.7],
    "filter": {
        "must": [ { "key": "city", "match": { "value": "Seoul" } } ]
    },
    "limit": 3
  }'

def extract_frames_ffmpeg(
    *,  # 키워드 전용
    video_path: Path,  # 입력 동영상 경로
    output_dir: Path,  # 추출 JPG 저장 디렉터리
    ffmpeg_bin: str | None = None,  # ffmpeg 실행 경로(None이면 자동 해석)
    fps: float | None = None,  # 초당 추출 장수(-vf fps, None이면 생략)
    max_frames: int | None = None,  # 최대 장수(None이면 제한 없음)
    jpeg_quality: int = 2,  # JPEG -q:v(작을수록 보통 고화질)
) -> int:  # 성공 시 frame_*.jpg 개수
    if not video_path.is_file():
        raise FileNotFoundError(f"동영상 파일이 없습니다: {video_path}")

    output_dir.mkdir(parents=True, exist_ok=True)
    for old in output_dir.glob("frame_*.jpg"):
        old.unlink()

    exe = resolve_ffmpeg_exe(ffmpeg_bin)
    # ffmpeg 옵션
    cmd: list[str] = [
        exe,
        "-hide_banner",
        "-loglevel",
        "error",
        "-y",
        "-i",
        str(video_path),
    ]
    if fps is not None and fps > 0:
        cmd.extend(["-vf", f"fps={fps}"])
    cmd.extend(["-q:v", str(jpeg_quality)])
    if max_frames is not None and max_frames > 0:
        cmd.extend(["-frames:v", str(max_frames)])
    cmd.append(str(output_dir / "frame_%06d.jpg"))

    # 추출 수행
    proc = subprocess.run(cmd, capture_output=True, text=True, check=False)
    if proc.returncode != 0:
        err = (proc.stderr or proc.stdout or "").strip()
        raise RuntimeError(f"ffmpeg 실패 (exit {proc.returncode}): {err or 'no stderr'}")

    return len(list(output_dir.glob("frame_*.jpg")))

def extract_features(
        self,  # CLIP 서비스(self.model·preprocess·device)
        frames_dir: str,  # 프레임 JPG 디렉터리
        *,  # 키워드 전용
        batch_size: int,  # 배치 크기(한 스텝에 넣을 이미지 수)
        num_workers: int,  # DataLoader 워커 수(0이면 메인만)
        verbose: bool = True,  # 로그·요약 출력 여부
        log_batches_every: int = 10,  # N 배치마다 진행 로그(0이면 생략)
    ) -> Tuple[np.ndarray, float, float]:
        # jpg 목록 가져오기
        all_frame_files = list_frame_jpgs(frames_dir)
        if not all_frame_files:
            return self._empty_features(), 0.0, 0.0

        # dataset에서 인덱스마다 파일을 읽고 그때 self.preprocess를 적용하게 생성
        dataset = AnimeFrameDataset(all_frame_files, self.preprocess)
        pin = self.device == "cuda"
        # Dataloader로 배치, 병렬처리 지정
        dataloader = DataLoader(
            dataset,
            batch_size=batch_size,
            shuffle=False,
            num_workers=num_workers,
            pin_memory=pin,
        )

        all_features: List[np.ndarray] = []
        start_time = time.perf_counter()
        # 추론할 때 그래프/그래디언트 비활성화해서 속도 향상
        # 학습을 위해 켜지는 옵션을 비활성화해서 추론 비용을 가볍게 처리
        with torch.no_grad():
            # 프레임 파일을 배치 단위로 순회
            for batch_idx, images in enumerate(dataloader):
                images = images.to(self.device)
                # 모델로 벡터(임베딩) 추출
                features = self.model.encode_image(images)
                # L2 정규화
                features /= features.norm(dim=-1, keepdim=True)
                all_features.append(features.cpu().numpy())

                if verbose and log_batches_every > 0 and (batch_idx + 1) % log_batches_every == 0:
                    print(f"Processed batch {batch_idx + 1}/{len(dataloader)}...")

        _sync_device(self.device)
        # 배치 단위로 나뉜 임베딩을 “프레임 N개 × 차원 D”처럼 하나의 레이아웃으로 합침
        final_matrix = np.vstack(all_features)
        elapsed = time.perf_counter() - start_time
        frames_per_sec = len(all_frame_files) / elapsed if elapsed > 0 else 0.0

        if verbose:
            print(f"  >> Inference Time: {elapsed:.2f}s ({frames_per_sec:.2f} frames/sec)")
        return final_matrix, elapsed, frames_per_sec

def upsert_job_frame_vectors(
    *,  # 키워드 전용
    job_public_id: uuid.UUID,  # 잡 단위 UUID — 포인트 id 생성에 사용
    anime_slug: str,  # 작품 슬러그
    episode: int | None,  # 화 번호(없으면 None)
    genre_slugs: list[str],  # 장르 슬러그 목록
    frame_indices: list[int],  # 프레임 인덱스(N개)
    vectors: Any,  # 임베딩 (N, D)
    timestamps: list[float],  # 재생 시각 초(N개)
    frame_files: list[str],  # 프레임 파일 경로(N개)
) -> None:
    # _get_client: Qdrant Client 로드
    pair = _get_client()
    if pair[0] is None:
        return
    client, collection = pair
    try:
        # PointStruct 포인트를 만들 때 쓰는 클래스
        from qdrant_client.models import PointStruct
    except ImportError:
        return

    n = len(frame_indices)
    if n == 0:
        return
    if len(timestamps) != n or len(frame_files) != n:
        raise ValueError("frame_indices, timestamps, frame_files 길이가 일치해야 합니다.")

    # 필요한 경우에만 로드, 앞에서 반환할 수 있으니
    import numpy as np

    mat = np.asarray(vectors, dtype=np.float32)
    if mat.ndim != 2 or mat.shape[0] != n:
        raise ValueError(f"vectors 는 (N, D) 형태여야 합니다. N={n}, shape={mat.shape}")

    dim = int(mat.shape[1])
    # 컬렉션 생성 및 페이로드에 대한 인덱스 생성
    ensure_collection_and_indexes(vector_dim=dim)

    points: list[Any] = []
    for i in range(n):
        fid = frame_indices[i]
        vec = mat[i]
        # 포인트 id 생성
        pid = frame_point_uuid(job_public_id, fid)
        payload = build_frame_payload(
            anime_slug=anime_slug,
            job_public_id=job_public_id,
            frame_index=fid,
            frame_file=frame_files[i],
            episode=episode,
            genre_slugs=genre_slugs,
            timestamp_sec=float(timestamps[i]),
        )
        # 포인트: Qdrant의 데이터 단위
        points.append(
            PointStruct(
                id=str(pid),
                vector=vec.tolist(),
                payload=payload,
            )
        )

    client.upsert(collection_name=collection, points=points, wait=True)

curl -sS 'http://127.0.0.1:6333/collections/anime_clip'
{"status":{"error":"Not found: Collection `anime_clip` doesn't exist!"},"time":0.000179333}%

curl -sS 'http://127.0.0.1:6333/collections/anime_clip' | jq

{
  "result": {
    "status": "green",
    "optimizer_status": "ok",
    "indexed_vectors_count": 0,
    "points_count": 1542,
    "segments_count": 4,
    "config": {
      "params": {
        "vectors": { "size": 768, "distance": "Cosine" },
        "shard_number": 1,
        "replication_factor": 1,
        "write_consistency_factor": 1,
        "on_disk_payload": true
      },
      "hnsw_config": {
        "m": 16,
        "ef_construct": 100,
        "full_scan_threshold": 10000,
        "max_indexing_threads": 0,
        "on_disk": false
      },
      "optimizer_config": {
        "deleted_threshold": 0.2,
        "vacuum_min_vector_number": 1000,
        "default_segment_number": 0,
        "max_segment_size": null,
        "memmap_threshold": null,
        "indexing_threshold": 10000,
        "flush_interval_sec": 5,
        "max_optimization_threads": null,
        "prevent_unoptimized": null
      },
      "wal_config": {
        "wal_capacity_mb": 32,
        "wal_segments_ahead": 0,
        "wal_retain_closed": 1
      },
      "quantization_config": null
    },
    "payload_schema": {
      "frame_index": { "data_type": "integer", "points": 1542 },
      "frame_file": { "data_type": "keyword", "points": 1542 },
      "anime_id": { "data_type": "keyword", "points": 1542 },
      "genre": { "data_type": "keyword", "points": 1542 },
      "episode": { "data_type": "integer", "points": 1542 },
      "job_public_id": { "data_type": "keyword", "points": 1542 }
    },
    "update_queue": { "length": 0 }
  },
  "status": "ok",
  "time": 0.002697042
}

# 저장 건수만 다시 확인
curl -sS -X POST 'http://127.0.0.1:6333/collections/anime_clip/points/count' \
  -H 'Content-Type: application/json' \
  -d '{"exact": true}' | jq

{
  "result": {
    "count": 1542
  },
  "status": "ok",
  "time": 0.0015215
}

# payload만 샘플 3건 (id 확인용)
curl -sS -X POST 'http://127.0.0.1:6333/collections/anime_clip/points/scroll' \
  -H 'Content-Type: application/json' \
  -d '{"limit": 3, "with_payload": true, "with_vector": false}' | jq

{
  "result": {
    "points": [
      {
        "id": "0003386b-da20-5cd3-a190-c163c1dd8af7",
        "payload": {
          "anime_id": "123",
          "job_public_id": "0e0f35d4-bdcd-4a08-8f4d-746d861487c5",
          "frame_index": 1015,
          "frame_file": "frame_001016.jpg",
          "timestamp_sec": 42.291666666666664,
          "episode": 1,
          "genre": [
            "g-612048e5"
          ]
        }
      },
      {
        "id": "0077db65-24ab-5325-b904-f49ff6937df8",
        "payload": {
          "anime_id": "123",
          "job_public_id": "0e0f35d4-bdcd-4a08-8f4d-746d861487c5",
          "frame_index": 1393,
          "frame_file": "frame_001394.jpg",
          "timestamp_sec": 58.041666666666664,
          "episode": 1,
          "genre": [
            "g-612048e5"
          ]
        }
      },
      {
        "id": "00b12149-49e8-505c-9af6-44e0dfd38f5e",
        "payload": {
          "anime_id": "123",
          "job_public_id": "0e0f35d4-bdcd-4a08-8f4d-746d861487c5",
          "frame_index": 327,
          "frame_file": "frame_000328.jpg",
          "timestamp_sec": 13.625,
          "episode": 1,
          "genre": [
            "g-612048e5"
          ]
        }
      }
    ],
    "next_page_offset": "00b80937-7c2c-5bb9-9b12-e262b2350f31"
  },
  "status": "ok",
  "time": 0.011248125
}

# Redis Docker
docker run -d --name redis -p 6379:6379 redis:7-alpine

# Celery worker (Django 프로젝트 루트, 가상환경 활성화 후)
celery -A anime_search worker -l info --concurrency=1

# settings.py — Celery (임베딩 잡 비동기 실행)
CELERY_BROKER_URL = os.environ.get("CELERY_BROKER_URL", "redis://127.0.0.1:6379/0").strip()
CELERY_RESULT_BACKEND = os.environ.get("CELERY_RESULT_BACKEND", CELERY_BROKER_URL).strip()
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = int(os.environ.get("CELERY_TASK_TIME_LIMIT", "3600"))
CELERY_WORKER_CONCURRENCY = int(os.environ.get("CELERY_WORKER_CONCURRENCY", "1"))
# 로컬(M1)에서는 위 값보다 `celery worker --concurrency=1` CLI가 실제 동시 실행 수를 정함
CELERY_TASK_ALWAYS_EAGER = os.environ.get("CELERY_TASK_ALWAYS_EAGER", "").strip().lower() in (
    "1",
    "true",
    "yes",
)

app = Celery("anime_search")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

sqlite> .schema catalog_anime_genres

CREATE TABLE IF NOT EXISTS "catalog_anime_genres" (
    "id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
    "anime_id" bigint NOT NULL
        REFERENCES "catalog_anime" ("id")
        DEFERRABLE INITIALLY DEFERRED,
    "genre_id" bigint NOT NULL
        REFERENCES "catalog_genre" ("id")
        DEFERRABLE INITIALLY DEFERRED
);

CREATE UNIQUE INDEX "catalog_anime_genres_anime_id_genre_id_c8680c2f_uniq"
    ON "catalog_anime_genres" ("anime_id", "genre_id");

CREATE INDEX "catalog_anime_genres_anime_id_c2f96e25"
    ON "catalog_anime_genres" ("anime_id");

CREATE INDEX "catalog_anime_genres_genre_id_8b6a7b27"
    ON "catalog_anime_genres" ("genre_id");

...

urlpatterns = [
    # 메인 페이지, 잡 목록 보기 페이지랑 뷰 겸용
    path("", views.job_console, name="catalog_home"),
    # 장르 리스트랑 수정, 등록 기능
    path("genres/", views.genre_list, name="catalog_genre_list"),
    path("genres/new/", views.genre_new, name="catalog_genre_new"),
    path("genres/bulk-new/", views.genre_bulk_new, name="catalog_genre_bulk_new"),
    path("genres/<int:pk>/edit/", views.genre_edit, name="catalog_genre_edit"),
    # 업로드
    path("upload/", views.anime_upload, name="catalog_anime_upload"),
    # 업로드 후 영상 미리보기
    path(
        "jobs/<uuid:public_id>/preview/<str:filename>",
        views.serve_uploaded_video,
        name="catalog_job_video_preview",
    ),
    # Job 목록 페이지
    path("jobs/", views.job_console, name="catalog_jobs"),
    # Job JSON API
    # Job 목록 가져오기
    path("api/jobs/", views.job_api_list, name="catalog_job_api_list"),
    # Job 수동 실행(자동 enqueue 외에 재실행·지연 실행용): job_api_run → enqueue_run_job
    path(
        "api/jobs/<uuid:public_id>/run/",
        views.job_api_run,
        name="catalog_job_api_run",
    ),
    # Job 삭제하기
    path(
        "api/jobs/<uuid:public_id>/delete/",
        views.job_api_delete,
        name="catalog_job_api_delete",
    ),
]

@require_http_methods(["GET", "POST"])
def anime_upload(request: HttpRequest) -> HttpResponse:
    # 로그인 필요 여부 체크, 개발 단계에서는 체크 안함
    gate = _ui_gate(request)
    if gate:
        return gate

    # 애니메이션 시리즈 구분하기 위한 리스트 가져오기
    animes = list(Anime.objects.order_by("slug"))
    # 장르 리스트 가져오기
    genres = list(Genre.objects.order_by("sort_order", "slug"))

    # 업로드
    if request.method == "POST":
        slug = (request.POST.get("anime_slug") or "").strip()
        # slug 유효성 검사
        if not is_valid_anime_id(slug):
            msg = "시리즈 slug는 영문·숫자·_- 만 1~255자여야 합니다."
            if _wants_json(request):
                return JsonResponse({"detail": msg}, status=400)
            messages.error(request, msg)
            return redirect("catalog_anime_upload")

        # 애니메이션이 없으면 새로 만들고 있으면 그대로 사용
        title = (request.POST.get("title") or "").strip()
        anime, _ = Anime.objects.get_or_create(slug=slug, defaults={"title": title})
        # title은 선택(없으면 slug만으로 생성)
        if title:
            anime.title = title
            anime.save(update_fields=["title", "updated_at"])

        # 장르 N:M 연결
        ids = [int(x) for x in request.POST.getlist("genre_ids") if x.isdigit()]
        anime.genres.set(Genre.objects.filter(pk__in=ids))

        # 에피소드 유효성 검사
        ep_raw = (request.POST.get("episode") or "").strip()
        if not ep_raw:
            msg = "episode(화수)는 필수입니다."
            if _wants_json(request):
                return JsonResponse({"detail": msg}, status=400)
            messages.error(request, msg)
            return redirect("catalog_anime_upload")
        try:
            episode_number = int(ep_raw)
            if episode_number < 1:
                raise ValueError
        except ValueError:
            msg = "episode는 1 이상 정수여야 합니다."
            if _wants_json(request):
                return JsonResponse({"detail": msg}, status=400)
            messages.error(request, msg)
            return redirect("catalog_anime_upload")

        episode_row = get_or_create_episode(anime=anime, number=episode_number)
        # Job 생성(anime·episode FK로 연결)
        job = EmbeddingJob.objects.create(anime=anime, episode=episode_row)
        # 잡 스테이징 디렉터리(jobs/<id>/frames, input/) 생성
        frames_leaf = ensure_job_staging_dirs(job)
        input_dir = staging_input_dir_for_job_frames(frames_leaf)

        dest_name: str | None = None
        f = request.FILES.get("video")
        if f:
            try:
                dest_name = _safe_video_name(f.name)
            except ValueError as exc:
                job.delete()
                if _wants_json(request):
                    return JsonResponse({"detail": str(exc)}, status=400)
                messages.error(request, str(exc))
                return redirect("catalog_anime_upload")
            # pathlib로 경로 조합(문자열 split/join 대신)
            dest = input_dir / dest_name
            with dest.open("wb") as out:
                # Django의 chunks()로 스트리밍 저장(기본 청크 64KB)
                for chunk in f.chunks():
                    out.write(chunk)

        # TODO: DB·파일 저장을 transaction.atomic() 등으로 묶기
        # DB 커밋 후 Celery enqueue 예약(on_commit; atomic 블록 안이면 즉시 실행)
        schedule_auto_enqueue_on_commit(job.public_id)

        if _wants_json(request):
            payload: dict[str, str | None] = {
                "public_id": str(job.public_id),
                "filename": dest_name,
                "preview_url": _video_preview_url(request, job, dest_name) if dest_name else None,
                "jobs_url": request.build_absolute_uri(reverse("catalog_jobs")),
            }
            if dest_name:
                payload["message"] = f"작업 생성됨 · {job.public_id} · 동영상 저장됨 · 처리 큐에 넣는 중"
            else:
                payload["message"] = (
                    f"작업 생성됨 · {job.public_id} · input 폴더에 동영상을 넣은 뒤 "
                    f"/jobs/ 에서 실행하세요"
                )
            return JsonResponse(payload)

        if dest_name:
            messages.success(
                request,
                f"작업 생성됨 · {job.public_id} · 동영상 저장됨 · 처리 큐에 넣는 중",
            )
        else:
            messages.success(
                request,
                f"작업 생성됨 · {job.public_id} · input 폴더에 동영상을 넣은 뒤 /jobs/ 에서 실행하세요",
            )
        return redirect("catalog_jobs")

    # 업로드 페이지 보여주기
    return render(
        request,
        "catalog/anime_upload.html",
        {"animes": animes, "genres": genres},
    )

def schedule_auto_enqueue_on_commit(public_id: UUID) -> None:
    """DB 커밋 후 자동 enqueue."""
    pid = public_id

    def _enqueue() -> None:
        try_auto_enqueue_job(pid)

    transaction.on_commit(_enqueue)

def try_auto_enqueue_job(public_id: UUID) -> bool:
    """
    pending 잡을 Celery에 넣는다.
    input/ 동영상 없음 등으로 거부되면 False, enqueue 성공 시 True.
    """
    from embeddings.services.job_dispatch import JobDispatchError, enqueue_run_job

    try:
        enqueue_run_job(public_id)
    except JobDispatchError as exc:
        logger.info("auto enqueue skipped public_id=%s: %s", public_id, exc)
        return False
    except Exception:
        logger.exception("auto enqueue failed public_id=%s", public_id)
        return False
    return True

...

class JobDispatchError(Exception):
    """enqueue 거부 (상태·중복 등)."""

    def __init__(self, message: str, *, status_code: int = 400) -> None:
        super().__init__(message)
        self.status_code = status_code


@dataclass(frozen=True)
class EnqueueResult:
    public_id: str
    celery_task_id: str
    status: str


def _save_task_id(job: EmbeddingJob, async_result) -> str:
    task_id = str(async_result.id)
    job.celery_task_id = task_id
    job.save(update_fields=["celery_task_id", "updated_at"])
    return task_id


def _require_input_video(job: EmbeddingJob) -> None:
    err = check_job_input_video(job)
    if err:
        raise JobDispatchError(err, status_code=400)


def enqueue_run_job(public_id: UUID) -> EnqueueResult:
    """
    pending 잡을 검증·processing 표시 후 Celery에 단건 실행을 넣음
    """
    from embeddings.tasks import run_embedding_job_task

    with transaction.atomic():
        # 잡 조회 (select_for_update로 행 잠금)
        job = (
            EmbeddingJob.objects.select_for_update()
            .filter(public_id=public_id)
            .first()
        )
        if job is None:
            raise JobDispatchError("작업을 찾을 수 없습니다.", status_code=404)

        # 잡 상태 검증
        if job.status == EmbeddingJob.Status.PROCESSING:
            raise JobDispatchError(
                f"이미 처리 중입니다 (task={job.celery_task_id or '—'})",
                status_code=409,
            )
        if job.status != EmbeddingJob.Status.PENDING:
            raise JobDispatchError(
                f"pending 만 실행 가능합니다 (현재 {job.status})",
                status_code=400,
            )
        # 스테이징 폴더(frames/, input/)만 생성
        ensure_job_staging_dirs(job)
        # 큐에 넣기 전 input/ 동영상 필수 체크
        _require_input_video(job)

        # 잡 상태 업데이트(processing)
        job.status = EmbeddingJob.Status.PROCESSING
        job.last_error = ""
        job.save(update_fields=["status", "last_error", "updated_at"])

    # Celery에 해당 작업을 넣음, Redis 브로커에 메시지 적재
    async_result = run_embedding_job_task.delay(str(public_id))
    task_id = _save_task_id(job, async_result)
    return EnqueueResult(
        public_id=str(public_id),
        celery_task_id=task_id,
        status=job.status,
    )

{
  "body": "W1siMGQzOGFjNjMtZDM5OS00MzcxLWJjM2EtOTRlMjJhMjIyNGQwIl0sIHt9LCB7ImNhbGxiYWNrcyI6IG51bGwsICJlcnJiYWNrcyI6IG51bGwsICJjaGFpbiI6IG51bGwsICJjaG9yZCI6IG51bGx9XQ==",
  "content-encoding": "utf-8",
  "content-type": "application/json",
  "headers": {
    "lang": "py",
    "task": "embeddings.run_embedding_job",
    "id": "2ed7a75a-4c40-459a-9275-7c49f3cddf17",
    "shadow": null,
    "eta": null,
    "expires": null,
    "group": null,
    "group_index": null,
    "retries": 0,
    "timelimit": [null, null],
    "root_id": "2ed7a75a-4c40-459a-9275-7c49f3cddf17",
    "parent_id": null,
    "argsrepr": "('0d38ac63-d399-4371-bc3a-94e22a2224d0',)",
    "kwargsrepr": "{}",
    "origin": "gen32524@jiseunglyeol-ui-MacBookAir.local",
    "ignore_result": false,
    "replaced_task_nesting": 0,
    "stamped_headers": null,
    "stamps": {}
  },
  "properties": {
    "correlation_id": "2ed7a75a-4c40-459a-9275-7c49f3cddf17",
    "reply_to": "9950199d-4293-39ea-bd9e-5a400d686cef",
    "delivery_mode": 2,
    "delivery_info": {
      "exchange": "",
      "routing_key": "celery"
    },
    "priority": 0,
    "body_encoding": "base64",
    "delivery_tag": "88c17946-d9c8-4704-9035-d8e204de4c28"
  }
}

@shared_task(name="embeddings.run_embedding_job")
def run_embedding_job_task(public_id: str) -> dict[str, str]:
    """단건 잡: enqueue_run_job에서 processing으로 올린 뒤 워커가 실행."""
    started = time.monotonic()
    uid = UUID(public_id)

    # job에서 select_related를 통해 anime와 같이 join 해서 가져옴
    job = EmbeddingJob.objects.select_related("anime").filter(public_id=uid).first()

    if job is None:
        logger.error("run_embedding_job_task: job not found public_id=%s", public_id)
        return {"public_id": public_id, "status": "not_found"}

    slug = job.anime.slug
    logger.info(
        "embedding job start public_id=%s anime=%s status=%s",
        public_id,
        slug,
        job.status,
    )

    try:
        # Job 상태 체크 
        if job.status != EmbeddingJob.Status.PROCESSING:
            logger.warning(
                "embedding job skip unexpected status public_id=%s status=%s",
                public_id,
                job.status,
            )
            return {"public_id": public_id, "status": job.status}

        # enqueue와 동일: input/ 동영상 필수(ffmpeg 추출 전 재검증)
        input_err = check_job_input_video(job)
        if input_err:
            _mark_failed(job, RuntimeError(input_err))
            return {"public_id": public_id, "status": EmbeddingJob.Status.FAILED}

        # 4장 파이프라인 (보통 running → done/failed는 여기서 갱신)
        run_single_embedding_job(job)
        # 파이프라인이 바꾼 status·필드를 DB에서 다시 읽음
        job.refresh_from_db()
        elapsed = time.monotonic() - started
        logger.info(
            "embedding job done public_id=%s anime=%s status=%s elapsed_sec=%.2f",
            public_id,
            slug,
            job.status,
            elapsed,
        )
        return {"public_id": public_id, "status": job.status}
    except Exception as exc:  # noqa: BLE001
        logger.exception(
            "embedding job failed public_id=%s anime=%s",
            public_id,
            slug,
        )
        job.refresh_from_db()
        if job.status != EmbeddingJob.Status.FAILED:
            _mark_failed(job, exc)
        return {"public_id": public_id, "status": EmbeddingJob.Status.FAILED}

@csrf_exempt
@require_POST
def chat_api(request: HttpRequest) -> JsonResponse:
    # session·message 파싱, get_or_create_session (생략)

    try:
        reply, scenes = run_chat_turn(session=session, user_message=message, request=request)
    except ChatOrchestratorError as exc:
        return JsonResponse({"detail": str(exc)}, status=503)
    except Exception as exc:  # noqa: BLE001
        return JsonResponse({"detail": str(exc)}, status=500)

    return JsonResponse(
        {
            "session_id": str(session.id),
            "reply": reply,
            "scenes": scenes,
        }
    )