"""私信消息内容解析、存储与展示（文本 / 图片 / 表情 / 语音 / 视频）。"""

from __future__ import annotations

import json
import re
from typing import Any

MSG_TYPE_TEXT = 7
MSG_TYPE_STICKER = 5
MSG_TYPE_VOICE = 17
MSG_TYPE_IMAGE = 27
MSG_TYPE_VIDEO = 8
MSG_TYPE_LINK_CARD = 70

_TYPE_LABELS = {
    "text": "文本",
    "image": "图片",
    "sticker": "表情",
    "voice": "语音",
    "video": "视频",
    "link": "链接",
    "link_card": "链接卡片",
}

_PLACEHOLDER_MARKERS = {
    "[表情包]",
    "[语音]",
    "[图片]",
    "[视频]",
    "[未读消息]",
}

_URI_HINT_RE = re.compile(
    r"(tos-cn|aweme-|voice/|ies-music|\.mp3|\.m4a|\.aac|\.mpeg|\.webp|\.jpeg|\.jpg|\.png|\.gif)",
    re.I,
)

_AUDIO_URL_RE = re.compile(
    r"(douyin-user-audio|/audio/|sc=audio|voice/|ies-music|\.mp3|\.m4a|\.aac|\.mpeg|\.wav|\.ogg)",
    re.I,
)

_VIDEO_URL_RE = re.compile(
    r"(sc=video|/video/|\.mp4|\.mov|\.webm|\.m3u8)",
    re.I,
)

_MEDIA_STRING_KEYS = (
    "url",
    "uri",
    "main_url",
    "download_url",
    "remote_url",
    "encrypt_url",
    "play_url",
    "secret_url",
    "audio_url",
    "video_url",
    "cover_url",
    "local_path",
)

_MEDIA_NESTED_KEYS = (
    "resource_url",
    "static_url",
    "animate_url",
    "cover_url",
    "thumb_url",
    "origin_url",
    "image",
    "picture",
    "pic",
    "sticker",
    "emoji",
    "audio",
    "voice",
    "video",
    "media",
    "large_url",
    "medium_url",
    "thumb",
    "avatar_thumb",
    "play_url",
)


def _normalize_uri_path(raw: str) -> str:
    path = (raw or "").strip().lstrip("/")
    if path.startswith("obj/"):
        path = path[4:]
    return path


def uri_to_cdn_urls(uri: str, *, prefer_voice: bool = False) -> list[str]:
    """将抖音 IM 中的 uri / tos 路径转为可访问的 CDN URL 候选列表。"""
    raw = (uri or "").strip()
    if not raw:
        return []
    if raw.startswith("//"):
        return [f"https:{raw}"]
    if raw.startswith("http://") or raw.startswith("https://"):
        return [raw]

    path = _normalize_uri_path(raw)
    if not path:
        return []

    candidates: list[str] = []
    seen: set[str] = set()

    def add(url: str) -> None:
        url = (url or "").strip()
        if url and url not in seen:
            seen.add(url)
            candidates.append(url)

    lower = path.lower()
    is_voice = prefer_voice or lower.startswith("voice/") or lower.endswith((".mp3", ".m4a", ".aac"))
    is_image = (
        not is_voice
        and (
            "tos-cn-i" in lower
            or "aweme-" in lower
            or lower.endswith((".jpeg", ".jpg", ".png", ".webp", ".gif"))
        )
    )

    if is_voice:
        for host in (
            "sf6-cdn-tos.douyinstatic.com",
            "sf3-cdn-tos.douyinstatic.com",
            "lf3-static.bytednsdoc.com",
        ):
            add(f"https://{host}/obj/{path}")
        add(f"https://p3.douyinpic.com/obj/{path}")

    if is_image or "tos-cn" in lower or "aweme" in lower:
        add(f"https://p3.douyinpic.com/obj/{path}")
        for size in ("720x720", "480x480", "300x300", "200x200", "100x100"):
            add(f"https://p3.douyinpic.com/aweme/{size}/{path}")
        add(f"https://p9-dy.byteimg.com/img/{path}")
        add(f"https://p6-dy.byteimg.com/img/{path}")

    add(f"https://p3.douyinpic.com/obj/{path}")
    add(f"https://p3-sign.douyinpic.com/obj/{path}".replace("-sign", ""))
    return candidates


def resolve_media_uri(uri: str, *, prefer_voice: bool = False) -> str:
    urls = uri_to_cdn_urls(uri, prefer_voice=prefer_voice)
    return urls[0] if urls else ""


def _looks_like_audio_url(value: str) -> bool:
    raw = (value or "").strip()
    return bool(raw and _AUDIO_URL_RE.search(raw))


def _looks_like_video_url(value: str) -> bool:
    raw = (value or "").strip()
    return bool(raw and _VIDEO_URL_RE.search(raw))


def _infer_media_type_from_url(url: str) -> str:
    """根据 URL 特征推断媒体类型（语音/视频/图片）。"""
    if _looks_like_audio_url(url):
        return "voice"
    if _looks_like_video_url(url):
        return "video"
    return "image"


def _looks_like_media_uri(value: str) -> bool:
    raw = (value or "").strip()
    if not raw or raw.startswith("{"):
        return False
    if raw.startswith("http://") or raw.startswith("https://") or raw.startswith("//"):
        return True
    return bool(_URI_HINT_RE.search(raw))


def _resolve_string_media(value: str, *, prefer_voice: bool = False) -> tuple[str, str]:
    raw = (value or "").strip()
    if not raw:
        return "", ""
    if raw.startswith("//"):
        return f"https:{raw}", raw
    if raw.startswith("http://") or raw.startswith("https://"):
        return raw, ""
    if _looks_like_media_uri(raw):
        return resolve_media_uri(raw, prefer_voice=prefer_voice), raw
    return "", ""


def _valid_sticker_id(value: Any) -> str:
    if value is None or value == "":
        return ""
    try:
        if int(value) == 0:
            return ""
    except (TypeError, ValueError):
        pass
    sticker_id = str(value).strip()
    return "" if sticker_id in ("0", "null", "None") else sticker_id


def _collect_media_candidates(value: Any, out: list[str], *, prefer_voice: bool = False, depth: int = 0) -> None:
    if depth > 12:
        return
    if isinstance(value, str):
        raw = value.strip()
        if raw.startswith("http://") or raw.startswith("https://") or raw.startswith("//"):
            url, _ = _resolve_string_media(raw, prefer_voice=prefer_voice)
            if url:
                out.append(url)
        return
    if isinstance(value, dict):
        for list_key in ("url_list", "urls", "urlList"):
            urls = value.get(list_key)
            if isinstance(urls, list):
                for item in urls:
                    _collect_media_candidates(item, out, prefer_voice=prefer_voice, depth=depth + 1)
        for key in _MEDIA_STRING_KEYS:
            direct = value.get(key)
            if isinstance(direct, str):
                url, _ = _resolve_string_media(direct, prefer_voice=prefer_voice)
                if url:
                    out.append(url)
        for nested_key in _MEDIA_NESTED_KEYS:
            _collect_media_candidates(value.get(nested_key), out, prefer_voice=prefer_voice, depth=depth + 1)
        for nested in value.values():
            if isinstance(nested, (dict, list)):
                _collect_media_candidates(nested, out, prefer_voice=prefer_voice, depth=depth + 1)
        return
    if isinstance(value, list):
        for item in value:
            _collect_media_candidates(item, out, prefer_voice=prefer_voice, depth=depth + 1)


def _pick_http_url(value: Any, *, prefer_voice: bool = False) -> str:
    candidates: list[str] = []
    _collect_media_candidates(value, candidates, prefer_voice=prefer_voice)
    return candidates[0] if candidates else ""


def _pick_media_uri(value: Any) -> str:
    if isinstance(value, str) and _looks_like_media_uri(value) and not value.strip().startswith("http"):
        return _normalize_uri_path(value)
    if isinstance(value, dict):
        for key in ("uri", "local_path", "remote_url"):
            direct = value.get(key)
            if isinstance(direct, str) and _looks_like_media_uri(direct) and not direct.strip().startswith("http"):
                return _normalize_uri_path(direct)
        for nested_key in _MEDIA_NESTED_KEYS:
            uri = _pick_media_uri(value.get(nested_key))
            if uri:
                return uri
        for nested in value.values():
            if isinstance(nested, (dict, list)):
                uri = _pick_media_uri(nested)
                if uri:
                    return uri
    if isinstance(value, list):
        for item in value:
            uri = _pick_media_uri(item)
            if uri:
                return uri
    return ""


def _coerce_message_type(value: Any, default: int = MSG_TYPE_TEXT) -> int:
    try:
        if value is None or value == "":
            return default
        return int(value)
    except (TypeError, ValueError):
        return default


def _parse_content_json(content_raw: str | dict | None) -> dict[str, Any]:
    if isinstance(content_raw, dict):
        data = content_raw
    else:
        raw = str(content_raw or "").strip()
        if raw.startswith("{"):
            try:
                parsed = json.loads(raw)
                if isinstance(parsed, dict):
                    data = parsed
                else:
                    return {}
            except json.JSONDecodeError:
                return {}
        else:
            return {}

    for nested_key in ("ext", "ai_ext", "extra", "payload", "data"):
        nested = data.get(nested_key)
        if isinstance(nested, str) and nested.strip().startswith("{"):
            try:
                nested_data = json.loads(nested)
                if isinstance(nested_data, dict):
                    merged = {**nested_data, **data}
                    data = merged
            except json.JSONDecodeError:
                pass
    return data


def _safe_int(value: Any) -> int | None:
    try:
        if value is None or value == "":
            return None
        return int(value)
    except (TypeError, ValueError):
        return None


def format_im_message(content_raw: str | dict | None, message_type: int = MSG_TYPE_TEXT) -> dict[str, Any]:
    """将 IM 原始 content 解析为统一结构 {type, text, url, ...}。"""
    content_json = _parse_content_json(content_raw)
    embedded_type = _coerce_message_type(
        content_json.get("message_type")
        or content_json.get("messageType")
        or content_json.get("msg_type"),
        message_type,
    )
    if embedded_type != MSG_TYPE_TEXT:
        message_type = embedded_type

    _EMPTY_MEDIA_DEFAULTS = {
        MSG_TYPE_IMAGE: ("image", "[图片]"),
        MSG_TYPE_STICKER: ("sticker", "[表情包]"),
        MSG_TYPE_VOICE: ("voice", "[语音]"),
        MSG_TYPE_VIDEO: ("video", "[视频]"),
        MSG_TYPE_LINK_CARD: ("link_card", "[链接卡片]"),
    }

    if not content_json:
        raw = str(content_raw or "").strip()
        if not raw:
            # 抖音相册图片(type 27)/部分语音等会以「空 content」推送，URL 不随推送下发。
            # 不能直接丢弃，否则消息「收不到」；这里按类型返回占位，URL 留空待后续补取。
            if message_type in _EMPTY_MEDIA_DEFAULTS:
                t, txt = _EMPTY_MEDIA_DEFAULTS[message_type]
                return {"type": t, "text": txt}
            return {"type": "text", "text": ""}
        if raw in _PLACEHOLDER_MARKERS:
            mapping = {
                "[图片]": "image",
                "[表情包]": "sticker",
                "[语音]": "voice",
                "[视频]": "video",
            }
            return {"type": mapping.get(raw, "text"), "text": raw}
        if message_type == MSG_TYPE_TEXT:
            return {"type": "text", "text": raw}
        content_json = {"text": raw}

    prefer_voice = message_type == MSG_TYPE_VOICE or bool(content_json.get("audio") or content_json.get("voice"))
    url = _pick_http_url(content_json, prefer_voice=prefer_voice)
    media_uri = _pick_media_uri(content_json)
    if not url and media_uri:
        url = resolve_media_uri(media_uri, prefer_voice=prefer_voice)
    duration = _safe_int(
        content_json.get("duration")
        or content_json.get("audio_duration")
        or content_json.get("video_duration")
    )
    width = _safe_int(content_json.get("width") or content_json.get("w"))
    height = _safe_int(content_json.get("height") or content_json.get("h"))

    def _media_payload(msg_type: str, text: str, **extra: Any) -> dict[str, Any]:
        payload: dict[str, Any] = {"type": msg_type, "text": text}
        if url:
            payload["url"] = url
        elif media_uri:
            payload["uri"] = media_uri
        payload.update({k: v for k, v in extra.items() if v not in (None, "", 0)})
        return payload

    # 语音/视频也会带 resource_url，不能仅凭该字段判为图片；优先按类型与 URL 特征识别。
    is_voice = (
        message_type == MSG_TYPE_VOICE
        or bool(content_json.get("audio") or content_json.get("voice"))
        or (url and _looks_like_audio_url(url))
    )
    is_video = (
        message_type == MSG_TYPE_VIDEO
        or bool(content_json.get("video"))
        or (url and _looks_like_video_url(url))
    )
    if is_voice and not is_video:
        return _media_payload("voice", "[语音]", duration=duration)
    if is_video:
        return _media_payload("video", "[视频]", duration=duration, width=width, height=height)

    has_image_hint = (
        message_type == MSG_TYPE_IMAGE
        or content_json.get("image")
        or content_json.get("inline_pic")
        or (
            content_json.get("resource_url")
            and not (url and (_looks_like_audio_url(url) or _looks_like_video_url(url)))
        )
    )
    if has_image_hint:
        # 抖音相册私图(biz_tag=aweme_im)的大图 URL 是加密内容，浏览器无法直接渲染；
        # 但 content 内嵌 inline_pic（base64 WEBP 缩略图），直接转 data URI 即可显示。
        inline = content_json.get("inline_pic")
        if isinstance(inline, str) and inline.strip():
            b64 = re.sub(r"\s+", "", inline)
            data_uri = f"data:image/webp;base64,{b64}"
            payload = {"type": "image", "text": "[图片]", "url": data_uri}
            if width:
                payload["width"] = width
            if height:
                payload["height"] = height
            return payload
        return _media_payload("image", "[图片]", width=width, height=height)
    if message_type == MSG_TYPE_STICKER or content_json.get("static_url") or content_json.get("animate_url") or _valid_sticker_id(
        content_json.get("sticker_id") or content_json.get("id")
    ):
        sticker_id = _valid_sticker_id(content_json.get("sticker_id") or content_json.get("id"))
        return _media_payload(
            "sticker",
            "[表情包]",
            sticker_id=sticker_id,
            name=str(content_json.get("display_name") or content_json.get("name") or ""),
        )

    link_card = _parse_link_card_payload(content_json, message_type)
    if link_card:
        return link_card

    rich_link = _parse_rich_text_link(content_json)
    if rich_link:
        return rich_link

    text = (
        str(content_json.get("text") or content_json.get("content") or content_json.get("message") or "")
    ).strip()
    if not text and url:
        inferred = _infer_media_type_from_url(url)
        if inferred == "voice":
            return {"type": "voice", "text": "[语音]", "url": url, "duration": duration}
        if inferred == "video":
            return {"type": "video", "text": "[视频]", "url": url, "duration": duration}
        if message_type == MSG_TYPE_IMAGE:
            return {"type": "image", "text": "[图片]", "url": url, "width": width, "height": height}
        if message_type == MSG_TYPE_STICKER:
            return {"type": "sticker", "text": "[表情包]", "url": url}
        if message_type == MSG_TYPE_VOICE:
            return {"type": "voice", "text": "[语音]", "url": url, "duration": duration}
        if message_type == MSG_TYPE_VIDEO:
            return {"type": "video", "text": "[视频]", "url": url, "duration": duration}

    final_text = text or str(content_raw or "").strip()
    emoji = _resolve_text_emoji(final_text)
    if emoji:
        return emoji
    return {"type": "text", "text": final_text}


def _parse_link_card_payload(content_json: dict[str, Any], message_type: int) -> dict[str, Any] | None:
    link_info = content_json.get("link_info")
    if not isinstance(link_info, dict):
        link_info = {}
    has_link = (
        message_type == MSG_TYPE_LINK_CARD
        or link_info
        or content_json.get("link_url")
        or content_json.get("cover_url")
    )
    if not has_link:
        return None
    title = str(content_json.get("title") or link_info.get("title") or "").strip()
    desc = str(
        content_json.get("desc")
        or content_json.get("description")
        or link_info.get("desc")
        or link_info.get("description")
        or ""
    ).strip()
    url = str(
        content_json.get("link_url")
        or content_json.get("url")
        or link_info.get("url")
        or link_info.get("link_url")
        or ""
    ).strip()
    cover = str(content_json.get("cover_url") or link_info.get("cover_url") or "").strip()
    text = title or desc or url or "[链接卡片]"
    payload: dict[str, Any] = {
        "type": "link_card",
        "text": text,
        "title": title,
        "desc": desc,
    }
    if url:
        payload["url"] = url
    if cover:
        payload["cover_url"] = cover
    return payload


def _parse_rich_text_link(content_json: dict[str, Any]) -> dict[str, Any] | None:
    rich = content_json.get("richTextInfos")
    if not isinstance(rich, list):
        return None
    for item in rich:
        if not isinstance(item, dict):
            continue
        link = str(item.get("link") or item.get("url") or "").strip()
        if not link:
            continue
        text = str(item.get("text") or item.get("display_text") or link).strip()
        return {"type": "link", "text": text or link, "url": link}
    return None


def _resolve_text_emoji(text: str) -> dict[str, Any] | None:
    """文字表情 [酷拽] 等：查标准表情表，命中则转成可显示的贴纸。"""
    stripped = (text or "").strip()
    if not stripped or not (stripped.startswith("[") and stripped.endswith("]")):
        return None
    try:
        from .emoji_pack import looks_like_emoji_token, lookup_emoji_url

        if not looks_like_emoji_token(stripped):
            return None
        url = lookup_emoji_url(stripped)
        if url:
            return {
                "type": "sticker",
                "text": stripped,
                "url": url,
                "name": stripped.strip("[]"),
            }
    except Exception:
        pass
    return None


def parse_incoming_message(data: dict[str, Any]) -> str:
    """从 IM API / WebSocket 消息 dict 提取并序列化展示内容。"""
    if not isinstance(data, dict):
        return str(data or "").strip()

    msg_type = _coerce_message_type(
        data.get("message_type") or data.get("messageType") or data.get("msg_type"),
        MSG_TYPE_TEXT,
    )
    content_raw = (
        data.get("content")
        or data.get("message")
        or data.get("msg")
        or data.get("lastMessage")
        or data.get("last_msg")
        or data.get("preview")
        or data.get("brief")
        or ""
    )
    if isinstance(content_raw, dict):
        if not msg_type or msg_type == MSG_TYPE_TEXT:
            msg_type = _coerce_message_type(
                content_raw.get("message_type")
                or content_raw.get("messageType")
                or content_raw.get("msg_type"),
                msg_type,
            )
        parsed = format_im_message(content_raw, msg_type)
        return serialize_message_content(parsed)

    if isinstance(content_raw, str):
        raw = content_raw.strip()
        if raw.startswith("{"):
            parsed = format_im_message(raw, msg_type)
            if parsed.get("type") != "text" or parsed.get("url") or msg_type != MSG_TYPE_TEXT:
                return serialize_message_content(parsed)
        if raw in _PLACEHOLDER_MARKERS and msg_type != MSG_TYPE_TEXT:
            parsed = format_im_message(raw, msg_type)
            return serialize_message_content(parsed)
        if msg_type != MSG_TYPE_TEXT:
            parsed = format_im_message(raw, msg_type)
            return serialize_message_content(parsed)
        return raw

    return ""


def serialize_message_content(msg: dict[str, Any]) -> str:
    """序列化写入 message_logs / reply_content。"""
    msg_type = (msg.get("type") or "text").strip()
    if msg_type == "text":
        text = str(msg.get("text") or "").strip()
        return text
    cleaned = {k: v for k, v in msg.items() if v not in (None, "", [], {})}
    return json.dumps(cleaned, ensure_ascii=False, separators=(",", ":"))


def parse_stored_content(raw: str | None) -> dict[str, Any]:
    """解析数据库中的 message_content / reply_content。"""
    text = (raw or "").strip()
    if not text:
        return {"type": "text", "text": ""}
    if text.startswith("{"):
        try:
            data = json.loads(text)
            if isinstance(data, dict) and data.get("type"):
                if not data.get("url") and data.get("uri"):
                    prefer_voice = data.get("type") == "voice"
                    resolved = resolve_media_uri(str(data["uri"]), prefer_voice=prefer_voice)
                    if resolved:
                        data = {**data, "url": resolved}
                url = str(data.get("url") or "").strip()
                if url:
                    inferred = _infer_media_type_from_url(url)
                    current = data.get("type")
                    if current == "image" and inferred in ("voice", "video"):
                        data = {
                            **data,
                            "type": inferred,
                            "text": "[语音]" if inferred == "voice" else "[视频]",
                        }
                    elif current not in ("voice", "video", "image", "sticker") and inferred:
                        data = {**data, "type": inferred}
                return data
        except json.JSONDecodeError:
            pass
    if text in _PLACEHOLDER_MARKERS:
        mapping = {
            "[图片]": "image",
            "[表情包]": "sticker",
            "[语音]": "voice",
            "[视频]": "video",
        }
        return {"type": mapping.get(text, "text"), "text": text}
    return {"type": "text", "text": text}


def message_preview(raw: str | None) -> str:
    """会话列表/日志摘要。"""
    msg = parse_stored_content(raw)
    msg_type = msg.get("type") or "text"
    if msg_type == "text":
        return str(msg.get("text") or "")
    label = _TYPE_LABELS.get(msg_type, msg.get("text") or "[消息]")
    extra = str(msg.get("name") or "").strip()
    if extra and msg_type == "sticker":
        return f"[表情] {extra}"
    return str(msg.get("text") or label)


def normalize_outgoing_content(
    content: str = "",
    message_type: str | None = None,
    media_url: str | None = None,
    sticker_url: str | None = None,
    width: int | None = None,
    height: int | None = None,
    sticker_id: str | None = None,
) -> str:
    """构造可发送/可落库的 content 字符串。"""
    raw = (content or "").strip()
    parsed_json: dict[str, Any] | None = None
    if raw.startswith("{"):
        try:
            data = json.loads(raw)
            if isinstance(data, dict) and data.get("type"):
                parsed_json = data
        except json.JSONDecodeError:
            pass

    if parsed_json:
        merged = dict(parsed_json)
        if width and not merged.get("width"):
            merged["width"] = width
        if height and not merged.get("height"):
            merged["height"] = height
        if sticker_id and not merged.get("sticker_id"):
            merged["sticker_id"] = sticker_id
        url = (media_url or sticker_url or "").strip()
        if url and not merged.get("url"):
            merged["url"] = url
        return serialize_message_content(merged)

    explicit_type = (message_type or "").strip().lower()
    if explicit_type in ("image", "sticker", "voice", "video", "text"):
        if explicit_type == "text":
            return (content or "").strip()
        payload: dict[str, Any] = {"type": explicit_type}
        url = (media_url or sticker_url or "").strip()
        if url:
            payload["url"] = url
        if explicit_type == "sticker" and sticker_id:
            payload["sticker_id"] = sticker_id
        if width:
            payload["width"] = width
        if height:
            payload["height"] = height
        text = (content or "").strip()
        if text:
            payload["text"] = text
        elif explicit_type == "image":
            payload["text"] = "[图片]"
        elif explicit_type == "sticker":
            payload["text"] = "[表情包]"
        return serialize_message_content(payload)

    raw = (content or "").strip()
    if raw.startswith("{"):
        try:
            data = json.loads(raw)
            if isinstance(data, dict) and data.get("type"):
                return serialize_message_content(data)
        except json.JSONDecodeError:
            pass
    return raw


def is_media_message(raw: str | None) -> bool:
    return parse_stored_content(raw).get("type") not in (None, "text")


def format_system_log_message(raw: str | None) -> str:
    """系统诊断日志中的消息摘要。"""
    msg = parse_stored_content(raw)
    msg_type = msg.get("type") or "text"
    if msg_type == "text":
        return str(msg.get("text") or "")
    parts = [message_preview(raw)]
    url = str(msg.get("url") or "").strip()
    if url:
        parts.append(f"URL: {url}")
    duration = msg.get("duration")
    if duration:
        parts.append(f"时长: {duration}s")
    return " | ".join(parts)


def extract_urls_from_detail(detail: str | None) -> list[str]:
    if not detail:
        return []
    return re.findall(r"https?://[^\s\]|）)\"']+", detail)