MoviePilot/app/db/oper/plugininstance.py

"""
插件实例的数据访问原语。
"""

from __future__ import annotations

import copy
from datetime import datetime, timezone
from typing import Any, Optional, Union

from sqlalchemy import JSON, delete, or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Session

from app.db.base import DbOper
from app.db.models.plugininstance import PluginInstance
from app.schemas.plugin import validate_plugin_id


def _now() -> str:
    """
    返回实例行使用的 ISO-8601 时间戳。
    """
    return datetime.now(timezone.utc).isoformat()


def _apply_config_data(
    session: Union[Session, AsyncSession],
    record: Optional[PluginInstance],
    *,
    instance_id: str,
    source_plugin_id: str,
    config_data: Any,
) -> Optional[bool]:
    """
    把业务参数写进已查出的行，行不存在则按本次身份建出一行。

    同步与异步写入只有查询方式不同，落库决策共用这一份：两边各写一份会让
    「值未变化不写」和「建行用哪个源插件」在两条路径上悄悄分叉。

    :param session: 调用方事务所持的会话，同步与异步的 ``add`` 都是同步方法
    :param record: 已查出的实例行，None 表示该实例尚无行
    :param instance_id: 实例 ID
    :param source_plugin_id: 该行不存在时用于建行的源插件 ID
    :param config_data: 业务参数
    :return: True 已写入，None 值未变化无需写入
    """
    if record is None:
        now = _now()
        session.add(
            PluginInstance(
                instance_id=instance_id,
                source_plugin_id=source_plugin_id,
                config_data=copy.deepcopy(config_data),
                created_at=now,
                updated_at=now,
            )
        )
        return True
    if record.config_data == config_data:
        return None
    record.config_data = copy.deepcopy(config_data)
    record.updated_at = _now()
    return True


class PluginInstanceOper(DbOper):
    """
    在调用方 Session 或独占事务中查询并暂存插件实例。

    一行由 ``instance_id`` 唯一确定，``source_plugin_id`` 指向提供代码的源插件；
    两者相等即源插件本体，不等即共享其源码的分身。
    """

    def get(self, instance_id: str) -> Optional[PluginInstance]:
        """
        按实例 ID 查询单行，分身与本体共用同一张表和同一个查询。
        """
        return self._execute_sync_query(
            lambda session: PluginInstance.get_by_instance_id(session, instance_id)
        )

    def list_by_source(self, source_plugin_id: str) -> list[PluginInstance]:
        """
        按源插件 ID 列举其全部实例，含分身与本体。
        """
        return list(
            self._execute_sync_query(
                lambda session: PluginInstance.list_by_source_plugin_id(
                    session, source_plugin_id
                )
            )
            or []
        )

    def list_all(self) -> list[PluginInstance]:
        """
        列举全部实例，供目录投影与兜底导入判空使用。
        """
        return list(
            self._execute_sync_query(
                lambda session: session.execute(select(PluginInstance)).scalars().all()
            )
        )

    def list_enabled(self) -> list[PluginInstance]:
        """
        列举应当被实例化并启动的实例，供运行期批量装载使用。
        """
        return list(
            self._execute_sync_query(
                lambda session: session.execute(
                    select(PluginInstance).where(PluginInstance.is_enabled.is_(True))
                ).scalars().all()
            )
        )

    def save(self, **fields: Any) -> PluginInstance:
        """
        按 ``instance_id`` 新增或更新一行，只写入本次给出的列。

        查询与写入收在同一事务内，避免两个并发的首次写入各自读到空、双双插入而
        撞上 ``instance_id`` 唯一键。

        已存在的行不允许改写 ``source_plugin_id``：它连同 ``instance_id`` 构成这一行
        的身份，改写会把整行的归属换掉，同时把本体与分身的角色判定一并翻转，只可能
        来自调用方错把分身自身实例 ID 当作源插件 ID 使用，因而在持久化层直接拒绝，
        不依赖上层纪律。

        :param fields: 实例字段，须含 ``instance_id``
        :return: 写入后的实例行
        :raise ValueError: 已存在的行与本次写入的 ``source_plugin_id`` 不一致
        """
        now = _now()
        instance_id = fields["instance_id"]

        def stage(session: Session) -> PluginInstance:
            """
            在同一事务内查询并新增或更新实例行。
            """
            existing = PluginInstance.get_by_instance_id(session, instance_id)
            if existing is None:
                record = PluginInstance(**fields, created_at=now, updated_at=now)
                session.add(record)
                return record
            incoming_source = fields.get("source_plugin_id")
            if incoming_source is not None and existing.source_plugin_id != incoming_source:
                raise ValueError(
                    f"插件实例 {instance_id} 已归属于 {existing.source_plugin_id}，"
                    f"不能改写为 {incoming_source}"
                )
            for key, value in {**fields, "updated_at": now}.items():
                setattr(existing, key, value)
            return existing

        return self._execute_sync_write(stage)

    def delete(self, instance_id: str) -> bool:
        """
        按实例 ID 删除整行，返回删除前是否存在。

        这会连同该实例的业务参数与展示信息一起清除：分身的存在本身就由这一行表达，
        删行即卸载分身。
        """

        def stage(session: Session) -> bool:
            """
            在同一事务内查询并删除，避免读写跨两个独立事务。
            """
            existing = PluginInstance.get_by_instance_id(session, instance_id)
            if existing is None:
                return False
            session.delete(existing)
            return True

        return bool(self._execute_sync_write(stage))

    def purge_invalid_empty_host(self, instance_id: str) -> bool:
        """
        原子删除非法 ID 的空本体行，不接触插件文件或其它数据表。

        该路径专门服务历史迁移残留：只有 ID 非法、行是未启用本体、没有展示或日志
        覆盖且配置为 SQL NULL/JSON null 时才删除。所有条件进入同一条 DELETE，避免
        检查后并发保存配置导致误删；空对象仍是一份配置，不能当作无配置回收。
        """
        try:
            validate_plugin_id(instance_id)
        except ValueError:
            pass
        else:
            return False

        def stage(session: Session) -> bool:
            """
            在调用方事务中删除满足历史空行条件的记录。
            """
            result = session.execute(
                delete(PluginInstance).where(
                    PluginInstance.instance_id == instance_id,
                    PluginInstance.source_plugin_id == instance_id,
                    PluginInstance.is_enabled.is_(False),
                    PluginInstance.is_default_target.is_(False),
                    or_(
                        PluginInstance.config_data.is_(None),
                        PluginInstance.config_data == JSON.NULL,
                    ),
                    *(
                        or_(field.is_(None), field == "")
                        for field in (
                            PluginInstance.plugin_name,
                            PluginInstance.plugin_desc,
                            PluginInstance.plugin_icon,
                            PluginInstance.log_level,
                            PluginInstance.log_expires_at,
                        )
                    ),
                ).execution_options(synchronize_session=False)
            )
            return bool(result.rowcount)

        return bool(self._execute_sync_write(stage))

    def set_enabled(self, *, instance_id: str, is_enabled: bool) -> bool:
        """
        写入启用位，它同时就是卸载与恢复的开关。

        置假即卸载：业务参数与展示信息原样留在这一行等待再次启用，因而不需要另设
        一个卸载时间列。同时清掉两项只对在册实例才有意义的状态——默认调用目标置位，
        否则未指定实例的外部调用会被路由到一个不会被实例化的实例；日志等级覆盖，
        它带失效时间、本就是临时调试设置，而运行期在停用时会同步清掉进程内的等级
        覆盖表，库里留着只会让两边不一致。

        :param instance_id: 实例 ID
        :param is_enabled: 目标启用状态
        :return: 该行是否存在
        """

        def stage(session: Session) -> bool:
            """
            在同一事务内写入启用位，停用时一并清掉仅对在册实例有意义的状态。
            """
            record = PluginInstance.get_by_instance_id(session, instance_id)
            if record is None:
                return False
            record.is_enabled = is_enabled
            if not is_enabled:
                record.is_default_target = False
                record.log_level = None
                record.log_expires_at = None
            record.updated_at = _now()
            return True

        return bool(self._execute_sync_write(stage))

    def set_default_target(self, source_plugin_id: str, instance_id: str) -> bool:
        """
        原子地把某源插件的默认调用目标改为指定实例，同一事务内清旧置新。

        目标行须已经落盘：这里只按 ``instance_id`` 与 ``source_plugin_id`` 双重匹配
        定位目标行，不做隐式创建；命中失败原样返回，不动同插件原有的置位。命中时
        先清后置，两条 DML 处在同一 session、同一事务内提交，中途不会出现两行同时
        为真；并发写入下的唯一性最终由表上的条件唯一索引兜底。

        :param source_plugin_id: 源插件 ID
        :param instance_id: 要设为默认调用目标的实例 ID
        :return: 目标行存在并已置位为 True，目标行不存在时为 False
        """

        def stage(session: Session) -> bool:
            """
            在同一事务内定位目标行、清除同插件其余置位、置位目标行。
            """
            target = session.execute(
                select(PluginInstance).where(
                    PluginInstance.instance_id == instance_id,
                    PluginInstance.source_plugin_id == source_plugin_id,
                )
            ).scalars().first()
            if target is None:
                return False
            session.execute(
                update(PluginInstance)
                .where(
                    PluginInstance.source_plugin_id == source_plugin_id,
                    PluginInstance.instance_id != instance_id,
                    PluginInstance.is_default_target.is_(True),
                )
                .values(is_default_target=False)
            )
            target.is_default_target = True
            target.updated_at = _now()
            return True

        return bool(self._execute_sync_write(stage))

    def clear_default_target(self, source_plugin_id: str) -> None:
        """
        清除某源插件的默认调用目标置位，重复调用保持幂等。

        置位被清掉后该行若已只剩身份列则不在这里回收：本体行的回收统一由清空业务
        参数与清除日志等级两处判定，多一处入口只会让「行何时消失」变得难以预期。
        """

        def stage(session: Session) -> None:
            """
            在同一事务内清除该源插件全部置位的行。
            """
            session.execute(
                update(PluginInstance)
                .where(
                    PluginInstance.source_plugin_id == source_plugin_id,
                    PluginInstance.is_default_target.is_(True),
                )
                .values(is_default_target=False, updated_at=_now())
            )

        self._execute_sync_write(stage)

    def get_config_data(self, instance_id: str) -> Any:
        """
        读取某实例的业务参数；该行不存在或从未存过参数都返回 None。

        返回深拷贝：调用方拿到的是一份独立副本，就地改动不会顺着 ORM 行漏进
        下一次写入的「值未变化」判定。

        :param instance_id: 实例 ID
        :return: 业务参数，无则 None
        """
        record = self.get(instance_id)
        return copy.deepcopy(record.config_data) if record is not None else None

    def save_config_data(
        self,
        *,
        instance_id: str,
        source_plugin_id: str,
        config_data: Any,
    ) -> Optional[bool]:
        """
        写入业务参数，保留该行已有的身份与展示信息。

        :param instance_id: 实例 ID
        :param source_plugin_id: 该行不存在时用于建行的源插件 ID
        :param config_data: 业务参数
        :return: True 已写入，None 值未变化无需写入
        """

        def stage(session: Session) -> Optional[bool]:
            """
            在同一事务内创建或更新业务参数。
            """
            return _apply_config_data(
                session,
                PluginInstance.get_by_instance_id(session, instance_id),
                instance_id=instance_id,
                source_plugin_id=source_plugin_id,
                config_data=config_data,
            )

        return self._execute_sync_write(stage)

    async def async_save_config_data(
        self,
        *,
        instance_id: str,
        source_plugin_id: str,
        config_data: Any,
    ) -> Optional[bool]:
        """
        在异步事务中写入业务参数，行为与同步写入一致。

        :param instance_id: 实例 ID
        :param source_plugin_id: 该行不存在时用于建行的源插件 ID
        :param config_data: 业务参数
        :return: True 已写入，None 值未变化无需写入
        """

        async def stage(session: AsyncSession) -> Optional[bool]:
            """
            在同一异步事务内创建或更新业务参数。
            """
            return _apply_config_data(
                session,
                await PluginInstance.async_get_by_instance_id(session, instance_id),
                instance_id=instance_id,
                source_plugin_id=source_plugin_id,
                config_data=config_data,
            )

        return await self._execute_async_write(stage)

    def set_log_level(
        self,
        *,
        instance_id: str,
        log_level: Optional[str],
        log_expires_at: Optional[str],
    ) -> bool:
        """
        写入或清除该实例的日志等级覆盖，没有实例行时按本体建出。

        只设过等级、从没存过参数的插件同样要有一行，否则覆盖无处落盘；反过来，
        清除覆盖后若该行是本体且各列皆空，整行一并回收，与清空业务参数同一规则，
        不让只剩一对身份列的空行堆积。

        :param instance_id: 实例 ID
        :param log_level: 目标等级名，None 表示清除覆盖
        :param log_expires_at: 覆盖失效时间的 ISO-8601 文本，None 表示不过期
        :return: 是否发生了写入；清除一个本来就没有覆盖的实例返回 False
        """

        def stage(session: Session) -> bool:
            """
            在同一事务内写入等级覆盖，并回收清空后只剩身份的本体行。
            """
            record = PluginInstance.get_by_instance_id(session, instance_id)
            if record is None:
                if log_level is None and log_expires_at is None:
                    return False
                now = _now()
                session.add(
                    PluginInstance(
                        instance_id=instance_id,
                        source_plugin_id=instance_id,
                        log_level=log_level,
                        log_expires_at=log_expires_at,
                        created_at=now,
                        updated_at=now,
                    )
                )
                return True
            record.log_level = log_level
            record.log_expires_at = log_expires_at
            record.updated_at = _now()
            if record.is_host and record.carries_only_identity:
                session.delete(record)
            return True

        return bool(self._execute_sync_write(stage))

    def clear_config_data(self, instance_id: str) -> bool:
        """
        清空某实例的业务参数，保留其身份与展示信息。

        清空后若该行是本体且各列皆空，整行一并移除，避免只剩一对身份列的空行堆积。

        :param instance_id: 实例 ID
        :return: 该行存在并已处理
        """

        def stage(session: Session) -> bool:
            """
            在同一事务内清空业务参数并回收空的本体行。
            """
            record = PluginInstance.get_by_instance_id(session, instance_id)
            if record is None:
                return False
            record.config_data = None
            record.updated_at = _now()
            if record.is_host and record.carries_only_identity:
                session.delete(record)
            return True

        return bool(self._execute_sync_write(stage))