group_fqcd_jr/app/service/projection_cleanup_service.py

"""投影清理：记忆失效或删除时，清掉它在图库与向量库里的派生数据。

**为什么以画像为准，而不是按标识直接删边**：一条记忆可能对应多条派生边（同一 key 的
证据累积会让 tag 取值变化），直接按 tag_key 删容易删不干净或误删。这里走
「删掉该记忆对应的长期事实 → 重建画像 → 对账修复」，让投影跟着权威源收敛——
最终图里剩下的内容必然与画像一致，而不依赖"我记得这条记忆当初投影成了什么"。

**Milvus 向量**按 `memory_uuid` 删除；集合不存在或客户端不可用时**如实报告未清理**，
绝不伪造成功。这与 `_cleanup_projection` 的审计语义一致：审计只记录适配器返回的真实结论。
"""

import logging
from dataclasses import dataclass

from sqlalchemy import delete, select

from app.core.config import get_settings
from app.infrastructure.db import SessionFactory
from app.infrastructure.milvus_profile_projection import PROFILE_COLLECTION
from app.model.memory import MemoryUnit
from app.model.profile import UserFact
from app.service.profile_assembly_service import ProfileAssemblyService
from app.service.profile_graph_projection_service import ProfileGraphProjectionService
from app.service.relationship_service import RelationshipService

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class ProjectionCleanupOutcome:
    """与 `app/worker/runtime.py` 的同名结构保持字段一致（该处只按属性取用）。

    `cleaned=True` 表示派生数据已确实清理；为假时 `detail` 必须说明真实缺口，
    审计会原样记录它。
    """

    cleaned: bool
    detail: str = ""


class ProjectionCleanupService:
    def __init__(self, relationships: RelationshipService | None = None) -> None:
        self.relationships = relationships

    async def cleanup(self, *, memory_uuid: str, operation: str) -> ProjectionCleanupOutcome:
        del operation  # 失效与删除对投影的处理一致：都以权威源为准重建
        async with SessionFactory() as session:
            memory = await session.scalar(
                select(MemoryUnit).where(MemoryUnit.memory_uuid == memory_uuid)
            )
            if memory is None:
                # 记忆已经不在库里：投影清理的目标已消失，视为完成（幂等）
                return ProjectionCleanupOutcome(True, "memory_not_found")
            customer_id = int(memory.customer_id)
            fact_key = str(memory.memory_key)

        # 该记忆已失效/删除，它对应的事实不应再参与画像
        async with SessionFactory() as session, session.begin():
            await session.execute(delete(UserFact).where(
                UserFact.customer_id == customer_id, UserFact.fact_key == fact_key
            ))

        # 以剩余证据重建画像，再让图跟着画像收敛
        details: list[str] = []
        try:
            async with SessionFactory() as session, session.begin():
                await ProfileAssemblyService(session).rebuild(customer_id)
            details.append("profile_rebuilt")
        except Exception:
            logger.warning("profile rebuild failed during cleanup customer_id=%s",
                           customer_id, exc_info=True)
            details.append("profile_rebuild_failed")

        graph_ok = await self._cleanup_graph(customer_id, details)
        vector_ok = await self._cleanup_vector(memory_uuid, details)
        return ProjectionCleanupOutcome(
            cleaned=graph_ok and vector_ok and "profile_rebuild_failed" not in details,
            detail="; ".join(details),
        )

    async def _cleanup_graph(self, customer_id: int, details: list[str]) -> bool:
        if self.relationships is None:
            details.append("graph_client_unavailable")
            return False
        try:
            async with SessionFactory() as session:
                outcome = await ProfileGraphProjectionService(
                    session, self.relationships
                ).reconcile_customer(customer_id, repair=True)
        except Exception:
            logger.warning("graph cleanup failed customer_id=%s", customer_id, exc_info=True)
            details.append("graph_cleanup_failed")
            return False
        if outcome.degraded:
            details.append(f"graph_degraded:{outcome.reason}")
            return False
        if not outcome.consistent:
            # repair 之后仍不一致：如实报告，不写成清理成功
            details.append(
                f"graph_still_inconsistent:missing={len(outcome.missing)},"
                f"orphaned={len(outcome.orphaned)}"
            )
            return False
        details.append("graph_cleaned")
        return True

    async def _cleanup_vector(self, memory_uuid: str, details: list[str]) -> bool:
        """删除该记忆的向量。集合不存在（未启用语义召回）时视为无需清理。"""
        settings = get_settings()
        if not settings.milvus_uri:
            details.append("vector_store_not_configured")
            return True
        try:
            from pymilvus import MilvusClient  # type: ignore[import-untyped]

            client = MilvusClient(
                uri=settings.milvus_uri, token=settings.milvus_token or None
            )
        except Exception:
            logger.warning("milvus client unavailable during cleanup", exc_info=True)
            details.append("vector_client_unavailable")
            return False
        try:
            # 集合名与投影写入路径**同一个常量**（`PROFILE_COLLECTION`）。此前读的是
            # 已删除的 `settings.milvus_collection`（`jr_memory`），该集合不存在
            # ⇒ 这里每次都走下面的 `vector_collection_absent` 分支、
            # **报清理成功但一个向量都没删**，陈旧向量永久留存。
            if PROFILE_COLLECTION not in set(client.list_collections()):
                # 记忆向量集合尚未启用：没有需要清理的派生数据
                details.append("vector_collection_absent")
                return True
            client.delete(
                collection_name=PROFILE_COLLECTION,
                filter=f'memory_uuid == "{memory_uuid}"',
            )
            details.append("vector_cleaned")
            return True
        except Exception:
            logger.warning("milvus delete failed memory_uuid=%s", memory_uuid, exc_info=True)
            details.append("vector_delete_failed")
            return False