from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from typing import Optional
import aiosqlite, json
from datetime import datetime
from urllib.parse import quote
from database import get_db
from services.message_formatter import extract_quote

router = APIRouter(prefix="/api/topics", tags=["topics"])
CHATLOG_BATCH_SIZE = 80
NAME_LOOKUP_PAGE_SIZE = 500
NAME_LOOKUP_MAX_ITEMS = 5000
STALE_SUMMARIZE_MINUTES = 15

class TopicCreate(BaseModel):
    group_id: int
    title: str

class TopicPatch(BaseModel):
    title: Optional[str] = None
    status: Optional[str] = None

class MessageAdd(BaseModel):
    msg_seq: int
    talker: str


async def _mark_stale_summarize_tasks(db: aiosqlite.Connection, group_id: int, topic_id: int) -> None:
    error = "AI 报告生成任务超过 15 分钟未完成，已自动标记为失败，可重新生成"
    stale_window = f"-{STALE_SUMMARIZE_MINUTES} minutes"
    await db.execute(
        """
        UPDATE ai_tasks
        SET status='error', error=?, updated_at=CURRENT_TIMESTAMP
        WHERE group_id=?
          AND type='summarize'
          AND status='running'
          AND datetime(updated_at) <= datetime('now', ?)
        """,
        (error, group_id, stale_window),
    )
    await db.execute(
        """
        UPDATE topics
        SET status='error', updated_at=CURRENT_TIMESTAMP
        WHERE id=?
          AND status='processing'
          AND datetime(updated_at) <= datetime('now', ?)
        """,
        (topic_id, stale_window),
    )


def _normalize_chatlog_message(item: dict, fallback_seq: int = 0) -> dict:
    contents = item.get("contents") or item.get("Contents") or {}
    if not isinstance(contents, dict):
        contents = {}
    media_key = (
        contents.get("rawmd5")
        or contents.get("md5")
        or contents.get("path")
        or item.get("media_key")
        or item.get("mediaKey")
        or ""
    )
    voice_key = (
        str(contents.get("voice"))
        if contents.get("voice")
        else item.get("voice_key") or item.get("voiceKey") or ""
    )
    raw_type = item.get("type") or item.get("Type") or 1
    raw_sub_type = item.get("sub_type") or item.get("subType") or item.get("SubType") or 0
    try:
        is_file = int(raw_type) == 49 and int(raw_sub_type) == 6
    except Exception:
        is_file = False
    file_md5 = str(contents.get("md5") or item.get("file_md5") or item.get("fileMd5") or "") if is_file else ""
    file_name = (
        contents.get("title")
        or contents.get("fileName")
        or contents.get("filename")
        or item.get("file_name")
        or item.get("fileName")
        or ""
    ) if is_file else ""
    file_url = f"/api/files/{quote(file_md5, safe='')}?filename={quote(file_name or file_md5, safe='')}" if file_md5 else ""
    return {
        "seq": item.get("seq") or item.get("Seq") or item.get("sort_seq") or fallback_seq or 0,
        "sender": item.get("sender") or item.get("Sender") or "",
        "sender_name": (
            item.get("sender_name")
            or item.get("senderName")
            or item.get("SenderName")
            or item.get("sender")
            or item.get("Sender")
            or ""
        ),
        "create_time": item.get("create_time") or item.get("time") or item.get("CreateTime") or "",
        "content": item.get("content") or item.get("Content") or "",
        "type": raw_type,
        "sub_type": raw_sub_type,
        "contents": contents,
        "media_key": media_key,
        "voice_key": voice_key,
        "image_path": media_key,
        "voice_path": voice_key,
        "video_path": media_key,
        "file_path": media_key,
        "link_url": contents.get("url") or item.get("link_url") or "",
        "link_title": contents.get("title") or item.get("link_title") or "",
        "link_desc": contents.get("desc") or item.get("link_desc") or "",
        "link_thumb": contents.get("thumbUrl") or contents.get("thumb_url") or item.get("link_thumb") or "",
        "link_source": contents.get("sourceName") or contents.get("source_name") or item.get("link_source") or "",
        "quote": item.get("quote") or extract_quote(item),
        "is_file": is_file,
        "file_name": file_name,
        "file_md5": file_md5,
        "file_url": file_url,
    }


def _message_from_snapshot(raw: str | None, fallback_seq: int) -> dict | None:
    if not raw:
        return None
    try:
        item = json.loads(raw)
    except Exception:
        return None
    if not isinstance(item, dict):
        return None
    return _normalize_chatlog_message(item, fallback_seq)


def _looks_like_raw_sender_id(value: str | None) -> bool:
    value = (value or "").strip()
    return (
        not value
        or value.startswith("wxid_")
        or value.startswith("gh_")
        or value.endswith("@chatroom")
        or value.startswith("chatroom_")
    )


def _sender_display_name(item: dict, sender: str = "") -> str:
    for key in (
        "sender_name",
        "senderName",
        "SenderName",
        "accountName",
        "groupNickname",
        "displayName",
        "nickName",
        "remark",
    ):
        value = str(item.get(key) or "").strip()
        if value and value != sender and not _looks_like_raw_sender_id(value):
            return value
    return ""


def _message_date(value: str | None) -> str | None:
    value = (value or "").strip()
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).strftime("%Y-%m-%d")
    except Exception:
        return value[:10] if len(value) >= 10 and value[4:5] == "-" else None


async def _build_sender_name_map(talker: str, messages: list[dict]) -> dict[str, str]:
    names: dict[str, str] = {}

    for msg in messages:
        sender = str(msg.get("sender") or "").strip()
        name = _sender_display_name(msg, sender)
        if sender and name:
            names[sender] = name

    missing = {
        str(msg.get("sender") or "").strip()
        for msg in messages
        if str(msg.get("sender") or "").strip()
        and _looks_like_raw_sender_id(str(msg.get("sender_name") or "").strip())
    }
    missing = {sender for sender in missing if sender not in names}
    if not missing:
        return names

    from services.chatlog_client import chatlog_client

    dates = sorted({
        date
        for msg in messages
        for date in [_message_date(str(msg.get("create_time") or ""))]
        if date
    })
    if dates:
        time_range = f"{dates[0]},{dates[-1]}"
        offset = 0
        seen = 0
        while missing and seen < NAME_LOOKUP_MAX_ITEMS:
            try:
                data = await chatlog_client.get_messages(
                    talker,
                    time=time_range,
                    limit=NAME_LOOKUP_PAGE_SIZE,
                    offset=offset,
                )
            except Exception as e:
                print(f"Failed to fetch sender names talker={talker}: {e}")
                break
            items = data.get("items", []) or []
            if not items:
                break
            for item in items:
                sender = str(item.get("sender") or item.get("Sender") or "").strip()
                if sender not in missing:
                    continue
                name = _sender_display_name(item, sender)
                if name:
                    names[sender] = name
                    missing.discard(sender)
            seen += len(items)
            offset += len(items)
            total = int(data.get("total") or 0)
            if total and offset >= total:
                break
            if len(items) < NAME_LOOKUP_PAGE_SIZE:
                break

    if missing:
        try:
            members = await chatlog_client.get_chatroom_members(talker)
            raw_members = members.get("members", []) if isinstance(members, dict) else []
            for member in raw_members:
                sender = str(member.get("userName") or member.get("UserName") or "").strip()
                if sender not in missing:
                    continue
                name = _sender_display_name(
                    {
                        "displayName": member.get("displayName") or member.get("DisplayName"),
                        "nickName": member.get("nickName") or member.get("NickName"),
                        "remark": member.get("remark") or member.get("Remark"),
                    },
                    sender,
                )
                if name:
                    names[sender] = name
                    missing.discard(sender)
        except Exception as e:
            print(f"Failed to fetch chatroom members talker={talker}: {e}")

    return names


async def _fill_sender_names(talker: str, messages: list[dict]) -> None:
    if not talker or not messages:
        return
    names = await _build_sender_name_map(talker, messages)
    if not names:
        return
    for msg in messages:
        sender = str(msg.get("sender") or "").strip()
        if not sender or sender not in names:
            continue
        current = str(msg.get("sender_name") or "").strip()
        if not current or current == sender or _looks_like_raw_sender_id(current):
            msg["sender_name"] = names[sender]
            msg["senderName"] = names[sender]

@router.get("")
async def list_topics(
    group_id: Optional[int] = None,
    status: Optional[str] = None,
    keyword: Optional[str] = None,
    db: aiosqlite.Connection = Depends(get_db)
):
    sql = "SELECT * FROM topics WHERE 1=1"
    params = []
    if group_id:
        sql += " AND group_id=?"; params.append(group_id)
    if status:
        sql += " AND status=?"; params.append(status)
    if keyword:
        sql += " AND title LIKE ?"; params.append(f"%{keyword}%")
    async with db.execute(sql, params) as cur:
        return [dict(r) for r in await cur.fetchall()]

@router.post("")
async def create_topic(body: TopicCreate, db: aiosqlite.Connection = Depends(get_db)):
    await db.execute(
        "INSERT INTO topics (group_id, title, source) VALUES (?, ?, 'manual')",
        (body.group_id, body.title),
    )
    await db.commit()
    async with db.execute("SELECT * FROM topics ORDER BY id DESC LIMIT 1") as cur:
        return dict(await cur.fetchone())

@router.get("/{topic_id}")
async def get_topic(topic_id: int, db: aiosqlite.Connection = Depends(get_db)):
    async with db.execute("SELECT * FROM topics WHERE id=?", (topic_id,)) as cur:
        row = await cur.fetchone()
    if not row:
        raise HTTPException(404, "not found")
    
    # 拿到该话题下关联的所有 seq
    async with db.execute("SELECT * FROM topic_messages WHERE topic_id=? ORDER BY msg_seq ASC", (topic_id,)) as cur:
        msg_rows = await cur.fetchall()
        
    msgs = []
    if msg_rows:
        # 获取群聊 ID
        talker = msg_rows[0]["talker"]
        seq_list = [r["msg_seq"] for r in msg_rows]
        
        # 分批调用 5030 接口获取真实消息内容，避免大批量 batch 触发 500。
        from services.chatlog_client import chatlog_client
        fetched_by_seq: dict[int, dict] = {}
        for i in range(0, len(seq_list), CHATLOG_BATCH_SIZE):
            chunk = seq_list[i: i + CHATLOG_BATCH_SIZE]
            try:
                msgs_data = await chatlog_client.get_messages_batch(talker, chunk)
                raw_items = msgs_data.get("items", []) or msgs_data.get("Items", [])
                for item in raw_items:
                    normalized = _normalize_chatlog_message(item)
                    seq = normalized.get("seq")
                    if seq:
                        fetched_by_seq[int(seq)] = normalized
            except Exception as e:
                print(f"Failed to fetch real messages chunk topic={topic_id}: {e}")

        for r in msg_rows:
            seq = int(r["msg_seq"])
            if seq in fetched_by_seq:
                msgs.append(fetched_by_seq[seq])
                continue
            snap = _message_from_snapshot(r["message_json"] if "message_json" in r.keys() else None, seq)
            if snap:
                msgs.append(snap)
                continue
            msgs.append({
                "seq": seq,
                "sender": "",
                "sender_name": "系统提示",
                "create_time": "",
                "content": f"原始消息无法从 chatlog 找回 (seq: {seq})",
                "type": 1,
                "sub_type": 0,
            })
            
    # 获取知识文档
    if msg_rows:
        await _fill_sender_names(talker, msgs)

    async with db.execute("SELECT id, topic_id, content, created_at, updated_at FROM knowledge_docs WHERE topic_id=?", (topic_id,)) as cur:
        doc = await cur.fetchone()

    return {**dict(row), "messages": msgs, "knowledge_doc": dict(doc) if doc else None}

@router.patch("/{topic_id}")
async def patch_topic(topic_id: int, body: TopicPatch, db: aiosqlite.Connection = Depends(get_db)):
    if body.title:
        await db.execute(
            "UPDATE topics SET title=?, source='manual', updated_at=CURRENT_TIMESTAMP WHERE id=?",
            (body.title, topic_id),
        )
    if body.status:
        await db.execute(
            "UPDATE topics SET status=?, source='manual', updated_at=CURRENT_TIMESTAMP WHERE id=?",
            (body.status, topic_id),
        )
    await db.commit()
    async with db.execute("SELECT * FROM topics WHERE id=?", (topic_id,)) as cur:
        return dict(await cur.fetchone())

@router.delete("/{topic_id}")
async def delete_topic(topic_id: int, db: aiosqlite.Connection = Depends(get_db)):
    # 先拿到 doc_id，用于清 FTS
    async with db.execute("SELECT id FROM knowledge_docs WHERE topic_id=?", (topic_id,)) as cur:
        doc_row = await cur.fetchone()
    if doc_row:
        await db.execute("DELETE FROM knowledge_fts WHERE doc_id=?", (doc_row["id"],))
    await db.execute("DELETE FROM topic_messages WHERE topic_id=?", (topic_id,))
    await db.execute("DELETE FROM knowledge_docs WHERE topic_id=?", (topic_id,))
    await db.execute("DELETE FROM topics WHERE id=?", (topic_id,))
    await db.commit()
    return {"ok": True}

@router.post("/{topic_id}/messages")
async def add_message(topic_id: int, body: MessageAdd, db: aiosqlite.Connection = Depends(get_db)):
    await db.execute(
        "INSERT OR IGNORE INTO topic_messages (topic_id, msg_seq, talker, added_by) VALUES (?, ?, ?, 'user')",
        (topic_id, body.msg_seq, body.talker)
    )
    await db.execute(
        "UPDATE topics SET source='manual', updated_at=CURRENT_TIMESTAMP WHERE id=?",
        (topic_id,),
    )
    await db.commit()
    return {"ok": True}

@router.delete("/{topic_id}/messages/{seq}")
async def remove_message(topic_id: int, seq: int, db: aiosqlite.Connection = Depends(get_db)):
    await db.execute("DELETE FROM topic_messages WHERE topic_id=? AND msg_seq=?", (topic_id, seq))
    await db.execute(
        "UPDATE topics SET source='manual', updated_at=CURRENT_TIMESTAMP WHERE id=?",
        (topic_id,),
    )
    await db.commit()
    return {"ok": True}

@router.post("/{topic_id}/summarize")
async def summarize(topic_id: int, db: aiosqlite.Connection = Depends(get_db)):
    async with db.execute("SELECT * FROM topics WHERE id=?", (topic_id,)) as cur:
        topic = await cur.fetchone()
    if not topic:
        raise HTTPException(404, "not found")
    topic_data = dict(topic)
    await _mark_stale_summarize_tasks(db, topic_data["group_id"], topic_id)
    # 创建 ai_tasks 记录以追踪进度
    await db.execute(
        "INSERT INTO ai_tasks (group_id, type, status, progress) VALUES (?, 'summarize', 'running', ?)",
        (topic_data["group_id"], json.dumps({"processed": 0, "total": 1}))
    )
    await db.commit()
    async with db.execute("SELECT last_insert_rowid() AS id") as cur:
        task_row = await cur.fetchone()
    task_id = task_row["id"]
    from services.summary_engine import run_summarize
    import asyncio
    asyncio.create_task(run_summarize(topic_id, topic_data, task_id))
    return {"ok": True, "task_id": task_id}