"""Netflix 시청 기록 분석 모듈.

탐색적 데이터 분석(EDA)에서 반복적으로 수행하던 작업들을
**재사용 가능한 함수**와 **클래스**로 정리한 교육용 모듈입니다.

설계 의도 (수업용)
------------------
1. **순수 함수 (pure functions)**: 입력 DataFrame을 받아 새로운 DataFrame /
   결과를 돌려줍니다. 부수효과(side effect)가 없어 ``.pipe()`` 로 연결하거나
   단독으로 테스트하기 좋습니다.
2. **NetflixAnalyzer 클래스**: 데이터와 분석 동작을 하나로 묶어(캡슐화),
   ``analyzer.top_rated_titles()`` 처럼 객체 메서드로 호출합니다.
   파생 변수 계산과 장르 집계 결과를 내부에 캐싱하고, 필터링 메서드는
   새 ``NetflixAnalyzer`` 를 돌려주어 메서드 체이닝이 가능합니다.
3. **SummaryReport 데이터클래스**: 종합 리포트를 보기 좋게 출력합니다.

이 데이터셋의 주요 컬럼
-----------------------
- ``viewer_id``   : 시청자 ID
- ``title``       : 작품 제목
- ``rating``      : 시청자가 매긴 평점 (1~5)
- ``watch_date``  : 시청일 (datetime)
- ``release_year``: 개봉 연도
- ``runtime``     : 러닝타임 (분)
- ``genre``       : 장르 리스트 (예: ["Comedies", "Romantic Movies"])

사용 예시
---------
>>> import netflix_analysis as na
>>> analyzer = na.NetflixAnalyzer.from_parquet("data/netflix_ratings_titles.parquet")
>>> analyzer.basic_stats()
>>> analyzer.top_rated_titles(n=10, min_ratings=3)
>>> print(analyzer.summary_report())
"""

from __future__ import annotations

from dataclasses import dataclass, field

import pandas as pd
import seaborn.objects as so

# =============================================================================
# 0. 상수 (Constants)
# =============================================================================

#: 월요일~일요일 순서. Categorical 정렬에 사용.
WEEKDAY_ORDER = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]

#: 평점 컬럼 기본 이름.
RATING_COL = "rating"


# =============================================================================
# 1. 데이터 로딩 및 출력 설정
# =============================================================================

def setup_display(max_rows: int = 6, float_precision: int = 3) -> None:
    """수업용으로 pandas/numpy 출력 환경을 설정합니다.

    Parameters
    ----------
    max_rows : int
        DataFrame 출력 시 최대 행 수.
    float_precision : int
        실수 출력 자릿수.
    """
    import numpy as np

    pd.set_option("mode.copy_on_write", True)
    pd.options.display.float_format = f"{{:.{float_precision}f}}".format
    pd.options.display.max_rows = max_rows
    np.set_printoptions(precision=float_precision, suppress=True, legacy="1.25")


def load_netflix_data(path: str = "data/netflix_ratings_titles.parquet") -> pd.DataFrame:
    """parquet 파일에서 Netflix 시청 기록 데이터를 로드합니다."""
    return pd.read_parquet(path)


# =============================================================================
# 2. 데이터 랭글링 (Wrangling) — .pipe() 로 연결 가능한 transform 함수들
# =============================================================================

def add_decade(df: pd.DataFrame, year_col: str = "release_year") -> pd.DataFrame:
    """개봉 연도를 10년 단위(decade)로 묶은 ``decade`` 열을 추가합니다."""
    return df.assign(decade=lambda x: x[year_col] // 10 * 10)


def add_weekday(df: pd.DataFrame, date_col: str = "watch_date",
                ordered: bool = True) -> pd.DataFrame:
    """시청일에서 요일(3글자) ``weekday`` 열을 추가합니다.

    ``ordered=True`` 이면 월~일 순서를 가진 Categorical 로 변환하여
    정렬·시각화 시 요일 순서가 자연스럽게 유지됩니다.
    """
    weekday = df[date_col].dt.day_name().str[:3]
    if ordered:
        weekday = weekday.astype("category").cat.set_categories(WEEKDAY_ORDER)
    return df.assign(weekday=weekday)


def add_year_month(df: pd.DataFrame, date_col: str = "watch_date") -> pd.DataFrame:
    """시청일을 월 단위 기간(Period)으로 묶은 ``year_month`` 열을 추가합니다."""
    return df.assign(year_month=lambda x: x[date_col].dt.to_period("M"))


def add_runtime_hours(df: pd.DataFrame, runtime_col: str = "runtime") -> pd.DataFrame:
    """러닝타임(분)을 시간 단위로 바꾼 ``runtime_hours`` 열을 추가합니다."""
    return df.assign(runtime_hours=lambda x: x[runtime_col] / 60)


def enrich(df: pd.DataFrame) -> pd.DataFrame:
    """분석에 자주 쓰는 파생 변수를 한 번에 추가합니다.

    ``decade``, ``weekday``, ``year_month``, ``runtime_hours`` 를 모두 더한
    새 DataFrame 을 반환합니다. (원본은 변경하지 않음)

    Examples
    --------
    >>> netflix = enrich(netflix)
    """
    return (
        df
        .pipe(add_decade)
        .pipe(add_weekday)
        .pipe(add_year_month)
        .pipe(add_runtime_hours)
    )


# =============================================================================
# 3. 집계 / 통계 함수
# =============================================================================

def basic_stats(df: pd.DataFrame) -> dict:
    """데이터셋 전반의 기본 통계를 딕셔너리로 반환합니다.

    Returns
    -------
    dict
        총 시청 기록 수, 총 시청자 수, 고유 작품 수,
        총 시청 시간(시간 단위), 전체 평균 평점.
    """
    return {
        "총 시청 기록 수": int(df.shape[0]),
        "총 시청자 수": int(df["viewer_id"].nunique()),
        "고유 작품 수": int(df["title"].nunique()),
        "총 시청 시간(시간)": float(df["runtime"].sum() / 60),
        "전체 평균 평점": float(df[RATING_COL].mean()),
    }


def rating_stats_by(df: pd.DataFrame, group_col, min_count: int = 1) -> pd.DataFrame:
    """임의의 그룹 컬럼별 평점 통계(mean, std, count)를 계산하는 범용 함수.

    Parameters
    ----------
    df : pd.DataFrame
    group_col : str or list[str]
        그룹화 기준 컬럼(들).
    min_count : int
        최소 평가 수. 이 수 미만인 그룹은 제외됩니다.
    """
    stats = (
        df
        .groupby(group_col)[RATING_COL]
        .agg(["mean", "std", "count"])
        .reset_index()
    )
    return stats.query("count >= @min_count")


def viewer_stats(df: pd.DataFrame, min_count: int = 1) -> pd.DataFrame:
    """시청자별 평점 통계(mean, std, count)를 계산합니다."""
    return rating_stats_by(df, "viewer_id", min_count=min_count)


def top_viewers(df: pd.DataFrame, n: int = 5) -> pd.DataFrame:
    """가장 활발한(시청 횟수 기준) 시청자 TOP N 을 반환합니다."""
    return df.value_counts("viewer_id").head(n).reset_index(name="count")


def top_rated_titles(df: pd.DataFrame, n: int = 10, min_ratings: int = 3) -> pd.DataFrame:
    """평균 평점이 높은 작품 TOP N 을 반환합니다.

    ``min_ratings`` 명 이상이 평가한 작품만 후보로 삼아, 소수의 평가로
    인한 왜곡을 줄입니다.
    """
    return (
        df
        .groupby("title")[RATING_COL]
        .agg(["mean", "count"])
        .query("count >= @min_ratings")
        .nlargest(n, "mean")
        .reset_index()
    )


def genre_stats(df: pd.DataFrame, genre_col: str = "genre") -> pd.DataFrame:
    """장르별 통계를 계산합니다 (리스트형 ``genre`` 열을 explode 처리).

    Returns
    -------
    pd.DataFrame
        컬럼: genre, count(시청 횟수), mean_rating(평균 평점),
        total_viewers(고유 시청자 수), total_runtime(총 시청 시간, 시간 단위).
    """
    return (
        df.explode(genre_col)
        .groupby(genre_col)
        .agg(
            count=(RATING_COL, "size"),
            mean_rating=(RATING_COL, "mean"),
            total_viewers=("viewer_id", "nunique"),
            total_runtime=("runtime", "sum"),
        )
        .assign(total_runtime=lambda x: x["total_runtime"] / 60)
        .reset_index()
    )


def monthly_trend(df: pd.DataFrame, date_col: str = "watch_date") -> pd.DataFrame:
    """월별 시청 횟수 추이를 계산합니다."""
    return (
        df.assign(year_month=lambda x: x[date_col].dt.to_period("M"))
        .value_counts("year_month")
        .sort_index()
        .reset_index(name="count")
    )


def weekday_pattern(df: pd.DataFrame, date_col: str = "watch_date") -> pd.DataFrame:
    """요일별 시청 횟수 패턴을 월~일 순서로 계산합니다."""
    return (
        add_weekday(df, date_col=date_col)
        .value_counts("weekday")
        .sort_index()
        .reset_index(name="count")
    )


def rating_distribution(df: pd.DataFrame) -> pd.DataFrame:
    """평점값별 빈도(분포)를 평점 오름차순으로 계산합니다."""
    return df.value_counts(RATING_COL).sort_index().reset_index(name="count")


# =============================================================================
# 4. groupby + apply 변환 함수
# =============================================================================

def standardize(series: pd.Series) -> pd.Series:
    """시리즈를 z-score 로 표준화합니다: (x - mean) / std."""
    return (series - series.mean()) / series.std()


def standardize_rating_by(df: pd.DataFrame, group_col: str = "viewer_id") -> pd.DataFrame:
    """그룹별로 평점을 표준화합니다.

    시청자마다 평점을 주는 기준(후하거나 박함)이 다르므로, 시청자별
    z-score 를 계산하면 시청자 간 비교가 공정해집니다.
    """
    return (
        df
        .groupby(group_col)[RATING_COL]
        .apply(standardize)
        .reset_index(level=0)
    )


def min_max_by(df: pd.DataFrame, group_col: str = "title",
               value_col: str = RATING_COL) -> pd.Series:
    """그룹별 최솟값/최댓값을 한 번에 계산합니다."""
    def _min_max(x):
        return pd.Series([x.min(), x.max()], index=["min", "max"])

    return df.groupby(group_col)[value_col].apply(_min_max)


# =============================================================================
# 5. 시각화 헬퍼 (seaborn.objects API)
# =============================================================================

def plot_genre_ranking(genre_stats_df: pd.DataFrame, metric: str = "count",
                       title: str = "장르별 순위") -> so.Plot:
    """장르 통계를 가로 막대그래프로 시각화합니다 (metric 내림차순 정렬)."""
    order = genre_stats_df.sort_values(metric, ascending=False)["genre"].values
    return (
        so.Plot(genre_stats_df, y="genre", x=metric)
        .add(so.Bar())
        .scale(y=so.Nominal(order=order))
        .label(title=title)
    )


def plot_monthly_trend(monthly_df: pd.DataFrame, title: str = "월별 시청 추이") -> so.Plot:
    """월별 시청 추이를 막대그래프로 시각화합니다."""
    plot_df = monthly_df.assign(year_month=lambda x: x["year_month"].astype(str))
    return (
        so.Plot(plot_df, x="count", y="year_month")
        .add(so.Bar())
        .layout(size=(6, 10))
        .label(title=title)
    )


def plot_weekday_pattern(weekday_df: pd.DataFrame, title: str = "요일별 시청 패턴") -> so.Plot:
    """요일별 시청 패턴을 가로 막대그래프로 시각화합니다."""
    return (
        so.Plot(weekday_df, y="weekday", x="count")
        .add(so.Bar())
        .label(title=title)
    )


def plot_rating_distribution(df: pd.DataFrame, title: str = "평점 분포") -> so.Plot:
    """평점 분포를 막대 히스토그램으로 시각화합니다."""
    return (
        so.Plot(df, y=RATING_COL)
        .add(so.Bar(), so.Hist(discrete=True))
        .label(title=title)
    )


# =============================================================================
# 6. 종합 리포트 데이터클래스
# =============================================================================

@dataclass
class SummaryReport:
    """종합 시청 리포트.

    ``NetflixAnalyzer.summary_report()`` 가 생성하며, ``print()`` 하면
    사람이 읽기 좋은 형태로 출력됩니다 (``__str__`` 재정의).
    """

    total_records: int
    total_viewers: int
    unique_titles: int
    total_hours: float
    average_rating: float
    favorite_genre: str
    top_titles: list = field(default_factory=list)  # [{"title", "mean", "count"}, ...]

    def __str__(self) -> str:
        lines = [
            "=" * 40,
            "  시청 기록 종합 리포트",
            "=" * 40,
            f"- 총 시청 기록 수 : {self.total_records:,}개",
            f"- 총 시청자 수    : {self.total_viewers:,}명",
            f"- 고유 작품 수    : {self.unique_titles:,}개",
            f"- 총 시청 시간    : {self.total_hours:,.1f}시간",
            f"- 전체 평균 평점  : {self.average_rating:.2f}/5",
            "",
            f"가장 인기 있는 장르: {self.favorite_genre}",
            "",
            "평균 평점이 가장 높은 작품:",
        ]
        for i, row in enumerate(self.top_titles, start=1):
            lines.append(
                f"  {i}. {row['title']} "
                f"(평균: {row['mean']:.2f}/5, {row['count']:,}명 평가)"
            )
        return "\n".join(lines)

    def to_dict(self) -> dict:
        """리포트를 딕셔너리로 변환합니다 (JSON 저장 등에 사용)."""
        return {
            "총 시청 기록 수": self.total_records,
            "총 시청자 수": self.total_viewers,
            "고유 작품 수": self.unique_titles,
            "총 시청 시간(시간)": self.total_hours,
            "전체 평균 평점": self.average_rating,
            "가장 인기 있는 장르": self.favorite_genre,
            "평균 평점 TOP": self.top_titles,
        }


# =============================================================================
# 7. NetflixAnalyzer 클래스 — 데이터 + 분석 동작을 묶은 핵심 클래스
# =============================================================================

class NetflixAnalyzer:
    """Netflix 시청 기록을 분석하는 클래스.

    DataFrame 하나를 감싸(wrap), EDA 에서 하던 집계·시각화 작업을
    메서드로 제공합니다.

    특징
    ----
    - 생성 시 파생 변수(decade, weekday, year_month, runtime_hours)를
      자동으로 추가합니다.
    - ``genre_stats`` 처럼 비용이 큰 집계 결과는 내부에 **캐싱**하여
      반복 호출 시 다시 계산하지 않습니다.
    - ``filter()`` 계열 메서드는 새 ``NetflixAnalyzer`` 를 돌려주므로
      ``analyzer.filter_by_genre("Dramas").top_rated_titles()`` 처럼
      체이닝할 수 있습니다.

    Examples
    --------
    >>> analyzer = NetflixAnalyzer.from_parquet("data/netflix_ratings_titles.parquet")
    >>> analyzer.basic_stats()
    >>> analyzer.top_rated_titles(n=5)
    >>> print(analyzer.summary_report())
    """

    def __init__(self, df: pd.DataFrame, enrich_data: bool = True):
        self._df = enrich(df) if enrich_data else df.copy()
        self._genre_stats_cache: pd.DataFrame | None = None

    # --- 생성자 (alternative constructors) ---------------------------------
    @classmethod
    def from_parquet(cls, path: str = "data/netflix_ratings_titles.parquet") -> "NetflixAnalyzer":
        """parquet 파일에서 바로 분석기를 생성합니다."""
        return cls(load_netflix_data(path))

    # --- 기본 속성 / 던더(dunder) 메서드 -----------------------------------
    @property
    def df(self) -> pd.DataFrame:
        """파생 변수가 추가된 내부 DataFrame (읽기 전용 접근)."""
        return self._df

    def __len__(self) -> int:
        return len(self._df)

    def __repr__(self) -> str:
        return (
            f"<NetflixAnalyzer: 기록 {len(self._df):,}개, "
            f"시청자 {self._df['viewer_id'].nunique():,}명, "
            f"작품 {self._df['title'].nunique():,}개>"
        )

    # --- 기본 통계 ---------------------------------------------------------
    def basic_stats(self) -> dict:
        """데이터셋 기본 통계 딕셔너리."""
        return basic_stats(self._df)

    # --- 시청자 분석 -------------------------------------------------------
    def viewer_stats(self, min_count: int = 1) -> pd.DataFrame:
        """시청자별 평점 통계."""
        return viewer_stats(self._df, min_count=min_count)

    def top_viewers(self, n: int = 5) -> pd.DataFrame:
        """가장 활발한 시청자 TOP N."""
        return top_viewers(self._df, n=n)

    # --- 작품 분석 ---------------------------------------------------------
    def top_rated_titles(self, n: int = 10, min_ratings: int = 3) -> pd.DataFrame:
        """평균 평점 TOP N 작품."""
        return top_rated_titles(self._df, n=n, min_ratings=min_ratings)

    # --- 장르 분석 (캐싱) --------------------------------------------------
    def genre_stats(self, force: bool = False) -> pd.DataFrame:
        """장르별 통계. 한 번 계산하면 캐싱하여 재사용합니다.

        Parameters
        ----------
        force : bool
            True 이면 캐시를 무시하고 다시 계산합니다.
        """
        if self._genre_stats_cache is None or force:
            self._genre_stats_cache = genre_stats(self._df)
        return self._genre_stats_cache

    def favorite_genre(self) -> str:
        """시청 횟수가 가장 많은(인기) 장르 이름."""
        gs = self.genre_stats()
        return gs.set_index("genre")["count"].idxmax()

    # --- 시간대 분석 -------------------------------------------------------
    def monthly_trend(self) -> pd.DataFrame:
        """월별 시청 추이."""
        return monthly_trend(self._df)

    def weekday_pattern(self) -> pd.DataFrame:
        """요일별 시청 패턴."""
        return weekday_pattern(self._df)

    # --- 평점 분석 ---------------------------------------------------------
    def rating_distribution(self) -> pd.DataFrame:
        """평점 분포."""
        return rating_distribution(self._df)

    def standardized_ratings(self, by: str = "viewer_id") -> pd.DataFrame:
        """그룹(기본: 시청자)별로 표준화한 평점."""
        return standardize_rating_by(self._df, group_col=by)

    # --- 필터링 (새 분석기 반환 → 체이닝 가능) -----------------------------
    def filter(self, query_str: str) -> "NetflixAnalyzer":
        """pandas ``query()`` 문자열로 필터링한 새 분석기를 반환합니다.

        Examples
        --------
        >>> analyzer.filter("release_year >= 2000")
        """
        return NetflixAnalyzer(self._df.query(query_str), enrich_data=False)

    def filter_by_year_range(self, start: int, end: int) -> "NetflixAnalyzer":
        """개봉 연도 [start, end] 범위로 필터링한 새 분석기를 반환합니다."""
        return self.filter(f"{start} <= release_year <= {end}")

    def filter_by_genre(self, genre: str) -> "NetflixAnalyzer":
        """특정 장르를 포함하는 기록만 남긴 새 분석기를 반환합니다."""
        mask = self._df["genre"].apply(lambda gs: genre in gs)
        return NetflixAnalyzer(self._df[mask], enrich_data=False)

    # --- 시각화 (헬퍼 함수에 위임) -----------------------------------------
    def plot_genre_ranking(self, metric: str = "count") -> so.Plot:
        return plot_genre_ranking(self.genre_stats(), metric=metric)

    def plot_monthly_trend(self) -> so.Plot:
        return plot_monthly_trend(self.monthly_trend())

    def plot_weekday_pattern(self) -> so.Plot:
        return plot_weekday_pattern(self.weekday_pattern())

    def plot_rating_distribution(self) -> so.Plot:
        return plot_rating_distribution(self._df)

    # --- 종합 리포트 -------------------------------------------------------
    def summary_report(self, top_n: int = 3, min_ratings: int = 3) -> SummaryReport:
        """기본 통계 + 인기 장르 + 평균 평점 TOP 을 묶은 종합 리포트."""
        stats = self.basic_stats()
        top = self.top_rated_titles(n=top_n, min_ratings=min_ratings)
        return SummaryReport(
            total_records=stats["총 시청 기록 수"],
            total_viewers=stats["총 시청자 수"],
            unique_titles=stats["고유 작품 수"],
            total_hours=stats["총 시청 시간(시간)"],
            average_rating=stats["전체 평균 평점"],
            favorite_genre=self.favorite_genre(),
            top_titles=top.to_dict(orient="records"),
        )