import os
import io
import json
from typing import Tuple

import numpy as np
import pandas as pd
import gradio as gr
import plotly.express as px
import requests
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer

APP_TITLE = "StreamSmart Recommender"
APP_SUBTITLE = (
    "Improve streaming recommendations by combining viewer review sentiment "
    "with watch-time and engagement metrics."
)

analyzer = SentimentIntensityAnalyzer()

REQUIRED_REVIEW_COLS = ["title", "review_text"]
REQUIRED_WATCH_COLS = [
    "title",
    "genre",
    "avg_watch_time",
    "completion_rate",
    "drop_off_rate",
    "rewatch_rate",
    "click_through_rate",
]


def clean_text(text: str) -> str:
    if pd.isna(text):
        return ""
    text = str(text).strip().replace("\n", " ")
    return " ".join(text.split())


def compute_sentiment(text: str) -> float:
    return analyzer.polarity_scores(clean_text(text))["compound"]


def minmax(series: pd.Series) -> pd.Series:
    series = pd.to_numeric(series, errors="coerce").fillna(0)
    min_v = series.min()
    max_v = series.max()
    if max_v == min_v:
        return pd.Series(np.full(len(series), 0.5), index=series.index)
    return (series - min_v) / (max_v - min_v)


def sentiment_label(score: float) -> str:
    if score >= 0.2:
        return "Positive"
    if score <= -0.2:
        return "Negative"
    return "Neutral"


def action_label(score: float) -> str:
    if score >= 80:
        return "Promote strongly"
    if score >= 65:
        return "Promote selectively"
    if score >= 45:
        return "Investigate mismatch"
    return "Reduce priority"


def business_explanation(row: pd.Series) -> str:
    s = row["avg_sentiment"]
    c = row["completion_rate"]
    d = row["drop_off_rate"]
    score = row["recommendation_score"]

    if s >= 0.2 and c >= 0.7 and d <= 0.3:
        return (
            f"{row['title']} has strong viewer satisfaction and high completion, so it is a good candidate "
            "for broader recommendation placement."
        )
    if s >= 0.2 and c < 0.7:
        return (
            f"{row['title']} gets positive reactions from viewers who engage with it, but completion is weaker. "
            "This suggests the title may perform better with more targeted audience matching."
        )
    if s < 0.2 and c >= 0.7:
        return (
            f"{row['title']} keeps viewers watching, but sentiment is not especially strong. This may indicate "
            "good initial appeal with weaker perceived quality or expectation mismatch."
        )
    if score < 45:
        return (
            f"{row['title']} shows weak satisfaction and engagement signals overall, so it should not be prioritized "
            "in recommendation slots until content positioning improves."
        )
    return (
        f"{row['title']} is a mixed case: some engagement indicators are promising, but the platform should review "
        "audience fit, metadata, or recommendation placement before scaling promotion."
    )


def validate_columns(df: pd.DataFrame, required_cols: list, name: str) -> None:
    missing = [c for c in required_cols if c not in df.columns]
    if missing:
        raise gr.Error(f"{name} is missing required columns: {missing}")


def make_demo_data() -> Tuple[pd.DataFrame, pd.DataFrame]:
    reviews = pd.DataFrame(
        {
            "title": [
                "Midnight City", "Midnight City", "Ocean Echoes", "Ocean Echoes",
                "Crimson Truth", "Crimson Truth", "Quiet Orbit", "Quiet Orbit",
                "Laugh Track", "Laugh Track", "Golden Hour", "Golden Hour",
            ],
            "review_text": [
                "Amazing pacing and really addictive storyline.",
                "Loved the characters and watched it in one sitting.",
                "Beautiful idea but too slow in the middle.",
                "Strong visuals, but I almost stopped halfway.",
                "Suspenseful and smart, one of the best thrillers.",
                "Great acting and excellent ending.",
                "Interesting concept but not very engaging.",
                "Felt too long and the story did not pull me in.",
                "Funny and light, easy to keep watching.",
                "Very entertaining and rewatchable.",
                "Good cast but the episodes drag a bit.",
                "Not bad, but I expected more excitement.",
            ],
            "genre": [
                "Sci-Fi", "Sci-Fi", "Drama", "Drama", "Thriller", "Thriller",
                "Sci-Fi", "Sci-Fi", "Comedy", "Comedy", "Drama", "Drama",
            ],
        }
    )

    watch = pd.DataFrame(
        {
            "title": ["Midnight City", "Ocean Echoes", "Crimson Truth", "Quiet Orbit", "Laugh Track", "Golden Hour"],
            "genre": ["Sci-Fi", "Drama", "Thriller", "Sci-Fi", "Comedy", "Drama"],
            "avg_watch_time": [83, 58, 79, 41, 72, 54],
            "completion_rate": [0.86, 0.61, 0.81, 0.39, 0.76, 0.57],
            "drop_off_rate": [0.18, 0.33, 0.21, 0.48, 0.24, 0.37],
            "rewatch_rate": [0.31, 0.15, 0.27, 0.08, 0.25, 0.11],
            "click_through_rate": [0.42, 0.36, 0.39, 0.29, 0.41, 0.34],
        }
    )
    return reviews, watch


def run_analysis(reviews_file, watch_file, use_demo: bool):
    if use_demo:
        reviews_df, watch_df = make_demo_data()
    else:
        if reviews_file is None or watch_file is None:
            raise gr.Error("Upload both CSV files or use the demo dataset.")
        reviews_df = pd.read_csv(reviews_file.name)
        watch_df = pd.read_csv(watch_file.name)

    validate_columns(reviews_df, REQUIRED_REVIEW_COLS, "Reviews CSV")
    validate_columns(watch_df, REQUIRED_WATCH_COLS, "Watch-time CSV")

    reviews = reviews_df.copy()
    watch = watch_df.copy()

    reviews["review_text"] = reviews["review_text"].apply(clean_text)
    reviews["sentiment_score"] = reviews["review_text"].apply(compute_sentiment)
    reviews["sentiment_label"] = reviews["sentiment_score"].apply(sentiment_label)

    if "genre" in reviews.columns:
        review_agg = reviews.groupby("title", as_index=False).agg(
            avg_sentiment=("sentiment_score", "mean"),
            review_count=("sentiment_score", "count"),
            dominant_genre=("genre", lambda s: s.mode().iat[0] if not s.mode().empty else s.iloc[0]),
        )
    else:
        review_agg = reviews.groupby("title", as_index=False).agg(
            avg_sentiment=("sentiment_score", "mean"),
            review_count=("sentiment_score", "count"),
        )
        review_agg["dominant_genre"] = "Unknown"

    merged = pd.merge(watch, review_agg, on="title", how="left")
    merged["avg_sentiment"] = merged["avg_sentiment"].fillna(0)
    merged["review_count"] = merged["review_count"].fillna(0).astype(int)
    merged["genre"] = merged["genre"].fillna(merged["dominant_genre"]).fillna("Unknown")

    merged["sentiment_norm"] = minmax(merged["avg_sentiment"])
    merged["completion_norm"] = minmax(merged["completion_rate"])
    merged["watch_norm"] = minmax(merged["avg_watch_time"])
    merged["rewatch_norm"] = minmax(merged["rewatch_rate"])
    merged["ctr_norm"] = minmax(merged["click_through_rate"])
    merged["dropoff_norm"] = minmax(merged["drop_off_rate"])

    raw_score = (
        0.35 * merged["sentiment_norm"]
        + 0.30 * merged["completion_norm"]
        + 0.20 * merged["watch_norm"]
        + 0.10 * merged["rewatch_norm"]
        + 0.05 * merged["ctr_norm"]
        - 0.15 * merged["dropoff_norm"]
    )

    merged["recommendation_score"] = (raw_score.clip(lower=0) * 100).round(2)
    merged["action"] = merged["recommendation_score"].apply(action_label)
    merged["explanation"] = merged.apply(business_explanation, axis=1)
    merged = merged.sort_values("recommendation_score", ascending=False).reset_index(drop=True)

    summary = (
        f"Reviews analyzed: {len(reviews)} | Titles scored: {merged['title'].nunique()} | "
        f"Average sentiment: {merged['avg_sentiment'].mean():.2f} | "
        f"Average completion rate: {merged['completion_rate'].mean():.2f}"
    )

    table_cols = [
        "title", "genre", "avg_sentiment", "avg_watch_time", "completion_rate",
        "drop_off_rate", "rewatch_rate", "click_through_rate", "review_count",
        "recommendation_score", "action"
    ]
    top_table = merged[table_cols]

    top_plot = px.bar(
        merged.head(10),
        x="title",
        y="recommendation_score",
        title="Top Titles by Recommendation Score",
    )
    scatter_plot = px.scatter(
        merged,
        x="avg_sentiment",
        y="completion_rate",
        size="avg_watch_time",
        hover_name="title",
        color="genre",
        title="Sentiment vs Completion Rate",
    )
    genre_plot = px.bar(
        merged.groupby("genre", as_index=False)["recommendation_score"].mean().sort_values("recommendation_score", ascending=False),
        x="genre",
        y="recommendation_score",
        title="Average Recommendation Score by Genre",
    )

    processed_csv = io.StringIO()
    top_table.to_csv(processed_csv, index=False)

    payload = merged.to_json(orient="records")
    return summary, top_table, top_plot, scatter_plot, genre_plot, payload, processed_csv.getvalue()


def inspect_title(payload: str, selected_title: str):
    if not payload:
        raise gr.Error("Run the analysis first.")
    records = json.loads(payload)
    df = pd.DataFrame(records)
    if selected_title not in df["title"].values:
        raise gr.Error("Title not found.")
    row = df[df["title"] == selected_title].iloc[0]
    return (
        f"Title: {row['title']}\n"
        f"Genre: {row['genre']}\n"
        f"Average sentiment: {row['avg_sentiment']:.2f}\n"
        f"Average watch time: {row['avg_watch_time']:.2f}\n"
        f"Completion rate: {row['completion_rate']:.2f}\n"
        f"Drop-off rate: {row['drop_off_rate']:.2f}\n"
        f"Recommendation score: {row['recommendation_score']:.2f}\n"
        f"Suggested action: {row['action']}\n\n"
        f"Explanation: {row['explanation']}"
    )


def update_title_choices(payload: str):
    if not payload:
        return gr.Dropdown(choices=[], value=None)
    df = pd.DataFrame(json.loads(payload))
    choices = sorted(df["title"].dropna().unique().tolist())
    value = choices[0] if choices else None
    return gr.Dropdown(choices=choices, value=value)


def _post_to_webhook(env_name: str, body: dict, success_label: str) -> str:
    webhook_url = os.getenv(env_name, "").strip()
    if not webhook_url:
        return f"{env_name} is not set yet. Add it as a Hugging Face Space secret, then try again."

    response = requests.post(webhook_url, json=body, timeout=60)
    response.raise_for_status()

    try:
        result = response.json()
        return f"{success_label} ran successfully. Response: {json.dumps(result, indent=2)}"
    except Exception:
        return f"{success_label} ran successfully. Raw response: {response.text}"


def send_to_processing_workflow(payload: str):
    if not payload:
        raise gr.Error("Run the analysis first.")

    data = json.loads(payload)
    return _post_to_webhook(
        "N8N_PROCESS_WEBHOOK_URL",
        {"app": APP_TITLE, "records": data, "record_count": len(data)},
        "n8n processing workflow",
    )


def send_to_report_workflow(payload: str):
    if not payload:
        raise gr.Error("Run the analysis first.")

    data = json.loads(payload)
    top5 = data[:5]
    return _post_to_webhook(
        "N8N_REPORT_WEBHOOK_URL",
        {"app": APP_TITLE, "top_recommendations": top5, "all_results": data},
        "n8n report workflow",
    )


with gr.Blocks(title=APP_TITLE) as demo:
    gr.Markdown(f"# {APP_TITLE}\n\n{APP_SUBTITLE}")
    gr.Markdown(
        "This app combines qualitative viewer review sentiment with quantitative watch-time metrics "
        "to score how strongly each title should be recommended on a streaming platform."
    )

    payload_state = gr.State("")
    csv_state = gr.State("")

    with gr.Tab("1. Upload & Run"):
        use_demo = gr.Checkbox(label="Use built-in demo dataset", value=True)
        reviews_file = gr.File(label="Upload reviews CSV", file_types=[".csv"])
        watch_file = gr.File(label="Upload watch-time CSV", file_types=[".csv"])
        run_btn = gr.Button("Run Analysis", variant="primary")
        summary_box = gr.Textbox(label="Processing Summary", lines=2)

    with gr.Tab("2. Dashboard"):
        results_table = gr.Dataframe(label="Scored Titles")
        chart_1 = gr.Plot(label="Top Recommendation Scores")
        chart_2 = gr.Plot(label="Sentiment vs Completion")
        chart_3 = gr.Plot(label="Genre Performance")

    with gr.Tab("3. Title Drilldown"):
        title_dropdown = gr.Dropdown(label="Select a title", choices=[])
        detail_box = gr.Textbox(label="Title Recommendation Detail", lines=10)
        inspect_btn = gr.Button("Explain Selected Title")

    with gr.Tab("4. n8n Automation"):
        gr.Markdown(
            "Set these Hugging Face Space secrets before using the buttons below:\n\n"
            "- `N8N_PROCESS_WEBHOOK_URL`\n"
            "- `N8N_REPORT_WEBHOOK_URL`"
        )
        process_btn = gr.Button("Send Full Results to n8n Processing Workflow")
        process_status = gr.Textbox(label="Processing Workflow Status", lines=5)
        report_btn = gr.Button("Send Top Recommendations to n8n Report Workflow")
        report_status = gr.Textbox(label="Report Workflow Status", lines=5)

    with gr.Tab("5. Download"):
        download_file = gr.File(label="Download processed CSV")

    def save_csv_text(csv_text: str):
        path = "/tmp/processed_streamsmart_results.csv"
        with open(path, "w", encoding="utf-8") as f:
            f.write(csv_text)
        return path

    run_btn.click(
        fn=run_analysis,
        inputs=[reviews_file, watch_file, use_demo],
        outputs=[summary_box, results_table, chart_1, chart_2, chart_3, payload_state, csv_state],
    ).then(
        fn=update_title_choices,
        inputs=[payload_state],
        outputs=[title_dropdown],
    ).then(
        fn=save_csv_text,
        inputs=[csv_state],
        outputs=[download_file],
    )

    inspect_btn.click(
        fn=inspect_title,
        inputs=[payload_state, title_dropdown],
        outputs=[detail_box],
    )

    process_btn.click(
        fn=send_to_processing_workflow,
        inputs=[payload_state],
        outputs=[process_status],
    )

    report_btn.click(
        fn=send_to_report_workflow,
        inputs=[payload_state],
        outputs=[report_status],
    )

if __name__ == "__main__":
    demo.launch()