deerflow-code/offline-backend-20260512/backend/app/gateway/routers/_agentfx_seed.py

"""Seeder for the built-in agentfx analyst agent and its report skills.

agentfx 任务深链（新系统 iframe 嵌入的智能体问答）固定使用内置智能体
``agentfx-analyst``：根据任务详情与用户输入搜集信息，把每次检索 JSON 落盘后
用 ``task-report-*`` 技能函数转换成 14 类 ``task-reports.json``，再写 md 研判
报告，总结后经前端「入库」协助卡或 ``task-report-import`` 写入任务详情。

种子内容与 ``forced-research-responder`` 同一套模式：
- ``_agentfx_seed_assets/agentfx-analyst/{config.yaml, SOUL.md}`` → 复制到
  ``{DEER_FLOW_HOME}/agents/agentfx-analyst/``（目录存在则不动，尊重现场修改），
  由启动流程的 ``_sync_legacy_agents`` upsert 进 ``agents`` 表（user_id IS NULL）。
- ``_agentfx_seed_assets/skills/task-report-{enemy,our,env,judge,import}/`` →
  复制到 ``skills/public/``（目录存在则不动）；共用引擎
  ``task-report-lib/``（无 SKILL.md，不进技能列表）一并补建。
  ``skills/`` 被 .gitignore 排除，public 类技能在 extensions_config 缺省即启用。

必须在 ``_sync_legacy_agents`` **之前**调用。
"""

from __future__ import annotations

import logging
import shutil
from pathlib import Path

from deerflow.config.paths import get_paths

logger = logging.getLogger(__name__)

AGENTFX_AGENT_ID = "agentfx-analyst"
TASK_REPORT_IMPORT_SKILL_NAME = "task-report-import"
TASK_REPORT_LIB_NAME = "task-report-lib"
TASK_REPORT_SKILL_NAMES: tuple[str, ...] = (
    "task-report-import",
    "task-report-enemy",
    "task-report-our",
    "task-report-env",
    "task-report-judge",
)

_ASSETS_DIR = Path(__file__).parent / "_agentfx_seed_assets"
_SEED_FILES: tuple[str, ...] = ("config.yaml", "SOUL.md")


def _ensure_agentfx_agent_dir() -> bool:
    """Create the built-in agent directory when missing; never overwrite it."""
    try:
        agent_dir = get_paths().agent_dir(AGENTFX_AGENT_ID)
        if agent_dir.exists():
            return False

        src_dir = _ASSETS_DIR / AGENTFX_AGENT_ID
        if not src_dir.is_dir():
            logger.error(
                "Missing seed assets for agent '%s' (expected at %s)",
                AGENTFX_AGENT_ID,
                src_dir,
            )
            return False

        agent_dir.mkdir(parents=True, exist_ok=True)
        for filename in _SEED_FILES:
            src = src_dir / filename
            if src.exists():
                (agent_dir / filename).write_bytes(src.read_bytes())
        logger.info("Seeded built-in agent '%s'", AGENTFX_AGENT_ID)
        return True
    except Exception:
        logger.exception("Failed to seed built-in agent '%s'", AGENTFX_AGENT_ID)
        return False


def _skills_public_dir() -> Path:
    from deerflow.skills.storage import get_or_new_skill_storage

    return Path(get_or_new_skill_storage().get_skills_root_path()) / "public"


def _copy_tree_if_missing(src_dir: Path, dest_dir: Path, label: str) -> bool:
    if dest_dir.exists():
        return False
    shutil.copytree(src_dir, dest_dir)
    logger.info("Seeded built-in %s '%s'", label, dest_dir.name)
    return True


def _ensure_task_report_skills() -> None:
    """Seed convert/import skills + shared lib into ``skills/public/`` when missing."""
    try:
        public_dir = _skills_public_dir()
        public_dir.mkdir(parents=True, exist_ok=True)

        lib_src = _ASSETS_DIR / "skills" / TASK_REPORT_LIB_NAME
        if not (lib_src / "convert_lib.py").is_file():
            logger.error(
                "Missing seed assets for '%s' (expected convert_lib.py at %s)",
                TASK_REPORT_LIB_NAME,
                lib_src,
            )
        else:
            _copy_tree_if_missing(lib_src, public_dir / TASK_REPORT_LIB_NAME, "lib")

        for name in TASK_REPORT_SKILL_NAMES:
            src_dir = _ASSETS_DIR / "skills" / name
            if not (src_dir / "SKILL.md").exists():
                logger.error(
                    "Missing seed assets for skill '%s' (expected at %s)",
                    name,
                    src_dir,
                )
                continue
            _copy_tree_if_missing(src_dir, public_dir / name, "skill")
    except Exception:
        logger.exception("Failed to seed agentfx task-report skills")


def _ensure_task_report_import_skill() -> bool:
    """Backward-compatible alias: seed the whole task-report skill set."""
    before = None
    try:
        dest = _skills_public_dir() / TASK_REPORT_IMPORT_SKILL_NAME
        before = dest.exists()
    except Exception:
        before = None
    _ensure_task_report_skills()
    return before is False


def ensure_agentfx_agent() -> bool:
    """Seed the agentfx analyst agent + report skills (idempotent, non-fatal).

    Returns True if the agent directory was just created.
    """
    _ensure_task_report_skills()
    return _ensure_agentfx_agent_dir()


__all__ = [
    "AGENTFX_AGENT_ID",
    "TASK_REPORT_IMPORT_SKILL_NAME",
    "TASK_REPORT_LIB_NAME",
    "TASK_REPORT_SKILL_NAMES",
    "ensure_agentfx_agent",
]