group_fqcd_jr/app/service/runtime_config_service.py

from datetime import UTC, datetime

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.infrastructure.db import SessionFactory
from app.model.configuration import (
    AgentIntentConfig,
    ConfigRelease,
    PlatformConfigItem,
    PromptTemplateVersion,
)
from app.service.intent_classifier import IntentConfigEntry


class RuntimeConfigService:
    """Reads only configuration rows bound to the caller's released version."""

    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def prompt(
        self, release_id: int, prompt_code: str, task_type: str, agent_type: str | None
    ) -> PromptTemplateVersion | None:
        result = await self.session.scalar(
            select(PromptTemplateVersion).where(
                PromptTemplateVersion.release_id == release_id,
                PromptTemplateVersion.prompt_code == prompt_code,
                PromptTemplateVersion.task_type == task_type,
                (PromptTemplateVersion.agent_type == agent_type)
                | PromptTemplateVersion.agent_type.is_(None),
            )
        )
        return result

    async def intent(self, agent_type: str, intent_code: str) -> AgentIntentConfig | None:
        result = await self.session.scalar(
            select(AgentIntentConfig).where(
                AgentIntentConfig.agent_type == agent_type,
                AgentIntentConfig.intent_code == intent_code,
                AgentIntentConfig.status == "active",
            )
        )
        return result

    async def active_intents(self, agent_type: str) -> tuple[IntentConfigEntry, ...]:
        """读取某 Agent 当前生效的意图配置，供意图分类链路使用。

        生效判定与 `intent()` 一致：`status='active'`；另外尊重 `effective_at/expire_at`
        的有效期窗口（未填写视为立即生效、不过期）。该表通过生成列唯一键
        `uk_intent_config_active_one` 保证同一 `agent_type:intent_code` 最多一个 active
        版本，因此按 intent_code 匹配不会歧义；仍按 `priority` 排序以保证结果稳定。
        """
        now = datetime.now(UTC).replace(tzinfo=None)
        rows = await self.session.scalars(
            select(AgentIntentConfig)
            .where(
                AgentIntentConfig.agent_type == agent_type,
                AgentIntentConfig.status == "active",
                (AgentIntentConfig.effective_at.is_(None))
                | (AgentIntentConfig.effective_at <= now),
                (AgentIntentConfig.expire_at.is_(None)) | (AgentIntentConfig.expire_at > now),
            )
            .order_by(AgentIntentConfig.priority, AgentIntentConfig.intent_code)
        )
        return tuple(self._entry(row) for row in rows)

    async def collection_routes(self, agent_type: str) -> dict[str, str]:
        """读取某 Agent 当前生效的「意图 → 收窄集合」路由。

        **与 `active_intents()` 使用同一套生效判定**（`status='active'` + 有效期窗口），
        因此"配置里生效的意图"与"路由里生效的意图"不会变成两套口径。

        `collection_routes` 是**既有 JSON 列**（`app/model/configuration.py:94`），本方法
        是对它的**首次消费**（此前全仓零消费，属"有列无消费"）。列里允许是字符串、
        字符串数组或 `{"collection": ...}` 形态，统一投影成 `{intent_code: collection}`。

        **拿不到可用集合名的行直接跳过** —— 缺失时由调用方**回落代码常量**，
        而不是塞一个空值把整条路由打断。这是本函数唯一的降级口径。
        """
        now = datetime.now(UTC).replace(tzinfo=None)
        rows = await self.session.scalars(
            select(AgentIntentConfig)
            .where(
                AgentIntentConfig.agent_type == agent_type,
                AgentIntentConfig.status == "active",
                (AgentIntentConfig.effective_at.is_(None))
                | (AgentIntentConfig.effective_at <= now),
                (AgentIntentConfig.expire_at.is_(None)) | (AgentIntentConfig.expire_at > now),
                AgentIntentConfig.collection_routes.is_not(None),
            )
            .order_by(AgentIntentConfig.priority, AgentIntentConfig.intent_code)
        )
        routes: dict[str, str] = {}
        for row in rows:
            name = _first_collection_name(row.collection_routes)
            if name:
                # 同一 intent_code 理论上只有一个 active 版本（唯一键保证）；
                # `setdefault` 是防御性的：真出现两条也只认优先级最高的那条。
                routes.setdefault(row.intent_code, name)
        return routes

    @staticmethod
    def _entry(row: AgentIntentConfig) -> IntentConfigEntry:
        return IntentConfigEntry(
            intent_code=row.intent_code,
            intent_name=row.intent_name or "",
            description=row.description,
            examples=tuple(str(item) for item in (row.examples or ())),
            classifier_instruction=row.classifier_instruction,
            confidence_threshold=float(row.confidence_threshold),
        )

    async def allowed_tools(
        self, release_id: int, agent_type: str, intent_code: str
    ) -> tuple[str, ...]:
        config_key = f"{agent_type}:{intent_code}"
        item = await self.session.scalar(
            select(PlatformConfigItem).where(
                PlatformConfigItem.release_id == release_id,
                PlatformConfigItem.namespace == "agent_tools",
                PlatformConfigItem.config_key == config_key,
            )
        )
        if item is None:
            return ()
        raw_tools = item.value_json.get("allowed_tools", [])
        if not isinstance(raw_tools, list) or not all(isinstance(tool, str) for tool in raw_tools):
            raise ValueError("invalid tool whitelist configuration")
        return tuple(raw_tools)

    async def fund_quote(self, release_id: int) -> dict[str, object]:
        """读取已发布行情配置；缺失时返回空映射，由 Service 使用安全默认值。"""
        item = await self.session.scalar(
            select(PlatformConfigItem).where(
                PlatformConfigItem.release_id == release_id,
                PlatformConfigItem.namespace == "fund_market",
                PlatformConfigItem.config_key == "default",
            )
        )
        return item.value_json if item is not None else {}


async def load_fund_quote_config() -> dict[str, object]:
    """读取当前激活行情配置；配置中心不可用时交给调用方使用默认值。"""
    async with SessionFactory() as session:
        release = await session.scalar(
            select(ConfigRelease).where(ConfigRelease.status == "active")
        )
        if release is None:
            return {}
        return await RuntimeConfigService(session).fund_quote(release.id)


def _first_collection_name(value: object) -> str:
    """从 `collection_routes` 的 JSON 值里取**第一个可用集合名**；取不到返回空串。

    兼容三种已见过的写法，避免把列格式钉死（列是既有的，历史/人工写入的形态不可控）：
      - `"fin_faq_collection"`            —— 裸字符串
      - `["fin_faq_collection", ...]`     —— 数组（取首个非空）
      - `{"collection": "fin_faq_collection"}`
    """
    if isinstance(value, str):
        return value.strip()
    if isinstance(value, (list, tuple)):
        for item in value:
            name = _first_collection_name(item)
            if name:
                return name
        return ""
    if isinstance(value, dict):
        for key in ("collection", "name", "target"):
            if key in value:
                name = _first_collection_name(value[key])
                if name:
                    return name
        return ""
    return ""


async def load_collection_routes(agent_type: str) -> dict[str, str]:
    """「意图 → 收窄集合」路由的运行期装载器。

    与 `load_active_intent_configs()` 并列、互不影响；**读不到（无 active 行、列全空、
    配置中心不可用）一律返回空映射**，由调用方回落代码常量
    （`customer_service.NARROW_COLLECTION_BY_INTENT`）—— 即"配置优先、缺失回落"。
    """
    async with SessionFactory() as session:
        return await RuntimeConfigService(session).collection_routes(agent_type)


async def load_active_intent_configs(agent_type: str) -> tuple[IntentConfigEntry, ...]:
    """意图分类链路的运行期配置装载器：只读 `agent_intent_config` 的 active 行。

    该表不绑定 `config_release`（没有 `release_id` 外键），因此这里按 Agent 直接读取；
    没有配置时返回空元组，分类退化为代码声明的意图清单——**行为与接入前一致**。
    """
    async with SessionFactory() as session:
        return await RuntimeConfigService(session).active_intents(agent_type)


async def load_active_prompt(
    prompt_code: str, task_type: str, agent_type: str
) -> PromptTemplateVersion | None:
    """读取当前生效版本里的提示词；没有 active 版本或没有该提示词时返回 None。

    `prompt_template_version` 绑定 `release_id`，所以必须先定位 active 的 `config_release`
    ——这也正是「提示词变更要经过审核与发布」的落地方式：改话术走发布流程，而不是改代码。

    返回 None 是**正常路径**而非异常：调用方（Agent）据此回落到代码内置的默认提示词，
    保证即使配置中心还没发布过这条提示词，功能也能工作。
    """
    async with SessionFactory() as session:
        release = await session.scalar(
            select(ConfigRelease).where(ConfigRelease.status == "active")
        )
        if release is None:
            return None
        return await RuntimeConfigService(session).prompt(
            release.id, prompt_code, task_type, agent_type
        )