chore: reset repository history - ArtifactNet HF Spaces Demo v8.0

.gitattributes

@@ -0,0 +1,35 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,40 @@
+# Python
+__pycache__/
+*.py[cod]
+*.egg-info/
+*.egg
+dist/
+build/
+
+# IP Protection note: core/proprietary.py contains obfuscated algorithms
+# (난독화된 알고리즘으로 특허 핵심 보호)
+
+# Models (downloaded at runtime from HF Hub)
+*.onnx
+*.pt
+*.onnx.data
+
+# Environment
+.env
+.venv/
+venv/
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Gradio
+flagged/
+
+# Development files (not needed in HF Spaces)
+CLAUDE.md
+.claude/
+local_demo_v77.py
+testing/
+trash/

Dockerfile.youtube-proxy

@@ -0,0 +1,62 @@
+# Multi-stage Dockerfile for YouTube Proxy Server
+
+FROM python:3.11-slim as builder
+
+# Install build dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    && rm -rf /var/lib/apt/lists/*
+
+# Create virtual environment
+RUN python -m venv /opt/venv
+ENV PATH="/opt/venv/bin:$PATH"
+
+# Copy and install Python dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# ============================================================
+# Final stage
+# ============================================================
+
+FROM python:3.11-slim
+
+# Install runtime dependencies (ffmpeg for yt-dlp)
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    ffmpeg \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy virtual environment from builder
+COPY --from=builder /opt/venv /opt/venv
+ENV PATH="/opt/venv/bin:$PATH"
+
+# Create non-root user for security (use UID 1001 to avoid conflicts)
+RUN useradd -m -u 1001 appuser 2>/dev/null || true
+
+# Set working directory
+WORKDIR /app
+
+# Copy application
+COPY youtube_proxy_server.py .
+
+# Change ownership
+RUN chown -R appuser:appuser /app 2>/dev/null || true
+
+# Switch to non-root user
+USER appuser
+
+# Expose port
+EXPOSE 8765
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
+    CMD python -c "import requests; requests.get('http://localhost:8765/health')" || exit 1
+
+# Default environment variables
+ENV HOST=0.0.0.0
+ENV PORT=8765
+ENV LOG_LEVEL=INFO
+ENV YOUTUBE_PROXY_API_KEY=default-key
+
+# Run application
+CMD ["python", "youtube_proxy_server.py"]

HF_SPACES_ENV.md

@@ -0,0 +1,139 @@
+# HF Spaces 환경변수 설정 가이드
+
+YouTube 프록시를 통해 HF Spaces 앱에서 YouTube URL 다운로드를 활성화하려면 다음 환경변수를 설정하세요.
+
+## 설정 단계
+
+### 1. cloudflared를 통한 외부 접근
+
+youtube-proxy 서비스는 `youtube-proxy.intrect.io`를 통해 접근 가능합니다 (Cloudflare Tunnel 역프록시).
+
+### 2. HF Spaces 시크릿 설정
+
+HF Spaces 설정에서 다음 환경변수를 추가하세요:
+
+#### `YOUTUBE_PROXY_URL`
+```
+https://youtube-proxy.intrect.io
+```
+
+#### `YOUTUBE_PROXY_API_KEY`
+```
+c60ba3dc9f26cfc700958983f82b997eac084743aad9f5be4db7bb625ae6dbbd
+```
+
+이는 `docker-compose.youtube-proxy.yml`의 `YOUTUBE_PROXY_API_KEY` 환경변수와 **정확히 동일**해야 합니다.
+
+## 인증 흐름
+
+1. HF Spaces 앱이 YouTube URL을 받으면
+2. `YOUTUBE_PROXY_URL` 및 `YOUTUBE_PROXY_API_KEY` 사용
+3. `https://youtube-proxy.intrect.io/download-youtube` 엔드포인트로 POST 요청
+4. `Authorization: Bearer {YOUTUBE_PROXY_API_KEY}` 헤더 포함
+5. 프록시 서버가 yt-dlp로 다운로드
+6. WAV 파일 반환
+
+## 보안 고려사항
+
+- API 키는 **절대 공개하지 마세요**
+- cloudflared 역프록시를 통해서만 접근 가능 (외부 포트 노출 없음)
+- 컨테이너는 `proxy` 사용자로 실행 (root 아님)
+- 최소 권한 원칙 준수
+
+## 문제 해결
+
+### HF Spaces에서 연결 실패
+
+1. cloudflared 상태 확인:
+   ```bash
+   sudo systemctl status cloudflared
+   ```
+
+2. youtube-proxy 컨테이너 상태 확인:
+   ```bash
+   docker ps | grep youtube-proxy
+   docker logs artifactnet-youtube-proxy
+   ```
+
+3. DNS 확인:
+   ```bash
+   curl -I https://youtube-proxy.intrect.io/health
+   ```
+
+### API 키 불일치
+
+`docker-compose.youtube-proxy.yml`의 `YOUTUBE_PROXY_API_KEY`와 HF Spaces의 `YOUTUBE_PROXY_API_KEY`가 **정확히 동일**한지 확인하세요.
+
+## Rate Limiting 설정 (권장)
+
+과도한 요청과 연속 스팸으로부터 HF Spaces 및 ubuntu-mini 보호:
+
+#### `RATE_LIMIT_REQUESTS`
+```
+5
+```
+(기본값: 5회, 1시간당)
+
+#### `RATE_LIMIT_MINUTES`
+```
+60
+```
+(기본값: 60분 윈도우)
+
+#### `BURST_LIMIT_PER_MINUTE`
+```
+2
+```
+(기본값: 최대 2회/분, 연속 요청 방지)
+
+**동작:**
+- **Burst 제한**: 사용자당 2회/분 (연속 요청 방지)
+- **시간 제한**: 사용자당 5회/60분 (장기 남용 방지)
+- 둘 다 만족해야 요청 허용
+
+---
+
+## 에지 케이스 수집 설정 (선택사항)
+
+Uncertain 판정 곡의 분석 데이터를 자동으로 수집하려면:
+
+#### `UBUNTU_MINI_ENABLED`
+```
+true
+```
+
+#### `UBUNTU_MINI_HOST`
+```
+ubuntu-mini.local
+```
+
+#### `UBUNTU_MINI_PORT`
+```
+9000
+```
+
+**수집되는 것:**
+- Mel-spectrogram (30초 미만)
+- 판정 통계
+- 타임스탬프
+
+**수집되지 않는 것:**
+- 원본 오디오 파일
+- 개인 정보
+
+## 다음 단계
+
+1. Docker 컨테이너 실행:
+   ```bash
+   docker-compose -f docker-compose.youtube-proxy.yml up -d
+   ```
+
+2. 건강 체크:
+   ```bash
+   curl -H "Authorization: Bearer <your-key>" \
+     https://youtube-proxy.intrect.io/health
+   ```
+
+3. HF Spaces에서 YouTube URL 탭이 나타나면 작동 중입니다.
+
+4. Uncertain 곡이 자동으로 ubuntu-mini로 전송되는지 확인합니다.

README.md

@@ -0,0 +1,17 @@
+---
+title: ArtifactNet
+emoji: 🔍
+colorFrom: indigo
+colorTo: yellow
+sdk: gradio
+sdk_version: 5.20.1
+app_file: app.py
+pinned: false
+license: mit
+hardware: zero-a10g
+---
+
+# ArtifactNet — AI Music Forensic Detector
+
+Detect AI-generated music using deep spectral analysis and neural networks.
+# Model sync check - 20260225

app.py

@@ -0,0 +1,611 @@
+#!/usr/bin/env python3
+# Purpose: ArtifactNet HF Spaces demo — Gradio Blocks UI (3-tier verdict)
+
+"""ArtifactNet — AI Music Forensic Detector.
+
+v8.0: ArtifactUNet ONNX — CPU-only, no GPU required.
+"""
+
+import sys
+import os
+import json
+import time
+import subprocess
+import tempfile
+import warnings
+from collections import defaultdict
+from datetime import datetime, timedelta
+
+import numpy as np
+import torch
+import gradio as gr
+
+# Add demo/ directory to module path
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from config import SR
+from inference.audio_utils import load_audio_mono_tensor, get_audio_info
+from inference.e2e_model import get_model, run_e2e_inference
+from visualization.spectrogram import plot_spectrograms
+from visualization.timeline import plot_timeline
+from ui import VerdictCardBuilder, create_theme, create_header, create_about_section
+from core import compute_stats, classify
+
+warnings.filterwarnings("ignore")
+
+IS_HF_SPACES = os.environ.get("SPACE_ID") is not None
+YOUTUBE_PROXY_URL = os.environ.get("YOUTUBE_PROXY_URL", "")
+UBUNTU_MINI_HOST = os.environ.get("UBUNTU_MINI_HOST", "ubuntu-mini.local")
+UBUNTU_MINI_PORT = int(os.environ.get("UBUNTU_MINI_PORT", "9000"))
+UBUNTU_MINI_ENABLED = os.environ.get("UBUNTU_MINI_ENABLED", "false").lower() == "true"
+
+# Rate limiting settings
+RATE_LIMIT_REQUESTS = int(os.environ.get("RATE_LIMIT_REQUESTS", "5"))  # requests per hour
+RATE_LIMIT_MINUTES = int(os.environ.get("RATE_LIMIT_MINUTES", "60"))  # time window in minutes
+BURST_LIMIT_PER_MINUTE = int(os.environ.get("BURST_LIMIT_PER_MINUTE", "2"))  # max requests per minute
+
+
+# ============================================================
+# Rate Limiter (dual-window: long-term + burst protection)
+# ============================================================
+
+class RateLimiter:
+    """Per-user rate limiting with both long-term and burst protection."""
+
+    def __init__(self, max_requests: int, window_minutes: int, burst_per_minute: int):
+        self.max_requests = max_requests
+        self.window_secs = window_minutes * 60
+        self.burst_per_minute = burst_per_minute
+        self.requests = defaultdict(list)  # long-term tracking
+        self.minute_requests = defaultdict(list)  # burst tracking
+
+    def _get_client_id(self) -> str:
+        """Get client ID from Gradio request context (IP-based)."""
+        try:
+            import gradio.context as ctx
+            request = ctx.get_request()
+            if request and hasattr(request, 'client'):
+                return str(request.client[0])  # IP address
+        except Exception:
+            pass
+        return "unknown"
+
+    def is_allowed(self, client_id: str = None) -> tuple:
+        """
+        Check if request is allowed. Returns (allowed: bool, reason: str).
+        Enforces both long-term limit (5/hour) and burst limit (2/minute).
+        """
+        if client_id is None:
+            client_id = self._get_client_id()
+
+        now = datetime.now()
+
+        # ===== Check 1: Burst limit (requests per minute) =====
+        minute_cutoff = now - timedelta(seconds=60)
+        self.minute_requests[client_id] = [
+            req_time for req_time in self.minute_requests[client_id]
+            if req_time > minute_cutoff
+        ]
+
+        if len(self.minute_requests[client_id]) >= self.burst_per_minute:
+            return False, f"Too many requests in the last minute. Please wait 60 seconds."
+
+        # ===== Check 2: Long-term limit (requests per hour) =====
+        long_cutoff = now - timedelta(seconds=self.window_secs)
+        self.requests[client_id] = [
+            req_time for req_time in self.requests[client_id]
+            if req_time > long_cutoff
+        ]
+
+        if len(self.requests[client_id]) >= self.max_requests:
+            if self.requests[client_id]:
+                reset_time = self.requests[client_id][0] + timedelta(seconds=self.window_secs)
+                reset_str = reset_time.strftime("%H:%M UTC")
+                return False, f"Hourly limit reached ({self.max_requests}). Try again at {reset_str}."
+            return False, f"Hourly limit reached. Please wait."
+
+        # Both checks passed - add request
+        self.requests[client_id].append(now)
+        self.minute_requests[client_id].append(now)
+        return True, ""
+
+    def get_remaining(self, client_id: str = None) -> dict:
+        """Get remaining requests for both limits."""
+        if client_id is None:
+            client_id = self._get_client_id()
+
+        now = datetime.now()
+
+        # Long-term
+        long_cutoff = now - timedelta(seconds=self.window_secs)
+        long_reqs = [r for r in self.requests[client_id] if r > long_cutoff]
+        long_remaining = self.max_requests - len(long_reqs)
+
+        # Burst
+        minute_cutoff = now - timedelta(seconds=60)
+        minute_reqs = [r for r in self.minute_requests[client_id] if r > minute_cutoff]
+        minute_remaining = self.burst_per_minute - len(minute_reqs)
+
+        return {
+            "hourly": long_remaining,
+            "per_minute": minute_remaining
+        }
+
+
+# Global rate limiter
+rate_limiter = RateLimiter(RATE_LIMIT_REQUESTS, RATE_LIMIT_MINUTES, BURST_LIMIT_PER_MINUTE)
+
+
+# ============================================================
+# Uncertain case collection (edge case detection)
+# ============================================================
+
+def _extract_mel_spectrogram(audio_np: np.ndarray, mono_np: np.ndarray) -> np.ndarray:
+    """Extract mel-spectrogram for uncertain cases (CNN training)."""
+    from librosa import feature
+
+    mel_spec = feature.melspectrogram(
+        y=mono_np,
+        sr=SR,
+        n_fft=2048,
+        hop_length=512,
+        n_mels=128
+    )
+    log_mel = np.log(np.clip(mel_spec, 1e-9, None))
+    return log_mel.astype(np.float32)
+
+
+def _send_uncertain_to_ubuntu_mini(
+    mel_spec: np.ndarray,
+    verdict_stats: dict,
+    duration_sec: float,
+    source: str  # "youtube" or "upload"
+) -> bool:
+    """Send uncertain case mel-spectrogram to ubuntu-mini for edge case collection."""
+    if not UBUNTU_MINI_ENABLED:
+        return False
+
+    try:
+        import requests
+
+        # 30초 미만만 수집
+        if duration_sec > 30:
+            return False
+
+        # mel-spectrogram 직렬화 (base64)
+        mel_bytes = mel_spec.tobytes()
+        mel_b64 = __import__('base64').b64encode(mel_bytes).decode('utf-8')
+
+        payload = {
+            "mel_spectrogram": mel_b64,
+            "mel_shape": mel_spec.shape,
+            "verdict_stats": verdict_stats,
+            "duration_sec": duration_sec,
+            "source": source,
+            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
+        }
+
+        response = requests.post(
+            f"http://{UBUNTU_MINI_HOST}:{UBUNTU_MINI_PORT}/collect-uncertain",
+            json=payload,
+            timeout=5
+        )
+        return response.status_code == 200
+    except Exception as e:
+        print(f"[WARNING] Failed to send uncertain case: {e}")
+        return False
+
+
+def _send_edge_case_report(
+    verdict: str,
+    reported_verdict: str,
+    mel_spec: np.ndarray,
+    verdict_stats: dict,
+    duration_sec: float,
+    user_comment: str = ""
+) -> bool:
+    """Send edge case report (wrong verdict) to ubuntu-mini for training correction."""
+    if not UBUNTU_MINI_ENABLED:
+        return False
+
+    try:
+        import requests
+
+        # mel-spectrogram 직렬화 (base64)
+        mel_bytes = mel_spec.tobytes()
+        mel_b64 = __import__('base64').b64encode(mel_bytes).decode('utf-8')
+
+        payload = {
+            "report_type": "edge_case_correction",
+            "predicted_verdict": verdict,
+            "true_verdict": reported_verdict,
+            "mel_spectrogram": mel_b64,
+            "mel_shape": mel_spec.shape,
+            "verdict_stats": verdict_stats,
+            "duration_sec": duration_sec,
+            "user_comment": user_comment,
+            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
+        }
+
+        response = requests.post(
+            f"http://{UBUNTU_MINI_HOST}:{UBUNTU_MINI_PORT}/report-edge-case",
+            json=payload,
+            timeout=5
+        )
+        return response.status_code == 200
+    except Exception as e:
+        print(f"[WARNING] Failed to send edge case report: {e}")
+        return False
+
+
+# Proprietary algorithms moved to core module for IP protection
+
+
+# ============================================================
+# Inference wrapper (CPU-only)
+# ============================================================
+
+def _run_e2e(wav_mono_tensor):
+    """E2E inference — ArtifactUNet ONNX (CPU)."""
+    return run_e2e_inference(wav_mono_tensor)
+
+
+# ============================================================
+# YouTube URL -> audio download (local only)
+# ============================================================
+
+def _download_youtube_audio(url: str) -> str:
+    """Download audio from YouTube URL as WAV. Returns temporary file path."""
+    if YOUTUBE_PROXY_URL:
+        import requests
+        api_key = os.environ.get("YOUTUBE_PROXY_API_KEY", "").strip()
+        if not api_key:
+            raise RuntimeError("YOUTUBE_PROXY_API_KEY not set")
+
+        response = requests.post(
+            f"{YOUTUBE_PROXY_URL}/download-youtube",
+            json={"url": url},
+            headers={"Authorization": f"Bearer {api_key}"},
+            timeout=180
+        )
+
+        if response.status_code != 200:
+            try:
+                error_msg = response.json().get('detail', response.text[:200])
+            except Exception as e:
+                error_msg = response.text[:500] if response.text else f"HTTP {response.status_code} (empty response)"
+                if not error_msg or error_msg.startswith("<"):
+                    error_msg = f"HTTP {response.status_code}: {type(e).__name__}"
+            raise RuntimeError(f"Proxy server error: {error_msg}")
+
+        if not response.content:
+            raise RuntimeError("Proxy returned empty file (no audio data)")
+
+        tmpdir = tempfile.mkdtemp(prefix="artifactnet_yt_")
+        out_path = os.path.join(tmpdir, "audio.wav")
+        with open(out_path, 'wb') as f:
+            f.write(response.content)
+        return out_path
+    else:
+        tmpdir = tempfile.mkdtemp(prefix="artifactnet_yt_")
+        out_path = os.path.join(tmpdir, "audio.wav")
+        cmd = [
+            "yt-dlp",
+            "--no-playlist",
+            "-x", "--audio-format", "wav",
+            "--audio-quality", "0",
+            "--max-filesize", "50M",
+            "-o", out_path,
+            url,
+        ]
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
+        if result.returncode != 0:
+            raise RuntimeError(f"yt-dlp error: {result.stderr[:300]}")
+        for f in os.listdir(tmpdir):
+            return os.path.join(tmpdir, f)
+        raise RuntimeError("Download completed but no file found")
+
+
+def analyze_youtube(url: str):
+    """Analyze via YouTube URL (local only)."""
+    # Rate limiting check (both burst and hourly)
+    allowed, reason = rate_limiter.is_allowed()
+    if not allowed:
+        err = (
+            f"<p style='color:#ff4757;'>"
+            f"⏱️ Rate limit: {reason}<br>"
+            f"<small>Limits: {BURST_LIMIT_PER_MINUTE}/min, {RATE_LIMIT_REQUESTS}/{RATE_LIMIT_MINUTES}min</small>"
+            f"</p>"
+        )
+        return err, None, None, None, None, None, None, None, None
+
+    if not url or not url.strip():
+        return (
+            VerdictCardBuilder.build_empty_card(),
+            None, None, None, None, None, None, None, None,
+        )
+    url = url.strip()
+    try:
+        audio_path = _download_youtube_audio(url)
+    except Exception as e:
+        err = f"<p style='color:#ff4757'>Download failed: {e}</p>"
+        return err, None, None, None, None, None, None, None, None
+    return analyze_audio(audio_path)
+
+
+# ============================================================
+# Main analysis function
+# ============================================================
+
+def analyze_audio(audio_path: str):
+    """Analyze audio file -> (verdict_html, spectrogram, timeline, json_file, verdict, stats, mel_spec, duration, audio_path)."""
+    # Rate limiting check (skip if called from analyze_youtube, which already checked)
+    # Only check for direct file uploads
+    if audio_path and "artifactnet_yt_" not in audio_path:
+        allowed, reason = rate_limiter.is_allowed()
+        if not allowed:
+            err = (
+                f"<p style='color:#ff4757;'>"
+                f"⏱️ Rate limit: {reason}<br>"
+                f"<small>Limits: {BURST_LIMIT_PER_MINUTE}/min, {RATE_LIMIT_REQUESTS}/{RATE_LIMIT_MINUTES}min</small>"
+                f"</p>"
+            )
+            return err, None, None, None, None, None, None, None, None
+
+    if audio_path is None:
+        return (
+            VerdictCardBuilder.build_empty_card(),
+            None, None, None, None, None, None, None, None,
+        )
+
+    t0 = time.time()
+
+    # 1. Load audio
+    try:
+        mono_tensor, audio_np, is_stereo = load_audio_mono_tensor(audio_path)
+    except Exception as e:
+        err = f"<p style='color:#ff4757'>Error loading audio: {e}</p>"
+        return err, None, None, None, None, None, None, None, None
+
+    info = get_audio_info(audio_np, is_stereo)
+    mono_np = mono_tensor.numpy()
+
+    # 2. E2E inference (ONNX — GPU if available, else CPU)
+    chunk_probs, _ = _run_e2e(mono_tensor)
+
+    # 3. Distribution-based verdict (LGBM 2nd-stage)
+    seg_stats = compute_stats(chunk_probs)
+    elapsed = time.time() - t0
+
+    # 4. Generate visualizations
+    verdict = classify(seg_stats, seg_probs=chunk_probs, audio_path=audio_path)
+    verdict_html = VerdictCardBuilder.build(
+        verdict, seg_stats, is_stereo,
+        duration=info["duration"], elapsed=elapsed,
+    )
+
+    # Extract mel-spectrogram for edge case reporting
+    mel_spec = None
+    try:
+        mel_spec = _extract_mel_spectrogram(audio_np, mono_np)
+    except Exception as e:
+        print(f"[WARNING] Mel-spectrogram extraction failed: {e}")
+
+    # 4.5. Edge case collection (Uncertain cases only)
+    collected = False
+    source_type = "upload"
+    if verdict == "Uncertain" and UBUNTU_MINI_ENABLED:
+        try:
+            if mel_spec is not None:
+                source_type = "youtube" if "tmp" in audio_path and "yt" in audio_path else "upload"
+                collected = _send_uncertain_to_ubuntu_mini(
+                    mel_spec, seg_stats, info["duration"], source_type
+                )
+                if collected:
+                    # Add collection notice to verdict HTML
+                    verdict_html += (
+                        "<div style='background:#e8f5e9;padding:12px;border-radius:4px;margin-top:12px;'>"
+                        f"<p style='color:#2e7d32;font-size:12px;margin:0;'>"
+                        "✓ 이 분석 데이터(30초 미만의 스펙트로그램)는 모델 개선용으로 수집되었습니다.</p>"
+                        "</div>"
+                    )
+        except Exception as e:
+            print(f"[WARNING] Edge case collection failed: {e}")
+
+    spec_fig = plot_spectrograms(mono_np)
+    timeline_fig = plot_timeline(chunk_probs)
+
+    # 5. Save result JSON
+    filename = os.path.basename(audio_path) if audio_path else "unknown"
+    result_json = {
+        "filename": filename,
+        "verdict": verdict,
+        "duration_sec": round(info["duration"], 2),
+        "is_stereo": is_stereo,
+        "elapsed_sec": round(elapsed, 2),
+        "segment_stats": {k: round(v, 4) if isinstance(v, float) else v
+                          for k, v in seg_stats.items()},
+        "segment_probs": [round(p, 4) for p in chunk_probs],
+    }
+    json_path = os.path.join(tempfile.gettempdir(), "artifactnet_result.json")
+    with open(json_path, "w") as f:
+        json.dump(result_json, f, indent=2)
+
+    return verdict_html, spec_fig, timeline_fig, json_path, verdict, seg_stats, mel_spec, info["duration"], audio_path
+
+
+
+# ============================================================
+# Gradio UI
+# ============================================================
+
+def build_ui():
+    """Build Gradio Blocks UI."""
+    theme = create_theme()
+
+    with gr.Blocks(theme=theme, title="ArtifactNet — AI Music Forensic Detector") as demo:
+        # Hidden state variables to track current analysis
+        current_verdict = gr.State(value=None)
+        current_stats = gr.State(value=None)
+        current_mel_spec = gr.State(value=None)
+        current_duration = gr.State(value=None)
+        current_audio_path = gr.State(value=None)
+
+        # Header
+        gr.HTML(create_header(IS_HF_SPACES))
+
+        # Row 1: Input + Verdict
+        with gr.Row():
+            with gr.Column(scale=1):
+                if IS_HF_SPACES and not YOUTUBE_PROXY_URL:
+                    # HF Spaces without proxy: file upload only
+                    audio_input = gr.Audio(
+                        label="WAV / MP3 / FLAC (max 5 min)",
+                        type="filepath",
+                        sources=["upload"],
+                    )
+                    analyze_btn = gr.Button(
+                        "Analyze", variant="primary", size="lg",
+                    )
+                else:
+                    # Local or HF Spaces with proxy: file upload + YouTube URL tabs
+                    with gr.Tabs():
+                        with gr.TabItem("Upload File"):
+                            audio_input = gr.Audio(
+                                label="WAV / MP3 / FLAC (max 5 min)",
+                                type="filepath",
+                                sources=["upload"],
+                            )
+                            analyze_btn = gr.Button(
+                                "Analyze", variant="primary", size="lg",
+                            )
+                        with gr.TabItem("YouTube URL"):
+                            yt_url_input = gr.Textbox(
+                                label="YouTube URL",
+                                placeholder="https://www.youtube.com/watch?v=...",
+                            )
+                            yt_analyze_btn = gr.Button(
+                                "Download & Analyze", variant="primary", size="lg",
+                            )
+            with gr.Column(scale=1):
+                verdict_output = gr.HTML(
+                    value=VerdictCardBuilder.build_empty_card(),
+                    label="Verdict",
+                )
+
+        # Row 2: Spectrograms
+        with gr.Row():
+            spec_output = gr.Plot(label="Spectral Analysis")
+
+        # Row 3: Timeline + JSON download
+        with gr.Row():
+            timeline_output = gr.Plot(label="P(AI) Timeline")
+        with gr.Row():
+            json_output = gr.File(label="Result JSON", visible=True)
+
+        # Row 4: Edge case reporting (if ubuntu-mini enabled)
+        if UBUNTU_MINI_ENABLED:
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("### 틀린 판정 보고하기")
+                    report_error_type = gr.Radio(
+                        choices=["맞음", "AI인데 Human이라고 함", "Human인데 AI라고 함"],
+                        value="맞음",
+                        label="판정 결과가...",
+                        info="잘못된 판정 결과를 보고해주세요. (데이터 수집용)"
+                    )
+                    report_comment = gr.Textbox(
+                        label="추가 의견 (선택사항)",
+                        placeholder="예: 매우 압축된 음악입니다. / 너무 짧은 샘플입니다.",
+                        lines=2
+                    )
+                    report_btn = gr.Button("보고하기", variant="secondary", size="md")
+                    report_status = gr.Textbox(
+                        label="상태",
+                        interactive=False,
+                        visible=False
+                    )
+
+        with gr.Accordion("About ArtifactNet", open=False):
+            gr.HTML(create_about_section())
+
+        # Event handler for edge case reporting
+        def report_edge_case_fn(error_type, comment, verdict, stats, mel_spec, duration):
+            """Submit edge case report."""
+            if not UBUNTU_MINI_ENABLED:
+                msg = "Edge case reporting is not enabled."
+                return gr.update(value=msg, visible=True)
+
+            if error_type == "맞음" or verdict is None:
+                msg = "판정 결과를 먼저 선택해주세요."
+                return gr.update(value=msg, visible=True)
+
+            if error_type not in ["AI인데 Human이라고 함", "Human인데 AI라고 함"]:
+                msg = "오류: 유효하지 않은 선택입니다."
+                return gr.update(value=msg, visible=True)
+
+            # Convert error_type to verdict
+            true_verdict = "AI Generated" if error_type == "AI인데 Human이라고 함" else "Human-Made"
+
+            try:
+                success = _send_edge_case_report(
+                    verdict=verdict,
+                    reported_verdict=true_verdict,
+                    mel_spec=mel_spec,
+                    verdict_stats=stats,
+                    duration_sec=duration,
+                    user_comment=comment
+                )
+                if success:
+                    msg = "✓ 보고가 접수되었습니다. 감사합니다!"
+                else:
+                    msg = "△ 데이터 수집 서버에 연결할 수 없습니다. 나중에 다시 시도해주세요."
+                return gr.update(value=msg, visible=True)
+            except Exception as e:
+                msg = f"△ 오류 발생: {str(e)[:100]}"
+                return gr.update(value=msg, visible=True)
+
+        # Events
+        outputs = [verdict_output, spec_output, timeline_output, json_output, current_verdict, current_stats, current_mel_spec, current_duration, current_audio_path]
+
+        analyze_btn.click(
+            fn=analyze_audio,
+            inputs=[audio_input],
+            outputs=outputs,
+            api_name=False,
+        )
+        if not IS_HF_SPACES or YOUTUBE_PROXY_URL:
+            yt_analyze_btn.click(
+                fn=analyze_youtube,
+                inputs=[yt_url_input],
+                outputs=outputs,
+                api_name=False,
+            )
+
+        # Report button event
+        if UBUNTU_MINI_ENABLED:
+            report_btn.click(
+                fn=report_edge_case_fn,
+                inputs=[report_error_type, report_comment, current_verdict, current_stats, current_mel_spec, current_duration],
+                outputs=[report_status],
+                api_name=False,
+            )
+
+    return demo
+
+
+# ============================================================
+# Entry point — module-level demo object (required for HF Spaces)
+# ============================================================
+
+print("Loading model...", flush=True)
+get_model()
+print("Model ready.", flush=True)
+
+demo = build_ui()
+
+if __name__ == "__main__":
+    launch_kwargs = dict(server_name="0.0.0.0", server_port=7860)
+    if IS_HF_SPACES:
+        launch_kwargs["root_path"] = "/ArtifactNet"
+    demo.launch(**launch_kwargs)

config.py

@@ -0,0 +1,30 @@
+# Purpose: ArtifactNet HF Spaces demo — constants and configuration
+
+"""Global constants and HF Hub model paths."""
+
+from core import get_params
+
+# ============================================================
+# HF Hub model paths
+# ============================================================
+HF_MODEL_REPO = "intrect/artifactnet-models"
+UNET_ONNX_FILENAME = "artifactnet_e2e.onnx"
+
+# ============================================================
+# Audio constants (proprietary parameters from core module)
+# ============================================================
+SR = get_params('sr')
+MAX_DURATION_SEC = get_params('max_dur')
+CHUNK_SEC = get_params('chunk_sec')
+CHUNK_SAMPLES = int(CHUNK_SEC * SR)
+
+# ============================================================
+# E2E model constants (proprietary parameters)
+# ============================================================
+N_FFT = get_params('n_fft')
+HOP_LENGTH = get_params('hop')
+
+# ============================================================
+# E2E inference batch size
+# ============================================================
+E2E_BATCH_SIZE = get_params('batch')

core/__init__.py

@@ -0,0 +1,7 @@
+# Proprietary core algorithms (IP protected)
+
+"""Core algorithms for ArtifactNet — CONFIDENTIAL."""
+
+from .proprietary import compute_stats, classify, get_params
+
+__all__ = ['compute_stats', 'classify', 'get_params']

core/codec_aware.py

@@ -0,0 +1,32 @@
+"""Codec-aware classification module."""
+
+from pathlib import Path
+import numpy as np
+
+
+def detect_codec(audio_path):
+    """Detect if audio is lossless or lossy."""
+    if audio_path is None:
+        return 'unknown'
+
+    if isinstance(audio_path, str):
+        audio_path = Path(audio_path)
+
+    ext = audio_path.suffix.lower()
+
+    if ext in {'.wav', '.flac', '.aiff', '.aif'}:
+        return 'lossless'
+    elif ext in {'.mp3', '.aac', '.m4a', '.ogg', '.opus', '.wma'}:
+        return 'lossy'
+    else:
+        return 'unknown'
+
+
+def get_codec_thresholds(codec_mode):
+    """Get thresholds based on codec mode."""
+    if codec_mode == 'lossless':
+        return {'ai': 0.5, 'real': 0.5, 'name': 'Lossless (High Sensitivity)'}
+    elif codec_mode == 'lossy':
+        return {'ai': 0.8, 'real': 0.3, 'name': 'Lossy (Conservative)'}
+    else:
+        return {'ai': 0.7, 'real': 0.4, 'name': 'Unknown (Moderate)'}

core/proprietary.py

@@ -0,0 +1,322 @@
+# CONFIDENTIAL - ArtifactNet Proprietary Algorithms
+# Copyright (c) 2026. All rights reserved.
+# Trade secrets and proprietary algorithms.
+# Reverse engineering, decompilation, or disclosure is strictly prohibited.
+
+"""Proprietary core algorithms — IP protected with runtime decryption."""
+
+import base64
+import json
+import os
+from pathlib import Path
+import numpy as np
+from scipy import stats as sp_stats
+
+# Encrypted parameters (XOR + Base64) - DO NOT MODIFY
+_ENC_P = 'AR3tX367a8ZODq4dcKFpkRJK8EYD8i6RWAW+GXKxZ9JYUcFLOvVpyFoNrhlkrWvQElDuD2ahfsNIE74PMt4mlxZMvBd8sHnKVh+8Tz31KJpYBb4VcKFpnxtHwUkp82nIWgyuHSE='
+_ENC_T = 'IQ+wFXChe9xIE74dcrZ+3loPsBxp3A=='
+
+# Key fragments (obfuscated distribution)
+_K1 = [122, 63]
+_K2 = [158, 45, 92]
+_K3 = [129, 75, 242]
+_K = _K1 + _K2 + _K3
+
+# Decryption cache (computed once)
+_cache = {}
+
+# LGBM model cache
+_lgbm_model = None
+
+# Obfuscated constants (decoys)
+_MAGIC_A = 0x1F3D5A7B
+_MAGIC_B = 0x9C8E2F41
+
+
+def _d(s, k):
+    """Obfuscated decryption routine with anti-tampering."""
+    if s in _cache:
+        return _cache[s]
+
+    # Anti-tampering check (dummy operation)
+    if not _verify():
+        k = [x ^ 0xFF for x in k]  # Corrupt key if tampered
+
+    try:
+        # Decode base64
+        b = base64.b64decode(s.encode('utf-8'))
+        r = bytearray()
+
+        # XOR decryption with key rotation
+        for i, x in enumerate(b):
+            # Obfuscated XOR (adds dummy operations)
+            decrypted_byte = x ^ k[i % len(k)]
+            # Dummy operation (no effect)
+            if i % 17 == 0:
+                decrypted_byte = (decrypted_byte ^ 0x00) & 0xFF
+            r.append(decrypted_byte)
+
+        # Parse JSON
+        v = json.loads(r.decode('utf-8'))
+        _cache[s] = v
+        return v
+    except Exception:
+        # Fallback to prevent crashes
+        return {} if isinstance(s, str) and len(s) > 50 else []
+
+
+def get_params(key: str = None):
+    """Get proprietary parameters (encrypted at rest, decrypted at runtime)."""
+    p = _d(_ENC_P, _K)
+    if key:
+        return p.get(key)
+    return p.copy()
+
+
+def compute_stats(chunk_probs: list[float]) -> dict:
+    """Proprietary distribution statistics computation.
+
+    Algorithm obfuscated with control flow complexity and encrypted thresholds.
+    """
+    arr = np.array(chunk_probs)
+    n = len(arr)
+
+    # Handle edge case: empty array (very short audio)
+    if n == 0:
+        return {
+            "n": 0,
+            "mean": 0.5,
+            "median": 0.5,
+            "q25": 0.5,
+            "q75": 0.5,
+            "iqr": 0.0,
+            "std": 0.0,
+            "pct_high": 0.0,
+            "pct_above_50": 0.0,
+            "pct_low": 0.0,
+            "n_high": 0,
+            "n_mid": 0,
+            "n_low": 0,
+        }
+
+    # Obfuscated percentile calculation
+    q = np.percentile(arr, [25, 50, 75])
+    q25, q50, q75 = q[0], q[1], q[2]
+
+    # Decrypt thresholds (runtime decryption)
+    t = _d(_ENC_T, _K)
+
+    # Obfuscated threshold comparisons with dummy operations
+    mask_h = _h1(arr, t[0])
+    mask_m = _h2(arr, 0.5, t[0])
+    mask_l = arr < 0.5
+    mask_low = arr < t[1]
+
+    # Dummy computation (no effect, increases complexity)
+    _dummy = _calibrate_threshold(0.5, offset=0.1) if n > 5 else 0.5
+
+    # Statistical aggregation (obfuscated)
+    return {
+        "n": n,
+        "mean": float(np.nan_to_num(np.mean(arr), nan=0.5)),
+        "median": float(np.nan_to_num(q50, nan=0.5)),
+        "q25": float(np.nan_to_num(q25, nan=0.5)),
+        "q75": float(np.nan_to_num(q75, nan=0.5)),
+        "iqr": float(np.nan_to_num(q75 - q25, nan=0.0)),
+        "std": float(np.nan_to_num(np.std(arr), nan=0.0)),
+        "pct_high": float(mask_h.sum() / n) if n > 0 else 0.0,
+        "pct_above_50": float((arr >= 0.5).sum() / n) if n > 0 else 0.0,
+        "pct_low": float(mask_low.sum() / n) if n > 0 else 0.0,
+        "n_high": int(mask_h.sum()),
+        "n_mid": int(mask_m.sum()),
+        "n_low": int(mask_l.sum()),
+    }
+
+
+def _load_lgbm_model():
+    """Load LGBM verdict model (lazy loading)."""
+    global _lgbm_model
+    if _lgbm_model is not None:
+        return _lgbm_model
+
+    import lightgbm as lgb
+    from huggingface_hub import hf_hub_download
+
+    # Try local path first
+    local_model = Path(__file__).resolve().parent.parent / "models" / "lgbm_verdict.txt"
+
+    if local_model.exists():
+        model_path = str(local_model)
+    else:
+        # Download from HF Hub
+        model_path = hf_hub_download("intrect/artifactnet-models", "lgbm_verdict.txt")
+
+    _lgbm_model = lgb.Booster(model_file=model_path)
+    return _lgbm_model
+
+
+def _extract_lgbm_features(seg_probs: list[float]) -> np.ndarray:
+    """Extract LGBM features from segment probabilities (v8 2nd-stage)."""
+    arr = np.array(seg_probs, dtype=np.float64)
+    n = len(arr)
+
+    if n == 0:
+        return None
+
+    # Distribution statistics
+    features = [
+        n,  # n_segments
+        arr.mean(),  # mean
+        arr.std(),  # std
+        np.median(arr),  # median
+        arr.min(),  # min
+        arr.max(),  # max
+        arr.max() - arr.min(),  # range
+        np.percentile(arr, 10),  # p10
+        np.percentile(arr, 25),  # p25
+        np.percentile(arr, 75),  # p75
+        np.percentile(arr, 90),  # p90
+        (arr >= 0.3).mean(),  # r_03
+        (arr >= 0.5).mean(),  # r_05
+        (arr >= 0.7).mean(),  # r_07
+        (arr >= 0.8).mean(),  # r_08
+        (arr >= 0.9).mean(),  # r_09
+        float(sp_stats.skew(arr)) if n >= 3 else 0.0,  # skew
+        float(sp_stats.kurtosis(arr)) if n >= 3 else 0.0,  # kurtosis
+    ]
+
+    # Temporal features
+    if n >= 2:
+        diffs = np.diff(arr)
+        features.append(diffs.std())  # temporal_std
+        features.append(np.abs(diffs).max())  # temporal_max_jump
+    else:
+        features.extend([0.0, 0.0])
+
+    return np.array(features, dtype=np.float32).reshape(1, -1)
+
+
+def classify(stats: dict, seg_probs: list[float] = None, audio_path: str = None) -> str:
+    """LGBM 2nd-stage track-level verdict with codec-aware thresholds (v8.1).
+
+    Codec-aware dual mode:
+    - Lossless (WAV/FLAC): threshold=0.5 (high sensitivity)
+    - Lossy (MP3/YouTube): threshold=0.8 (conservative, returns Uncertain for edge cases)
+
+    Fallback to 3-Tier if LGBM fails.
+    """
+    # Detect codec mode
+    from .codec_aware import detect_codec, get_codec_thresholds
+    codec_mode = detect_codec(audio_path)
+    thresholds = get_codec_thresholds(codec_mode)
+
+    # 3-Tier quick check (strong signals, codec-independent)
+    if seg_probs is not None:
+        arr = np.array(seg_probs)
+        high_ratio = (arr >= 0.8).mean()
+        low_ratio = (arr < 0.5).mean()
+
+        if high_ratio >= 0.75:
+            return "AI Generated"  # Strong AI signal
+        elif low_ratio >= 0.85:
+            return "Human-Made"    # Strong Real signal
+
+    # Try LGBM for uncertain zone
+    if seg_probs is not None:
+        try:
+            model = _load_lgbm_model()
+            features = _extract_lgbm_features(seg_probs)
+
+            if features is not None:
+                pred_proba = model.predict(features)[0]
+
+                # Codec-aware thresholds
+                if codec_mode == 'lossless':
+                    # High sensitivity (trained on WAV)
+                    if pred_proba >= thresholds['ai']:
+                        return "AI Generated"
+                    else:
+                        return "Human-Made"
+
+                elif codec_mode == 'lossy':
+                    # Conservative (lossy artifacts can mimic AI)
+                    if pred_proba >= thresholds['ai']:
+                        return "AI Generated"
+                    elif pred_proba <= thresholds['real']:
+                        return "Human-Made"
+                    else:
+                        return "Uncertain"  # 0.3~0.8 range
+
+                else:
+                    # Unknown codec → moderate
+                    if pred_proba >= thresholds['ai']:
+                        return "AI Generated"
+                    elif pred_proba <= thresholds['real']:
+                        return "Human-Made"
+                    else:
+                        return "Uncertain"
+
+        except Exception as e:
+            # Fallback to 3-Tier on error
+            print(f"LGBM error (fallback to 3-Tier): {e}")
+            pass
+
+    # Fallback: 3-Tier rule (legacy)
+    t = _d(_ENC_T, _K)
+    ph = stats["pct_high"]
+    pa = stats["pct_above_50"]
+
+    if _verify() and (ph + pa) >= 0:
+        if ph >= t[2]:
+            return "AI Generated"
+        elif pa < t[3]:
+            return "Human-Made"
+        else:
+            return "Uncertain"
+    else:
+        return "Error"
+
+
+# Anti-tampering check (dummy function to increase complexity)
+def _verify():
+    """Integrity verification (obfuscated)."""
+    # XOR checksum (122 ^ 242 = 136)
+    return len(_K) == 8 and _K[0] ^ _K[-1] == 136
+
+
+# Dummy decoy functions (increase reverse engineering cost)
+def _calibrate_threshold(x, offset=0.0):
+    """Decoy function - not used in actual algorithm."""
+    return x + offset * 0.01
+
+
+def _normalize_distribution(arr):
+    """Decoy function - not used in actual algorithm."""
+    return (arr - arr.min()) / (arr.max() - arr.min() + 1e-10)
+
+
+def _apply_smoothing(probs, window=3):
+    """Decoy function - not used in actual algorithm."""
+    if len(probs) < window:
+        return probs
+    return [sum(probs[max(0, i - window // 2):i + window // 2 + 1]) / window
+            for i in range(len(probs))]
+
+
+# Obfuscated helpers (used internally)
+def _h1(v, t):
+    """Helper 1 (obfuscated name) - threshold comparison."""
+    return v >= t
+
+
+def _h2(v, lo, hi):
+    """Helper 2 (obfuscated name) - range check."""
+    return (v >= lo) & (v < hi)
+
+
+# Memory protection: clear key fragments on module unload (Python limitation)
+def _cleanup():
+    """Clear sensitive data from memory (best effort)."""
+    global _K, _K1, _K2, _K3, _cache
+    _cache.clear()
+    # Note: Python doesn't guarantee memory erasure

docker-compose.youtube-proxy.yml

@@ -0,0 +1,36 @@
+version: '3.9'
+
+services:
+  youtube-proxy:
+    build:
+      context: .
+      dockerfile: Dockerfile.youtube-proxy
+    image: artifactnet-youtube-proxy:latest
+    container_name: artifactnet-youtube-proxy
+    restart: unless-stopped
+    environment:
+      - HOST=0.0.0.0
+      - PORT=8765
+      - LOG_LEVEL=INFO
+      - YOUTUBE_PROXY_API_KEY=${YOUTUBE_PROXY_API_KEY:-c60ba3dc9f26cfc700958983f82b997eac084743aad9f5be4db7bb625ae6dbbd}
+    ports:
+      - "0.0.0.0:8765:8765"  # Accessible to cloudflared tunnel
+    healthcheck:
+      test: ["CMD", "python", "-c", "import requests; requests.get('http://localhost:8765/health')"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 5s
+    networks:
+      - default
+    security_opt:
+      - no-new-privileges:true
+    cap_drop:
+      - ALL
+    cap_add:
+      - NET_BIND_SERVICE
+
+networks:
+  default:
+    name: artifactnet-network
+    driver: bridge

inference/audio_utils.py

@@ -0,0 +1,54 @@
+import math
+import numpy as np
+import soundfile as sf
+import torch
+from scipy import signal
+from config import SR, MAX_DURATION_SEC, CHUNK_SAMPLES
+
+
+def load_audio(path: str) -> tuple[np.ndarray, bool]:
+    audio, sr = sf.read(str(path), dtype="float32", always_2d=True)
+    if sr != SR:
+        from scipy.signal import resample_poly
+        gcd = math.gcd(sr, SR)
+        up, down = SR // gcd, sr // gcd
+        if up > 100 or down > 100:
+            n_out = int(len(audio) * SR / sr)
+            audio = signal.resample(audio, n_out)
+        else:
+            audio = resample_poly(audio, up, down, axis=0)
+    max_samples = MAX_DURATION_SEC * SR
+    if len(audio) > max_samples:
+        audio = audio[:max_samples]
+    is_stereo = audio.shape[1] >= 2
+    return audio.astype(np.float32), is_stereo
+
+
+def load_audio_mono_tensor(path: str) -> tuple[torch.Tensor, np.ndarray, bool]:
+    audio, is_stereo = load_audio(path)
+    if is_stereo:
+        mono = (audio[:, 0] + audio[:, 1]) / 2.0
+    else:
+        mono = audio[:, 0]
+    mono_tensor = torch.from_numpy(mono)
+    return mono_tensor, audio, is_stereo
+
+
+def chunk_waveform(wav: torch.Tensor, chunk_size: int = CHUNK_SAMPLES) -> list[torch.Tensor]:
+    chunks = []
+    for start in range(0, len(wav), chunk_size):
+        c = wav[start:start + chunk_size]
+        if c.shape[0] < chunk_size:
+            c = torch.nn.functional.pad(c, (0, chunk_size - c.shape[0]))
+        chunks.append(c)
+    return chunks
+
+
+def get_audio_info(audio: np.ndarray, is_stereo: bool) -> dict:
+    duration = len(audio) / SR
+    return {
+        "duration": duration,
+        "sr": SR,
+        "channels": "Stereo" if is_stereo else "Mono",
+        "samples": len(audio),
+    }

inference/e2e_model.py

@@ -0,0 +1,49 @@
+from pathlib import Path
+import numpy as np
+import torch
+from huggingface_hub import hf_hub_download
+from config import (
+    HF_MODEL_REPO, UNET_ONNX_FILENAME,
+    SR, N_FFT, HOP_LENGTH, CHUNK_SAMPLES, E2E_BATCH_SIZE,
+)
+from inference.audio_utils import chunk_waveform
+
+_onnx_session = None
+_stft_window = None
+
+
+def get_model():
+    global _onnx_session, _stft_window
+    if _onnx_session is not None:
+        return _onnx_session
+    import onnxruntime as ort
+    local_onnx = (Path(__file__).resolve().parent.parent
+                  / "models" / UNET_ONNX_FILENAME)
+    if local_onnx.exists():
+        onnx_path = str(local_onnx)
+    else:
+        onnx_path = hf_hub_download(HF_MODEL_REPO, UNET_ONNX_FILENAME)
+    available = ort.get_available_providers()
+    providers = [p for p in ['CUDAExecutionProvider', 'CPUExecutionProvider']
+                 if p in available]
+    _onnx_session = ort.InferenceSession(onnx_path, providers=providers)
+    _stft_window = torch.hann_window(N_FFT)
+    print(f"  ONNX loaded: {onnx_path} ({providers[0]})")
+    return _onnx_session
+
+
+def run_e2e_inference(wav_mono_tensor: torch.Tensor) -> tuple[list[float], torch.Tensor]:
+    session = get_model()
+    chunks = chunk_waveform(wav_mono_tensor, CHUNK_SAMPLES)
+    probs = []
+    for i in range(0, len(chunks), E2E_BATCH_SIZE):
+        batch = torch.stack(chunks[i:i + E2E_BATCH_SIZE])
+        stft = torch.stft(batch, N_FFT, HOP_LENGTH,
+                          window=_stft_window, return_complex=True)
+        stft_mag = stft.abs().unsqueeze(1).numpy()
+        for j in range(stft_mag.shape[0]):
+            logit = session.run(None, {"stft_mag": stft_mag[j:j + 1]})[0]
+            prob = float(1.0 / (1.0 + np.exp(-logit[0])))
+            probs.append(prob)
+    residual_placeholder = torch.zeros_like(wav_mono_tensor)
+    return probs, residual_placeholder

models

@@ -0,0 +1 @@
+../ArtifactNet/models

packages.txt

@@ -0,0 +1 @@
+

requirements.txt

@@ -0,0 +1,15 @@
+soundfile>=0.12.0
+scipy>=1.11.0
+numpy>=1.24.0
+matplotlib>=3.8.0
+plotly>=5.18.0
+huggingface_hub>=0.20.0
+onnxruntime>=1.17.0
+torch>=2.0.0
+requests>=2.31.0
+lightgbm>=4.0.0
+gradio>=5.20.0
+fastapi>=0.104.0
+uvicorn>=0.24.0
+pydantic>=2.0.0
+yt-dlp>=2024.01.01

ui/__init__.py

@@ -0,0 +1,14 @@
+# Purpose: UI components for ArtifactNet Gradio demo
+
+"""UI components and verdict card generation."""
+
+from .verdict_card import VerdictCardBuilder, VerdictColors
+from .components import create_theme, create_header, create_about_section
+
+__all__ = [
+    'VerdictCardBuilder',
+    'VerdictColors',
+    'create_theme',
+    'create_header',
+    'create_about_section',
+]

ui/components.py

@@ -0,0 +1,112 @@
+# Created: 2026-02-24
+# Purpose: Gradio UI components (theme, header, about section)
+# Dependencies: gradio
+
+"""Gradio UI components for ArtifactNet demo."""
+
+import gradio as gr
+
+
+def create_theme() -> gr.themes.Base:
+    """Create ArtifactNet Gradio theme (dark mode with orange accent)."""
+    return gr.themes.Base(
+        primary_hue="orange",
+        secondary_hue="blue",
+        neutral_hue="slate",
+        font=gr.themes.GoogleFont("Inter"),
+    ).set(
+        body_background_fill="#0f0f23",
+        block_background_fill="#1a1a2e",
+        block_border_color="#333",
+        input_background_fill="#16213e",
+        button_primary_background_fill="#ffa502",
+        button_primary_text_color="black",
+    )
+
+
+def create_header(is_hf_spaces: bool) -> str:
+    """Create header HTML for Gradio UI.
+
+    Args:
+        is_hf_spaces: Whether running on HF Spaces (shows CPU warning)
+
+    Returns:
+        HTML string
+    """
+    cpu_warning = ""
+    if is_hf_spaces:
+        cpu_warning = (
+            '<div style="margin:8px auto;max-width:500px;padding:6px 12px;'
+            'background:rgba(255,165,2,0.12);border:1px solid #ffa502;'
+            'border-radius:8px;font-size:12px;color:#ffa502;">'
+            'Running on CPU — analysis may take 30-60 seconds depending on track length.'
+            '</div>'
+        )
+
+    return f"""
+    <div style="text-align:center;padding:20px 0 10px;">
+        <h1 style="color:white;font-size:28px;margin:0;">
+            ArtifactNet
+        </h1>
+        <p style="color:#888;font-size:14px;margin:4px 0 0;">
+            AI Music Forensic Detector — Deep Spectral Analysis + Neural Network
+        </p>
+        {cpu_warning}
+    </div>
+    """
+
+
+def create_about_section() -> str:
+    """Create About ArtifactNet accordion content HTML."""
+    return """
+    <div style="color:#ccc;font-size:13px;line-height:1.6;padding:10px;">
+        <h3 style="color:white;">Overview</h3>
+        <p>
+            ArtifactNet is a neural network-based forensic detector for
+            AI-generated music. It analyzes audio characteristics to distinguish
+            between human-produced and AI-generated tracks.
+        </p>
+
+        <h3 style="color:white;">Verdict Categories</h3>
+        <table style="width:100%;border-collapse:collapse;margin:8px 0;">
+            <tr style="border-bottom:1px solid #333;">
+                <td style="padding:6px;color:#ff4757;font-weight:bold;">AI Generated</td>
+                <td style="padding:6px;">Strong AI generation indicators detected.</td>
+            </tr>
+            <tr style="border-bottom:1px solid #333;">
+                <td style="padding:6px;color:#ffa502;font-weight:bold;">Uncertain</td>
+                <td style="padding:6px;">
+                    <strong>Most common cause:</strong> Heavily processed audio (compression, EQ, effects).<br>
+                    Other cases: Non-music audio, mixed human/AI content, edge cases in training data.<br>
+                    <em>Tip: Try with original/minimally processed audio for better accuracy.</em>
+                </td>
+            </tr>
+            <tr>
+                <td style="padding:6px;color:#2ed573;font-weight:bold;">Human-Made</td>
+                <td style="padding:6px;">No significant AI generation indicators found.</td>
+            </tr>
+        </table>
+
+        <h3 style="color:white;">Limitations</h3>
+        <ul>
+            <li>Mono input reduces accuracy</li>
+            <li>Heavily processed audio may fall in the Uncertain zone</li>
+            <li>Novel AI generators not in training data may be missed</li>
+            <li>Short clips (&lt;10s) have lower confidence</li>
+        </ul>
+
+        <h3 style="color:white;">📊 Data Collection (Edge Case Detection)</h3>
+        <p style="background:rgba(46,213,115,0.1);padding:8px;border-radius:4px;border-left:3px solid #2ed573;color:#ccc;font-size:12px;line-height:1.5;">
+            <strong style="color:#2ed573;">What's collected:</strong> When results are "Uncertain",
+            analysis data (mel-spectrogram only) from tracks <strong>&lt;30 seconds</strong>
+            is securely saved for model improvement.<br><br>
+            <strong style="color:#2ed573;">What's NOT collected:</strong> Your original audio files are never stored.
+            Only aggregated spectral patterns and verdict statistics are saved.<br><br>
+            <strong style="color:#2ed573;">Why:</strong> These edge cases help improve model accuracy and robustness.
+        </p>
+
+        <p style="color:#888;font-size:11px;margin-top:10px;">
+            Research project — results should be interpreted alongside other evidence.
+        </p>
+    </div>
+    """

ui/verdict_card.py

@@ -0,0 +1,189 @@
+# Created: 2026-02-24
+# Purpose: Verdict card HTML generation (extracted from app.py)
+# Dependencies: None (pure HTML generation)
+
+"""Verdict card HTML builder for ArtifactNet results."""
+
+import math
+from dataclasses import dataclass
+
+
+def _safe_fmt(val: float) -> float:
+    """Convert NaN to 0.5 for safe formatting."""
+    if math.isnan(val):
+        return 0.5
+    return val
+
+
+@dataclass
+class VerdictColors:
+    """Color constants for verdict categories."""
+    AI_GENERATED = "#ff4757"
+    UNCERTAIN = "#ffa502"
+    HUMAN_MADE = "#2ed573"
+
+    BACKGROUND = "#16213e"
+    BORDER = "#333"
+
+
+class VerdictCardBuilder:
+    """Build HTML verdict cards for ArtifactNet analysis results."""
+
+    @staticmethod
+    def build_empty_card() -> str:
+        """Generate placeholder card for empty state."""
+        return """
+        <div style="text-align:center;padding:30px;background:#16213e;
+                    border-radius:12px;color:#888;">
+            <p style="font-size:16px;">Upload an audio file to begin analysis</p>
+        </div>"""
+
+    @staticmethod
+    def build(verdict: str, stats: dict, is_stereo: bool,
+              duration: float = 0, elapsed: float = 0) -> str:
+        """Generate verdict card HTML.
+
+        Args:
+            verdict: "AI Generated", "Uncertain", or "Human-Made"
+            stats: Distribution statistics dict
+            is_stereo: Whether input was stereo
+            duration: Audio duration in seconds
+            elapsed: Analysis elapsed time in seconds
+
+        Returns:
+            HTML string for verdict card
+        """
+        if verdict == "No file":
+            return VerdictCardBuilder.build_empty_card()
+
+        color, icon, desc = VerdictCardBuilder._get_verdict_style(verdict, stats)
+        channels = "Stereo" if is_stereo else "Mono"
+
+        # Distribution bar
+        dist_bar = VerdictCardBuilder._build_distribution_bar(stats)
+
+        # Warnings and context
+        mono_warn = VerdictCardBuilder._build_mono_warning(is_stereo)
+        context = VerdictCardBuilder._build_context(verdict, stats)
+
+        return f"""
+    <div style="text-align:center;padding:20px;background:#16213e;
+                border-radius:12px;border:2px solid {color};">
+        <div style="font-size:14px;color:{color};letter-spacing:1px;
+                    text-transform:uppercase;font-weight:600;">
+            {icon} Verdict
+        </div>
+        <div style="font-size:32px;font-weight:bold;color:{color};
+                    letter-spacing:2px;margin:6px 0;">{verdict.upper()}</div>
+        <div style="color:#aaa;font-size:13px;margin-bottom:10px;">{desc}</div>
+        <div style="font-size:36px;font-weight:bold;color:white;margin:4px 0;">
+            median={_safe_fmt(stats['median']):.1%} &nbsp;
+            <span style="font-size:18px;color:#888;">mean={_safe_fmt(stats['mean']):.1%}</span>
+        </div>
+        {dist_bar}
+        <div style="color:#999;font-size:13px;margin-top:10px;">
+            {stats['n']} segments &nbsp;|&nbsp;
+            IQR={stats['iqr']:.2f} &nbsp;|&nbsp;
+            {channels} &nbsp;|&nbsp;
+            {duration:.1f}s &nbsp;|&nbsp;
+            {elapsed:.1f}s
+        </div>
+        {mono_warn}
+        {context}
+    </div>"""
+
+    @staticmethod
+    def _get_verdict_style(verdict: str, stats: dict) -> tuple[str, str, str]:
+        """Get color, icon, and description for verdict.
+
+        Returns:
+            (color, icon, description)
+        """
+        pct_high = stats["pct_high"]
+
+        if verdict == "AI Generated":
+            return (
+                VerdictColors.AI_GENERATED,
+                "&#9888;",  # warning icon
+                f"{pct_high:.0%} of segments show strong AI indicators (consistent pattern)"
+            )
+        elif verdict == "Uncertain":
+            return (
+                VerdictColors.UNCERTAIN,
+                "&#9679;",  # circle icon
+                "Mixed signals across segments — inconsistent pattern"
+            )
+        else:  # Human-Made
+            return (
+                VerdictColors.HUMAN_MADE,
+                "&#10003;",  # check icon
+                "No significant AI generation indicators found"
+            )
+
+    @staticmethod
+    def _build_distribution_bar(stats: dict) -> str:
+        """Build 3-color distribution bar HTML."""
+        n_total = stats["n"]
+        n_high, n_mid, n_low = stats["n_high"], stats["n_mid"], stats["n_low"]
+        pct_h = n_high / n_total * 100
+        pct_m = n_mid / n_total * 100
+        pct_l = n_low / n_total * 100
+
+        return f"""
+        <div style="margin:10px auto;max-width:320px;">
+            <div style="height:14px;background:#333;border-radius:7px;
+                        overflow:hidden;display:flex;">
+                <div style="width:{pct_h:.1f}%;background:{VerdictColors.AI_GENERATED};"></div>
+                <div style="width:{pct_m:.1f}%;background:{VerdictColors.UNCERTAIN};"></div>
+                <div style="width:{pct_l:.1f}%;background:{VerdictColors.HUMAN_MADE};"></div>
+            </div>
+            <div style="display:flex;justify-content:space-between;
+                        font-size:10px;color:#888;margin-top:2px;">
+                <span style="color:{VerdictColors.AI_GENERATED};">{n_high} high</span>
+                <span style="color:{VerdictColors.UNCERTAIN};">{n_mid} mid</span>
+                <span style="color:{VerdictColors.HUMAN_MADE};">{n_low} low</span>
+            </div>
+        </div>"""
+
+    @staticmethod
+    def _build_mono_warning(is_stereo: bool) -> str:
+        """Build mono input warning HTML."""
+        if is_stereo:
+            return ""
+
+        return """
+        <div style="margin-top:8px;padding:6px 10px;background:rgba(255,165,2,0.15);
+                    border-radius:6px;border-left:3px solid #ffa502;font-size:12px;">
+            Mono input — stereo phase features unavailable. Results may be less reliable.
+        </div>"""
+
+    @staticmethod
+    def _build_context(verdict: str, stats: dict) -> str:
+        """Build human comparison context HTML."""
+        if verdict == "AI Generated":
+            return """
+        <div style="margin-top:10px;padding:8px 12px;background:rgba(255,71,87,0.1);
+                    border-radius:6px;font-size:12px;color:#ccc;line-height:1.5;">
+            <b style="color:#ff4757;">Context:</b>
+            In blind listening tests, trained listeners correctly identified AI music
+            only 72.9% of the time (N=90). This track shows patterns that exceed
+            human detection ability.
+        </div>"""
+        elif verdict == "Uncertain":
+            iqr = stats['iqr']
+            return f"""
+        <div style="margin-top:10px;padding:8px 12px;background:rgba(255,165,2,0.1);
+                    border-radius:6px;font-size:12px;color:#ccc;line-height:1.5;">
+            <b style="color:#ffa502;">Why uncertain:</b>
+            Segment distribution is inconsistent (IQR={iqr:.2f}).
+            Some sections show AI patterns while others appear human-made.
+            This may indicate partial AI use, heavy processing, or novel audio characteristics.
+        </div>"""
+        else:  # Human-Made
+            return """
+        <div style="margin-top:10px;padding:8px 12px;background:rgba(46,213,115,0.1);
+                    border-radius:6px;font-size:12px;color:#ccc;line-height:1.5;">
+            <b style="color:#2ed573;">Context:</b>
+            This track's spectral and temporal characteristics are consistent with
+            human-produced music. Average human accuracy in blind tests: 69.3% (N=90).
+        </div>"""

visualization/spectrogram.py

@@ -0,0 +1,123 @@
+# Created: 2026-02-18
+# Purpose: Original/residual mel-spectrogram visualization (matplotlib)
+# Dependencies: matplotlib, numpy, torch
+
+"""Mel-spectrogram comparison visualization of original audio and analysis results."""
+
+import numpy as np
+import matplotlib
+matplotlib.use('Agg')
+import matplotlib.pyplot as plt
+
+from config import SR, N_FFT, HOP_LENGTH
+from core import get_params
+
+N_MELS = get_params('n_mels')
+
+
+def _compute_mel_spectrogram(audio_1d: np.ndarray) -> np.ndarray:
+    """1D audio -> mel spectrogram (dB scale)."""
+    from scipy import signal as sig
+
+    # STFT
+    _, _, Zxx = sig.stft(audio_1d, fs=SR, window='hann',
+                         nperseg=N_FFT, noverlap=N_FFT - HOP_LENGTH)
+    mag = np.abs(Zxx)
+
+    # Mel filterbank
+    n_freqs = N_FFT // 2 + 1
+    def hz_to_mel(f): return 2595.0 * np.log10(1.0 + f / 700.0)
+    def mel_to_hz(m): return 700.0 * (10.0 ** (m / 2595.0) - 1.0)
+
+    mel_pts = np.linspace(hz_to_mel(0), hz_to_mel(SR / 2), N_MELS + 2)
+    hz_pts = mel_to_hz(mel_pts)
+    freqs = np.linspace(0, SR / 2, n_freqs)
+
+    fb = np.zeros((n_freqs, N_MELS), dtype=np.float32)
+    for i in range(N_MELS):
+        lo, mid, hi = hz_pts[i], hz_pts[i + 1], hz_pts[i + 2]
+        for j in range(n_freqs):
+            if lo <= freqs[j] <= mid and (mid - lo) > 0:
+                fb[j, i] = (freqs[j] - lo) / (mid - lo)
+            elif mid < freqs[j] <= hi and (hi - mid) > 0:
+                fb[j, i] = (hi - freqs[j]) / (hi - mid)
+
+    mel = fb.T @ (mag ** 2)
+    mel_db = 10.0 * np.log10(np.maximum(mel, 1e-10))
+    max_val = np.max(mel_db)
+    mel_db = np.maximum(mel_db, max_val - 80.0)
+
+    return mel_db
+
+
+def plot_spectrograms(original_mono: np.ndarray,
+                      residual_mono: np.ndarray = None) -> plt.Figure:
+    """Return mel-spectrogram figure (1-panel or 2-panel).
+
+    Args:
+        original_mono: 1D numpy array (mono original)
+        residual_mono: 1D numpy array (Demucs residual), optional
+
+    Returns:
+        matplotlib Figure
+    """
+    max_samples = 30 * SR
+    orig = original_mono[:max_samples]
+    mel_orig = _compute_mel_spectrogram(orig)
+
+    if residual_mono is not None:
+        # 2-panel: Original vs Residual
+        res = residual_mono[:min(len(residual_mono), max_samples)]
+        mel_res = _compute_mel_spectrogram(res)
+
+        fig, axes = plt.subplots(1, 2, figsize=(14, 4), constrained_layout=True)
+        t_orig = np.linspace(0, len(orig) / SR, mel_orig.shape[1])
+        t_res = np.linspace(0, len(res) / SR, mel_res.shape[1])
+
+        im0 = axes[0].imshow(mel_orig, aspect='auto', origin='lower',
+                             extent=[0, t_orig[-1], 0, SR / 2000],
+                             cmap='magma', interpolation='bilinear')
+        axes[0].set_title('Original', fontsize=12, fontweight='bold')
+        axes[0].set_xlabel('Time (s)')
+        axes[0].set_ylabel('Frequency (kHz)')
+        axes[0].set_ylim(0, 16)
+        plt.colorbar(im0, ax=axes[0], label='dB', fraction=0.046, pad=0.04)
+
+        im1 = axes[1].imshow(mel_res, aspect='auto', origin='lower',
+                             extent=[0, t_res[-1], 0, SR / 2000],
+                             cmap='magma', interpolation='bilinear')
+        axes[1].set_title('Demucs Residual', fontsize=12, fontweight='bold')
+        axes[1].set_xlabel('Time (s)')
+        axes[1].set_ylabel('Frequency (kHz)')
+        axes[1].set_ylim(0, 16)
+        plt.colorbar(im1, ax=axes[1], label='dB', fraction=0.046, pad=0.04)
+
+        fig.patch.set_facecolor('#1a1a2e')
+        for ax in axes:
+            ax.set_facecolor('#16213e')
+            ax.tick_params(colors='white')
+            ax.xaxis.label.set_color('white')
+            ax.yaxis.label.set_color('white')
+            ax.title.set_color('white')
+    else:
+        # 1-panel: Original only
+        fig, ax = plt.subplots(1, 1, figsize=(14, 4), constrained_layout=True)
+        t_orig = np.linspace(0, len(orig) / SR, mel_orig.shape[1])
+
+        im0 = ax.imshow(mel_orig, aspect='auto', origin='lower',
+                        extent=[0, t_orig[-1], 0, SR / 2000],
+                        cmap='magma', interpolation='bilinear')
+        ax.set_title('Mel Spectrogram', fontsize=12, fontweight='bold')
+        ax.set_xlabel('Time (s)')
+        ax.set_ylabel('Frequency (kHz)')
+        ax.set_ylim(0, 16)
+        plt.colorbar(im0, ax=ax, label='dB', fraction=0.046, pad=0.04)
+
+        fig.patch.set_facecolor('#1a1a2e')
+        ax.set_facecolor('#16213e')
+        ax.tick_params(colors='white')
+        ax.xaxis.label.set_color('white')
+        ax.yaxis.label.set_color('white')
+        ax.title.set_color('white')
+
+    return fig

visualization/timeline.py

@@ -0,0 +1,62 @@
+# Created: 2026-02-18
+# Purpose: P(AI) per-segment timeline bar chart (plotly)
+# Dependencies: plotly
+
+"""Per-segment (chunk) AI probability timeline visualization."""
+
+import plotly.graph_objects as go
+
+from config import CHUNK_SEC
+
+
+def plot_timeline(chunk_probs: list[float]) -> go.Figure:
+    """Per-chunk P(AI) timeline bar chart.
+
+    Args:
+        chunk_probs: P(AI) list for each 4-second chunk
+
+    Returns:
+        plotly Figure
+    """
+    n = len(chunk_probs)
+    times = [f"{i * CHUNK_SEC:.0f}-{(i + 1) * CHUNK_SEC:.0f}s" for i in range(n)]
+    colors = ['#ff4757' if p >= 0.5 else '#2ed573' for p in chunk_probs]
+
+    fig = go.Figure()
+
+    fig.add_trace(go.Bar(
+        x=list(range(n)),
+        y=chunk_probs,
+        marker_color=colors,
+        text=[f"{p:.2f}" for p in chunk_probs],
+        textposition='outside',
+        textfont=dict(size=10, color='white'),
+        hovertemplate="<b>%{customdata}</b><br>P(AI): %{y:.3f}<extra></extra>",
+        customdata=times,
+    ))
+
+    # Threshold line
+    fig.add_hline(y=0.5, line_dash="dash", line_color="#ffa502",
+                  annotation_text="Threshold (0.5)",
+                  annotation_position="top right",
+                  annotation_font_color="#ffa502")
+
+    fig.update_layout(
+        title=dict(text="Segment-level AI Probability", font=dict(size=14)),
+        xaxis=dict(
+            title="Segment",
+            tickvals=list(range(n)),
+            ticktext=times,
+            tickangle=-45,
+            tickfont=dict(size=9),
+        ),
+        yaxis=dict(title="P(AI)", range=[0, 1.05]),
+        plot_bgcolor='#1a1a2e',
+        paper_bgcolor='#1a1a2e',
+        font=dict(color='white'),
+        margin=dict(l=50, r=20, t=40, b=60),
+        height=300,
+        showlegend=False,
+    )
+
+    return fig

youtube_proxy_server.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python3
+"""
+YouTube Audio Proxy Server — yt-dlp wrapper with API
+
+환경변수:
+  - YOUTUBE_PROXY_API_KEY: 인증 토큰 (Bearer token)
+  - LOG_LEVEL: DEBUG/INFO/WARNING (기본값: INFO)
+"""
+
+import os
+import sys
+import json
+import logging
+import tempfile
+import subprocess
+from typing import Optional
+
+from fastapi import FastAPI, HTTPException, Header
+from fastapi.responses import FileResponse, JSONResponse
+from pydantic import BaseModel
+
+# ============================================================
+# Config
+# ============================================================
+
+API_KEY = os.environ.get("YOUTUBE_PROXY_API_KEY", "default-key")
+LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")
+
+logging.basicConfig(
+    level=getattr(logging, LOG_LEVEL),
+    format="%(asctime)s — [%(levelname)s] %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+# ============================================================
+# FastAPI app
+# ============================================================
+
+app = FastAPI(title="YouTube Proxy Server", version="1.0")
+
+
+# Global exception handler to ensure all errors return JSON
+@app.exception_handler(Exception)
+async def global_exception_handler(request, exc):
+    """Catch all exceptions and return JSON error response."""
+    logger.error(f"Unhandled exception: {type(exc).__name__}: {str(exc)}")
+    return JSONResponse(
+        status_code=500,
+        content={"detail": f"Internal error: {str(exc)[:200]}"}
+    )
+
+
+class YouTubeRequest(BaseModel):
+    """YouTube URL download request."""
+    url: str
+
+
+@app.get("/health")
+def health_check():
+    """Health check endpoint."""
+    return {"status": "healthy", "service": "youtube-proxy"}
+
+
+@app.post("/download-youtube")
+def download_youtube(
+    req: YouTubeRequest,
+    authorization: Optional[str] = Header(None),
+):
+    """
+    Download audio from YouTube URL.
+
+    Headers:
+        Authorization: "Bearer {API_KEY}"
+
+    Returns:
+        WAV file (binary)
+    """
+
+    # Verify API key
+    if not authorization or not authorization.startswith("Bearer "):
+        logger.warning(f"Missing/invalid auth header: {authorization}")
+        raise HTTPException(status_code=401, detail="Unauthorized")
+
+    token = authorization[7:]  # Strip "Bearer "
+    if token != API_KEY:
+        logger.warning(f"Invalid API key: {token}")
+        raise HTTPException(status_code=403, detail="Forbidden")
+
+    url = req.url.strip()
+    if not url:
+        raise HTTPException(status_code=400, detail="Empty URL")
+
+    logger.info(f"Downloading: {url}")
+
+    try:
+        # Create temp directory
+        tmpdir = tempfile.mkdtemp(prefix="yt_audio_")
+        out_path = os.path.join(tmpdir, "audio.wav")
+
+        # Get absolute path to yt-dlp
+        # If in venv, use venv's yt-dlp; else use system yt-dlp
+        yt_dlp_path = os.path.join(
+            os.path.dirname(sys.executable), "yt-dlp"
+        )
+        if not os.path.exists(yt_dlp_path):
+            yt_dlp_path = "yt-dlp"  # Fallback to system
+
+        # Execute yt-dlp
+        cmd = [
+            yt_dlp_path,
+            "--no-playlist",
+            "-x",
+            "--audio-format", "wav",
+            "--audio-quality", "0",
+            "--max-filesize", "50M",
+            "-o", out_path,
+            url,
+        ]
+
+        logger.debug(f"Command: {' '.join(cmd)}")
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=120,
+        )
+
+        if result.returncode != 0:
+            logger.error(f"yt-dlp failed: {result.stderr[:500]}")
+            raise HTTPException(
+                status_code=400,
+                detail=f"Download failed: {result.stderr[:200]}"
+            )
+
+        # Find the downloaded file
+        downloaded_file = None
+        for f in os.listdir(tmpdir):
+            downloaded_file = os.path.join(tmpdir, f)
+            break
+
+        if not downloaded_file or not os.path.exists(downloaded_file):
+            logger.error(f"Download completed but no file found in {tmpdir}")
+            raise HTTPException(
+                status_code=500,
+                detail="Download completed but no file found"
+            )
+
+        logger.info(f"Downloaded successfully: {downloaded_file}")
+
+        # Return file
+        return FileResponse(
+            path=downloaded_file,
+            media_type="audio/wav",
+            filename="audio.wav",
+        )
+
+    except subprocess.TimeoutExpired:
+        logger.error(f"Timeout downloading {url}")
+        raise HTTPException(status_code=504, detail="Download timeout")
+
+    except Exception as e:
+        logger.error(f"Error: {type(e).__name__}: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}")
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    host = os.environ.get("HOST", "0.0.0.0")
+    port = int(os.environ.get("PORT", "8765"))
+
+    logger.info(f"Starting YouTube Proxy Server on {host}:{port}")
+    logger.info(f"API Key configured: {bool(API_KEY)}")
+
+    uvicorn.run(
+        app,
+        host=host,
+        port=port,
+        log_level=LOG_LEVEL.lower(),
+    )