app/services/dedup.py

"""
Challenge deduplication utilities.

Before inserting a new challenge into any of the four tables,
we check whether a sufficiently similar challenge already exists
to prevent the pool from filling with near-duplicates.
"""

from __future__ import annotations

import os
import re
from difflib import SequenceMatcher
from typing import Dict, List, Optional, Set

import httpx

SUPABASE_URL = os.getenv("SUPABASE_URL", "")
SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY", "")

# ---------------------------------------------------------------------------
# Text normalisation helpers
# ---------------------------------------------------------------------------

_AR_STOP_WORDS: Set[str] = {
    "في", "من", "على", "إلى", "عن", "مع", "هذا", "هذه", "التي", "الذي",
    "أن", "إن", "لا", "ما", "لا", "كل", "بها", "به", "فيها", "فيه",
    "أو", "ثم", "لكن", "بل", "حتى", "قد", "لقد", "كان", "يجب", "يتم",
    "يمكن", "حيث", "كما", "أيضاً", "ذلك", "تلك", "اللذين", "اللتين",
}


def _normalise(text: str) -> str:
    """Lower-case, strip punctuation, collapse whitespace."""
    text = text.lower()
    text = re.sub(r"[^\w\s]", " ", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text


def _token_set(text: str) -> Set[str]:
    """Return de-duplicated token set, Arabic stop-words removed."""
    tokens = _normalise(text).split()
    return {t for t in tokens if t not in _AR_STOP_WORDS and len(t) > 1}


def _title_similarity(a: str, b: str) -> float:
    """
    Title similarity = max of:
      - token overlap (Jaccard-ish)
      - SequenceMatcher ratio on normalised strings
    """
    sa, sb = _normalise(a), _normalise(b)
    if not sa or not sb:
        return 0.0
    # SequenceMatcher on normalised strings
    seq = SequenceMatcher(None, sa, sb).ratio()
    # Token overlap
    ta, tb = _token_set(a), _token_set(b)
    if not ta or not tb:
        return seq
    jaccard = len(ta & tb) / max(len(ta | tb), 1)
    return max(seq, jaccard)


def _code_similarity(a: str, b: str) -> float:
    """
    For code: use SequenceMatcher on stripped lines.
    Code tends to have boilerplate; line-level matching works well.
    """
    def _strip(lines: str) -> List[str]:
        return [ln.strip() for ln in lines.splitlines() if ln.strip()]
    la, lb = _strip(a), _strip(b)
    if not la or not lb:
        return 0.0
    return SequenceMatcher(None, la, lb).ratio()


# ---------------------------------------------------------------------------
# DB fetch helpers
# ---------------------------------------------------------------------------

_HEADERS = {
    "apikey": SUPABASE_ANON_KEY,
    "Authorization": f"Bearer {SUPABASE_ANON_KEY}",
    "Content-Type": "application/json",
}


def _fetch_existing_titles(table: str, role_filter: Optional[str] = None,
                           limit: int = 200) -> List[str]:
    """Fetch recent titles from a challenge table."""
    url = f"{SUPABASE_URL}/rest/v1/{table}"
    params: Dict[str, str] = {
        "select": "title",
        "order": "created_at.desc",
        "limit": str(limit),
    }
    if role_filter:
        params["team_role"] = f"eq.{role_filter}"
    try:
        r = httpx.get(url, headers=_HEADERS, params=params, timeout=10)
        r.raise_for_status()
        return [row.get("title", "") for row in r.json()]
    except Exception:
        return []


def _fetch_existing_codes(table: str, code_col: str,
                          role_filter: Optional[str] = None,
                          limit: int = 100) -> List[str]:
    """Fetch recent code/story/log_content from a challenge table."""
    url = f"{SUPABASE_URL}/rest/v1/{table}"
    params: Dict[str, str] = {
        "select": f"title,{code_col}",
        "order": "created_at.desc",
        "limit": str(limit),
    }
    if role_filter:
        params["team_role"] = f"eq.{role_filter}"
    try:
        r = httpx.get(url, headers=_HEADERS, params=params, timeout=10)
        r.raise_for_status()
        return [row.get(code_col, "") for row in r.json() if row.get(code_col)]
    except Exception:
        return []

def fetch_existing_titles(table: str, role_filter: Optional[str] = None,
                          limit: int = 200) -> List[str]:
    """Fetch recent titles from a challenge table (public wrapper)."""
    return _fetch_existing_titles(table, role_filter, limit)


# ---------------------------------------------------------------------------
# Public API — one function per generator
# ---------------------------------------------------------------------------

def is_duplicate_encryption(title: str, story: str,
                            task_outline: str = "",
                            role_filter: Optional[str] = None) -> bool:
    """Check if an encryption challenge is too similar to existing ones."""
    existing_titles = _fetch_existing_titles("encryption_challenges", role_filter)
    for t in existing_titles:
        if _title_similarity(title, t) > 0.55:
            return True
    existing_stories = _fetch_existing_codes(
        "encryption_challenges", "story", role_filter)
    for s in existing_stories:
        if _title_similarity(story, s) > 0.55:
            return True
    if task_outline:
        existing_tasks = _fetch_existing_codes(
            "encryption_challenges", "task_outline", role_filter)
        for t in existing_tasks:
            if _title_similarity(task_outline, t) > 0.55:
                return True
    return False


def is_duplicate_code_fixing(title: str, vulnerable_code: str,
                             task_outline: str = "",
                             role_filter: Optional[str] = None) -> bool:
    """Check if a code-fixing challenge is too similar to existing ones."""
    existing_titles = _fetch_existing_titles(
        "code_fixing_challenges", role_filter, limit=300)
    for t in existing_titles:
        if _title_similarity(title, t) > 0.55:
            return True

    existing_codes = _fetch_existing_codes(
        "code_fixing_challenges", "vulnerable_code", role_filter, limit=100)
    for c in existing_codes:
        if _code_similarity(vulnerable_code, c) > 0.65:
            return True

    if task_outline:
        existing_tasks = _fetch_existing_codes(
            "code_fixing_challenges", "task_outline", role_filter, limit=200)
        for t in existing_tasks:
            if _title_similarity(task_outline, t) > 0.55:
                return True
    return False


def is_duplicate_log_analysis(title: str, log_content: str,
                              task_outline: str = "",
                              role_filter: Optional[str] = None) -> bool:
    """Check if a log-analysis challenge is too similar to existing ones."""
    existing_titles = _fetch_existing_titles(
        "log_analysis_challenges", role_filter, limit=300)
    for t in existing_titles:
        if _title_similarity(title, t) > 0.55:
            return True
    existing_logs = _fetch_existing_codes(
        "log_analysis_challenges", "log_content", role_filter, limit=100)
    for lc in existing_logs:
        if _code_similarity(log_content, lc) > 0.60:
            return True
    if task_outline:
        existing_tasks = _fetch_existing_codes(
            "log_analysis_challenges", "task_outline", role_filter, limit=200)
        for t in existing_tasks:
            if _title_similarity(task_outline, t) > 0.55:
                return True
    return False


def is_duplicate_vuln_hunter(title: str, vulnerable_code: str,
                             task_outline: str = "",
                             role_filter: Optional[str] = None) -> bool:
    """Check if a vulnerability-hunter challenge is too similar to existing ones."""
    existing_titles = _fetch_existing_titles(
        "vulnerability_hunter_challenges", role_filter, limit=300)
    for t in existing_titles:
        if _title_similarity(title, t) > 0.55:
            return True
    existing_codes = _fetch_existing_codes(
        "vulnerability_hunter_challenges", "vulnerable_code", role_filter, limit=100)
    for c in existing_codes:
        if _code_similarity(vulnerable_code, c) > 0.60:
            return True
    if task_outline:
        existing_tasks = _fetch_existing_codes(
            "vulnerability_hunter_challenges", "task_outline", role_filter, limit=200)
        for t in existing_tasks:
            if _title_similarity(task_outline, t) > 0.55:
                return True
    return False


def is_duplicate_web_exploit(title: str, http_request: str,
                             task_outline: str = "",
                             role_filter: Optional[str] = None) -> bool:
    """Check if a web-exploitation challenge is too similar to existing ones."""
    existing_titles = _fetch_existing_titles(
        "web_exploitation_challenges", role_filter, limit=300)
    for t in existing_titles:
        if _title_similarity(title, t) > 0.55:
            return True
    existing_requests = _fetch_existing_codes(
        "web_exploitation_challenges", "http_request", role_filter, limit=100)
    for r in existing_requests:
        if _code_similarity(http_request, r) > 0.60:
            return True
    if task_outline:
        existing_tasks = _fetch_existing_codes(
            "web_exploitation_challenges", "task_outline", role_filter, limit=200)
        for t in existing_tasks:
            if _title_similarity(task_outline, t) > 0.55:
                return True
    return False


def is_duplicate_steganography(title: str, story: str,
                              task_outline: str = "",
                              role_filter: Optional[str] = None) -> bool:
    """Check if a steganography challenge is too similar to existing ones."""
    existing_titles = _fetch_existing_titles(
        "steganography_challenges", role_filter, limit=300)
    for t in existing_titles:
        if _title_similarity(title, t) > 0.55:
            return True
    existing_stories = _fetch_existing_codes(
        "steganography_challenges", "story", role_filter, limit=100)
    for s in existing_stories:
        if _title_similarity(story, s) > 0.55:
            return True
    if task_outline:
        existing_tasks = _fetch_existing_codes(
            "steganography_challenges", "task_outline", role_filter, limit=200)
        for t in existing_tasks:
            if _title_similarity(task_outline, t) > 0.55:
                return True
    return False