Agentbay Monitor Skills

"""
无影舆情爬取服务主入口
本技能仅负责按关键词/平台爬取原始数据并返回；情感分析、报告生成由主 Agent 完成。
除 AGENTBAY_API_KEY 外，其余参数由主 Agent 通过传参/命令行传入。
"""
import sys
import json
from pathlib import Path
from datetime import datetime

# 从技能根目录运行 python scripts/crawl.py 时，将 scripts 加入 path 以便导入
_scripts_dir = Path(__file__).resolve().parent
if str(_scripts_dir) not in sys.path:
    sys.path.insert(0, str(_scripts_dir))

import asyncio
import argparse
import os
from typing import List, Optional, Dict, Any

from crawler import SocialMediaCrawler, get_platform_config, create_crawler_session
from reporter import ReportGenerator


def get_api_key():
    """从 ~/.config/agentbay/api_key 或环境变量 AGENTBAY_API_KEY 获取 API Key"""
    from pathlib import Path
    file_path = Path.home() / ".config" / "agentbay" / "api_key"
    file_path.parent.mkdir(parents=True, exist_ok=True)
    if not file_path.exists():
        file_path.touch()
    if os.environ.get("AGENTBAY_API_KEY"):
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(os.environ.get("AGENTBAY_API_KEY"))
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            api_key = f.read().strip()
        if not api_key:
            api_key = None
    except Exception as e:
        api_key = None
    return api_key


# ---------- 以下供主 Agent 在爬取完成后调用：报告生成 ----------


def generate_report(
    processed_results: Dict[str, Any],
    output_dir: str = "output",
    title: Optional[str] = None,
) -> Dict[str, Any]:
    """
    根据情感分析结果生成报告（Markdown/JSON/可选 PDF），供主 Agent 调用。

    Args:
        processed_results: 主 Agent 按提示词完成情感分析后写入的 JSON（需符合 sentiment_instruction.md 中的输出格式）
        output_dir: 报告输出目录
        title: 报告标题，可选

    Returns:
        含 markdown_path、json_path、pdf_path（若有）等的字典
    """
    generator = ReportGenerator(output_dir=output_dir)
    return generator.generate_report(processed_results=processed_results, title=title)


def _compute_crawl_timeout(keywords: List[str], max_results_per_keyword: int) -> int:
    """按「约 1 条/分钟」动态计算爬取超时（秒），至少 10 分钟。"""
    total_max = len(keywords) * max_results_per_keyword
    return max(1200, 180 + total_max * 60)


async def crawl_for_sentiment(
    platform: str,
    keywords: List[str],
    *,
    max_results_per_keyword: int = 10,
    output_dir: Optional[str] = None,
    report_title: Optional[str] = None,
    agentbay_api_key: Optional[str] = None,
    context_name: str = "sentiment-analysis",
    crawl_timeout: Optional[int] = None,
) -> Dict[str, Any]:
    """
    按关键词/平台爬取原始数据并返回。不在此做情感分析或报告生成，由主 Agent 基于返回数据完成。

    Args:
        platform: 平台标识（"xhs", "weibo", "bing" 等）
        keywords: 关键词列表
        max_results_per_keyword: 每个关键词的最大结果数
        output_dir: 原始数据输出目录，默认 "output"
        report_title: 用于生成输出文件名，可选
        agentbay_api_key: AgentBay API Key，未传则从环境变量 AGENTBAY_API_KEY 读取
        context_name: Browser Context 名称
        crawl_timeout: 爬取超时（秒）；不传或为 None 时按「约 1 条/分钟」动态计算，至少 10 分钟

    Returns:
        含 success、crawl_results、raw_output_path（可选）的字典
    """
    if crawl_timeout is None or crawl_timeout <= 0:
        crawl_timeout = _compute_crawl_timeout(keywords, max_results_per_keyword)
        print(f"⏱️ 爬取超时已动态设置为 {crawl_timeout} 秒（约 1 条/分钟，至少 10 分钟）\n")

    api_key = (agentbay_api_key or "").strip() or get_api_key()
    if not api_key:
        return {
            "success": False,
            "error": "未提供 AGENTBAY_API_KEY（请设置环境变量 AGENTBAY_API_KEY 或创建 ~/.config/agentbay/api_key 文件）",
        }

    if not keywords:
        return {"success": False, "error": "关键词 keywords 不能为空"}

    out_dir = output_dir or "output"

    print(f"\n{'='*60}")
    print(f"🚀 启动舆情爬取（情感分析由主 Agent 完成）")
    print(f"{'='*60}\n")

    try:
        platform_config = get_platform_config(platform)
        print(f"📱 目标平台: {platform_config.display_name}")
    except ValueError as e:
        return {"success": False, "error": str(e)}

    print(f"🔍 关键词: {', '.join(keywords)}")
    print(f"📊 每个关键词最大结果数: {max_results_per_keyword}\n")

    adapter = None
    try:
        print("=" * 60)
        print("步骤1: 创建爬取会话")
        print("=" * 60)
        adapter = await create_crawler_session(
            api_key=api_key,
            context_name=context_name,
            platform_config=platform_config,
        )

        print("\n" + "=" * 60)
        print("步骤2: 执行内容爬取")
        print("=" * 60)
        crawler = SocialMediaCrawler(adapter, platform_config)

        if len(keywords) == 1:
            crawl_results = await crawler.crawl_by_keyword(
                keyword=keywords[0],
                max_results=max_results_per_keyword,
                timeout=crawl_timeout,
            )
        else:
            crawl_results = await crawler.crawl_multiple_keywords(
                keywords=keywords,
                max_results_per_keyword=max_results_per_keyword,
                timeout=crawl_timeout,
            )

        if not crawl_results.get("success"):
            return crawl_results

        crawl_results["data_sources"] = [crawl_results.get("platform_display", "浏览器爬取")]

        # 将原始爬取结果写入 output_dir，供主 Agent 做情感分析/报告
        raw_output_path = None
        try:
            out_path = Path(out_dir)
            out_path.mkdir(parents=True, exist_ok=True)
            platform_display = crawl_results.get("platform_display", platform)
            title_part = (report_title or f"{platform_display}_{','.join(keywords[:2])}").replace(" ", "_")[:50]
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            raw_filename = f"{platform}_{title_part}_{ts}.json"
            raw_output_path = out_path / raw_filename
            with open(raw_output_path, "w", encoding="utf-8") as f:
                json.dump(crawl_results, f, ensure_ascii=False, indent=2)
            raw_output_path = str(raw_output_path)
        except Exception as e:
            print(f"   ⚠️ 写入原始数据文件失败（不影响返回）: {e}")

        print("\n" + "=" * 60)
        print("✅ 爬取完成，原始数据已返回（情感分析由主 Agent 完成）")
        print("=" * 60)

        return {
            "success": True,
            "crawl_results": crawl_results,
            "raw_output_path": raw_output_path,
        }

    except Exception as e:
        import traceback
        error_msg = f"爬取过程中发生错误: {str(e)}"
        print(f"\n❌ {error_msg}")
        print(f"详细错误:\n{traceback.format_exc()}")
        return {"success": False, "error": error_msg}

    finally:
        if adapter:
            await adapter.close()


def _parse_args():
    parser = argparse.ArgumentParser(
        description="舆情爬取：除 AGENTBAY_API_KEY 外，其余参数由主 Agent 传入；情感分析由主 Agent 完成。",
    )
    parser.add_argument("--keywords", "-k", required=True, help="关键词，多个用逗号分隔")
    parser.add_argument("--platform", "-p", default="baidu", help="平台: baidu/xhs/weibo/douyin/zhihu/bing（默认 baidu）")
    parser.add_argument("--max-results", type=int, default=10, help="每关键词最大结果数（默认10）")
    parser.add_argument("--output-dir", "-o", default="output", help="报告输出目录")
    parser.add_argument("--report-title", help="报告标题（可选）")
    parser.add_argument("--context-name", default="sentiment-analysis", help="Browser Context 名称")
    parser.add_argument(
        "--crawl-timeout",
        type=int,
        default=None,
        help="爬取超时（秒）；不传时按「约 1 条/分钟」动态计算，至少 10 分钟",
    )
    return parser.parse_args()


async def main():
    args = _parse_args()
    keywords = [kw.strip() for kw in args.keywords.split(",") if kw.strip()]
    if not keywords:
        print("❌ 错误: --keywords 不能为空")
        sys.exit(1)

    # 未传 --crawl-timeout 时由 crawl_for_sentiment 内部按「约 1 条/分钟」动态计算
    effective_timeout = args.crawl_timeout
    if effective_timeout is None:
        effective_timeout = _compute_crawl_timeout(keywords, args.max_results)
    print(f"\n📋 参数: 平台={args.platform}, 关键词={', '.join(keywords)}, 每关键词最大={args.max_results}, 输出={args.output_dir}, 爬取超时={effective_timeout} 秒")
    if args.report_title:
        print(f"   报告标题: {args.report_title}")
    print()

    result = await crawl_for_sentiment(
        platform=args.platform,
        keywords=keywords,
        max_results_per_keyword=args.max_results,
        output_dir=args.output_dir,
        report_title=args.report_title or None,
        context_name=args.context_name,
        crawl_timeout=args.crawl_timeout,
    )

    if result.get("success"):
        print("\n✅ 爬取完成！")
        path = result.get("raw_output_path")
        if path:
            print(f"📄 原始数据: {path}")
        print("   情感分析、报告由主 Agent 基于返回数据完成。")
    else:
        print(f"\n❌ 爬取失败: {result.get('error')}")
        sys.exit(1)


if __name__ == "__main__":
    asyncio.run(main())

"""
爬取模块
提供基于 wuying-agentbay-sdk 的社交媒体平台爬取功能
"""

from .agentbay_adapter import AgentBayAdapter, create_crawler_session
from .platform_config import PlatformConfig, get_platform_config, SUPPORTED_PLATFORMS
from .crawler import SocialMediaCrawler
from .prompts import build_search_prompt, get_search_prompt_template

__all__ = [
    "AgentBayAdapter",
    "create_crawler_session",
    "PlatformConfig",
    "get_platform_config",
    "SUPPORTED_PLATFORMS",
    "SocialMediaCrawler",
    "build_search_prompt",
    "get_search_prompt_template",
]

"""
AgentBay 适配器模块
提供与 AgentBay 交互的核心功能，用于社交媒体平台爬取
"""
import asyncio
import json
from typing import Dict, Any, Optional

# AgentBay imports
try:
    from agentbay import AsyncAgentBay, CreateSessionParams, BrowserOption, BrowserContext
    AGENTBAY_AVAILABLE = True
except ImportError as e:
    AGENTBAY_AVAILABLE = False
    print("⚠️ 警告: wuying-agentbay-sdk 未安装，无法使用 AgentBay")
    print(f"   导入错误详情: {e}")

from .platform_config import PlatformConfig


class AgentBayAdapter:
    """AgentBay适配器类"""

    def __init__(self, api_key: str, context_name: str = "sentiment-analysis"):
        """
        初始化适配器

        Args:
            api_key: AgentBay API密钥
            context_name: Browser Context 名称
        """
        if not AGENTBAY_AVAILABLE:
            raise ImportError("wuying-agentbay-sdk 未安装，无法使用 AgentBay")

        self.api_key = api_key
        self.context_name = context_name
        self.agent_bay = None
        self.session = None
        self.context = None

    async def create_session(self, platform_config: PlatformConfig) -> Dict[str, Any]:
        """
        创建浏览器会话

        Args:
            platform_config: 平台配置

        Returns:
            包含 session、context、agent_bay 的字典，如果失败返回错误信息
        """
        try:
            self.agent_bay = AsyncAgentBay(api_key=self.api_key)

            # 创建或获取 Browser Context
            print(f"📦 创建/获取 Browser Context: {self.context_name}")
            context_result = await self.agent_bay.context.get(self.context_name, create=False)
            context_is_new = False

            if not context_result.success or not context_result.context:
                print(f"   Context 不存在，正在创建新的 Context...")
                context_result = await self.agent_bay.context.get(self.context_name, create=True)
                context_is_new = True

            if not context_result.success or not context_result.context:
                error_msg = context_result.error_message or 'Unknown error'
                return {
                    "success": False,
                    "error": f"创建 Context 失败: {error_msg}"
                }

            self.context = context_result.context
            if context_is_new:
                print(f"✅ 新 Context 已创建，ID: {self.context.id}")
            else:
                print(f"✅ 已存在的 Context 已加载，ID: {self.context.id}")

            # 创建 BrowserContext 配置
            browser_context = BrowserContext(self.context.id, auto_upload=True)

            # 创建浏览器会话
            print("📡 正在创建 AgentBay 浏览器会话...")
            params = CreateSessionParams(
                image_id="linux_latest",
                browser_context=browser_context
            )
            session_result = await self.agent_bay.create(params)

            if not session_result.success:
                return {
                    "success": False,
                    "error": session_result.error_message or "Failed to create session"
                }

            self.session = session_result.session
            print(f"✅ 会话已创建: {self.session.session_id}\n")

            # 初始化浏览器
            print("🌐 正在初始化浏览器...")
            browser_option = BrowserOption()
            browser_init = await self.session.browser.initialize(browser_option)
            if not browser_init:
                return {
                    "success": False,
                    "error": "Browser initialization failed"
                }
            print("✅ 浏览器已初始化\n")

            return {
                "success": True,
                "agent_bay": self.agent_bay,
                "session": self.session,
                "context": self.context
            }

        except Exception as e:
            error_msg = f"创建 Session 失败: {str(e)}"
            print(f"\n❌ 错误: {error_msg}")
            return {
                "success": False,
                "error": error_msg
            }

    async def execute_crawl_task(
        self,
        task_prompt: str,
        timeout: int = 600
    ) -> Dict[str, Any]:
        """
        执行爬取任务

        Args:
            task_prompt: 任务提示词
            timeout: 超时时间（秒）

        Returns:
            任务执行结果
        """
        if not self.session:
            return {
                "success": False,
                "error": "Session 未创建，请先调用 create_session"
            }

        try:
            print(f"🚀 正在执行爬取任务...")
            result = await self.session.agent.browser.execute_task_and_wait(
                task_prompt,
                timeout,
                True,
                None
            )

            if not result.success:
                return {
                    "success": False,
                    "error": result.error_message or result.task_status,
                    "task_status": result.task_status
                }

            # 解析结果
            task_result = result.task_result
            if isinstance(task_result, str):
                try:
                    task_result = json.loads(task_result)
                except json.JSONDecodeError:
                    pass

            return {
                "success": True,
                "result": task_result,
                "raw_result": result.task_result
            }

        except Exception as e:
            import traceback
            error_msg = f"执行爬取任务失败: {str(e)}"
            print(f"❌ {error_msg}")
            print(f"   详细错误:\n{traceback.format_exc()}")
            return {
                "success": False,
                "error": error_msg
            }

    async def close(self):
        """关闭会话"""
        if self.session and self.agent_bay:
            try:
                await asyncio.sleep(2)

                # 显式同步 Context（保存浏览器状态）
                try:
                    sync_result = await self.session.context.sync()
                    if sync_result.success:
                        print("✅ Context 已同步")
                except Exception as sync_error:
                    print(f"⚠️ Context 同步出错: {sync_error}")

                # 删除 session
                await self.agent_bay.delete(self.session, sync_context=False)
                print("✅ 会话已关闭")
            except Exception as e:
                print(f"⚠️ 关闭会话时出错: {e}")


async def create_crawler_session(
    api_key: str,
    context_name: str,
    platform_config: PlatformConfig
) -> AgentBayAdapter:
    """
    创建爬取会话的便捷函数

    Args:
        api_key: AgentBay API密钥
        context_name: Browser Context 名称
        platform_config: 平台配置

    Returns:
        AgentBayAdapter实例
    """
    adapter = AgentBayAdapter(api_key, context_name)
    result = await adapter.create_session(platform_config)
    if not result.get("success"):
        raise Exception(result.get("error", "创建会话失败"))
    return adapter

"""
社交媒体爬取器
使用 AgentBay 进行内容爬取
"""
import json
from typing import Dict, Any, List, Optional
from datetime import datetime

from .agentbay_adapter import AgentBayAdapter
from .platform_config import PlatformConfig, get_platform_config
from .prompts import build_search_prompt


class SocialMediaCrawler:
    """社交媒体爬取器"""

    def __init__(self, adapter: AgentBayAdapter, platform_config: PlatformConfig):
        """
        初始化爬取器

        Args:
            adapter: AgentBay适配器
            platform_config: 平台配置
        """
        self.adapter = adapter
        self.platform_config = platform_config

    def _build_search_prompt(self, keyword: str, max_results: int = 50) -> str:
        """
        构建搜索提示词

        Args:
            keyword: 搜索关键词
            max_results: 最大结果数

        Returns:
            搜索提示词
        """
        # 构造搜索URL
        search_url = self.platform_config.build_search_url(keyword)

        return build_search_prompt(
            platform_name=self.platform_config.display_name,
            keyword=keyword,
            base_url=self.platform_config.base_url,
            search_url=search_url,
            max_results=max_results,
            platform_id=self.platform_config.name
        )

    async def crawl_by_keyword(
        self,
        keyword: str,
        max_results: int = 50,
        timeout: int = 600
    ) -> Dict[str, Any]:
        """
        根据关键词爬取内容

        Args:
            keyword: 搜索关键词
            max_results: 最大结果数
            timeout: 超时时间（秒）

        Returns:
            爬取结果
        """
        print(f"\n{'='*60}")
        print(f"🔍 开始爬取 {self.platform_config.display_name} 平台")
        print(f"关键词: {keyword}")
        print(f"最大结果数: {max_results}")
        print(f"{'='*60}\n")

        # 结果文件由 Agent 按 JSON Lines 格式写入（每行一条，append 追加），无需预先创建

        # 构建搜索提示词
        prompt = self._build_search_prompt(keyword, max_results)

        # 执行爬取任务
        result = await self.adapter.execute_crawl_task(prompt, timeout)

        # 任务结束后从会话文件系统读取 /tmp/results.json（支持 JSON 数组或 JSON Lines）
        # 即使任务被标记为 success: false（如 Agent 因未满 30 条调用了 done(success: false)），
        # 仍尝试读取文件，若有数据则视为部分成功，避免丢弃已抓取内容
        results_from_file = None
        if self.adapter.session:
            try:
                print("📂 正在从会话读取 /tmp/results.json ...")
                file_result = await self.adapter.session.file_system.read_file("/tmp/results.json")
                if file_result.success:
                    raw = (file_result.content or "").strip()
                    if raw:
                        try:
                            # 兼容多种格式：
                            # 1) 整份为 JSON 数组 [ ... ]
                            if raw.strip().startswith("["):
                                results_from_file = json.loads(raw)
                                if not isinstance(results_from_file, list):
                                    results_from_file = None
                            else:
                                lines = [ln.strip() for ln in raw.split("\n") if ln.strip()]
                                results_from_file = []
                                # 2) 首行为 Agent 写的 header（以 "results":[ 结尾），其余每行一条 result JSON
                                if lines and lines[0].rstrip().endswith('"results":['):
                                    for line in lines[1:]:
                                        try:
                                            results_from_file.append(json.loads(line))
                                        except json.JSONDecodeError:
                                            pass
                                else:
                                    # 3) 纯 JSONL：每行一条完整 JSON
                                    for line in lines:
                                        try:
                                            results_from_file.append(json.loads(line))
                                        except json.JSONDecodeError:
                                            pass
                            if results_from_file is not None:
                                print(f"📄 已读取 /tmp/results.json，共 {len(results_from_file)} 条结果")
                        except json.JSONDecodeError as e:
                            print(f"⚠️ /tmp/results.json 解析失败: {e}，将使用任务返回结果")
                            results_from_file = None
                    else:
                        results_from_file = []
                        print("📄 已读取 /tmp/results.json，内容为空")
                else:
                    print(f"⚠️ 读取 /tmp/results.json 失败: {getattr(file_result, 'error_message', 'unknown')}")
            except Exception as e:
                print(f"⚠️ 读取 /tmp/results.json 异常: {e}，将使用任务返回结果")

        # 若任务标记为失败且未从文件读到任何结果，则直接返回失败
        if not result.get("success") and not (
            results_from_file is not None
            and isinstance(results_from_file, list)
            and len(results_from_file) > 0
        ):
            return result

        # 解析结果
        task_result = result.get("result", {})

        # 只要成功从文件读到列表（含空列表），就以文件为准
        if results_from_file is not None and isinstance(results_from_file, list):
            task_result = {
                "success": True,
                "platform": self.platform_config.name,
                "keyword": keyword,
                "total_count": len(results_from_file),
                "results": results_from_file
            }
            if len(results_from_file) > 0:
                print(f"✅ 已使用 /tmp/results.json 中的结果，共 {len(results_from_file)} 条")

        # 如果结果是字符串，尝试解析JSON
        if isinstance(task_result, str):
            try:
                task_result = json.loads(task_result)
            except json.JSONDecodeError:
                # 如果不是JSON，尝试从原始结果中提取
                task_result = {
                    "success": True,
                    "platform": self.platform_config.name,
                    "keyword": keyword,
                    "total_count": 0,
                    "results": []
                }

        # 确保结果格式正确
        if not isinstance(task_result, dict):
            task_result = {
                "success": True,
                "platform": self.platform_config.name,
                "keyword": keyword,
                "total_count": 0,
                "results": []
            }

        # 添加元数据
        task_result["crawl_time"] = datetime.now().isoformat()
        task_result["platform"] = self.platform_config.name
        task_result["platform_display"] = self.platform_config.display_name

        # 确保results是列表
        if "results" not in task_result or not isinstance(task_result["results"], list):
            task_result["results"] = []

        print(f"✅ 爬取完成，共获取 {len(task_result.get('results', []))} 条结果\n")

        return {
            "success": True,
            **task_result
        }

    async def crawl_multiple_keywords(
        self,
        keywords: List[str],
        max_results_per_keyword: int = 50,
        timeout: int = 600
    ) -> Dict[str, Any]:
        """
        爬取多个关键词

        Args:
            keywords: 关键词列表
            max_results_per_keyword: 每个关键词的最大结果数
            timeout: 超时时间（秒）

        Returns:
            合并后的爬取结果
        """
        all_results = []

        for i, keyword in enumerate(keywords, 1):
            print(f"\n处理关键词 {i}/{len(keywords)}: {keyword}")

            result = await self.crawl_by_keyword(
                keyword=keyword,
                max_results=max_results_per_keyword,
                timeout=timeout
            )

            if result.get("success") and "results" in result:
                all_results.extend(result["results"])

            # 添加延迟，避免请求过快
            import asyncio
            await asyncio.sleep(2)

        return {
            "success": True,
            "platform": self.platform_config.name,
            "platform_display": self.platform_config.display_name,
            "keywords": keywords,
            "total_count": len(all_results),
            "results": all_results,
            "crawl_time": datetime.now().isoformat()
        }

"""
平台配置模块
支持小红书、微博等社交媒体平台的配置
"""
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class PlatformConfig:
    """平台配置类"""
    # 平台标识
    name: str  # 平台名称，如 "xiaohongshu", "weibo"
    display_name: str  # 显示名称，如 "小红书", "微博"

    # URL配置
    base_url: str  # 平台首页URL
    search_url: str  # 搜索页面URL

    # 搜索相关配置
    search_box_selector: Optional[str] = None  # 搜索框选择器
    search_button_text: List[str] = None  # 搜索按钮文本

    # 内容选择器
    content_selector: Optional[str] = None  # 内容容器选择器
    title_selector: Optional[str] = None  # 标题选择器
    author_selector: Optional[str] = None  # 作者选择器
    time_selector: Optional[str] = None  # 时间选择器

    def __post_init__(self):
        """初始化后处理"""
        if self.search_button_text is None:
            self.search_button_text = ["搜索", "Search"]

    def build_search_url(self, keyword: str) -> str:
        """构造搜索 URL，关键词直接使用中文（由浏览器在请求时编码）。"""
        if self.name == "xhs":
            return f"{self.search_url}?keyword={keyword}&source=web_explore_feed"
        elif self.name == "weibo":
            return f"{self.search_url}?q={keyword}"
        elif self.name == "bing":
            return f"{self.search_url}?q={keyword}"
        elif self.name == "baidu":
            return f"{self.search_url}?rtt=1&bsst=1&cl=2&tn=news&ie=utf-8&word={keyword}"
        else:
            return f"{self.search_url}?keyword={keyword}"


# 平台配置字典
PLATFORM_CONFIGS = {
    "xhs": PlatformConfig(
        name="xhs",
        display_name="小红书",
        base_url="https://www.xiaohongshu.com",
        search_url="https://www.xiaohongshu.com/search_result",
        search_box_selector="input[placeholder*='搜索']",
        search_button_text=["搜索"],
        content_selector=".note-item",
        title_selector=".title",
        author_selector=".author",
        time_selector=".time"
    ),

    "weibo": PlatformConfig(
        name="weibo",
        display_name="微博",
        base_url="https://weibo.com",
        search_url="https://s.weibo.com",
        search_box_selector="input[type='text']",
        search_button_text=["搜索", "Search"],
        content_selector=".card-wrap",
        title_selector=".txt",
        author_selector=".name",
        time_selector=".from"
    ),

    "douyin": PlatformConfig(
        name="douyin",
        display_name="抖音",
        base_url="https://www.douyin.com",
        search_url="https://www.douyin.com/search",
        search_box_selector="input[placeholder*='搜索']",
        search_button_text=["搜索"],
        content_selector=".video-item",
        title_selector=".title",
        author_selector=".author",
        time_selector=".time"
    ),

    "zhihu": PlatformConfig(
        name="zhihu",
        display_name="知乎",
        base_url="https://www.zhihu.com",
        search_url="https://www.zhihu.com/search",
        search_box_selector="input[type='text']",
        search_button_text=["搜索"],
        content_selector=".ContentItem",
        title_selector=".ContentItem-title",
        author_selector=".AuthorInfo-name",
        time_selector=".ContentItem-time"
    ),

    "bing": PlatformConfig(
        name="bing",
        display_name="Bing",
        base_url="https://www.bing.com",
        search_url="https://www.bing.com/search",
        search_box_selector="input[name='q']",
        search_button_text=["搜索", "Search"],
        content_selector="li.b_algo",
        title_selector="h2 a",
        author_selector=None,
        time_selector=None
    ),

    "baidu": PlatformConfig(
        name="baidu",
        display_name="百度",
        base_url="https://www.baidu.com",
        search_url="https://www.baidu.com/s",
        search_box_selector="input[name='wd']",
        search_button_text=["百度一下", "搜索"],
        content_selector=".result-op, .c-container",
        title_selector="h3 a, .c-title a",
        author_selector=None,
        time_selector=None
    ),
}

# 支持的平台列表
SUPPORTED_PLATFORMS = list(PLATFORM_CONFIGS.keys())


def get_platform_config(platform: str) -> PlatformConfig:
    """
    获取平台配置

    Args:
        platform: 平台标识（"xhs", "weibo"等）

    Returns:
        PlatformConfig对象

    Raises:
        ValueError: 如果平台不存在
    """
    if platform not in PLATFORM_CONFIGS:
        available_platforms = ", ".join(PLATFORM_CONFIGS.keys())
        raise ValueError(
            f"不支持的平台: {platform}\n"
            f"支持的平台: {available_platforms}"
        )

    return PLATFORM_CONFIGS[platform]

"""
提示词管理模块
从 YAML 文件加载提示词模板
"""
import os
import yaml
from pathlib import Path
from typing import Dict, Any, Optional


# 提示词缓存
_prompts_cache: Dict[str, str] = {}


def _load_prompts() -> Dict[str, str]:
    """
    加载提示词模板

    Returns:
        提示词字典
    """
    global _prompts_cache

    if _prompts_cache:
        return _prompts_cache

    # 获取当前文件所在目录
    current_dir = Path(__file__).parent
    prompts_file = current_dir / "prompts.yaml"

    if not prompts_file.exists():
        raise FileNotFoundError(f"提示词文件不存在: {prompts_file}")

    try:
        with open(prompts_file, "r", encoding="utf-8") as f:
            prompts_data = yaml.safe_load(f)

        if not prompts_data:
            raise ValueError("提示词文件为空或格式错误")

        _prompts_cache = prompts_data
        return _prompts_cache

    except yaml.YAMLError as e:
        raise ValueError(f"解析提示词 YAML 文件失败: {e}")
    except Exception as e:
        raise RuntimeError(f"加载提示词文件失败: {e}")


def get_search_prompt_template() -> str:
    """
    获取搜索提示词模板

    Returns:
        提示词模板字符串
    """
    prompts = _load_prompts()
    template = prompts.get("search_prompt_template")

    if not template:
        raise ValueError("未找到 search_prompt_template 提示词模板")

    return template


def get_bing_search_prompt_template() -> str:
    """
    获取 Bing 搜索提示词模板

    Returns:
        Bing 搜索提示词模板字符串
    """
    prompts = _load_prompts()
    template = prompts.get("bing_search_prompt_template")

    if not template:
        raise ValueError("未找到 bing_search_prompt_template 提示词模板")

    return template


def get_baidu_search_prompt_template() -> str:
    """
    获取百度搜索提示词模板

    Returns:
        百度搜索提示词模板字符串
    """
    prompts = _load_prompts()
    template = prompts.get("baidu_search_prompt_template")

    if not template:
        raise ValueError("未找到 baidu_search_prompt_template 提示词模板")

    return template


def build_search_prompt(
    platform_name: str,
    keyword: str,
    base_url: str,
    search_url: str,
    max_results: int = 50,
    platform_id: Optional[str] = None
) -> str:
    """
    构建搜索提示词。根据 platform_id 选择社交媒体模板或 Bing 模板。

    Args:
        platform_name: 平台名称
        keyword: 搜索关键词
        base_url: 平台基础URL
        search_url: 构造好的搜索URL（已包含关键词）
        max_results: 最大结果数
        platform_id: 平台标识，为 "bing" 时使用 Bing 模板、"baidu" 时使用百度模板，否则使用社交媒体模板

    Returns:
        格式化后的提示词
    """
    if platform_id == "bing":
        template = get_bing_search_prompt_template()
    elif platform_id == "baidu":
        template = get_baidu_search_prompt_template()
    else:
        template = get_search_prompt_template()

    # 只替换已知占位符，避免模板中 JSON 示例的 { } 被 str.format() 误解析
    out = (
        template.replace("{platform_name}", platform_name)
        .replace("{keyword}", keyword)
        .replace("{base_url}", base_url)
        .replace("{search_url}", search_url)
        .replace("{max_results}", str(max_results))
    )
    # 还原 YAML 中为转义写的双花括号为单花括号（用于 JSON 示例）
    return out.replace("{{", "{").replace("}}", "}")

# 爬取提示词模板
# search_prompt_template: 社交媒体（小红书、微博、抖音、知乎等）
# bing_search_prompt_template: 搜索引擎（Bing）

# 社交媒体：进详情页点封面图，从详情页提取
search_prompt_template: |
  在{platform_name}平台搜索关键词"{keyword}"并爬取内容。

  **步骤**：
  1. `go_to_url` 跳转 {search_url}，等待 5-10 秒页面加载。
  2. 滚动加载更多结果，确保至少 {max_results} 条可提取。
  3. **循环提取**（满 {max_results} 条即停）：
     - 在列表页**点击该条封面图**（勿点标题）进详情页，等待 3-5 秒；
     - 从详情页 DOM 提取：title、content、author、publish_time、likes、shares、comments、comment_list、url、content_type（无评论则 comment_list=[]、comments=null）；
     - **立即** `write_file` 写入 `/tmp/results.json`：第一条 append=false，之后 append=true，每行一条 JSON；禁止全部写完再一次性写入；
     - 返回列表页再处理下一条。
  4. 写完第 {max_results} 条后**立即 `done`**，不要 summary_info、read_file 或汇总。

  **输出**：每条 JSON 含 title、content、author、publish_time、likes、shares、comments、comment_list、url、content_type；整体 success、platform、keyword、total_count、results[]。

  **约束**：直接跳转不点搜索框；进详情只点封面图；严格 {max_results} 条；缺字段填 null/空；关键词 "{keyword}"。

# 搜索引擎（Bing）：仅列表页提取资讯，不点进任何链接；不足条数不重试，能抓几条就几条。
bing_search_prompt_template: |
  在 Bing 搜索"{keyword}"，**仅爬取资讯类**结果，**不点击任何链接**（不进文章页、不点「资讯」选项卡——该选项卡会跳转 MSN，导致无法在 Bing 列表页提取）。

  **步骤**：
  1. `go_to_url` 跳转 {search_url}，等待 5-10 秒。
  2. **在当前 Bing 列表页**识别资讯类条目（带来源如光明网/新华网/XX 新闻及「X小时前」等时间戳的卡片），跳过官网、百科、广告。最多 {max_results} 条；若当前屏不足可滚动 1～2 屏，**仍不足则能抓几条就几条，抓完即结束，不要为凑满而反复滚动**。
  3. **仅在列表页**从 DOM 读取每条卡片的标题、摘要(content)、链接(url)，以及若有则 author、publish_time；无则 null/[]；content_type 填 "news"。**每提取一条立即** `write_file` 写入 `/tmp/results.json`：第一条 append=false，之后 append=true，每行一条 JSON；禁止点进链接后提取、禁止全部写完再写文件。
  4. 抓完当前能抓到的所有资讯条（最多 {max_results} 条）后**立即 `done`**（例如只有 3 条就写 3 条后 done）。不要 summary_info、read_file 或汇总。

  **输出**：每条 JSON 含 title、content、author、publish_time、likes、shares、comments、comment_list、url、content_type；整体 success、platform="bing"、keyword、total_count、results[]。

  **约束**：直接跳转不点搜索框；不点「资讯/新闻」选项卡；不点任何结果标题或链接；能抓几条就几条，抓完即 done；缺字段填 null；关键词 "{keyword}"。

# 搜索引擎（百度）：仅列表页提取资讯，不点进任何链接；资讯 URL 为 tn=news；支持翻页以凑满 max_results。
# 提取方式：从 DOM（Interactive elements）直接提取，禁止 extract_structured_data，避免页面内容被截断（约3000字）导致失败。
baidu_search_prompt_template: |
  在百度搜索"{keyword}"，**仅爬取资讯类**结果，**不点击任何结果标题/链接**（不进文章页）。

  **步骤**：
  1. `go_to_url` 跳转 {search_url}（该 URL 已为资讯搜索 tn=news），等待 5-10 秒。
  2. **在当前百度列表页**识别资讯类条目（带来源、时间戳的卡片），跳过官网、百科、广告。目标 **{max_results} 条**：若当前屏不足可先滚动 1～2 屏；**若本页抓完后仍不足 {max_results} 条，须点击页面底部的「下一页」翻页**，在新一页继续抓取，直至已抓满 {max_results} 条或没有「下一页」为止。**翻页前必须先把当前页所有已识别的资讯条全部写入 /tmp/results.json，否则下一页会覆盖或漏掉本页数据。**
  3. **提取方式（必读）**：**禁止使用 extract_structured_data**。请**仅根据当前步骤上下文中已有的「Interactive elements」**（即当前页面 DOM 的可交互元素列表，带索引 [N] 和文本）识别每条资讯卡片，从中解析出：标题(title)、链接(url)、来源(author)、发布时间(publish_time)、摘要(content)。每解析出一条就**立即**用 `write_file` 写入 `/tmp/results.json`：**只写每条资讯的 JSON 对象**，第一条 append=false（写入第一行），之后 append=true（每行追加一条）。**不要写** `{"platform":...}` 或 `"results":[` 等头部，文件内容应为「每行一条完整 JSON 对象」。当前页能识别的资讯全部写完后，再点击「下一页」；翻页后同样从新页的 Interactive elements 中逐条解析并立即 write_file。缺的字段填 null；likes、shares、comments、comment_list 无则 null/[]；content_type 填 "资讯"。
  4. 抓满 {max_results} 条或已无更多页/更多条后**立即 `done`**。不要 summary_info、read_file 或汇总。
  **重要**：无论抓满 {max_results} 条还是「已无更多页/搜索结果总量不足」而提前结束，**都必须**调用 `done(success: true, data=...)`，在 data.result 中说明「共抓取 N 条」即可。只有真正执行出错才用 success: false。

  **防止死循环（必读）**：同一批资讯**只写一次**，不要在多步中重复写入相同条目。**一旦在本轮中已经执行过一轮 write_file 且共写入了 {max_results} 条**（即先 1 次 append=false 再 {max_results}-1 次 append=true），**下一步有且仅有一个动作：调用 `done(success: true, data=...)`**，不得再次对同一页或同一批卡片执行 write_file。若你上一步已写入过 {max_results} 条（或任意已达到目标条数），则本步**只允许**调用 done，禁止再写文件。

  **输出**：每条 JSON 含 title、content、author、publish_time、likes、shares、comments、comment_list、url、content_type；整体 success、platform="baidu"、keyword、total_count、results[]。

  **约束**：直接跳转不点搜索框；不点任何结果标题或链接；**先写满当前页再翻页**；**仅从 DOM（Interactive elements）提取，禁止调用 extract_structured_data**；缺字段填 null；关键词 "{keyword}"。

#!/usr/bin/env python3
"""
登录模块

提供社交媒体平台的登录功能，使用 AgentBay 沙箱和 CDP 协议进行登录。

API Key 从 ~/.config/agentbay/api_key 或环境变量 AGENTBAY_API_KEY 读取。

使用方法：
    python scripts/login.py [--platform xhs] [--context-name sentiment-analysis]
"""

import asyncio
import os
import sys
import traceback
from pathlib import Path

# 从技能根目录运行 python scripts/login.py 时，将 scripts 加入 path 以便导入
_scripts_dir = Path(__file__).resolve().parent
if str(_scripts_dir) not in sys.path:
    sys.path.insert(0, str(_scripts_dir))


def get_api_key():
    """从 ~/.config/agentbay/api_key 或环境变量 AGENTBAY_API_KEY 获取 API Key"""
    file_path = Path.home() / ".config" / "agentbay" / "api_key"
    file_path.parent.mkdir(parents=True, exist_ok=True)
    if not file_path.exists():
        file_path.touch()
    if os.environ.get("AGENTBAY_API_KEY"):
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(os.environ.get("AGENTBAY_API_KEY"))
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            api_key = f.read().strip()
        if not api_key:
            api_key = None
    except Exception as e:
        api_key = None
    return api_key

# AgentBay imports
try:
    from agentbay import AsyncAgentBay, CreateSessionParams, BrowserOption, BrowserContext
    AGENTBAY_AVAILABLE = True
except ImportError as e:
    AGENTBAY_AVAILABLE = False
    print("⚠️ 警告: wuying-agentbay-sdk 未安装，无法使用 AgentBay")
    print(f"   导入错误详情: {e}")

# 导入平台配置
from crawler.platform_config import get_platform_config


async def login_only(
    agentbay_api_key: str,
    platform: str = None,
    context_name: str = None
) -> bool:
    """
    只执行登录流程，登录完成后保存状态并退出

    简化流程：
    1. 创建 Session 并初始化浏览器
    2. 使用 CDP 直接导航到登录页面（不使用 agent）
    3. 打开流化页面让用户登录
    4. 等待用户回车确认
    5. 保存登录状态到 Context

    Args:
        agentbay_api_key: AgentBay API密钥
        platform: 平台标识（"xhs", "weibo", "douyin", "zhihu"等），必需
        context_name: Browser Context 名称，必需

    Returns:
        True: 流程完成
        False: 流程失败
    """
    if not AGENTBAY_AVAILABLE:
        print("❌ wuying-agentbay-sdk 未安装，无法使用 AgentBay")
        return False

    # 获取平台配置
    if platform is None:
        platform = "xhs"

    try:
        platform_config = get_platform_config(platform)
    except ValueError as e:
        print(f"❌ {e}")
        return False

    # context_name 必须由调用方传入
    if context_name is None:
        print("❌ 错误: 未提供 context_name 参数")
        return False

    login_url = platform_config.base_url  # 使用平台首页作为登录入口
    platform_name = platform_config.display_name

    print(f"\n{'='*60}")
    print(f"🔐 登录模式：只执行登录流程")
    print(f"   平台: {platform_name}")
    print(f"   Context: {context_name}")
    print(f"{'='*60}\n")

    agent_bay = None
    session = None
    context = None

    try:
        # 创建 AgentBay 实例
        agent_bay = AsyncAgentBay(api_key=agentbay_api_key)

        # 创建或获取持久化的 Browser Context
        print(f"📦 创建/获取 Browser Context: {context_name}")
        context_result = await agent_bay.context.get(context_name, create=False)
        context_is_new = False

        if not context_result.success or not context_result.context:
            print(f"   Context 不存在，正在创建新的 Context...")
            context_result = await agent_bay.context.get(context_name, create=True)
            context_is_new = True

        if not context_result.success or not context_result.context:
            error_msg = context_result.error_message or 'Unknown error'
            print(f"❌ 创建 Context 失败: {error_msg}")
            return False

        context = context_result.context
        if context_is_new:
            print(f"✅ 新 Context 已创建，ID: {context.id}")
        else:
            print(f"✅ 已存在的 Context 已加载，ID: {context.id}")

        # 创建 BrowserContext 配置
        browser_context = BrowserContext(context.id, auto_upload=True)

        # 创建浏览器会话
        print("📡 正在创建 AgentBay 浏览器会话...")
        params = CreateSessionParams(
            image_id="linux_latest",
            browser_context=browser_context
        )
        session_result = await agent_bay.create(params)

        if not session_result.success:
            print(f"❌ 创建会话失败: {session_result.error_message}")
            return False

        session = session_result.session
        print(f"✅ 会话已创建: {session.session_id}\n")

        # 初始化浏览器
        print("🌐 正在初始化浏览器...")
        browser_init = await session.browser.initialize(BrowserOption())
        if not browser_init:
            print("❌ 浏览器初始化失败")
            return False
        print("✅ 浏览器已初始化\n")

        # 获取流化页面 URL
        resource_url = None
        try:
            if hasattr(session_result, 'resource_url') and session_result.resource_url:
                resource_url = session_result.resource_url
            elif hasattr(session, 'resource_url') and session.resource_url:
                resource_url = session.resource_url
        except:
            pass

        if resource_url:
            print(f"📱 流化页面 URL: {resource_url}")
            try:
                import webbrowser
                webbrowser.open(resource_url)
                print("✅ 流化页面已打开\n")
            except:
                print(f"   请手动复制以下链接在浏览器中打开: {resource_url}\n")

        # 使用 CDP 直接导航到登录页面（不使用 agent）
        print("=" * 60)
        print("🔐 登录流程")
        print("=" * 60)
        print(f"🚀 正在使用 CDP 导航到{platform_name}登录页面: {login_url}")

        try:
            from playwright.async_api import async_playwright

            # 获取 CDP endpoint URL
            endpoint_url = session.browser.get_endpoint_url()
            if asyncio.iscoroutine(endpoint_url):
                endpoint_url = await endpoint_url

            if endpoint_url:
                async with async_playwright() as p:
                    browser_pw = await p.chromium.connect_over_cdp(endpoint_url)
                    context_pw = browser_pw.contexts[0] if browser_pw.contexts else await browser_pw.new_context()
                    page_pw = context_pw.pages[0] if context_pw.pages else await context_pw.new_page()

                    # 使用 CDP 直接导航
                    await page_pw.goto(login_url, wait_until="domcontentloaded", timeout=30000)
                    print(f"✅ 已导航到登录页面\n")

                    await browser_pw.close()
            else:
                print(f"⚠️ 无法获取 CDP endpoint URL，跳过自动导航")
                print(f"   请在流化页面中手动导航到: {login_url}\n")
        except Exception as e:
            print(f"⚠️ CDP 导航失败: {e}")
            print(f"   请在流化页面中手动导航到: {login_url}\n")
            print(f"   详细错误:\n{traceback.format_exc()}")

        # 提示用户完成登录
        print(f"💡 请在已打开的流化页面中完成登录操作")
        print(f"   登录完成后，请在终端按 Enter 键继续\n")

        try:
            input("👉 登录完成后，请按 Enter 键继续: ")
            print("\n✅ 登录流程完成\n")
            return True
        except (EOFError, KeyboardInterrupt):
            print("\n⚠️ 用户取消操作")
            return False

    except Exception as e:
        print(f"❌ 登录过程中出错: {str(e)}")
        print(f"   详细错误:\n{traceback.format_exc()}")
        return False

    finally:
        # 删除 session 并同步 Context（保存登录状态）
        if session and agent_bay:
            try:
                await asyncio.sleep(2)  # 等待浏览器数据落盘

                # 显式同步 Context（保存登录状态）
                print("\n🔄 正在同步登录状态到 Context...")
                try:
                    sync_result = await session.context.sync()
                    if sync_result.success:
                        print("✅ Context 同步成功")
                    else:
                        print(f"⚠️ Context 同步失败: {sync_result.error_message if hasattr(sync_result, 'error_message') else 'Unknown error'}")
                except Exception as sync_error:
                    print(f"⚠️ Context 同步出错: {sync_error}")

                # 删除 session（不再需要 sync_context=True，因为已经手动同步了）
                delete_result = await agent_bay.delete(session, sync_context=False)
                if delete_result.success:
                    print(f"✅ Session 已删除 (RequestID: {delete_result.request_id})")
                    print("   💡 登录状态已保存，下次运行爬虫时会自动使用已保存的登录状态")
                else:
                    print(f"\n⚠️ 删除 Session 失败: {delete_result.error_message}")
            except Exception as e:
                print(f"\n⚠️ 删除会话/同步 Context 时出错: {e}")
                print(f"   详细错误:\n{traceback.format_exc()}")


def _parse_args():
    import argparse
    p = argparse.ArgumentParser(description="登录目标平台并保存 Browser Context。仅 AGENTBAY_API_KEY 从环境读取，其余传参。")
    p.add_argument("--platform", "-p", default="xhs", help="平台: xhs/weibo/douyin/zhihu")
    p.add_argument("--context-name", "-c", default="sentiment-analysis", help="Browser Context 名称")
    return p.parse_args()


async def main():
    """主函数：仅 AGENTBAY_API_KEY 从 ~/.config/agentbay/api_key 或环境变量读取，platform/context_name 由主 Agent 传参"""
    args = _parse_args()
    api_key = get_api_key()
    if not api_key:
        print("❌ 错误: 未提供 AgentBay API 密钥")
        print("请设置环境变量 AGENTBAY_API_KEY 或创建 ~/.config/agentbay/api_key 文件")
        print("获取 API Key: https://agentbay.console.aliyun.com/service-management")
        sys.exit(1)

    try:
        success = await login_only(
            agentbay_api_key=api_key,
            platform=args.platform,
            context_name=args.context_name,
        )
        if success:
            print("\n" + "=" * 60)
            print("✅ 登录流程完成")
            print("=" * 60)
            sys.exit(0)
        else:
            print("\n" + "=" * 60)
            print("❌ 登录流程失败")
            print("=" * 60)
            sys.exit(1)
    except KeyboardInterrupt:
        print("\n⚠️ 用户中断操作")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ 发生错误: {e}")
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    asyncio.run(main())

#!/usr/bin/env python3
"""
根据情感分析结果生成报告（Markdown/JSON/可选 PDF），供主 Agent 在情感分析完成后调用。
用法：python scripts/report.py --input processed.json [--output-dir output] [--title "报告标题"]
"""
import sys
import json
import argparse
from pathlib import Path

_scripts_dir = Path(__file__).resolve().parent
if str(_scripts_dir) not in sys.path:
    sys.path.insert(0, str(_scripts_dir))

from crawl import generate_report


def main():
    parser = argparse.ArgumentParser(description="根据情感分析结果生成舆情报告")
    parser.add_argument("--input", "-i", required=True, help="processed 结果 JSON 文件路径（主 Agent 按提示词完成情感分析后写入）")
    parser.add_argument("--output-dir", "-o", default="output", help="报告输出目录")
    parser.add_argument("--title", "-t", help="报告标题，可选")
    args = parser.parse_args()

    try:
        with open(args.input, "r", encoding="utf-8") as f:
            processed_results = json.load(f)
    except json.JSONDecodeError as e:
        sys.exit(
            f"输入 JSON 解析失败（{e.msg}，约第 {e.lineno} 行第 {e.colno} 列）。\n"
            "常见原因：内容字段（如 title、content）中含有未转义的双引号 \"。\n"
            "解决：请用程序生成 processed 文件（如 Python：先 json.load 爬取结果，添加 sentiment 后用 json.dump 写入），勿手写或直接粘贴含引号的内容。"
        )

    report = generate_report(
        processed_results=processed_results,
        output_dir=args.output_dir,
        title=args.title,
    )
    print(f"Markdown: {report.get('markdown_path', '')}")
    if report.get("json_path"):
        print(f"JSON: {report['json_path']}")
    if report.get("pdf_path"):
        print(f"PDF: {report['pdf_path']}")


if __name__ == "__main__":
    main()

"""
报告生成模块
提供舆情分析报告的生成功能
"""

from .generator import ReportGenerator
from .templates import ReportTemplate

__all__ = [
    "ReportGenerator",
    "ReportTemplate",
]

"""
报告生成器
生成舆情分析报告
"""
import os
import json
from typing import Dict, Any, List, Optional
from datetime import datetime
from pathlib import Path

from .templates import ReportTemplate


class ReportGenerator:
    """报告生成器"""

    def __init__(self, output_dir: str = "output"):
        """
        初始化报告生成器

        Args:
            output_dir: 输出目录
        """
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        self.template = ReportTemplate()

    def generate_report(
        self,
        processed_results: Dict[str, Any],
        title: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        生成舆情分析报告

        Args:
            processed_results: 处理后的舆情数据
            title: 报告标题，如果为None则自动生成

        Returns:
            包含报告内容和文件路径的字典
        """
        # 生成报告标题
        if title is None:
            platform = processed_results.get("platform_display", "未知平台")
            keyword = processed_results.get("keyword", "") or ", ".join(
                processed_results.get("keywords", [])
            )
            title = f"{platform} - {keyword} 舆情分析"

        # 获取统计数据
        stats = processed_results.get("sentiment_statistics", {})
        results = processed_results.get("results", [])

        # 格式化报告内容
        report_content = self.template.get_default_template().format(
            title=title,
            sentiment_summary=self.template.format_sentiment_summary(stats),
            total_count=stats.get("total_count", 0),
            positive_ratio=stats.get("positive_ratio", 0) * 100,
            negative_ratio=stats.get("negative_ratio", 0) * 100,
            neutral_ratio=stats.get("neutral_ratio", 0) * 100,
            average_score=stats.get("average_score", 0.0),
            platform_summary=self.template.format_platform_summary(processed_results),
            data_source_table=self.template.format_data_source_table(processed_results),
            sentiment_distribution_table=self.template.format_sentiment_distribution_table(stats),
            time_distribution=self._format_time_distribution(results),
            positive_content=self.template.format_content_section(results, "正面"),
            negative_content=self.template.format_content_section(results, "负面"),
            neutral_content=self.template.format_content_section(results, "中性"),
            trend_analysis=self._format_trend_analysis(stats),
            key_topics=self._format_key_topics(results),
            engagement_analysis=self._format_engagement_analysis(results),
            conclusion=self._format_conclusion(processed_results, stats),
            recommendations=self._format_recommendations(processed_results, stats),
            data_appendix=self._format_data_appendix(stats),
            raw_data_stats=self._format_raw_data_stats(processed_results),
            report_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            data_sources=", ".join(processed_results.get("data_sources", [processed_results.get("platform_display", "未知平台")]))
        )

        # 保存报告
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        platform = processed_results.get("platform", "unknown")
        safe_title = "".join(c for c in title if c.isalnum() or c in (" ", "-", "_"))[:50]
        filename = f"{platform}_{safe_title}_{timestamp}.md"
        filepath = self.output_dir / filename

        # 保存Markdown报告
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(report_content)

        # 保存JSON数据
        json_filename = filename.replace(".md", ".json")
        json_filepath = self.output_dir / json_filename
        with open(json_filepath, "w", encoding="utf-8") as f:
            json.dump(processed_results, f, ensure_ascii=False, indent=2)

        # 尝试生成 PDF（图文并茂）
        pdf_path = None
        try:
            from .pdf_export import export_to_pdf, is_pdf_available
            if is_pdf_available():
                pdf_filename = filename.replace(".md", ".pdf")
                pdf_filepath = self.output_dir / pdf_filename
                pdf_path = export_to_pdf(
                    report_content,
                    pdf_filepath,
                    title=title,
                    processed_results=processed_results,
                )
                if pdf_path:
                    print(f"   PDF: {pdf_path}")
        except Exception as e:
            print(f"   PDF 生成跳过: {e}")

        print(f"✅ 报告已生成:")
        print(f"   Markdown: {filepath}")
        print(f"   JSON数据: {json_filepath}")

        result = {
            "success": True,
            "title": title,
            "markdown_path": str(filepath),
            "json_path": str(json_filepath),
            "content": report_content,
            "statistics": stats
        }
        if pdf_path:
            result["pdf_path"] = str(pdf_path)
            result["html_path"] = str(Path(pdf_path).with_suffix(".html"))
        return result

    def _format_time_distribution(self, results: List[Dict[str, Any]]) -> str:
        """格式化时间分布"""
        if not results:
            return "暂无时间数据。"

        # 简单的时间分布统计
        return f"共收集 {len(results)} 条内容，时间范围: 待分析"

    def _format_trend_analysis(self, stats: Dict[str, Any]) -> str:
        """格式化趋势分析"""
        avg_score = stats.get("average_score", 0.0)
        positive_ratio = stats.get("positive_ratio", 0.0)
        negative_ratio = stats.get("negative_ratio", 0.0)

        if avg_score > 0.3:
            trend = "整体舆情趋势偏向正面，用户反馈较为积极。"
        elif avg_score < -0.3:
            trend = "整体舆情趋势偏向负面，需要关注用户反馈中的问题。"
        else:
            trend = "整体舆情趋势较为中性，用户反馈相对平衡。"

        return f"""
根据情感分析结果：
- 平均情感分数: {avg_score:.2f}
- 正面内容占比: {positive_ratio*100:.1f}%
- 负面内容占比: {negative_ratio*100:.1f}%

{trend}
"""

    def _format_key_topics(self, results: List[Dict[str, Any]]) -> str:
        """格式化关键话题"""
        if not results:
            return "暂无话题数据。"

        # 简单提取标题作为话题
        topics = [item.get("title", "") for item in results[:10] if item.get("title")]
        if topics:
            topics_str = "\n".join([f"- {topic}" for topic in topics[:5]])
            return f"主要话题包括：\n{topics_str}"
        else:
            return "暂无话题数据。"

    def _format_engagement_analysis(self, results: List[Dict[str, Any]]) -> str:
        """格式化参与度分析"""
        if not results:
            return "暂无参与度数据。"

        # 统计点赞、转发、评论
        total_likes = sum(item.get("likes", 0) or 0 for item in results)
        total_shares = sum(item.get("shares", 0) or 0 for item in results)
        total_comments = sum(item.get("comments", 0) or 0 for item in results)

        return f"""
- 总点赞数: {total_likes}
- 总转发数: {total_shares}
- 总评论数: {total_comments}
- 平均互动数: {(total_likes + total_shares + total_comments) / len(results):.1f}
"""

    def _format_conclusion(
        self, processed_results: Dict[str, Any], stats: Dict[str, Any]
    ) -> str:
        """格式化结论：优先使用主 Agent 撰写的 agent_summary，否则回退到基于统计的简短结论"""
        agent_summary = processed_results.get("agent_summary") or ""
        if agent_summary and agent_summary.strip():
            return agent_summary.strip()
        avg_score = stats.get("average_score", 0.0)
        total = stats.get("total_count", 0)
        if avg_score > 0.3:
            return f"整体舆情偏向正面，共分析 {total} 条内容，用户反馈积极。"
        if avg_score < -0.3:
            return f"整体舆情偏向负面，共分析 {total} 条内容，需要关注用户反馈中的问题。"
        return f"整体舆情较为中性，共分析 {total} 条内容，用户反馈相对平衡。"

    def _format_recommendations(
        self, processed_results: Dict[str, Any], stats: Dict[str, Any]
    ) -> str:
        """格式化建议：优先使用主 Agent 撰写的 agent_recommendations，否则回退到基于统计的简短建议"""
        agent_rec = processed_results.get("agent_recommendations") or ""
        if agent_rec and agent_rec.strip():
            return agent_rec.strip()
        avg_score = stats.get("average_score", 0.0)
        negative_ratio = stats.get("negative_ratio", 0.0)
        recommendations = []
        if negative_ratio > 0.3:
            recommendations.append("负面反馈占比较高，建议重点关注并采取应对措施。")
        if avg_score < -0.3:
            recommendations.append("整体情感偏向负面，建议加强用户沟通和服务改进。")
        elif avg_score > 0.3:
            recommendations.append("整体情感偏向正面，建议继续保持并放大正面影响。")
        if not recommendations:
            recommendations.append("舆情状况良好，建议持续监测。")
        return "\n".join([f"{i+1}. {rec}" for i, rec in enumerate(recommendations)])

    def _format_data_appendix(self, stats: Dict[str, Any]) -> str:
        """格式化数据附录"""
        return f"""
- 总内容数: {stats.get('total_count', 0)}
- 正面内容: {stats.get('positive_count', 0)}
- 负面内容: {stats.get('negative_count', 0)}
- 中性内容: {stats.get('neutral_count', 0)}
- 平均情感分数: {stats.get('average_score', 0.0):.3f}
"""

    def _format_raw_data_stats(self, processed_results: Dict[str, Any]) -> str:
        """格式化原始数据统计"""
        platform = processed_results.get("platform_display", "未知平台")
        keyword = processed_results.get("keyword", "") or ", ".join(
            processed_results.get("keywords", [])
        )
        crawl_time = processed_results.get("crawl_time", "")
        processed_time = processed_results.get("processed_time", "")

        return f"""
- 数据来源平台: {platform}
- 搜索关键词: {keyword}
- 爬取时间: {crawl_time}
- 处理时间: {processed_time}
"""

"""
PDF 导出模块
将舆情分析报告（Markdown）转为图文并茂的 PDF。
依赖：markdown、weasyprint（可选，缺失时仅跳过 PDF 生成）
中文显示：优先 reporter/fonts/（通用或按平台子目录 windows/mac/linux）→ 本机系统字体（按 OS 选择）→ 下载 WOFF → CDN woff2。
"""
from __future__ import annotations

import base64
import sys
import tempfile
import urllib.request
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

# 用于 @font-face 的 CJK 字体族名，正文统一使用该字体
_CJK_FONT_FAMILY = "ReportCJK"

# 各平台系统 CJK 字体路径（路径, format），按优先级排列；Python 根据 sys.platform 选用
_SYSTEM_FONT_CANDIDATES: Dict[str, List[Tuple[Path, str]]] = {
    "darwin": [
        (Path("/System/Library/Fonts/PingFang.ttc"), "truetype"),
        (Path("/System/Library/Fonts/Supplemental/Songti.ttc"), "truetype"),
        (Path("/System/Library/Fonts/Supplemental/STHeiti Medium.ttc"), "truetype"),
        (Path("/Library/Fonts/Arial Unicode.ttf"), "truetype"),
    ],
    "win32": [
        (Path("C:/Windows/Fonts/msyh.ttc"), "truetype"),   # 微软雅黑
        (Path("C:/Windows/Fonts/simsun.ttc"), "truetype"), # 宋体
        (Path("C:/Windows/Fonts/simhei.ttf"), "truetype"), # 黑体
    ],
    "linux": [
        (Path("/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc"), "truetype"),
        (Path("/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc"), "truetype"),
        (Path("/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc"), "truetype"),
    ],
}
# 兼容 win64 等
if sys.platform == "win32":
    _PLATFORM_KEY = "win32"
elif sys.platform == "darwin":
    _PLATFORM_KEY = "darwin"
else:
    _PLATFORM_KEY = "linux"

# Markdown → HTML（必选）
try:
    import markdown
    MARKDOWN_AVAILABLE = True
except ImportError:
    MARKDOWN_AVAILABLE = False

# HTML → PDF（可选，需系统安装 Pango/Cairo）
try:
    from weasyprint import HTML as WeasyHTML
    from weasyprint.text.fonts import FontConfiguration
    WEASYPRINT_AVAILABLE = True
except (ImportError, OSError):
    WEASYPRINT_AVAILABLE = False


def _markdown_to_html(md_content: str) -> str:
    """将 Markdown 转为 HTML 片段（不含 document 包装）。"""
    if not MARKDOWN_AVAILABLE:
        raise RuntimeError("请安装 markdown: pip install markdown")
    html_body = markdown.markdown(
        md_content,
        extensions=["tables", "fenced_code", "nl2br"],
        extension_configs={"tables": {}},
    )
    return html_body


def _sentiment_chart_svg(stats: Dict[str, Any]) -> str:
    """根据情感统计生成简单 SVG 柱状图（图文并茂）。
    将「非常正面」合并入「正面」、「非常负面」合并入「负面」，与执行摘要比例一致。
    """
    total = stats.get("total_count", 0)
    if total <= 0:
        return ""

    dist = stats.get("sentiment_distribution", {})
    # 合并细粒度标签：正面=正面+非常正面，负面=负面+非常负面，其余归中性
    if any(k in dist for k in ("正面", "负面", "中性", "非常正面", "非常负面")):
        pos = dist.get("正面", 0) + dist.get("非常正面", 0)
        neg = dist.get("负面", 0) + dist.get("非常负面", 0)
        neu = total - pos - neg
        if neu < 0:
            neu = dist.get("中性", 0)
        labels_map = [
            ("正面", pos, "#22c55e"),
            ("负面", neg, "#ef4444"),
            ("中性", neu, "#94a3b8"),
        ]
    else:
        pos = dist.get("positive", 0)
        neg = dist.get("negative", 0)
        neu = total - pos - neg
        if neu < 0:
            neu = dist.get("neutral", 0)
        labels_map = [
            ("正面", pos, "#22c55e"),
            ("负面", neg, "#ef4444"),
            ("中性", neu, "#94a3b8"),
        ]

    max_count = max((c for _, c, _ in labels_map), default=1) or 1
    w, h = 400, 140
    bar_h = 24
    gap = 12
    margin = 40

    parts = [
        f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {w} {h}" width="{w}" height="{h}">',
        f'<style>.bar-label{{font-family:{_CJK_FONT_FAMILY},sans-serif;font-size:12px;fill:#374151}}.bar-val{{font-family:{_CJK_FONT_FAMILY},sans-serif;font-size:11px;fill:#6b7280}}</style>',
    ]
    y = margin
    for label, count, color in labels_map:
        ratio = count / max_count
        bar_w = max(4, int(200 * ratio))
        parts.append(f'<rect x="{margin}" y="{y}" width="{bar_w}" height="{bar_h - 2}" rx="4" fill="{color}" opacity="0.85"/>')
        parts.append(f'<text class="bar-label" x="{margin}" y="{y + bar_h - 6}">{label}</text>')
        parts.append(f'<text class="bar-val" x="{margin + 210}" y="{y + bar_h - 6}">{count} ({100*count/total:.1f}%)</text>')
        y += bar_h + gap

    parts.append("</svg>")
    return "".join(parts)


def _font_path_to_file_url(path: Path) -> str:
    """将字体路径转为 WeasyPrint 可用的 file:// URL（含 Windows 盘符）。"""
    path = path.resolve()
    # Windows: file:///C:/Windows/Fonts/... ；Unix: file:///usr/share/...
    posix = path.as_posix()
    if posix.startswith("/"):
        return f"file://{posix}"
    return f"file:///{posix}"


def _get_cjk_font_css() -> str:
    """
    生成用于 PDF 的 CJK @font-face，避免中文乱码。
    优先级：1) reporter/fonts/ 通用字体 或 reporter/fonts/windows|mac|linux/ 按系统；
            2) 本机系统字体（按 Windows/macOS/Linux 选择）；
            3) 下载 Noto WOFF；4) CDN woff2。
    """
    report_dir = Path(__file__).resolve().parent
    fonts_dir = report_dir / "fonts"

    # 1a) 项目内 reporter/fonts/ 通用字体（任意 .ttf/.otf）
    if fonts_dir.is_dir():
        for name in ("NotoSansSC-Regular.otf", "NotoSansSC-Regular.ttf", "SourceHanSansSC-Regular.otf", "SimSun.ttf", "SimSun.otf"):
            path = fonts_dir / name
            if path.is_file():
                try:
                    data = path.read_bytes()
                    b64 = base64.b64encode(data).decode("ascii")
                    fmt = "opentype" if path.suffix.lower() == ".otf" else "truetype"
                    return f"""
        @font-face {{
            font-family: '{_CJK_FONT_FAMILY}';
            src: url(data:font/{fmt};base64,{b64}) format('{fmt}');
            font-weight: normal;
            font-style: normal;
        }}"""
                except Exception:
                    continue

    # 1b) 按系统使用 reporter/fonts/windows、reporter/fonts/mac、reporter/fonts/linux 下字体
    platform_subdir = {"win32": "windows", "darwin": "mac", "linux": "linux"}.get(_PLATFORM_KEY, "linux")
    platform_fonts_dir = fonts_dir / platform_subdir
    if platform_fonts_dir.is_dir():
        for ext in (".ttf", ".otf", ".ttc"):
            for path in sorted(platform_fonts_dir.glob(f"*{ext}")):
                if path.is_file():
                    try:
                        if path.suffix.lower() in (".ttf", ".otf"):
                            data = path.read_bytes()
                            b64 = base64.b64encode(data).decode("ascii")
                            fmt = "opentype" if path.suffix.lower() == ".otf" else "truetype"
                            return f"""
        @font-face {{
            font-family: '{_CJK_FONT_FAMILY}';
            src: url(data:font/{fmt};base64,{b64}) format('{fmt}');
            font-weight: normal;
            font-style: normal;
        }}"""
                        else:
                            url = _font_path_to_file_url(path)
                            return f"""
        @font-face {{
            font-family: '{_CJK_FONT_FAMILY}';
            src: url('{url}') format('truetype');
            font-weight: normal;
            font-style: normal;
        }}"""
                    except Exception:
                        continue

    # 2) 本机系统字体（按 Windows / macOS / Linux 选择）
    for path, fmt in _SYSTEM_FONT_CANDIDATES.get(_PLATFORM_KEY, []):
        if path.is_file():
            url = _font_path_to_file_url(path)
            return f"""
        @font-face {{
            font-family: '{_CJK_FONT_FAMILY}';
            src: url('{url}') format('{fmt}');
            font-weight: normal;
            font-style: normal;
        }}"""

    # 3) 下载 Noto Sans SC 到临时文件（CDN 仅有 woff/woff2；WeasyPrint 对远程 woff2 支持差）
    woff_url = (
        "https://cdn.jsdelivr.net/npm/@fontsource/noto-sans-sc@5.0.0/files/"
        "noto-sans-sc-5-400-normal.woff"
    )
    try:
        with tempfile.NamedTemporaryFile(suffix=".woff", delete=False) as f:
            req = urllib.request.Request(woff_url, headers={"User-Agent": "Mozilla/5.0 (compatible; WeasyPrint)"})
            with urllib.request.urlopen(req, timeout=15) as resp:
                f.write(resp.read())
            local_path = Path(f.name)
        return f"""
        @font-face {{
            font-family: '{_CJK_FONT_FAMILY}';
            src: url('file://{local_path}') format('woff');
            font-weight: normal;
            font-style: normal;
        }}"""
    except Exception:
        pass

    # 4) 最后回退：CDN woff2（WeasyPrint 可能无法正确加载，仅作兜底）
    woff2_url = (
        "https://cdn.jsdelivr.net/npm/@fontsource/noto-sans-sc@5.0.0/files/"
        "noto-sans-sc-5-400-normal.woff2"
    )
    return f"""
        @font-face {{
            font-family: '{_CJK_FONT_FAMILY}';
            src: url('{woff2_url}') format('woff2');
            font-weight: normal;
            font-style: normal;
        }}"""


def _wrap_html_document(html_body: str, title: str, chart_svg: str = "", cjk_font_css: str = "") -> str:
    """包装成完整 HTML 文档，带样式与可选图表。"""
    chart_block = ""
    if chart_svg:
        # 将 SVG 转为可嵌入的 data 或内联（避免外部依赖）
        chart_block = f"""
        <div class="chart-wrap">
            <h3>情感分布示意</h3>
            <div class="chart-inner">{chart_svg}</div>
        </div>
        """

    # 正文与图表统一使用 CJK 字体，避免乱码
    body_font = f"'{_CJK_FONT_FAMILY}', 'PingFang SC', 'Microsoft YaHei', 'SimSun', sans-serif"
    return f"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8"/>
    <title>{_escape_html(title)}</title>
    <style>
        {cjk_font_css}
        @page {{ size: A4; margin: 2cm; }}
        body, h1, h2, h3, table, th, td, blockquote, ul, ol, .chart-wrap, .chart-inner {{ font-family: {body_font}; }}
        body {{ font-size: 12pt; line-height: 1.6; color: #1f2937; }}
        h1 {{ font-size: 18pt; color: #111; border-bottom: 2px solid #3b82f6; padding-bottom: 6px; margin-top: 0; }}
        h2 {{ font-size: 14pt; color: #1e40af; margin-top: 1.2em; }}
        h3 {{ font-size: 12pt; color: #374151; margin-top: 1em; }}
        table {{ border-collapse: collapse; width: 100%; margin: 0.5em 0; font-size: 11pt; }}
        th, td {{ border: 1px solid #e5e7eb; padding: 6px 10px; text-align: left; }}
        th {{ background: #f3f4f6; font-weight: 600; }}
        blockquote {{ margin: 0.5em 0; padding: 0.5em 1em; background: #f9fafb; border-left: 4px solid #3b82f6; color: #4b5563; }}
        .chart-wrap {{ margin: 1em 0; padding: 1em; background: #f8fafc; border-radius: 8px; }}
        .chart-wrap h3 {{ margin-top: 0; }}
        .chart-inner {{ margin-top: 8px; }}
        ul, ol {{ margin: 0.4em 0; padding-left: 1.5em; }}
        strong {{ color: #111; }}
        hr {{ border: none; border-top: 1px solid #e5e7eb; margin: 1em 0; }}
    </style>
</head>
<body>
    <h1>{_escape_html(title)}</h1>
    {chart_block}
    <div class="content">
    {html_body}
    </div>
</body>
</html>
"""


def _escape_html(s: str) -> str:
    return (
        s.replace("&", "&amp;")
        .replace("<", "&lt;")
        .replace(">", "&gt;")
        .replace('"', "&quot;")
    )


def export_to_pdf(
    md_content: str,
    output_pdf_path: Path,
    title: str = "舆情分析报告",
    processed_results: Optional[Dict[str, Any]] = None,
    save_html: bool = True,
) -> Optional[Path]:
    """
    将 Markdown 报告内容导出为 PDF（图文并茂）。

    Args:
        md_content: 报告 Markdown 全文
        output_pdf_path: 输出 PDF 路径
        title: 报告标题，用于 HTML 标题与页眉
        processed_results: 可选，含 sentiment_statistics 时会在正文前插入情感分布图
        save_html: 是否同时保存中间生成的 HTML（与 PDF 同目录、同名 .html），默认 True

    Returns:
        成功时返回 output_pdf_path，依赖缺失或失败时返回 None
    """
    if not MARKDOWN_AVAILABLE:
        print("⚠ 未安装 markdown，无法生成 PDF。请运行: pip install markdown")
        return None

    html_body = _markdown_to_html(md_content)
    chart_svg = ""
    if processed_results:
        stats = processed_results.get("sentiment_statistics", {})
        if stats:
            chart_svg = _sentiment_chart_svg(stats)

    cjk_font_css = _get_cjk_font_css()
    full_html = _wrap_html_document(html_body, title, chart_svg, cjk_font_css=cjk_font_css)

    if not WEASYPRINT_AVAILABLE:
        print("⚠ 未安装或无法加载 weasyprint（需系统安装 Pango/Cairo），跳过 PDF 生成。")
        print("   macOS: brew install cairo pango gdk-pixbuf")
        print("   pip: pip install weasyprint")
        return None

    output_pdf_path = Path(output_pdf_path)
    output_pdf_path.parent.mkdir(parents=True, exist_ok=True)

    # 可选：保存中间生成的 HTML（与 PDF 同名 .html）
    if save_html:
        html_path = output_pdf_path.with_suffix(".html")
        try:
            html_path.write_text(full_html, encoding="utf-8")
            print(f"   HTML: {html_path}")
        except Exception as e:
            print(f"   HTML 保存跳过: {e}")

    try:
        font_config = FontConfiguration()
        doc = WeasyHTML(string=full_html, base_url=str(Path.cwd()))
        doc.write_pdf(
            output_pdf_path,
            font_config=font_config,
            presentational_hints=True,
        )
        return output_pdf_path
    except Exception as e:
        print(f"⚠ PDF 生成失败: {e}")
        return None


def is_pdf_available() -> bool:
    """是否具备 PDF 导出能力（markdown + weasyprint 均可用）。"""
    return bool(MARKDOWN_AVAILABLE and WEASYPRINT_AVAILABLE)