Immortal Skill

generated/
corpus/
__pycache__/
*.pyc
.DS_Store
.snapshots/
feishu.json
dingtalk.json
*.env

"""immortal-skill 多平台数据采集器。

统一接口：所有采集器继承 BaseCollector，输出标准 Message 格式，
最终转为 intake-protocol 兼容的 Markdown corpus。

按数据获取方式分三类：
  API:      飞书、钉钉、Slack、Discord、Telegram、Email (Gmail)
  Local DB: 微信 (SQLite)、iMessage (chat.db)
  Archive:  WhatsApp (txt)、Twitter/X (archive zip)、社交媒体 (Takeout)、手动导入
"""

from .base import BaseCollector, Message, Channel  # noqa: F401

"""采集器基类与统一消息格式。

所有平台采集器继承 BaseCollector，实现 authenticate / scan / collect 三个方法，
并使用 to_corpus() 将消息列表转为 intake-protocol 兼容的 Markdown。
"""

from __future__ import annotations

import json
from abc import ABC, abstractmethod
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional


@dataclass
class Message:
    """统一消息格式——各采集器输出此结构。"""
    sender: str
    text: str
    timestamp: Optional[datetime] = None
    channel_name: str = ""
    msg_type: str = "text"
    platform: str = ""
    raw: dict = field(default_factory=dict)

    @property
    def date_str(self) -> str:
        if self.timestamp:
            return self.timestamp.strftime("%Y-%m-%d %H:%M")
        return ""

    def to_dict(self) -> dict:
        d = asdict(self)
        if self.timestamp:
            d["timestamp"] = self.timestamp.isoformat()
        return d


@dataclass
class Channel:
    """统一频道/会话格式。"""
    channel_id: str
    name: str
    platform: str
    member_count: int = 0
    extra: dict = field(default_factory=dict)


class BaseCollector(ABC):
    """采集器基类。"""

    platform_name: str = "unknown"
    platform_name_zh: str = "未知平台"

    def __init__(self, config: Optional[dict] = None):
        self.config = config or {}
        self._authenticated = False

    @abstractmethod
    def authenticate(self) -> bool:
        """验证凭证，成功返回 True。"""
        ...

    @abstractmethod
    def scan(self, keyword: str = "") -> list[Channel]:
        """扫描可用的频道/会话列表。"""
        ...

    @abstractmethod
    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        """采集指定频道的消息。"""
        ...

    def to_corpus(
        self,
        messages: list[Message],
        channel_name: str = "",
        target: str = "",
    ) -> str:
        """将消息列表转为 intake-protocol 兼容的 Markdown 格式。"""
        meta_channel = f"{self.platform_name}_api"
        header = (
            f"<!-- channel: {meta_channel} | confidence: verbatim | scope: both -->\n"
            f"# {self.platform_name_zh}消息采集"
        )
        if channel_name:
            header += f"：{channel_name}"
        if target:
            header += f"（目标：{target}）"
        header += f"\n\n共 {len(messages)} 条消息\n\n---\n"

        lines = [header]
        for msg in messages:
            prefix = f"[{msg.date_str}]" if msg.date_str else "[]"
            sender_tag = f" **{msg.sender}**:" if msg.sender else ""
            lines.append(f"{prefix}{sender_tag} {msg.text}")

        return "\n".join(lines) + "\n"

    def save_corpus(
        self,
        messages: list[Message],
        output: Path,
        channel_name: str = "",
        target: str = "",
    ) -> Path:
        """将 corpus 写入文件。"""
        output = Path(output)
        output.parent.mkdir(parents=True, exist_ok=True)
        content = self.to_corpus(messages, channel_name, target)
        output.write_text(content, encoding="utf-8")
        return output

    def save_json(self, messages: list[Message], output: Path) -> Path:
        """将消息列表保存为 JSON（便于后续处理）。"""
        output = Path(output)
        output.parent.mkdir(parents=True, exist_ok=True)
        data = [msg.to_dict() for msg in messages]
        output.write_text(
            json.dumps(data, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )
        return output

"""钉钉数据采集器。

通过钉钉开放 API 采集群聊消息。
需要钉钉企业内部应用的 AppKey / AppSecret。

认证方式：
  - 环境变量 DINGTALK_APP_KEY / DINGTALK_APP_SECRET
  - 配置文件 ~/.immortal-skill/dingtalk.json

API 文档：https://open.dingtalk.com/
"""

from __future__ import annotations

import json
import os
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

from .base import BaseCollector, Message, Channel

try:
    import requests
except ImportError:
    requests = None  # type: ignore[assignment]

CONFIG_DIR = Path.home() / ".immortal-skill"
CONFIG_FILE = CONFIG_DIR / "dingtalk.json"
BASE_URL = "https://oapi.dingtalk.com"
API_NEW = "https://api.dingtalk.com"


class DingtalkCollector(BaseCollector):
    platform_name = "dingtalk"
    platform_name_zh = "钉钉"

    def __init__(self, config: Optional[dict] = None):
        super().__init__(config)
        self._token_cache: dict = {}

    def _require_requests(self) -> None:
        if requests is None:
            raise ImportError("需要 requests 库：pip3 install requests")

    def _load_credentials(self) -> tuple[str, str]:
        key = os.environ.get("DINGTALK_APP_KEY", "") or self.config.get("app_key", "")
        secret = os.environ.get("DINGTALK_APP_SECRET", "") or self.config.get("app_secret", "")
        if key and secret:
            return key, secret
        if CONFIG_FILE.is_file():
            cfg = json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
            return cfg.get("app_key", ""), cfg.get("app_secret", "")
        return "", ""

    def _get_access_token(self) -> str:
        if (
            self._token_cache.get("token")
            and self._token_cache.get("expires_at", 0) > time.time()
        ):
            return self._token_cache["token"]

        key, secret = self._load_credentials()
        if not key or not secret:
            raise RuntimeError("未配置钉钉凭证。请设置环境变量或运行 immortal_cli.py setup dingtalk")

        resp = requests.post(
            f"{API_NEW}/v1.0/oauth2/accessToken",
            json={"appKey": key, "appSecret": secret},
            timeout=10,
        )
        data = resp.json()
        token = data.get("accessToken")
        if not token:
            raise RuntimeError(f"获取钉钉 token 失败：{data}")

        self._token_cache["token"] = token
        self._token_cache["expires_at"] = time.time() + data.get("expireIn", 7200) - 300
        return token

    def _headers(self) -> dict:
        return {"x-acs-dingtalk-access-token": self._get_access_token()}

    def authenticate(self) -> bool:
        self._require_requests()
        try:
            self._get_access_token()
            self._authenticated = True
            return True
        except RuntimeError:
            return False

    def scan(self, keyword: str = "") -> list[Channel]:
        self._require_requests()
        channels: list[Channel] = []
        page_token = ""

        while True:
            params: dict = {"maxResults": 100}
            if page_token:
                params["nextToken"] = page_token

            resp = requests.get(
                f"{API_NEW}/v1.0/im/privilegedAccess/groupInfos",
                headers=self._headers(), params=params, timeout=15,
            )
            data = resp.json()
            if not data.get("success", True):
                break

            for group in data.get("result", {}).get("records", []):
                name = group.get("title", "(未命名)")
                if keyword and keyword.lower() not in name.lower():
                    continue
                channels.append(Channel(
                    channel_id=group.get("openConversationId", ""),
                    name=name,
                    platform="dingtalk",
                    member_count=group.get("memberCount", 0),
                ))

            page_token = data.get("result", {}).get("nextToken", "")
            if not page_token:
                break

        return channels

    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        self._require_requests()
        messages: list[Message] = []
        next_token: Optional[str] = None

        while len(messages) < limit:
            body: dict = {
                "openConversationId": channel_id,
                "maxResults": min(50, limit - len(messages)),
            }
            if next_token:
                body["nextToken"] = next_token

            resp = requests.post(
                f"{API_NEW}/v1.0/im/privilegedAccess/messages",
                headers=self._headers(), json=body, timeout=15,
            )
            data = resp.json()
            if not data.get("success", True):
                break

            for msg in data.get("result", {}).get("records", []):
                text = msg.get("text", "")
                if not text:
                    continue

                ts = self._parse_timestamp(msg.get("createAt"))
                messages.append(Message(
                    sender=msg.get("senderNick", msg.get("senderId", "")),
                    text=text,
                    timestamp=ts,
                    channel_name=channel_id,
                    platform="dingtalk",
                ))

            next_token = data.get("result", {}).get("nextToken")
            if not next_token:
                break

        return messages

    @staticmethod
    def _parse_timestamp(ts_ms: Optional[int]) -> Optional[datetime]:
        if not ts_ms:
            return None
        try:
            return datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
        except (ValueError, OSError, TypeError):
            return None

    @classmethod
    def setup_interactive(cls) -> None:
        """交互式配置钉钉凭证。"""
        print("钉钉凭证配置")
        print("需要钉钉企业内部应用。")
        print("开放平台：https://open.dingtalk.com/\n")

        app_key = input("AppKey: ").strip()
        app_secret = input("AppSecret: ").strip()
        if not app_key or not app_secret:
            raise ValueError("AppKey 和 AppSecret 不能为空")

        collector = cls({"app_key": app_key, "app_secret": app_secret})
        if not collector.authenticate():
            raise RuntimeError("凭证验证失败")

        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        CONFIG_FILE.write_text(
            json.dumps({"app_key": app_key, "app_secret": app_secret}, indent=2) + "\n",
            encoding="utf-8",
        )
        print(f"\n凭证已保存到 {CONFIG_FILE}")

"""Discord 采集器。

通过 Discord Bot API 采集频道消息。
需要 Discord Bot Token。

依赖：pip3 install requests

认证方式：
  - 环境变量 DISCORD_TOKEN
  - 配置 dict 中的 token 字段

注意：Discord ToS 禁止用户 token 自动化，推荐使用 Bot token。
"""

from __future__ import annotations

import os
from datetime import datetime, timezone
from typing import Optional

from .base import BaseCollector, Message, Channel

try:
    import requests
except ImportError:
    requests = None  # type: ignore[assignment]

API_BASE = "https://discord.com/api/v10"


class DiscordCollector(BaseCollector):
    platform_name = "discord"
    platform_name_zh = "Discord"

    def __init__(self, config: Optional[dict] = None):
        super().__init__(config)
        self._token = ""

    def _get_token(self) -> str:
        if self._token:
            return self._token
        self._token = (
            os.environ.get("DISCORD_TOKEN", "")
            or (self.config or {}).get("token", "")
        )
        return self._token

    def _headers(self) -> dict:
        return {"Authorization": f"Bot {self._get_token()}"}

    def authenticate(self) -> bool:
        if requests is None:
            return False
        token = self._get_token()
        if not token:
            return False
        resp = requests.get(f"{API_BASE}/users/@me", headers=self._headers(), timeout=10)
        ok = resp.status_code == 200
        self._authenticated = ok
        return ok

    def scan(self, keyword: str = "") -> list[Channel]:
        if requests is None:
            return []
        channels: list[Channel] = []
        guilds_resp = requests.get(
            f"{API_BASE}/users/@me/guilds", headers=self._headers(), timeout=10,
        )
        if guilds_resp.status_code != 200:
            return channels

        for guild in guilds_resp.json():
            guild_id = guild["id"]
            ch_resp = requests.get(
                f"{API_BASE}/guilds/{guild_id}/channels",
                headers=self._headers(), timeout=10,
            )
            if ch_resp.status_code != 200:
                continue
            for ch in ch_resp.json():
                if ch.get("type") not in (0, 5):
                    continue
                name = ch.get("name", "")
                if keyword and keyword.lower() not in name.lower():
                    continue
                channels.append(Channel(
                    channel_id=ch["id"],
                    name=f"{guild.get('name', '')}#{name}",
                    platform="discord",
                    extra={"guild_id": guild_id},
                ))

        return channels

    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        if requests is None:
            return []
        messages: list[Message] = []
        before: Optional[str] = None

        while len(messages) < limit:
            params: dict = {"limit": min(100, limit - len(messages))}
            if before:
                params["before"] = before

            resp = requests.get(
                f"{API_BASE}/channels/{channel_id}/messages",
                headers=self._headers(), params=params, timeout=15,
            )
            if resp.status_code != 200:
                break
            items = resp.json()
            if not items:
                break

            for msg in items:
                text = msg.get("content", "")
                if not text:
                    continue
                ts = self._parse_iso(msg.get("timestamp"))
                if since and ts and ts < since:
                    continue
                if until and ts and ts > until:
                    continue
                author = msg.get("author", {})
                sender = author.get("global_name") or author.get("username", "")
                messages.append(Message(
                    sender=sender,
                    text=text,
                    timestamp=ts,
                    channel_name=channel_id,
                    platform="discord",
                ))

            before = items[-1]["id"]
            if len(items) < 100:
                break

        return messages

    @staticmethod
    def _parse_iso(ts_str: Optional[str]) -> Optional[datetime]:
        if not ts_str:
            return None
        try:
            return datetime.fromisoformat(ts_str.replace("+00:00", "+00:00"))
        except ValueError:
            return None

"""Email 采集器。

支持两种数据源：
  1. mbox 文件（Gmail Takeout / Thunderbird 导出）
  2. Gmail API（OAuth2 认证）

对于大多数用户，推荐先通过 Google Takeout (https://takeout.google.com/) 导出
Gmail 的 mbox 文件，然后用 parse_mbox() 方法解析。
"""

from __future__ import annotations

import email
import email.header
import email.policy
import mailbox
import re
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from typing import Optional

from .base import BaseCollector, Message, Channel


class EmailCollector(BaseCollector):
    platform_name = "email"
    platform_name_zh = "邮件"

    def __init__(self, config: Optional[dict] = None):
        super().__init__(config)
        self._mbox_path: Optional[Path] = None
        if config and config.get("mbox_path"):
            self._mbox_path = Path(config["mbox_path"])

    def authenticate(self) -> bool:
        if self._mbox_path and self._mbox_path.is_file():
            self._authenticated = True
            return True
        return False

    def scan(self, keyword: str = "") -> list[Channel]:
        """扫描 mbox 文件中的通信对象列表。"""
        if not self._mbox_path:
            return []

        senders: dict[str, int] = {}
        try:
            mbox = mailbox.mbox(str(self._mbox_path))
            for msg in mbox:
                from_addr = self._decode_header(msg.get("From", ""))
                addr = self._extract_email(from_addr)
                if addr:
                    if keyword and keyword.lower() not in addr.lower():
                        continue
                    senders[addr] = senders.get(addr, 0) + 1
        except Exception:
            pass

        return [
            Channel(
                channel_id=addr,
                name=addr,
                platform="email",
                member_count=count,
            )
            for addr, count in sorted(senders.items(), key=lambda x: -x[1])
        ]

    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        """采集与指定邮箱地址的往来邮件。"""
        if not self._mbox_path:
            return []

        messages: list[Message] = []
        try:
            mbox = mailbox.mbox(str(self._mbox_path))
            for msg in mbox:
                from_addr = self._extract_email(self._decode_header(msg.get("From", "")))
                to_addr = self._extract_email(self._decode_header(msg.get("To", "")))

                if channel_id not in (from_addr, to_addr):
                    continue

                body = self._get_body(msg)
                if not body.strip():
                    continue

                ts = self._parse_date(msg.get("Date"))
                if since and ts and ts < since:
                    continue
                if until and ts and ts > until:
                    continue

                subject = self._decode_header(msg.get("Subject", ""))
                text = f"**{subject}**\n\n{body}" if subject else body

                messages.append(Message(
                    sender=from_addr or "",
                    text=text,
                    timestamp=ts,
                    channel_name=channel_id,
                    platform="email",
                ))

                if len(messages) >= limit:
                    break
        except Exception:
            pass

        return messages

    @classmethod
    def parse_mbox(cls, mbox_path: Path, target_email: str = "", limit: int = 1000) -> list[Message]:
        """直接解析 mbox 文件为 Message 列表。"""
        collector = cls({"mbox_path": str(mbox_path)})
        if target_email:
            return collector.collect(target_email, limit=limit)
        messages: list[Message] = []
        mbox = mailbox.mbox(str(mbox_path))
        for msg in mbox:
            body = cls._get_body(msg)
            if not body.strip():
                continue
            from_addr = cls._extract_email(cls._decode_header(msg.get("From", "")))
            subject = cls._decode_header(msg.get("Subject", ""))
            ts = cls._parse_date(msg.get("Date"))
            text = f"**{subject}**\n\n{body}" if subject else body
            messages.append(Message(
                sender=from_addr or "",
                text=text,
                timestamp=ts,
                platform="email",
            ))
            if len(messages) >= limit:
                break
        return messages

    @staticmethod
    def _decode_header(header_val: str) -> str:
        if not header_val:
            return ""
        decoded = email.header.decode_header(header_val)
        parts = []
        for part, charset in decoded:
            if isinstance(part, bytes):
                parts.append(part.decode(charset or "utf-8", errors="replace"))
            else:
                parts.append(part)
        return " ".join(parts)

    @staticmethod
    def _extract_email(text: str) -> str:
        m = re.search(r"[\w.+-]+@[\w-]+\.[\w.-]+", text)
        return m.group(0).lower() if m else text.strip().lower()

    @staticmethod
    def _get_body(msg: email.message.Message) -> str:
        if msg.is_multipart():
            for part in msg.walk():
                ct = part.get_content_type()
                if ct == "text/plain":
                    payload = part.get_payload(decode=True)
                    if payload:
                        charset = part.get_content_charset() or "utf-8"
                        return payload.decode(charset, errors="replace")
            for part in msg.walk():
                ct = part.get_content_type()
                if ct == "text/html":
                    payload = part.get_payload(decode=True)
                    if payload:
                        charset = part.get_content_charset() or "utf-8"
                        html = payload.decode(charset, errors="replace")
                        return re.sub(r"<[^>]+>", "", html)
        else:
            payload = msg.get_payload(decode=True)
            if payload:
                charset = msg.get_content_charset() or "utf-8"
                return payload.decode(charset, errors="replace")
        return ""

    @staticmethod
    def _parse_date(date_str: Optional[str]) -> Optional[datetime]:
        if not date_str:
            return None
        try:
            return parsedate_to_datetime(date_str)
        except Exception:
            return None

"""飞书数据采集器。

通过飞书开放 API 采集群聊消息与文档内容。
需要飞书自建应用（已开启机器人能力）的 App ID / App Secret。

认证方式：
  - 环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET
  - 配置文件 ~/.immortal-skill/feishu.json
"""

from __future__ import annotations

import json
import os
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

from .base import BaseCollector, Message, Channel

try:
    import requests
except ImportError:
    requests = None  # type: ignore[assignment]

CONFIG_DIR = Path.home() / ".immortal-skill"
CONFIG_FILE = CONFIG_DIR / "feishu.json"
BASE_URL = "https://open.feishu.cn/open-apis"


class FeishuCollector(BaseCollector):
    platform_name = "feishu"
    platform_name_zh = "飞书"

    def __init__(self, config: Optional[dict] = None):
        super().__init__(config)
        self._token_cache: dict = {}

    def _require_requests(self) -> None:
        if requests is None:
            raise ImportError("需要 requests 库：pip3 install requests")

    def _load_credentials(self) -> tuple[str, str]:
        app_id = os.environ.get("FEISHU_APP_ID", "") or self.config.get("app_id", "")
        app_secret = os.environ.get("FEISHU_APP_SECRET", "") or self.config.get("app_secret", "")
        if app_id and app_secret:
            return app_id, app_secret
        if CONFIG_FILE.is_file():
            cfg = json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
            return cfg.get("app_id", ""), cfg.get("app_secret", "")
        return "", ""

    def _get_tenant_token(self) -> str:
        if (
            self._token_cache.get("token")
            and self._token_cache.get("expires_at", 0) > time.time()
        ):
            return self._token_cache["token"]

        app_id, app_secret = self._load_credentials()
        if not app_id or not app_secret:
            raise RuntimeError("未配置飞书凭证。请设置环境变量或运行 immortal_cli.py setup feishu")

        resp = requests.post(
            f"{BASE_URL}/auth/v3/tenant_access_token/internal",
            json={"app_id": app_id, "app_secret": app_secret},
            timeout=10,
        )
        data = resp.json()
        if data.get("code") != 0:
            raise RuntimeError(f"获取飞书 token 失败：{data.get('msg', resp.text)}")

        token = data["tenant_access_token"]
        self._token_cache["token"] = token
        self._token_cache["expires_at"] = time.time() + data.get("expire", 7200) - 300
        return token

    def _headers(self) -> dict:
        return {"Authorization": f"Bearer {self._get_tenant_token()}"}

    def authenticate(self) -> bool:
        self._require_requests()
        try:
            self._get_tenant_token()
            self._authenticated = True
            return True
        except RuntimeError:
            return False

    def scan(self, keyword: str = "") -> list[Channel]:
        self._require_requests()
        channels: list[Channel] = []
        page_token = ""

        while True:
            params: dict = {"page_size": 100}
            if page_token:
                params["page_token"] = page_token

            resp = requests.get(
                f"{BASE_URL}/im/v1/chats",
                headers=self._headers(), params=params, timeout=15,
            )
            data = resp.json()
            if data.get("code") != 0:
                break

            for chat in data.get("data", {}).get("items", []):
                name = chat.get("name", "(未命名)")
                if keyword and keyword.lower() not in name.lower():
                    continue
                channels.append(Channel(
                    channel_id=chat.get("chat_id", ""),
                    name=name,
                    platform="feishu",
                    member_count=chat.get("user_count", 0),
                ))

            page_token = data.get("data", {}).get("page_token", "")
            if not data.get("data", {}).get("has_more", False):
                break

        return channels

    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        self._require_requests()
        messages: list[Message] = []
        page_token = ""

        while len(messages) < limit:
            params: dict = {
                "container_id_type": "chat",
                "container_id": channel_id,
                "page_size": 50,
                "sort_type": "ByCreateTimeDesc",
            }
            if page_token:
                params["page_token"] = page_token

            resp = requests.get(
                f"{BASE_URL}/im/v1/messages",
                headers=self._headers(), params=params, timeout=15,
            )
            data = resp.json()
            if data.get("code") != 0:
                break

            for item in data.get("data", {}).get("items", []):
                text = self._extract_text(item)
                if text is None:
                    continue

                ts = self._parse_timestamp(item.get("create_time", ""))
                sender = item.get("sender", {}).get("sender_id", {}).get("open_id", "")

                messages.append(Message(
                    sender=sender,
                    text=text,
                    timestamp=ts,
                    channel_name=channel_id,
                    platform="feishu",
                ))

                if len(messages) >= limit:
                    break

            page_token = data.get("data", {}).get("page_token", "")
            if not data.get("data", {}).get("has_more", False):
                break

        return messages

    def collect_document(self, doc_token: str) -> str:
        """导出飞书文档为 Markdown 文本。"""
        self._require_requests()
        blocks: list[dict] = []
        page_token = ""

        while True:
            params: dict = {"page_size": 500}
            if page_token:
                params["page_token"] = page_token

            resp = requests.get(
                f"{BASE_URL}/docx/v1/documents/{doc_token}/blocks",
                headers=self._headers(), params=params, timeout=15,
            )
            data = resp.json()
            if data.get("code") != 0:
                raise RuntimeError(f"飞书文档 API 错误：{data.get('msg', resp.text)}")

            blocks.extend(data.get("data", {}).get("items", []))
            page_token = data.get("data", {}).get("page_token", "")
            if not data.get("data", {}).get("has_more", False):
                break

        return self._blocks_to_markdown(blocks)

    @staticmethod
    def _extract_text(item: dict) -> Optional[str]:
        msg_type = item.get("msg_type", "")
        body = item.get("body", {})
        content_str = body.get("content", "")
        if not content_str:
            return None

        try:
            content = json.loads(content_str)
        except (json.JSONDecodeError, TypeError):
            return content_str if isinstance(content_str, str) else None

        if msg_type == "text":
            return content.get("text", "")
        if msg_type == "post":
            lines = []
            for lang_content in content.values():
                if isinstance(lang_content, dict):
                    title = lang_content.get("title", "")
                    if title:
                        lines.append(f"**{title}**")
                    for para in lang_content.get("content", []):
                        if isinstance(para, list):
                            parts = [
                                e.get("text", "")
                                for e in para
                                if isinstance(e, dict) and e.get("tag") == "text"
                            ]
                            if parts:
                                lines.append("".join(parts))
            return "\n".join(lines) if lines else None

        return f"[{msg_type} 消息]"

    @staticmethod
    def _parse_timestamp(ts_str: str) -> Optional[datetime]:
        if not ts_str:
            return None
        try:
            ts = int(ts_str) // 1000 if len(ts_str) > 10 else int(ts_str)
            return datetime.fromtimestamp(ts, tz=timezone.utc)
        except (ValueError, OSError):
            return None

    @staticmethod
    def _blocks_to_markdown(blocks: list[dict]) -> str:
        heading_prefix = {3: "# ", 4: "## ", 5: "### ", 9: "- ", 10: "1. "}
        lines = []
        for block in blocks:
            bt = block.get("block_type", 0)
            text_run = (
                block.get("text", {})
                or block.get("heading1", {})
                or block.get("heading2", {})
                or block.get("heading3", {})
            )
            if isinstance(text_run, dict):
                elements = text_run.get("elements", [])
                text = "".join(
                    e.get("text_run", {}).get("content", "") for e in elements
                )
                if text:
                    lines.append(heading_prefix.get(bt, "") + text)
            elif bt == 14:
                lines.append("---")
        return "\n".join(lines)

    @classmethod
    def setup_interactive(cls) -> None:
        """交互式配置飞书凭证。"""
        print("飞书凭证配置")
        print("需要一个飞书自建应用，且已开启机器人能力。")
        print("开放平台：https://open.feishu.cn/app\n")

        app_id = input("App ID: ").strip()
        app_secret = input("App Secret: ").strip()
        if not app_id or not app_secret:
            raise ValueError("App ID 和 App Secret 不能为空")

        collector = cls({"app_id": app_id, "app_secret": app_secret})
        if not collector.authenticate():
            raise RuntimeError("凭证验证失败")

        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        CONFIG_FILE.write_text(
            json.dumps({"app_id": app_id, "app_secret": app_secret}, indent=2) + "\n",
            encoding="utf-8",
        )
        print(f"\n凭证已保存到 {CONFIG_FILE}")

"""iMessage 采集器（macOS 本地 SQLite）。

直接读取 macOS 的 Messages 数据库 ~/Library/Messages/chat.db。
需要「完全磁盘访问权限」（System Settings > Privacy & Security > Full Disk Access）。

仅在 macOS 上可用。
"""

from __future__ import annotations

import sqlite3
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Optional

from .base import BaseCollector, Message, Channel

CHAT_DB = Path.home() / "Library" / "Messages" / "chat.db"
APPLE_EPOCH = datetime(2001, 1, 1, tzinfo=timezone.utc)


class IMessageCollector(BaseCollector):
    platform_name = "imessage"
    platform_name_zh = "iMessage"

    def __init__(self, config: Optional[dict] = None):
        super().__init__(config)
        self._db_path = Path(config.get("db_path", str(CHAT_DB))) if config else CHAT_DB

    def authenticate(self) -> bool:
        if self._db_path.is_file():
            try:
                conn = sqlite3.connect(str(self._db_path))
                conn.execute("SELECT COUNT(*) FROM message LIMIT 1")
                conn.close()
                self._authenticated = True
                return True
            except sqlite3.Error:
                pass
        return False

    def scan(self, keyword: str = "") -> list[Channel]:
        if not self._db_path.is_file():
            return []

        channels: list[Channel] = []
        try:
            conn = sqlite3.connect(str(self._db_path))
            query = """
                SELECT c.ROWID, c.chat_identifier, c.display_name,
                       COUNT(cm.message_id) as msg_count
                FROM chat c
                LEFT JOIN chat_message_join cm ON c.ROWID = cm.chat_id
                GROUP BY c.ROWID
                ORDER BY msg_count DESC
            """
            for row in conn.execute(query):
                name = row[2] or row[1] or str(row[0])
                if keyword and keyword.lower() not in name.lower():
                    continue
                channels.append(Channel(
                    channel_id=str(row[0]),
                    name=name,
                    platform="imessage",
                    member_count=row[3],
                ))
            conn.close()
        except sqlite3.Error:
            pass
        return channels

    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        if not self._db_path.is_file():
            return []

        messages: list[Message] = []
        try:
            conn = sqlite3.connect(str(self._db_path))
            query = """
                SELECT m.text, m.date, m.is_from_me,
                       COALESCE(h.id, '') as sender_id
                FROM message m
                JOIN chat_message_join cm ON m.ROWID = cm.message_id
                LEFT JOIN handle h ON m.handle_id = h.ROWID
                WHERE cm.chat_id = ?
                  AND m.text IS NOT NULL AND m.text != ''
                ORDER BY m.date DESC
                LIMIT ?
            """
            for row in conn.execute(query, (int(channel_id), limit)):
                text = row[0]
                if not text or not text.strip():
                    continue
                ts = self._apple_ts_to_datetime(row[1])
                if since and ts and ts < since:
                    continue
                if until and ts and ts > until:
                    continue
                sender = "我" if row[2] == 1 else (row[3] or "对方")
                messages.append(Message(
                    sender=sender,
                    text=text,
                    timestamp=ts,
                    channel_name=channel_id,
                    platform="imessage",
                ))
            conn.close()
        except sqlite3.Error:
            pass
        return messages

    @staticmethod
    def _apple_ts_to_datetime(ts: Optional[int]) -> Optional[datetime]:
        """Apple Core Data 时间戳转 datetime（纳秒自 2001-01-01）。"""
        if ts is None:
            return None
        try:
            seconds = ts / 1_000_000_000
            return APPLE_EPOCH + timedelta(seconds=seconds)
        except (ValueError, OverflowError):
            return None

"""手动导入采集器。

支持从文件或剪贴板粘贴导入数据：
  - 纯文本文件 (.txt, .md)
  - JSON 文件
  - CSV 文件
  - 剪贴板粘贴的文本

这是最通用的采集方式，适用于所有无法自动采集的数据源。
"""

from __future__ import annotations

import csv
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

from .base import BaseCollector, Message, Channel


class ManualCollector(BaseCollector):
    platform_name = "manual"
    platform_name_zh = "手动导入"

    def authenticate(self) -> bool:
        self._authenticated = True
        return True

    def scan(self, keyword: str = "") -> list[Channel]:
        return [Channel(
            channel_id="manual",
            name="手动导入",
            platform="manual",
        )]

    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        return []

    @classmethod
    def from_text(cls, text: str, source_label: str = "paste") -> list[Message]:
        """从粘贴的文本创建消息列表。按空行分段。"""
        messages: list[Message] = []
        paragraphs = re.split(r"\n\s*\n", text.strip())
        for para in paragraphs:
            para = para.strip()
            if not para:
                continue
            messages.append(Message(
                sender="",
                text=para,
                timestamp=datetime.now(timezone.utc),
                channel_name=source_label,
                platform="manual",
            ))
        return messages

    @classmethod
    def from_file(cls, file_path: Path) -> list[Message]:
        """根据文件后缀自动选择解析方式。"""
        suffix = file_path.suffix.lower()
        if suffix == ".json":
            return cls._from_json(file_path)
        if suffix == ".csv":
            return cls._from_csv(file_path)
        return cls._from_text_file(file_path)

    @classmethod
    def _from_text_file(cls, path: Path) -> list[Message]:
        """按空行分段解析纯文本/Markdown。"""
        content = path.read_text(encoding="utf-8")
        return cls.from_text(content, source_label=path.name)

    @classmethod
    def _from_json(cls, path: Path) -> list[Message]:
        """解析 JSON 文件（支持数组或带 messages 字段的对象）。"""
        data = json.loads(path.read_text(encoding="utf-8"))
        items = data if isinstance(data, list) else data.get("messages", [])
        messages: list[Message] = []
        for item in items:
            if isinstance(item, str):
                messages.append(Message(
                    sender="", text=item,
                    timestamp=datetime.now(timezone.utc),
                    platform="manual",
                ))
                continue
            if not isinstance(item, dict):
                continue
            text = item.get("text") or item.get("content") or item.get("message", "")
            if not text:
                continue
            ts = None
            ts_val = item.get("timestamp") or item.get("date") or item.get("time")
            if isinstance(ts_val, str):
                for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%d %H:%M"):
                    try:
                        ts = datetime.strptime(ts_val, fmt).replace(tzinfo=timezone.utc)
                        break
                    except ValueError:
                        continue
            messages.append(Message(
                sender=item.get("sender", item.get("from", "")),
                text=text,
                timestamp=ts,
                channel_name=path.name,
                platform="manual",
            ))
        return messages

    @classmethod
    def _from_csv(cls, path: Path) -> list[Message]:
        """解析 CSV 文件（预期列: sender, text, time）。"""
        messages: list[Message] = []
        with open(path, encoding="utf-8", newline="") as f:
            reader = csv.DictReader(f)
            for row in reader:
                text = row.get("text") or row.get("content") or row.get("message", "")
                if not text:
                    continue
                ts = None
                time_str = row.get("time") or row.get("date") or row.get("timestamp", "")
                if time_str:
                    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
                        try:
                            ts = datetime.strptime(time_str.strip(), fmt).replace(tzinfo=timezone.utc)
                            break
                        except ValueError:
                            continue
                messages.append(Message(
                    sender=row.get("sender", row.get("from", "")),
                    text=text,
                    timestamp=ts,
                    channel_name=path.name,
                    platform="manual",
                ))
        return messages

"""Slack 采集器。

通过 Slack Web API 采集频道消息。
需要 Slack Bot Token（xoxb-）或 User Token（xoxp-）。

依赖：pip3 install requests

认证方式：
  - 环境变量 SLACK_TOKEN
  - 配置 dict 中的 token 字段
"""

from __future__ import annotations

import os
from datetime import datetime, timezone
from typing import Optional

from .base import BaseCollector, Message, Channel

try:
    import requests
except ImportError:
    requests = None  # type: ignore[assignment]

API_BASE = "https://slack.com/api"


class SlackCollector(BaseCollector):
    platform_name = "slack"
    platform_name_zh = "Slack"

    def __init__(self, config: Optional[dict] = None):
        super().__init__(config)
        self._token = ""

    def _get_token(self) -> str:
        if self._token:
            return self._token
        self._token = (
            os.environ.get("SLACK_TOKEN", "")
            or (self.config or {}).get("token", "")
        )
        return self._token

    def _headers(self) -> dict:
        return {"Authorization": f"Bearer {self._get_token()}"}

    def authenticate(self) -> bool:
        if requests is None:
            return False
        token = self._get_token()
        if not token:
            return False
        resp = requests.get(f"{API_BASE}/auth.test", headers=self._headers(), timeout=10)
        ok = resp.json().get("ok", False)
        self._authenticated = ok
        return ok

    def scan(self, keyword: str = "") -> list[Channel]:
        if requests is None:
            return []
        channels: list[Channel] = []
        cursor = ""

        while True:
            params: dict = {"limit": 200, "types": "public_channel,private_channel,im,mpim"}
            if cursor:
                params["cursor"] = cursor
            resp = requests.get(
                f"{API_BASE}/conversations.list",
                headers=self._headers(), params=params, timeout=15,
            )
            data = resp.json()
            if not data.get("ok"):
                break
            for ch in data.get("channels", []):
                name = ch.get("name") or ch.get("user") or ch.get("id", "")
                if keyword and keyword.lower() not in name.lower():
                    continue
                channels.append(Channel(
                    channel_id=ch["id"],
                    name=name,
                    platform="slack",
                    member_count=ch.get("num_members", 0),
                ))
            cursor = data.get("response_metadata", {}).get("next_cursor", "")
            if not cursor:
                break

        return channels

    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        if requests is None:
            return []
        messages: list[Message] = []
        cursor = ""

        while len(messages) < limit:
            params: dict = {"channel": channel_id, "limit": min(200, limit - len(messages))}
            if cursor:
                params["cursor"] = cursor
            if since:
                params["oldest"] = str(since.timestamp())
            if until:
                params["latest"] = str(until.timestamp())

            resp = requests.get(
                f"{API_BASE}/conversations.history",
                headers=self._headers(), params=params, timeout=15,
            )
            data = resp.json()
            if not data.get("ok"):
                break
            for msg in data.get("messages", []):
                text = msg.get("text", "")
                if not text or msg.get("subtype") in ("channel_join", "channel_leave"):
                    continue
                ts = self._parse_ts(msg.get("ts"))
                messages.append(Message(
                    sender=msg.get("user", ""),
                    text=text,
                    timestamp=ts,
                    channel_name=channel_id,
                    platform="slack",
                ))
            cursor = data.get("response_metadata", {}).get("next_cursor", "")
            if not data.get("has_more") or not cursor:
                break

        return messages

    @staticmethod
    def _parse_ts(ts_str: Optional[str]) -> Optional[datetime]:
        if not ts_str:
            return None
        try:
            return datetime.fromtimestamp(float(ts_str), tz=timezone.utc)
        except (ValueError, OSError):
            return None

"""通用社交媒体归档解析器。

支持解析 Google Takeout、Facebook/Instagram 数据下载等标准归档格式。
每个平台的归档格式不同，本模块提供统一的入口和多种解析策略。

支持的归档类型：
  - Google Takeout (YouTube 评论、Google+ 等)
  - Facebook/Instagram 数据下载 (JSON 格式)
  - 微博导出
  - 知乎导出
  - 通用 JSON 数组格式
"""

from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

from .base import BaseCollector, Message, Channel


class SocialArchiveCollector(BaseCollector):
    platform_name = "social_archive"
    platform_name_zh = "社交媒体归档"

    def __init__(self, config: Optional[dict] = None):
        super().__init__(config)
        self._archive_path: Optional[Path] = None
        if config and config.get("archive_path"):
            self._archive_path = Path(config["archive_path"])

    def authenticate(self) -> bool:
        if self._archive_path and self._archive_path.exists():
            self._authenticated = True
            return True
        return False

    def scan(self, keyword: str = "") -> list[Channel]:
        if not self._archive_path:
            return []
        channels: list[Channel] = []
        if self._archive_path.is_dir():
            for f in self._archive_path.rglob("*.json"):
                name = f.stem
                if keyword and keyword.lower() not in name.lower():
                    continue
                channels.append(Channel(
                    channel_id=str(f),
                    name=name,
                    platform="social_archive",
                ))
        return channels

    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        path = Path(channel_id)
        if not path.is_file():
            return []
        return self.parse_json_file(path, since, until, limit)

    @classmethod
    def parse_json_file(
        cls,
        json_path: Path,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 1000,
    ) -> list[Message]:
        """通用 JSON 解析——自动检测格式。"""
        data = json.loads(json_path.read_text(encoding="utf-8"))
        if isinstance(data, list):
            return cls._parse_array(data, since, until, limit)
        if isinstance(data, dict):
            for key in ("messages", "posts", "comments", "items", "data", "statuses"):
                if key in data and isinstance(data[key], list):
                    return cls._parse_array(data[key], since, until, limit)
        return []

    @classmethod
    def parse_facebook_json(cls, json_path: Path, limit: int = 1000) -> list[Message]:
        """解析 Facebook/Instagram 数据下载的 JSON。"""
        data = json.loads(json_path.read_text(encoding="utf-8"))
        items = data.get("messages", data.get("posts", []))
        messages: list[Message] = []
        for item in items:
            content = item.get("content") or item.get("data", [{}])[0].get("post", "")
            if not content:
                continue
            ts = None
            ts_val = item.get("timestamp_ms") or item.get("timestamp")
            if ts_val:
                try:
                    if isinstance(ts_val, int) and ts_val > 1e12:
                        ts = datetime.fromtimestamp(ts_val / 1000, tz=timezone.utc)
                    elif isinstance(ts_val, (int, float)):
                        ts = datetime.fromtimestamp(ts_val, tz=timezone.utc)
                except (ValueError, OSError):
                    pass
            messages.append(Message(
                sender=item.get("sender_name", item.get("name", "")),
                text=content,
                timestamp=ts,
                platform="facebook",
            ))
            if len(messages) >= limit:
                break
        return messages

    @classmethod
    def _parse_array(
        cls,
        items: list,
        since: Optional[datetime],
        until: Optional[datetime],
        limit: int,
    ) -> list[Message]:
        messages: list[Message] = []
        for item in items:
            if not isinstance(item, dict):
                continue
            text = (
                item.get("text")
                or item.get("content")
                or item.get("body")
                or item.get("message")
                or item.get("post")
                or ""
            )
            if not text:
                continue
            ts = cls._extract_timestamp(item)
            if since and ts and ts < since:
                continue
            if until and ts and ts > until:
                continue
            sender = (
                item.get("sender")
                or item.get("sender_name")
                or item.get("author")
                or item.get("user")
                or item.get("from")
                or ""
            )
            if isinstance(sender, dict):
                sender = sender.get("name", sender.get("username", ""))
            messages.append(Message(
                sender=str(sender),
                text=text,
                timestamp=ts,
                platform="social_archive",
            ))
            if len(messages) >= limit:
                break
        return messages

    @staticmethod
    def _extract_timestamp(item: dict) -> Optional[datetime]:
        for key in ("timestamp", "timestamp_ms", "created_at", "date", "time", "createdAt"):
            val = item.get(key)
            if val is None:
                continue
            if isinstance(val, (int, float)):
                try:
                    if val > 1e12:
                        return datetime.fromtimestamp(val / 1000, tz=timezone.utc)
                    return datetime.fromtimestamp(val, tz=timezone.utc)
                except (ValueError, OSError):
                    continue
            if isinstance(val, str):
                for fmt in ("%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%d %H:%M:%S"):
                    try:
                        return datetime.strptime(val, fmt).replace(tzinfo=timezone.utc)
                    except ValueError:
                        continue
        return None

"""Telegram 采集器。

通过 Telethon（MTProto 客户端库）采集消息。
需要 Telegram API 凭证（从 https://my.telegram.org 获取）。

依赖：pip3 install telethon

认证方式：
  - 环境变量 TELEGRAM_API_ID / TELEGRAM_API_HASH
  - 配置文件 ~/.immortal-skill/telegram.json
"""

from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

from .base import BaseCollector, Message, Channel

CONFIG_DIR = Path.home() / ".immortal-skill"
CONFIG_FILE = CONFIG_DIR / "telegram.json"

try:
    from telethon import TelegramClient
    from telethon.tl.types import User, Chat, Channel as TGChannel
    HAS_TELETHON = True
except ImportError:
    HAS_TELETHON = False


class TelegramCollector(BaseCollector):
    platform_name = "telegram"
    platform_name_zh = "Telegram"

    def __init__(self, config: Optional[dict] = None):
        super().__init__(config)
        self._client: Optional[object] = None

    def _load_credentials(self) -> tuple[int, str]:
        api_id = os.environ.get("TELEGRAM_API_ID", "") or (self.config or {}).get("api_id", "")
        api_hash = os.environ.get("TELEGRAM_API_HASH", "") or (self.config or {}).get("api_hash", "")
        if api_id and api_hash:
            return int(api_id), api_hash
        if CONFIG_FILE.is_file():
            cfg = json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
            return int(cfg.get("api_id", 0)), cfg.get("api_hash", "")
        return 0, ""

    def _get_client(self) -> "TelegramClient":
        if self._client is not None:
            return self._client
        if not HAS_TELETHON:
            raise ImportError("需要 telethon 库：pip3 install telethon")
        api_id, api_hash = self._load_credentials()
        if not api_id or not api_hash:
            raise RuntimeError("未配置 Telegram 凭证")
        session_path = str(CONFIG_DIR / "telegram_session")
        self._client = TelegramClient(session_path, api_id, api_hash)
        return self._client

    def authenticate(self) -> bool:
        try:
            client = self._get_client()
            client.start()  # type: ignore[attr-defined]
            self._authenticated = True
            return True
        except Exception:
            return False

    def scan(self, keyword: str = "") -> list[Channel]:
        if not HAS_TELETHON:
            return []
        channels: list[Channel] = []
        try:
            client = self._get_client()
            with client:
                for dialog in client.iter_dialogs():  # type: ignore[attr-defined]
                    name = dialog.name or str(dialog.id)
                    if keyword and keyword.lower() not in name.lower():
                        continue
                    channels.append(Channel(
                        channel_id=str(dialog.id),
                        name=name,
                        platform="telegram",
                        member_count=getattr(dialog.entity, "participants_count", 0) or 0,
                    ))
        except Exception:
            pass
        return channels

    def collect(
        self,
        channel_id: str,
        since: Optional[datetime] = None,
        until: Optional[datetime] = None,
        limit: int = 500,
    ) -> list[Message]:
        if not HAS_TELETHON:
            return []
        messages: list[Message] = []
        try:
            client = self._get_client()
            entity = int(channel_id) if channel_id.lstrip("-").isdigit() else channel_id
            with client:
                for msg in client.iter_messages(  # type: ignore[attr-defined]
                    entity, limit=limit, offset_date=until, reverse=False,
                ):
                    if not msg.text:
                        continue
                    ts = msg.date.replace(tzinfo=timezone.utc) if msg.date else None
                    if since and ts and ts < since:
                        break
                    sender_name = ""
                    if msg.sender:
                        if hasattr(msg.sender, "first_name"):
                            sender_name = (msg.sender.first_name or "") + " " + (msg.sender.last_name or "")
                            sender_name = sender_name.strip()
                        elif hasattr(msg.sender, "title"):
                            sender_name = msg.sender.title or ""
                    messages.append(Message(
                        sender=sender_name or str(msg.sender_id or ""),
                        text=msg.text,
                        timestamp=ts,
                        channel_name=str(channel_id),
                        platform="telegram",
                    ))
        except Exception:
            pass
        return messages

    @classmethod
    def setup_interactive(cls) -> None:
        """交互式配置 Telegram 凭证。"""
        print("Telegram API 配置")
        print("获取凭证：https://my.telegram.org -> API development tools\n")

        api_id = input("API ID: ").strip()
        api_hash = input("API Hash: ").strip()
        if not api_id or not api_hash:
            raise ValueError("API ID 和 API Hash 不能为空")

        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        CONFIG_FILE.write_text(
            json.dumps({"api_id": api_id, "api_hash": api_hash}, indent=2) + "\n",
            encoding="utf-8",
        )
        print(f"\n凭证已保存到 {CONFIG_FILE}")
        print("首次连接时需要输入手机号和验证码。")