import json
import os
import sys
import threading
import traceback
import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from typing import Any, Dict, List, Optional

import redis
from flask import Flask, jsonify, request

base_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(base_path)


app = Flask(__name__)

MAX_WORKERS = 4
REDIS_TASK_KEY_PREFIX = "text_extract:task:"
executor = ThreadPoolExecutor(max_workers=MAX_WORKERS)
redis_client = redis.Redis(
    host="172.16.0.24",
    port=6379,
    password="aiyingli@@123",
    socket_connect_timeout=2,
    socket_timeout=10,
    decode_responses=True,
    db=0,
)
tasks: Dict[str, Dict[str, Any]] = {}
tasks_lock = threading.Lock()


def extract_events_from_url(url: str):
    from ai_chat_stream import parse_utils

    yield from parse_utils.extract_mp4_to_text(url)


def now_text() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def build_task(url: str) -> Dict[str, Any]:
    return {
        "task_id": str(uuid.uuid4()),
        "url": url,
        "status": "queued",
        "status_message": "任务已提交",
        "text": "",
        "video_info": None,
        "events": [],
        "error": None,
        "created_at": now_text(),
        "updated_at": now_text(),
        "finished_at": None,
    }


def get_task_key(task_id: str) -> str:
    return f"{REDIS_TASK_KEY_PREFIX}{task_id}"


def save_task_to_redis(task: Dict[str, Any]):
    try:
        redis_client.set(
            get_task_key(task["task_id"]),
            json.dumps(task, ensure_ascii=False),
        )
    except Exception as exc:
        print(f"save task to redis failed, task_id={task.get('task_id')}, error={exc}")


def load_task_from_redis(task_id: str) -> Optional[Dict[str, Any]]:
    try:
        task_text = redis_client.get(get_task_key(task_id))
    except Exception as exc:
        print(f"load task from redis failed, task_id={task_id}, error={exc}")
        return None
    if not task_text:
        return None
    return json.loads(task_text)


def update_task(task_id: str, **kwargs):
    task = None
    with tasks_lock:
        task = tasks.get(task_id)

    if task is None:
        task = load_task_from_redis(task_id)
    if task is None:
        return

    with tasks_lock:
        task.update(kwargs)
        task["updated_at"] = now_text()
        tasks[task_id] = task

    save_task_to_redis(task)


def get_task(task_id: str) -> Optional[Dict[str, Any]]:
    with tasks_lock:
        task = tasks.get(task_id)

    if task is None:
        task = load_task_from_redis(task_id)
    if task is None:
        return None

    with tasks_lock:
        tasks[task_id] = task
    return dict(task)


def build_public_task(task: Dict[str, Any]) -> Dict[str, Any]:
    return {
        "task_id": task.get("task_id"),
        "url": task.get("url"),
        "status": task.get("status"),
        "status_message": task.get("status_message"),
        "video_info": task.get("video_info"),
        "text": task.get("text", ""),
        "error": task.get("error"),
        "created_at": task.get("created_at"),
        "updated_at": task.get("updated_at"),
        "finished_at": task.get("finished_at"),
    }


def parse_sse_event(raw_event: Any) -> Optional[Dict[str, Any]]:
    if isinstance(raw_event, bytes):
        event_text = raw_event.decode("utf-8", errors="ignore")
    else:
        event_text = str(raw_event)

    for line in event_text.splitlines():
        line = line.strip()
        if not line.startswith("data:"):
            continue
        data_text = line[len("data:"):].strip()
        if not data_text or data_text == "[DONE]":
            continue
        try:
            return json.loads(data_text)
        except json.JSONDecodeError:
            return {
                "message": data_text,
                "type": 0,
            }
    return None


def extract_text_from_url(task_id: str, url: str):
    update_task(task_id, status="running", status_message="任务处理中")
    events: List[Dict[str, Any]] = []
    text_parts: List[str] = []

    try:
        for raw_event in extract_events_from_url(url):
            event = parse_sse_event(raw_event)
            if event is None:
                continue

            events.append(event)
            if event.get("type") == 5 and isinstance(event.get("message"), dict):
                update_task(
                    task_id,
                    video_info=event.get("message"),
                    status_message="已获取视频基础信息",
                )

            if event.get("type") == 3:
                message = event.get("message")
                if message is not None:
                    update_task(task_id, status_message=str(message))

            if event.get("type") == 1:
                message = event.get("message")
                if message is not None:
                    text_parts.append(str(message))

            update_task(
                task_id,
                events=list(events),
                text="".join(text_parts),
            )

        final_text = "".join(text_parts)
        if final_text:
            update_task(
                task_id,
                status="success",
                status_message="文案提取完成",
                text=final_text,
                events=events,
                finished_at=now_text(),
            )
        else:
            update_task(
                task_id,
                status="failed",
                status_message="文案提取失败",
                text="",
                events=events,
                error="未提取到文案",
                finished_at=now_text(),
            )
    except Exception as exc:
        update_task(
            task_id,
            status="failed",
            status_message="文案提取异常",
            error=f"{exc}\n{traceback.format_exc()}",
            events=events,
            text="".join(text_parts),
            finished_at=now_text(),
        )


def normalize_urls(payload: Dict[str, Any]) -> List[str]:
    urls = payload.get("urls")
    if urls is None:
        url = payload.get("url")
        urls = [url] if url else []

    if not isinstance(urls, list):
        raise ValueError("urls 必须是数组")

    cleaned_urls = []
    for item in urls:
        if not isinstance(item, str):
            raise ValueError("url 必须是字符串")
        url = item.strip()
        if url:
            cleaned_urls.append(url)

    if not cleaned_urls:
        raise ValueError("请提交 url 或 urls")
    return cleaned_urls


@app.route("/api/text-extract/tasks", methods=["POST"])
def submit_text_extract_tasks():
    payload = request.get_json(silent=True) or {}
    try:
        urls = normalize_urls(payload)
    except ValueError as exc:
        return jsonify({
            "code": 400,
            "msg": str(exc),
            "data": None,
        }), 400

    created_tasks = []
    for url in urls:
        task = build_task(url)
        task_id = task["task_id"]
        with tasks_lock:
            tasks[task_id] = task
        save_task_to_redis(task)
        executor.submit(extract_text_from_url, task_id, url)
        created_tasks.append({
            "url": url,
            "task_id": task_id,
            "status": task["status"],
        })

    return jsonify({
        "code": 200,
        "msg": "success",
        "data": created_tasks,
    }), 202


@app.route("/api/text-extract/tasks/<task_id>", methods=["GET"])
def get_text_extract_task(task_id: str):
    task = get_task(task_id)
    if task is None:
        return jsonify({
            "code": 404,
            "msg": "任务不存在",
            "data": None,
        }), 404

    return jsonify({
        "code": 200,
        "msg": "success",
        "data": build_public_task(task),
    })


@app.route("/api/text-extract/result", methods=["GET"])
def get_text_extract_result_by_query():
    task_id = request.args.get("task_id", "").strip()
    if not task_id:
        return jsonify({
            "code": 400,
            "msg": "请提供 task_id",
            "data": None,
        }), 400
    return get_text_extract_task(task_id)


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=10443)
