"""
Lyrics Moderation System v3

This module provides functionality for detecting potential copyright infringement in song lyrics
using Elasticsearch-based similarity search. It supports multiple languages and implements
a tiered protection system based on song popularity (view count).

The system uses different matching thresholds for viral songs (>10M views) vs regular songs,
and supports specialized text analysis for various languages including CJK.
"""

import logging
import time
from datadog import statsd

try:
    from elasticsearch import Elasticsearch
    from langdetect import detect
except ImportError:
    print("Elasticsearch or langdetect not installed.")
    Elasticsearch = None
    detect = None

NON_WHITESPACE_LANGS = frozenset({"zh", "ja", "th", "km", "lo", "my"})

DEFAULT_INDEX_NAME = "prod_lyrics_data"
STRICT_INDEX_NAME = "prod_lyrics_data_v2"
# Default ES client


# How many views a song needs to be protected by this system
MIN_VIEWS_FOR_PROTECTION = 1000000

# Songs with this many views are considered "viral" and get stricter protection
VIRAL_SONG_VIEW_THRESHOLD = 10000000

# For regular songs, what percentage of phrases must match to trigger detection
# Higher = stricter matching, fewer false positives
STANDARD_PHRASE_MATCH_THRESHOLD = "50%"

# For viral songs, use a lower threshold to catch partial copying
# Lower = catches more potential copies, more false positives
VIRAL_PHRASE_MATCH_THRESHOLD = "30%"

# Add a minimum required matches constant
MIN_VIRAL_MATCHES = 3

# Define supported languages with their corresponding field suffixes
SUPPORTED_LANGUAGES = {
    # CJK languages use cjk analyzer
    "ja": "text.cjk",
    "zh": "text.cjk",
    "ko": "text.cjk",
    # Default English/Latin script languages use base analyzer
    "en": "text.phrase",
    "es": "text.phrase",
    "fr": "text.phrase",
    # Add more supported languages as analyzers are added
}

# Common invisible/zero-width characters
INVISIBLE_CHARS = [
    "\u200e",  # Left-to-Right Mark
    "\u200f",  # Right-to-Left Mark
    "\u200b",  # Zero Width Space
    "\u200c",  # Zero Width Non-Joiner
    "\u200d",  # Zero Width Joiner
    "\u2060",  # Word Joiner
    "\u2061",  # Function Application
    "\u2062",  # Invisible Times
    "\u2063",  # Invisible Separator
    "\u2064",  # Invisible Plus
    "\ufeff",  # Zero Width No-Break Space
]

logger = logging.getLogger(__name__)


def _get_field_for_text(text: str) -> str:
    """Determine which text field to use based on language detection."""
    try:
        lang = detect(text)
    except:  # noqa: E722
        return "text"  # Default to standard analyzer if detection fails

    # Use specialized field if language is supported, otherwise fall back to standard
    return SUPPORTED_LANGUAGES.get(lang, "text.phrase")


# We could also do this in Elasticsearch but, to better capture when users actually
# add illegal characters, I opted for a pre-processing step instead.
def remove_invisible_chars(text: str) -> str:
    """Remove all invisible/zero-width characters and any spaces that follow them."""
    result = text
    for char in INVISIBLE_CHARS:
        # Remove invisible char + space combination
        result = result.replace(char + " ", "")
        # Remove just the invisible char
        result = result.replace(char, "")
    return result


def remove_repeated_noise_phrases(text: str, phrase_len: int = 5, max_word_len: int = 3) -> str:
    """
    Remove repeated noise phrases like 'la la la la la' from the text.
    Only removes phrases where all words are <= max_word_len and identical.
    """
    words = text.split()
    cleaned_words = []
    i = 0
    while i <= len(words) - phrase_len:
        phrase = words[i : i + phrase_len]
        if all(len(w) <= max_word_len for w in phrase) and len(set(phrase)) == 1:
            # Skip this phrase
            i += phrase_len
            continue
        cleaned_words.append(words[i])
        i += 1
    # Add any trailing words that weren't part of a phrase
    cleaned_words.extend(words[i:])
    return " ".join(cleaned_words)


class CopyrightDetector:
    def __init__(self, lyrics_mod_es_url, lyrics_mod_es_key, deployment_type):
        self.es_client = Elasticsearch(lyrics_mod_es_url, api_key=lyrics_mod_es_key)
        self.deployment_type = deployment_type

    def are_lyrics_copyrighted(self, lyrics: str, strict: bool = False) -> bool:
        print("Running copyright detector with strict mode?", strict)
        result = self._search_lyrics_v3(lyrics, threshold=0.9, timeout_ms=600, strict=strict)
        return result["is_copyright_infringement"]

    def _search_lyrics_v3(
        self, lyrics: str, threshold: float, timeout_ms: int = 600, strict: bool = False
    ) -> dict:
        """Optimized lyrics search for phrase detection."""
        start_time = time.time()
        timed_out = False

        # Clean lyrics before processing
        has_invisible = False
        try:
            # Certain unicode edgecases can throw here but it should be extremely rare
            # and shouldn't impact the rest of the copyright check
            len_before = len(lyrics)
            lyrics = remove_invisible_chars(lyrics)
            has_invisible = len(lyrics) < len_before
        except Exception as e:
            logger.warning(f"Could not remove invisible characters: {e}")

        has_noise = False
        try:
            len_before = len(lyrics)
            lyrics = remove_repeated_noise_phrases(lyrics)
            has_noise = len(lyrics) < len_before
        except Exception as e:
            logger.warning(f"Could not remove noise phrases: {e}")

        # Get detected language for metrics
        try:
            detected_lang = detect(lyrics)
        except:  # noqa: E722
            detected_lang = "unknown"

        field = _get_field_for_text(lyrics)

        # Scale threshold from 0-1 range to 0-500 range used by scoring
        scaled_min_score = threshold * 500
        boost_multiplier = 2 if len(lyrics.split()) < 50 else 1

        should_clauses = [
            # Add a high-boost exact phrase matcher using the appropriate field
            {
                "match_phrase": {
                    field: {
                        "query": lyrics,
                        "boost": 25.0 * boost_multiplier,
                        "_name": "exact_phrase_match",
                    }
                }
            },
            # For viral songs (>10M views), use appropriate field
            {
                "bool": {
                    "must": [
                        {"range": {"views": {"gte": VIRAL_SONG_VIEW_THRESHOLD}}},
                        {
                            "match": {
                                field: {
                                    "query": lyrics,
                                    # Require both percentage and minimum matches
                                    "minimum_should_match": f"{MIN_VIRAL_MATCHES}<{VIRAL_PHRASE_MATCH_THRESHOLD}",
                                    "boost": 20.0 * boost_multiplier,
                                    "_name": "viral_song_coverage",
                                }
                            }
                        },
                    ],
                    "_name": "viral_song_match",
                }
            },
            # For standard songs, use appropriate field
            {
                "bool": {
                    "must": [
                        {
                            "range": {
                                "views": {
                                    "gte": MIN_VIEWS_FOR_PROTECTION,
                                    "lt": VIRAL_SONG_VIEW_THRESHOLD,
                                }
                            }
                        },
                        {
                            "match": {
                                field: {
                                    "query": lyrics,
                                    "minimum_should_match": f"6<{STANDARD_PHRASE_MATCH_THRESHOLD}",
                                    "boost": 15.0 * boost_multiplier,
                                    "_name": "regular_song_sliding_window",
                                }
                            }
                        },
                    ],
                    "_name": "regular_song_match",
                }
            },
        ]

        if strict:
            print("Adding match_phrase clauses for unique 5-word phrases")
            phrases = [" ".join(lyrics.split()[i : i + 5]) for i in range(len(lyrics.split()) - 4)]
            should_clauses += [{"match_phrase": {"text.phrase": phrase}} for phrase in set(phrases)]

        query = {
            "_source": ["text", "views", "lang"],
            "size": 1,
            "min_score": scaled_min_score,
            "explain": True,
            "query": {
                "bool": {
                    "should": should_clauses,
                    "minimum_should_match": 1,
                },
            },
        }

        index = STRICT_INDEX_NAME

        try:
            results = self.es_client.search(
                index=index,
                body=query,
                _source=True,
                stored_fields=["text.phrase"],
                timeout=f"{timeout_ms}ms",
            )
            # Check if the query timed out
            timed_out = results.get("timed_out", False)
            if timed_out:
                logger.warning(
                    f"Lyrics search timed out after {timeout_ms}ms for language: {detected_lang}"
                )
        except Exception as e:
            logger.error(f"Error in lyrics search: {e}")
            timed_out = True
            results = {"hits": {"hits": [], "max_score": 0}}

        hits = results["hits"]["hits"]

        def _extract_matched_tokens(explanation):
            """Extract just the shingle tokens from the explanation details."""
            tokens = set()
            if isinstance(explanation, dict):
                desc = explanation.get("description", "")
                if "weight(text.phrase:" in desc:
                    # Extract token between 'text.phrase:' and ' in'
                    token = desc.split("text.phrase:")[1].split(" in")[0]
                    if len(token.split()) >= 4 if strict else 6:  # Only include 4-6-word shingles
                        tokens.add(token)
                for detail in explanation.get("details", []):
                    tokens.update(_extract_matched_tokens(detail))
            return tokens

        def _is_valid_shingle(shingle: str) -> bool:
            """Check if a shingle is valid."""
            words = shingle.split()

            # Return False if all words are short characters
            if all(len(word) <= 3 for word in words):
                return False

            # Check for hyphenated repetitions (e.g. "yeah-yeah-yeah") that may
            # reach the phrase match threshold but are clearly not copyright infringement
            if all(word.count("-") > 0 for word in words):
                parts = [part for word in words for part in word.split("-")]
                unique_parts = set(parts)
                if len(unique_parts) == 1:
                    return False

            # Check for minimum unique words (handles both normal and hyphenated cases)
            unique_words = set(words)
            if len(unique_words) < 4:
                return False

            return True

        similarities = [
            {
                "lyrics_id": hit["_id"],
                "lyrics": hit["_source"]["text"],
                "similarity_score": float(hit["_score"]),
                "normalized_score": 100 * (float(hit["_score"]) / results["hits"]["max_score"])
                if results["hits"]["max_score"]
                else 0,
                "matched_tokens": list(_extract_matched_tokens(hit.get("_explanation", {}))),
                "matched_queries": hit.get("matched_queries", []),
            }
            for hit in hits
            # Only include matches that have at least one valid shingle
            if any(
                _is_valid_shingle(shingle)
                for shingle in _extract_matched_tokens(hit.get("_explanation", {}))
            )
        ]

        # If all matches were filtered out, return empty results
        query_time = round(time.time() - start_time, 3)

        # default
        is_copyright_infringement = False

        if not similarities:
            result = {
                "model_name": "elasticsearch-v1",
                "similarities": [],
                "metadata": {
                    "score": 0.0,
                    "normalized_score": 0.0,
                    "query_time_seconds": query_time,
                    "matched_queries": [],
                    "timed_out": timed_out,
                },
                "is_copyright_infringement": False,
            }
        else:
            # Adjust thresholds for better detection
            is_copyright_infringement = any(
                (
                    # Strong exact match
                    s["similarity_score"] > 500
                    or
                    # High normalized score indicates a good match
                    s["normalized_score"] > 80
                )
                for s in similarities
            )

            # If timed out, force non-copyright infringement response
            if timed_out:
                result = {
                    "model_name": "elasticsearch-v1",
                    "similarities": [],
                    "metadata": {
                        "score": 0.0,
                        "normalized_score": 0.0,
                        "query_time_seconds": query_time,
                        "matched_queries": [],
                        "timed_out": True,
                    },
                    "is_copyright_infringement": False,
                }
            else:
                result = {
                    "model_name": "elasticsearch-v1",
                    "similarities": similarities,
                    "metadata": {
                        "score": similarities[0]["similarity_score"] if similarities else 0.0,
                        "normalized_score": similarities[0]["normalized_score"] if similarities else 0.0,
                        "query_time_seconds": query_time,
                        "matched_queries": similarities[0].get("matched_queries", [])
                        if similarities
                        else [],
                        "timed_out": False,
                    },
                    "is_copyright_infringement": is_copyright_infringement,
                }

        # Track metrics with language and environment tags
        tags = [
            f"env:{self.deployment_type}",
            f"lang:{detected_lang}",
            f"has_invisible:{has_invisible}",
            f"has_noise:{has_noise}",
            f"is_infringement:{is_copyright_infringement}",
            f"timed_out:{timed_out}",
        ]

        # Track throughput and results with all tags
        statsd.increment("copyright_detector.requests", tags=tags)
        if timed_out:  # Only track timeouts when they occur
            statsd.increment("copyright_detector.timeouts", tags=tags)

        statsd.distribution(
            "copyright_detector.query_time",
            query_time * 1000,  # Convert to milliseconds
            tags=tags,
        )

        if similarities:  # Only track similarity score if we have matches
            statsd.distribution(
                "copyright_detector.similarity_score", similarities[0]["similarity_score"], tags=tags
            )

        return result