import base64
import json
import mimetypes
import os
from datetime import datetime

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

from app.config import settings
from app.models.model_config import ModelConfig
from app.models.token_usage import TokenUsage
from app.utils.id_gen import generate_id
from app.services.log_config import is_enabled, LOG_DIR, LOG_DATE_FORMAT, encrypt_data


def _sanitize_for_log(data):
    """Replace base64 data URIs with placeholder for readable logs."""
    if isinstance(data, str):
        if data.startswith("data:") and ";base64," in data:
            return "[base64 image data]"
        return data
    if isinstance(data, dict):
        return {k: _sanitize_for_log(v) for k, v in data.items()}
    if isinstance(data, list):
        return [_sanitize_for_log(item) for item in data]
    return data


def _log_ai_request_response(config, request_data: dict, response_data: dict | None, error: str | None = None):
    """Log AI model request/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")
        request_encrypted = encrypt_data(_sanitize_for_log(request_data), True)
        response_encrypted = encrypt_data(_sanitize_for_log(response_data), True) if response_data else ""
        entry = {
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "model_name": config.name,
            "model_id": config.model_name,
            "provider": config.provider,
            "api_base": config.api_base,
            "request": request_encrypted,
            "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


MOCK_OPTIMIZED_PROMPTS = {
    "直播": "专业直播间场景，45度斜角机位，暖色柔光打光，主播居中构图，背景虚化处理，产品特写切换流畅，镜头推进节奏感强，画面色彩饱和度高，适合电商直播推广视频。",
    "产品": "高端产品展示视频，360度旋转环绕拍摄，纯白/深色渐变背景，柔光箱打光消除阴影，微距镜头捕捉产品细节，金属/玻璃材质高光反射，品牌Logo水印角落显示。",
    "课程": "在线教育课程片头，明亮书房环境，讲师半身出镜，板书/屏幕录制无缝切换，字幕条底部滚动，知识要点动画弹出，背景轻音乐辅助，整体色调清新专业。",
    "美食": "美食制作过程记录，俯拍+侧拍双机位切换，暖色灯光突出食材质感，慢镜头捕捉烹饪瞬间（蒸汽、油花、酱汁淋洒），成品摆盘精致特写，色调偏暖黄增进食欲。",
    "品牌": "品牌形象宣传片，电影级调色（青橙对比色），航拍+地面多角度取景，城市/自然场景交替，人物情感特写穿插，品牌故事旁白叠加，片尾Logo定版动画。",
    "游戏": "游戏宣传CG风格视频，高速运镜+粒子特效，角色动态捕捉流畅，技能释放光效炸裂，UI界面模拟叠加，BGM史诗感配乐，画面帧率60fps丝滑体验。",
    "product": "Premium product showcase video with 360-degree rotation, clean gradient background, professional studio lighting, macro lens capturing fine details, metallic and glass material highlights, brand watermark in corner.",
    "live": "Professional livestream scene with 45-degree angle camera, warm soft lighting, host centered with bokeh background, smooth product close-up transitions, vibrant saturated colors.",
    "course": "Online education intro with bright study environment, instructor half-body shot, seamless screen recording transitions, animated key points overlay, clean professional tone.",
    "food": "Food preparation recording with overhead and side camera switching, warm lighting highlighting textures, slow-motion cooking moments, elegant plating close-up.",
    "brand": "Cinematic brand film with teal-orange color grading, aerial and ground multi-angle shots, urban and nature scenes alternating, emotional character close-ups, brand story narration.",
    "game": "Game promotional CG-style video with dynamic camera movements, particle effects, character motion capture, explosive skill light effects, epic BGM, smooth 60fps visuals.",
}


def _get_default_prompt(prompt: str, gen_type: str = "video") -> tuple[str, dict]:
    if gen_type == "image":
        optimized = (
            f"Professional high-quality image with expert composition, precise color grading, "
            f"sharp focus and rich details. Theme: {prompt}. Photographic style with "
            f"professional lighting and strong visual impact, suitable for commercial use."
        )
    else:
        optimized = (
            f"Professionally crafted video with expert composition, precise color grading, "
            f"smooth camera movements. Theme: {prompt}. Cinematic shooting techniques with "
            f"rich lighting layers and strong visual impact, suitable for commercial distribution."
        )
    return optimized, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}


async def optimize_prompt(
    db: AsyncSession,
    original_prompt: str,
    user_id: str | None = None,
    industry_key: str | None = None,
    duration: int | None = None,
    image_size: str | None = None,
    image_proportion: str | None = None,
    image_px: str | None | None = None,
    references: list[dict] | None = None,
    gen_type: str = "video",
) -> tuple[str, dict]:
    """Optimize user prompt using LLM. Returns (optimized_text, token_usage_dict)."""

    result = await db.execute(
        select(ModelConfig)
        .where(ModelConfig.is_active == True, ModelConfig.deleted_at.is_(None))
        .order_by(ModelConfig.priority.desc())
    )
    configs = list(result.scalars().all())
    # Release the read transaction before the external LLM request. Callers
    # must commit their business claim before invoking optimize_prompt.
    await db.commit()

    if configs:
        # 按 priority 从大到小依次尝试，跳过 mock，失败则用下一个
        for selected in configs:
            if selected.provider == "mock":
                continue
            if selected.provider in ("openai_compatible", "sdk"):
                try:
                    return await _call_openai_compatible(
                        selected, original_prompt, db, user_id, industry_key, duration,
                        references=references,
                        gen_type=gen_type,
                        image_size=image_size,
                        image_proportion=image_proportion,
                        image_px=image_px,
                    )
                except Exception:
                    continue

        # 所有真实模型都失败，降级到 mock
        mock_cfg = next((c for c in configs if c.provider == "mock"), None)
        if mock_cfg:
            return _mock_optimize(original_prompt, gen_type)

    if settings.LLM_MOCK:
        return _mock_optimize(original_prompt, gen_type)

    return _get_default_prompt(original_prompt, gen_type)


def _mock_optimize(prompt: str, gen_type: str = "video") -> tuple[str, dict]:
    """Return a keyword-matched mock optimized prompt."""
    for keyword, optimized in MOCK_OPTIMIZED_PROMPTS.items():
        if keyword in prompt:
            return optimized, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
    return _get_default_prompt(prompt, gen_type)


async def _call_openai_compatible(
    config: ModelConfig,
    original_prompt: str,
    db: AsyncSession | None = None,
    user_id: str | None = None,
    industry_key: str | None = None,
    duration: int | None = None,
    references: list[dict] | None = None,
    gen_type: str = "video",
    image_size: str | None = None,
    image_proportion: str | None = None,
    image_px: str | None | None = None,
) -> tuple[str, dict]:
    """Call an OpenAI-compatible API to optimize the prompt. Returns (content, token_usage)."""
    system_prompt = None
    if industry_key and db is not None:
        from app.models.industry_config import IndustryConfig
        result = await db.execute(
            select(IndustryConfig).where(IndustryConfig.key == industry_key).limit(1)
        )
        ind = result.scalar_one_or_none()
        if ind and ind.skills:
            try:
                skills = json.loads(ind.skills)
                if gen_type == "image":
                    skill_keys = ("文图理解生成图片提示词", "文图理解生成图片", "文图理解")
                else:
                    skill_keys = ("文图理解生成视频提示词", "文图理解")
                for s in skills:
                    if s.get("key") in skill_keys and s.get("label"):
                        system_prompt = s["label"]
                        break
            except (json.JSONDecodeError, TypeError):
                pass
        await db.commit()

    if not system_prompt:
        if gen_type == "image":
            system_prompt = (
                "你是一位专业的摄影师和图像创作专家。请根据用户提供的主题，"
                "生成一段详细的、专业的图片生成提示词。要求：\n"
                "1. 包含具体的画面构图\n"
                "2. 描述光影效果和色调\n"
                "3. 指定拍摄角度和镜头类型\n"
                "4. 画面风格和质感\n"
                "5. 整体不超过200字"
            )
        else:
            system_prompt = (
                "你是一位专业的视频导演和文案专家。请根据用户提供的视频主题，"
                "生成一段详细的、专业的视频生成提示词。要求：\n"
                "1. 包含具体的镜头语言（机位、运镜方式）\n"
                "2. 描述光影效果和色调\n"
                "3. 画面构图和视觉层次\n"
                "4. 适合的节奏感和转场\n"
                "5. 整体不超过200字"
            )

    if gen_type == "video" and duration:
        system_prompt += f"\n\n请根据视频总时长{duration}秒，合理分配镜头节奏，生成适合{duration}秒视频的提示词。"
        user_content = f"{original_prompt}\n\n请生成一段{duration}秒的视频提示词。"
    elif gen_type == "image" and image_size:
        system_prompt += f"\n\n请根据画面分辨率:{image_size}，宽高比:{image_proportion}，宽高像素值:{image_px}，生成适合该分辨率的图片提示词。"
        user_content = f"{original_prompt}\n\n请生成适合分辨率:{image_size}，宽高比:{image_proportion}，宽高像素值:{image_px}的图片提示词。"
    else:
        user_content = original_prompt

    # Build multimodal user message content when images are present
    image_urls = []
    video_urls = []
    if references:
        for ref in references:
            if not isinstance(ref, dict):
                continue

            ref_type = ref.get("type")
            ref_url = ref.get("url")

            if not ref_url:
                continue

            if ref_type == "image":
                image_urls.append(ref_url)

            if ref_type == "video":
                video_urls.append(ref_url)

    async def _build_multimodal_content(
        user_content: str,
        image_urls: list[str],
        video_urls: list[str],
    ) -> tuple[dict, dict | None]:
        """构建多模态 user_message。返回 (actual_message, log_message)。"""
        from app.utils.media import media_to_base64

        content_parts = [{"type": "text", "text": user_content}]

        from app.utils.media import get_llm_media_as_base64
        as_base64 = await get_llm_media_as_base64(db)
        for img in image_urls:
            if as_base64:
                url = await media_to_base64(img, "image/png")
            else:
                url = _file_url_or_data_uri(img, "image/png")
            content_parts.append({
                "type": "image_url",
                "image_url": {"url": url},
            })

        for video in video_urls:
            if as_base64:
                url = await media_to_base64(video, "video/mp4")
            else:
                url = _file_url_or_data_uri(video, "video/mp4")
            content_parts.append({
                "type": "video_url",
                "video_url": {"url": url},
            })

        user_message = {
            "role": "user",
            "content": content_parts,
        }

        # Log-friendly version: keep original paths instead of base64
        log_content_parts = [{"type": "text", "text": user_content}]
        for img in image_urls:
            log_content_parts.append({
                "type": "image_url",
                "image_url": {"url": img},
            })
        for video in video_urls:
            log_content_parts.append({
                "type": "video_url",
                "video_url": {"url": video},
            })

        log_user_message = {
            "role": "user",
            "content": log_content_parts,
        }
        return user_message, log_user_message

    def _file_url_or_data_uri(file_url: str, fallback_mime: str) -> str:
        """
        Convert local upload path to base64 data URI.
        Keep remote http/https/data URLs as-is.
        """
        if file_url.startswith(("http://", "https://", "data:")):
            return file_url

        # Compatible with /uploads/xxx and plain relative paths
        relative_path = file_url.replace("/uploads/", "", 1).lstrip("/")
        file_path = os.path.join(settings.UPLOAD_LOCAL_PATH, relative_path)

        mime = mimetypes.guess_type(file_path)[0] or fallback_mime

        with open(file_path, "rb") as f:
            b64 = base64.b64encode(f.read()).decode()

        return f"data:{mime};base64,{b64}"

    if image_urls or video_urls:
        user_message, log_user_message = await _build_multimodal_content(
            user_content, image_urls, video_urls
        )
    else:
        user_message = {
            "role": "user",
            "content": user_content,
        }
        log_user_message = None

    async with httpx.AsyncClient(timeout=120) as client:
        request_data = {
            "model": config.model_name,
            "messages": [
                {"role": "system", "content": system_prompt},
                user_message,
            ],
            "max_tokens": config.max_tokens,
            "temperature": config.temperature,
        }
        # Build log-friendly request data (image paths instead of base64)
        if log_user_message:
            log_request_data = {**request_data, "messages": [
                {"role": "system", "content": system_prompt},
                log_user_message,
            ]}
        else:
            log_request_data = request_data
        try:
            response = await client.post(
                f"{config.api_base}/chat/completions",
                headers={
                    "Authorization": f"Bearer {config.api_key}",
                    "Content-Type": "application/json",
                },
                json=request_data,
            )
            if response.status_code >= 400:
                error_body = response.text
                _log_ai_request_response(config, log_request_data, None, error=f"HTTP {response.status_code}: {error_body}")
                raise RuntimeError(f"HTTP {response.status_code}: {error_body}")
            data = response.json()
        except RuntimeError:
            raise
        except Exception as e:
            _log_ai_request_response(config, log_request_data, None, error=str(e))
            raise RuntimeError(f"{type(e).__name__}: {e}")

        # Log request/response
        _log_ai_request_response(config, log_request_data, data)

        # Record token usage
        usage = data.get("usage", {})
        input_tokens = usage.get("prompt_tokens", 0)
        output_tokens = usage.get("completion_tokens", 0)
        total_tokens = usage.get("total_tokens", input_tokens + output_tokens)

        token_usage_id = None
        if db is not None:
            token_usage_id = generate_id()
            record = TokenUsage(
                id=token_usage_id,
                model_config_id=config.id,
                user_id=user_id,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                total_tokens=total_tokens,
            )
            db.add(record)
            await db.flush()

        content = data["choices"][0]["message"]["content"].strip()
        token_usage = {
            "token_usage_id": token_usage_id,
            "model_config_id": config.id,
            "model_config_name": config.name,
            "model_provider": config.provider,
            "model_name": config.model_name,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": total_tokens,
        }
        return content, token_usage