backend/app/services/feishu_service.py

from __future__ import annotations

import json
import logging
import re
import time
from typing import Any, Dict, Optional, Tuple

import requests

from app.services.feishu_config_manager import FeishuConfigManager

logger = logging.getLogger(__name__)

DEFAULT_TIMEOUT = 30  # seconds
_POLL_INTERVAL = 1.5  # seconds between import-task polls
_POLL_MAX_ATTEMPTS = 40  # ~60s 上限，导入大文档也够用

# import_tasks 的 job_status：0=成功，1=初始化中，2=处理中，其余为错误码。
_JOB_STATUS_SUCCESS = 0
_JOB_STATUS_IN_PROGRESS = {1, 2}


class FeishuError(Exception):
    """飞书推送相关错误，message 直接面向用户（前端会原样展示）。"""

    def __init__(self, message: str):
        super().__init__(message)
        self.message = message


class FeishuService:
    """飞书（Lark）开放平台客户端：把 Markdown 笔记导入为飞书云文档（docx）。

    走「导入任务」接口（drive/v1/import_tasks），由飞书原生把 Markdown 转成 docx，
    标题 / 列表 / 代码块 / 表格等格式保真度最好，比手动拼 block 更可靠。

    流程（官方 import-user-guide）：
      1. medias/upload_all 上传 .md 源文件 → file_token
      2. import_tasks 创建导入任务（type=docx, file_extension=md）→ ticket
      3. 轮询 import_tasks/{ticket} 直到 job_status=0 → 拿到新文档 url / token

    鉴权用自建应用的 tenant_access_token（应用身份）。注意：以应用身份创建的文档
    归应用所有，普通用户要能看到，需要把目标文件夹的协作者加上该应用——这点在前端
    配置页有说明。
    """

    # tenant_access_token 进程内缓存：key=(base_url, app_id) -> (token, expire_ts)
    _token_cache: Dict[Tuple[str, str], Tuple[str, float]] = {}
    # 应用根目录 token 缓存：key=(base_url, app_id) -> folder_token
    _root_folder_cache: Dict[Tuple[str, str], str] = {}

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.cfg = config or FeishuConfigManager().get_config()
        self.base_url = (self.cfg.get("base_url") or "https://open.feishu.cn").rstrip("/")
        self.app_id = (self.cfg.get("app_id") or "").strip()
        self.app_secret = (self.cfg.get("app_secret") or "").strip()
        self.folder_token = (self.cfg.get("folder_token") or "").strip()
        # 知识库节点 token：填了就把笔记导入到该 wiki 节点下（先导云空间再移动挂载）
        self.wiki_token = (self.cfg.get("wiki_token") or "").strip()

    @property
    def _cache_key(self) -> Tuple[str, str]:
        return (self.base_url, self.app_id)

    # ─── 鉴权 ────────────────────────────────────────────────────────────────
    def _get_tenant_access_token(self) -> str:
        if not self.app_id or not self.app_secret:
            raise FeishuError(
                "飞书未配置：请到「设置 → 飞书推送」填写 App ID 与 App Secret"
            )

        cached = FeishuService._token_cache.get(self._cache_key)
        if cached and cached[1] > time.time():
            return cached[0]

        url = f"{self.base_url}/open-apis/auth/v3/tenant_access_token/internal"
        try:
            resp = requests.post(
                url,
                json={"app_id": self.app_id, "app_secret": self.app_secret},
                timeout=DEFAULT_TIMEOUT,
            )
            data = resp.json()
        except Exception as exc:
            raise FeishuError(f"连接飞书失败：{exc}") from exc

        if data.get("code") != 0:
            raise FeishuError(
                f"飞书鉴权失败（code={data.get('code')}）：{data.get('msg')}。"
                "请检查 App ID / App Secret 是否正确、应用是否启用"
            )

        # tenant_access_token / expire 在响应顶层（这个老接口不包在 data 里）
        token = data.get("tenant_access_token")
        if not token:
            raise FeishuError(f"飞书鉴权异常：响应缺少 tenant_access_token（{data}）")
        expire_in = int(data.get("expire", 7200))
        # 提前 5 分钟过期，避开临界点
        FeishuService._token_cache[self._cache_key] = (token, time.time() + expire_in - 300)
        return token

    def _auth_headers(self) -> Dict[str, str]:
        return {"Authorization": f"Bearer {self._get_tenant_access_token()}"}

    @staticmethod
    def _fmt_api_error(prefix: str, payload: Dict[str, Any]) -> str:
        """统一格式化飞书业务错误，带上 code/msg 方便用户/开发定位。"""
        return f"{prefix}（code={payload.get('code')}）：{payload.get('msg')}"

    def _root_folder_token(self) -> str:
        """未配置目标文件夹时，取应用云空间根目录 token 兜底。"""
        cached = FeishuService._root_folder_cache.get(self._cache_key)
        if cached:
            return cached
        url = f"{self.base_url}/open-apis/drive/explorer/v2/root_folder/meta"
        try:
            resp = requests.get(url, headers=self._auth_headers(), timeout=DEFAULT_TIMEOUT)
            payload = resp.json()
        except Exception as exc:
            raise FeishuError(f"获取飞书根目录失败：{exc}") from exc
        if payload.get("code") != 0:
            raise FeishuError(
                self._fmt_api_error("获取飞书根目录失败", payload)
                + "。建议在「设置 → 飞书推送」直接填写目标文件夹 token"
            )
        token = (payload.get("data") or {}).get("token", "")
        if not token:
            raise FeishuError("飞书根目录 token 为空，请在配置里指定目标文件夹 token")
        FeishuService._root_folder_cache[self._cache_key] = token
        return token

    # ─── 公有方法 ────────────────────────────────────────────────────────────
    def test_connection(self) -> Dict[str, Any]:
        """验证凭证：能成功换取 tenant_access_token 即视为连接成功。"""
        self._get_tenant_access_token()
        return {"success": True, "message": "飞书连接成功，凭证有效"}

    def push_markdown(
        self,
        title: str,
        markdown: str,
        image_base_url: Optional[str] = None,
    ) -> Dict[str, Any]:
        """把 Markdown 导入为飞书云文档。返回 {url, token, type, title}。

        :param title: 文档标题（取视频标题）
        :param markdown: 笔记 Markdown 正文
        :param image_base_url: 把正文里 /static、/uploads 等相对图片链接补成绝对地址的前缀，
                               飞书服务端导入时会按 http(s) 抓图（本机/内网地址抓不到则跳过）
        """
        if not (markdown or "").strip():
            raise FeishuError("笔记内容为空，无法推送")

        safe_title = self._safe_title(title)
        prepared = self._prepare_markdown(markdown, image_base_url)
        content = prepared.encode("utf-8")

        folder_token = self.folder_token or self._root_folder_token()

        # rpc_failed 多为飞书侧瞬时错误：整条导入流程自动重试几次。
        last_exc: Optional[FeishuError] = None
        for attempt in range(3):
            try:
                file_token = self._upload_media(safe_title, content, folder_token)
                ticket = self._create_import_task(safe_title, file_token, folder_token)
                result = self._poll_import_task(ticket)
                token = result.get("token") or ""
                doc_type = result.get("type") or "docx"
                # 配了知识库节点 → 把这篇 docx 移动/挂载到该 wiki 节点下
                if self.wiki_token:
                    moved = self._move_to_wiki(token)
                    logger.info(f"飞书导入知识库成功：{safe_title} -> {moved['url']}")
                    return {**moved, "title": safe_title}
                url = result.get("url") or self._fallback_doc_url(doc_type, token)
                logger.info(f"飞书导入成功：{safe_title} -> {url}")
                return {"url": url, "token": token, "type": doc_type, "title": safe_title}
            except FeishuError as exc:
                last_exc = exc
                if "rpc_failed" in (exc.message or "") and attempt < 2:
                    logger.warning(f"飞书导入 rpc_failed，第 {attempt + 1} 次重试…")
                    time.sleep(2)
                    continue
                raise
        if last_exc:
            raise last_exc
        raise FeishuError("飞书导入失败：未知错误")

    # ─── 导入流程内部步骤 ─────────────────────────────────────────────────────
    def _upload_media(self, title: str, content: bytes, folder_token: str) -> str:
        """步骤 1：上传 .md 源文件，拿 file_token。"""
        url = f"{self.base_url}/open-apis/drive/v1/medias/upload_all"
        file_name = f"{title}.md"
        data = {
            "file_name": file_name,
            "parent_type": "ccm_import_open",  # 导入专用素材类型
            "parent_node": folder_token,
            "size": str(len(content)),
            "extra": json.dumps({"obj_type": "docx", "file_extension": "md"}),
        }
        files = {"file": (file_name, content, "text/markdown")}
        try:
            resp = requests.post(
                url,
                headers=self._auth_headers(),
                data=data,
                files=files,
                timeout=DEFAULT_TIMEOUT,
            )
            payload = resp.json()
        except Exception as exc:
            raise FeishuError(f"上传 Markdown 到飞书失败：{exc}") from exc
        if payload.get("code") != 0:
            raise FeishuError(self._fmt_api_error("上传 Markdown 到飞书失败", payload))
        file_token = (payload.get("data") or {}).get("file_token")
        if not file_token:
            raise FeishuError(f"飞书上传异常：响应缺少 file_token（{payload}）")
        return file_token

    def _create_import_task(self, title: str, file_token: str, folder_token: str) -> str:
        """步骤 2：创建导入任务（md → docx），拿 ticket。"""
        url = f"{self.base_url}/open-apis/drive/v1/import_tasks"
        body = {
            "file_extension": "md",
            "file_token": file_token,
            "type": "docx",
            "file_name": title,
            "point": {
                "mount_type": 1,  # 1 = 挂载到云空间文件夹
                "mount_key": folder_token,
            },
        }
        try:
            resp = requests.post(
                url,
                headers={**self._auth_headers(), "Content-Type": "application/json"},
                json=body,
                timeout=DEFAULT_TIMEOUT,
            )
            payload = resp.json()
        except Exception as exc:
            raise FeishuError(f"创建飞书导入任务失败：{exc}") from exc
        if payload.get("code") != 0:
            raise FeishuError(self._fmt_api_error("创建飞书导入任务失败", payload))
        ticket = (payload.get("data") or {}).get("ticket")
        if not ticket:
            raise FeishuError(f"飞书导入异常：响应缺少 ticket（{payload}）")
        return ticket

    def _poll_import_task(self, ticket: str) -> Dict[str, Any]:
        """步骤 3：轮询导入结果，成功返回 result 字典（含 url/token）。"""
        url = f"{self.base_url}/open-apis/drive/v1/import_tasks/{ticket}"
        last_result: Dict[str, Any] = {}
        for _ in range(_POLL_MAX_ATTEMPTS):
            try:
                resp = requests.get(url, headers=self._auth_headers(), timeout=DEFAULT_TIMEOUT)
                payload = resp.json()
            except Exception as exc:
                raise FeishuError(f"查询飞书导入结果失败：{exc}") from exc
            if payload.get("code") != 0:
                raise FeishuError(self._fmt_api_error("查询飞书导入结果失败", payload))

            last_result = (payload.get("data") or {}).get("result") or {}
            job_status = last_result.get("job_status")

            if job_status == _JOB_STATUS_SUCCESS and last_result.get("token"):
                return last_result
            if job_status in _JOB_STATUS_IN_PROGRESS or job_status is None:
                time.sleep(_POLL_INTERVAL)
                continue
            # 其余 job_status 视为失败
            err = last_result.get("job_error_msg") or "未知错误"
            raise FeishuError(f"飞书导入失败（job_status={job_status}）：{err}")

        # 轮询超时：可能仍在处理，给出可理解的提示
        raise FeishuError(
            "飞书导入超时（文档可能仍在生成中）。请稍后到飞书云空间查看，"
            f"或重试推送（job_status={last_result.get('job_status')}）"
        )

    def _fallback_doc_url(self, doc_type: str, token: str) -> str:
        """飞书偶尔不回 url，按文档类型 + token 拼一个可访问地址。"""
        if not token:
            return ""
        # open.feishu.cn → 主站 feishu.cn；open.larksuite.com → larksuite.com
        host = self.base_url.replace("https://open.", "https://").replace("http://open.", "http://")
        return f"{host}/{doc_type}/{token}"

    def _site_host(self) -> str:
        return self.base_url.replace("https://open.", "https://").replace("http://open.", "http://")

    # ─── 知识库（Wiki）导入：把已生成的 docx 移动/挂载到 wiki 节点下 ────────────
    def _wiki_space_id(self, node_token: str) -> str:
        """由 wiki 节点 token 解析其所属知识库 space_id。"""
        url = f"{self.base_url}/open-apis/wiki/v2/spaces/get_node"
        try:
            resp = requests.get(
                url, headers=self._auth_headers(), params={"token": node_token}, timeout=DEFAULT_TIMEOUT
            )
            payload = resp.json()
        except Exception as exc:
            raise FeishuError(f"解析知识库节点失败：{exc}") from exc
        if payload.get("code") != 0:
            raise FeishuError(
                self._fmt_api_error("解析知识库节点失败", payload)
                + "。请确认「知识库节点」token 正确，且应用已加入该知识库并有编辑权限"
            )
        node = (payload.get("data") or {}).get("node") or {}
        space_id = node.get("space_id")
        if not space_id:
            raise FeishuError(f"知识库节点缺少 space_id（{payload}）")
        return space_id

    def _move_to_wiki(self, docx_token: str) -> Dict[str, Any]:
        """把 docx 移动/挂载到配置的 wiki 节点下，返回 {url, token, type}。"""
        space_id = self._wiki_space_id(self.wiki_token)
        url = f"{self.base_url}/open-apis/wiki/v2/spaces/{space_id}/nodes/move_docs_to_wiki"
        body = {
            "parent_wiki_token": self.wiki_token,
            "obj_type": "docx",
            "obj_token": docx_token,
            "apply": True,
        }
        try:
            resp = requests.post(
                url,
                headers={**self._auth_headers(), "Content-Type": "application/json"},
                json=body,
                timeout=DEFAULT_TIMEOUT,
            )
            payload = resp.json()
        except Exception as exc:
            raise FeishuError(f"移动文档到知识库失败：{exc}") from exc
        if payload.get("code") != 0:
            raise FeishuError(
                self._fmt_api_error("移动文档到知识库失败", payload)
                + "。请确认应用已开通 wiki 权限、已加入该知识库并有编辑权限"
            )
        data = payload.get("data") or {}
        wiki_token = data.get("wiki_token")
        if not wiki_token and data.get("task_id"):
            wiki_token = self._poll_wiki_move(data["task_id"])
        if not wiki_token:
            raise FeishuError(f"知识库导入异常：未拿到 wiki_token（{payload}）")
        return {"url": f"{self._site_host()}/wiki/{wiki_token}", "token": wiki_token, "type": "wiki"}

    def _poll_wiki_move(self, task_id: str) -> str:
        """move_docs_to_wiki 异步时轮询任务，返回新 wiki_token。"""
        url = f"{self.base_url}/open-apis/wiki/v2/tasks/{task_id}"
        for _ in range(_POLL_MAX_ATTEMPTS):
            try:
                resp = requests.get(
                    url, headers=self._auth_headers(), params={"task_type": "move"}, timeout=DEFAULT_TIMEOUT
                )
                payload = resp.json()
            except Exception as exc:
                raise FeishuError(f"查询知识库导入结果失败：{exc}") from exc
            if payload.get("code") != 0:
                raise FeishuError(self._fmt_api_error("查询知识库导入结果失败", payload))
            task = (payload.get("data") or {}).get("task") or {}
            for item in task.get("move_result") or []:
                wt = (item.get("node") or {}).get("wiki_token")
                if wt:
                    return wt
            time.sleep(_POLL_INTERVAL)
        raise FeishuError("知识库导入超时，请稍后到知识库查看")

    # ─── Markdown 预处理 ──────────────────────────────────────────────────────
    @staticmethod
    def _safe_title(title: str, fallback: str = "VideoMemo 笔记") -> str:
        """清掉标题里的换行/控制字符，截到 120 字（飞书文档标题有长度限制）。"""
        cleaned = re.sub(r"[\r\n\t]", " ", (title or "").strip())
        cleaned = cleaned.strip()
        return (cleaned[:120] or fallback)

    @staticmethod
    def _prepare_markdown(markdown: str, image_base_url: Optional[str]) -> str:
        """处理站内相对图片（/static、/uploads）：

        - 若 image_base_url 是飞书服务端能抓到的「公网 http(s)」地址 → 补成绝对地址，图片入文档。
        - 若是本机/内网（localhost / 127.* / 内网段）或为空 → 飞书根本抓不到，直接**删掉**这些图片，
          只保留正文。避免飞书导入时为抓一张抓不到的图而整单失败（rpc_failed）。
        """
        base = (image_base_url or "").rstrip("/")
        public = bool(re.match(r"^https?://", base)) and not re.search(
            r"://(localhost|127\.|0\.0\.0\.0|10\.|192\.168\.|172\.(1[6-9]|2\d|3[01])\.|\[?::1)",
            base,
        )

        def _repl(m: "re.Match[str]") -> str:
            alt, path = m.group(1), m.group(2)
            return f"![{alt}]({base}{path})" if public else ""

        # 仅处理 ](/static...) 与 ](/uploads...) 这类站内相对图片
        return re.sub(r"!\[([^\]]*)\]\((/(?:static|uploads)/[^)]+)\)", _repl, markdown)