import base64
import json
import logging
import mimetypes
import os
from datetime import datetime, timezone

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from volcenginesdkarkruntime import AsyncArk

from app.config import settings
from app.models.video_engine import VideoEngine
from app.services.log_config import is_enabled, LOG_DIR, LOG_DATE_FORMAT, encrypt_data
from app.services.generation_provider_types import (
    ProviderGenerationRecordLike,
    ProviderVideoEngineLike,
)

logger = logging.getLogger("videogen")


def _log_video_request(engine: ProviderVideoEngineLike, record_id: str, request_data: dict):
    """Log video generation request to log/AiModel/YYYY-MM-DD.log"""
    if not is_enabled():
        return
    try:
        os.makedirs(LOG_DIR, exist_ok=True)
        today = datetime.now().strftime(LOG_DATE_FORMAT)
        log_file = os.path.join(LOG_DIR, f"{today}.log")
        request_str = json.dumps(request_data, ensure_ascii=False)
        request_encrypted = encrypt_data(request_data)
        entry = {
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "type": "video_gen_request",
            "engine": engine.name,
            "model": engine.model_name,
            "record_id": record_id,
            "request": request_encrypted,
            "request_length": len(request_str),
        }
        with open(log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    except Exception:
        pass


def _log_video_response(record_id: str, response_data: dict, error: str | None = None):
    """Log video generation response to log/AiModel/YYYY-MM-DD.log"""
    if not is_enabled():
        return
    try:
        os.makedirs(LOG_DIR, exist_ok=True)
        today = datetime.now().strftime(LOG_DATE_FORMAT)
        log_file = os.path.join(LOG_DIR, f"{today}.log")
        response_encrypted = encrypt_data(response_data) if response_data else ""

        entry = {
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "type": "video_gen_response",
            "record_id": record_id,
            "response": response_encrypted,
            "error": error,
        }
        with open(log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    except Exception:
        pass





async def get_active_engine(db: AsyncSession) -> VideoEngine:
    """Get the active video engine with highest priority."""
    result = await db.execute(
        select(VideoEngine)
        .where(VideoEngine.is_active == True)
        .order_by(VideoEngine.priority.desc())
        .limit(1)
    )
    engine = result.scalar_one_or_none()
    if not engine:
        raise ValueError("没有可用的视频引擎，请联系管理员配置")
    return engine


def _resolve_url(url: str) -> str:
    """Convert local path to base64 data URI, pass through remote URLs."""
    # if url.startswith("http"):
    #     return url
    # # Local file: read and encode as base64 data URI
    # file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, url.replace("/uploads/", ""))
    # if not os.path.exists(file_path):
    #     return url
    # mime = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
    # with open(file_path, "rb") as f:
    #     b64 = base64.b64encode(f.read()).decode()
    # return f"data:{mime};base64,{b64}"
    if url.startswith(("http://", "https://", "data:")):
        return url
    return f"{settings.BASE_URL.rstrip('/')}/{url.lstrip('/')}"


async def submit_video_task(
    db: AsyncSession,
    engine: ProviderVideoEngineLike,
    record: ProviderGenerationRecordLike,
    *,
    include_media_references: bool,
) -> str:
    """Submit a video generation task via Ark SDK. Returns task_id."""
    client = AsyncArk(
        base_url=engine.api_base,
        api_key=engine.api_key,
    )

    content = [{"type": "text", "text": record.optimized_prompt or record.original_prompt}]

    # GenerationRecord 的附件已在 API 提词优化阶段参与过 optimized_prompt 生成；
    # ChatGenerationTask 不走 API 提词优化，所以创建供应商任务时仍需携带附件。
    if include_media_references and record.media_references:
        try:
            refs = json.loads(record.media_references)
            for ref in refs:
                ref_type = ref.get("type")
                ref_url = ref.get("url", "")
                if ref_type == "image" and ref_url:
                    resolved = _resolve_url(ref_url)
                    content.append({"type": "image_url", "image_url": {"url": resolved},"role":"reference_image"})
                elif ref_type == "video" and ref_url:
                    resolved = _resolve_url(ref_url)
                    content.append({"type": "video_url", "video_url": {"url": resolved},"role":"reference_video"})
        except (json.JSONDecodeError, TypeError):
            pass

    request_payload = {
        "model": engine.model_name,
        "content": content,
        "ratio": record.aspect_ratio,
        "duration": record.duration,
        "resolution": record.resolution,
        "generate_audio": True,
        "watermark": False,
    }

    # Log request to AiModel log. include_media_references 只用于排查日志，不传给供应商 API。
    _log_video_request(
        engine,
        record.id,
        {**request_payload, "include_media_references": include_media_references},
    )

    try:
        result = await client.content_generation.tasks.create(**request_payload)
        task_id = result.id
        _log_video_response(record.id, {"task_id": task_id})
    except Exception as e:
        _log_video_response(record.id, {}, str(e))
        raise
    finally:
        await client.close()
    
    return task_id


async def poll_task_status(engine: VideoEngine, task_id: str) -> dict:
    """Query task status via Ark SDK. Returns {status, video_url, response_data}."""
    client = AsyncArk(
        base_url=engine.api_base,
        api_key=engine.api_key,
    )

    result = await client.content_generation.tasks.get(task_id=task_id)
    await client.close()

    # Serialize response
    response_dict = {
        "id": result.id,
        "model": result.model,
        "status": result.status,
        "created_at": result.created_at,
        "updated_at": result.updated_at,
    }

    video_url = None
    video_tokens = 0
    if result.status == "succeeded" and result.content:
        video_url = getattr(result.content, "video_url", None)
        response_dict["video_url"] = video_url
        response_dict["duration"] = getattr(result, "duration", None)
        response_dict["ratio"] = getattr(result, "ratio", None)
        response_dict["resolution"] = getattr(result, "resolution", None)
        # Extract usage info if present
        usage = getattr(result, "usage", None)
        if usage:
            response_dict["usage"] = {
                "input_tokens": getattr(usage, "input_tokens", 0),
                "output_tokens": getattr(usage, "output_tokens", 0),
                "total_tokens": getattr(usage, "total_tokens", 0),
            }
            video_tokens = getattr(usage, "total_tokens", 0)
    elif result.status == "failed":
        response_dict["error"] = str(getattr(result, "error", "视频生成失败"))

    return {
        "status": result.status,
        "video_url": video_url,
        "video_tokens": video_tokens,
        "response_data": json.dumps(response_dict, ensure_ascii=False, default=str),
        "error": response_dict.get("error"),
    }


async def download_video(video_url: str, dest_path: str) -> str:
    """Download video to local storage."""
    import os

    os.makedirs(os.path.dirname(dest_path), exist_ok=True)

    async with httpx.AsyncClient(timeout=300) as client:
        async with client.stream("GET", video_url) as response:
            response.raise_for_status()
            with open(dest_path, "wb") as f:
                async for chunk in response.aiter_bytes(chunk_size=8192):
                    f.write(chunk)
    return dest_path
