awoooi/apps/api/src/repositories/knowledge_repository.py

"""
Knowledge Repository - PostgreSQL 實作
=======================================
Knowledge Base Phase 1: CRUD + 搜尋

建立時間: 2026-04-02 (台北時區)
建立者: Claude Code (Knowledge Base Phase 1)

遵循 leWOOOgo 積木化原則:
- 實作 IKnowledgeRepository Protocol
- 只做資料存取，業務邏輯在 Service 層
"""

import json

import structlog
from sqlalchemy import String, func, or_, select, update
from sqlalchemy import text as sa_text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from src.db.models import KnowledgeEntryRecord
from src.models.knowledge import (
    EntrySource,
    EntryStatus,
    EntryType,
    KnowledgeEntry,
    KnowledgeEntryCreate,
)
from src.utils.timezone import now_taipei

logger = structlog.get_logger(__name__)

_DEFAULT_CATEGORY = "general"

_ASSET_TAXONOMY_KEYS = (
    "project",
    "product",
    "website",
    "service",
    "package",
    "tool",
    "log",
    "alert",
    "playbook",
    "rag",
    "mcp",
    "schedule",
)

_ASSET_TAXONOMY_TERMS: dict[str, tuple[str, ...]] = {
    "project": (
        "project",
        "repo",
        "repository",
        "gitea",
        "branch",
        "workflow",
        "source_control",
    ),
    "product": (
        "product",
        "awoooi",
        "awooop",
        "iwooos",
        "stockplatform",
        "momo",
        "awooogo",
        "agent-bounty",
        "agent_bounty",
        "tsenyang",
        "vibework",
        "2026fifa",
    ),
    "website": ("website", "site", "nginx", "ssl", "domain", "route", "frontend"),
    "service": ("service", "daemon", "api", "worker", "runtime", "container", "pod"),
    "package": (
        "package",
        "dependency",
        "npm",
        "pnpm",
        "node",
        "python",
        "pip",
        "prisma",
        "next",
        "library",
    ),
    "tool": (
        "tool",
        "ansible",
        "mcp",
        "playbook",
        "telegram",
        "wazuh",
        "kali",
        "sentry",
        "signoz",
        "runner",
    ),
    "log": (
        "log",
        "logs",
        "event",
        "timeline",
        "trace",
        "audit",
        "callback",
        "conversation_event",
        "telemetry",
    ),
    "alert": (
        "alert",
        "telegram",
        "sentry",
        "signoz",
        "notification",
        "warning",
        "critical",
    ),
    "playbook": ("playbook", "runbook", "sop"),
    "rag": ("rag", "vector", "embedding", "semantic", "retrieval"),
    "mcp": ("mcp", "connector", "gateway", "tool integration", "tool-integration"),
    "schedule": ("schedule", "cron", "job", "worker", "patrol", "recurrence", "cadence"),
}

_ASSET_TAXONOMY_CATEGORY_HINTS: dict[str, tuple[str, ...]] = {
    "website": ("external_site",),
    "service": (
        "infrastructure",
        "application",
        "ai_system",
        "database",
        "host_resource",
        "kubernetes",
        "alert_handling",
    ),
    "tool": ("devops_tool", "AI自動化/Ansible受控修復"),
    "alert": ("alert_handling",),
    "playbook": ("auto_repair",),
}


def _enum_text(column):
    """Return a lowercase text view for enum/varchar columns across old KM schemas."""
    return func.lower(column.cast(String))


_ENUM_FALLBACKS = {
    EntryType: EntryType.BEST_PRACTICE,
    EntrySource: EntrySource.AI_EXTRACTED,
    EntryStatus: EntryStatus.REVIEW,
}


def _knowledge_enum_db_label(value):
    """Match the mixed-case labels in the deployed PostgreSQL enums."""

    if isinstance(value, EntryStatus) and value is EntryStatus.PUBLISHED:
        return value.value
    if isinstance(value, EntryType | EntrySource | EntryStatus):
        return value.name
    return value


def _knowledge_enum_db_value(value):
    if isinstance(value, EntryStatus) and value is EntryStatus.PUBLISHED:
        return sa_text("'published'::entrystatus")
    return _knowledge_enum_db_label(value)


def _enum_member(value, enum_cls):
    if value is None:
        return _ENUM_FALLBACKS.get(enum_cls)
    raw = str(value.value if hasattr(value, "value") else value)
    normalized = raw.lower()
    for member in enum_cls:
        if normalized in {member.value.lower(), member.name.lower()}:
            return member
    logger.warning(
        "knowledge_read_model_unknown_enum",
        enum=enum_cls.__name__,
        value=raw,
        fallback=getattr(_ENUM_FALLBACKS.get(enum_cls), "value", None),
    )
    return _ENUM_FALLBACKS.get(enum_cls)


def _active_status_filter():
    return _enum_text(KnowledgeEntryRecord.status) != EntryStatus.ARCHIVED.value


def _normalize_category(value: object) -> str:
    """Keep legacy KM rows with NULL/blank category from breaking read models."""
    category = str(value or "").strip()
    return category or _DEFAULT_CATEGORY


def _normalize_tags(value: object) -> list[str]:
    """Accept legacy JSON/string tag shapes without breaking KM list readback."""
    if value is None:
        return []
    if isinstance(value, list | tuple | set):
        return [str(tag).strip() for tag in value if str(tag).strip()]
    if isinstance(value, str):
        stripped = value.strip()
        if not stripped:
            return []
        try:
            parsed = json.loads(stripped)
        except json.JSONDecodeError:
            parsed = None
        if isinstance(parsed, list):
            return [str(tag).strip() for tag in parsed if str(tag).strip()]
        return [stripped]
    return [str(value).strip()] if str(value).strip() else []


def _normalize_int(value: object, *, fallback: int = 0) -> int:
    try:
        return int(value or fallback)
    except (TypeError, ValueError):
        return fallback


def _normalized_category_expr():
    return func.coalesce(
        func.nullif(func.trim(KnowledgeEntryRecord.category), ""),
        _DEFAULT_CATEGORY,
    )


def _knowledge_entry_columns():
    return (
        KnowledgeEntryRecord.id.label("id"),
        KnowledgeEntryRecord.title.label("title"),
        KnowledgeEntryRecord.content.label("content"),
        KnowledgeEntryRecord.entry_type.cast(String).label("entry_type"),
        _normalized_category_expr().label("category"),
        KnowledgeEntryRecord.tags.label("tags"),
        KnowledgeEntryRecord.source.cast(String).label("source"),
        KnowledgeEntryRecord.status.cast(String).label("status"),
        KnowledgeEntryRecord.related_incident_id.label("related_incident_id"),
        KnowledgeEntryRecord.related_playbook_id.label("related_playbook_id"),
        KnowledgeEntryRecord.related_approval_id.label("related_approval_id"),
        KnowledgeEntryRecord.path_type.label("path_type"),
        KnowledgeEntryRecord.symptoms_hash.label("symptoms_hash"),
        KnowledgeEntryRecord.view_count.label("view_count"),
        KnowledgeEntryRecord.created_by.label("created_by"),
        KnowledgeEntryRecord.created_at.label("created_at"),
        KnowledgeEntryRecord.updated_at.label("updated_at"),
    )


def _asset_taxonomy_condition(key: str):
    """Build a broad-but-readable taxonomy matcher for AI automation surfaces."""
    category_expr = _normalized_category_expr()
    category_text = func.lower(category_expr.cast(String))
    text_columns = (
        KnowledgeEntryRecord.title,
        KnowledgeEntryRecord.content,
        KnowledgeEntryRecord.tags.cast(String),
        category_expr.cast(String),
    )
    conditions = []
    category_hints = tuple(
        hint.lower() for hint in _ASSET_TAXONOMY_CATEGORY_HINTS.get(key, ())
    )
    if category_hints:
        conditions.append(category_text.in_(category_hints))
    for term in _ASSET_TAXONOMY_TERMS.get(key, ()):
        pattern = f"%{term.lower()}%"
        conditions.extend(func.lower(column).like(pattern) for column in text_columns)
    if key == "playbook":
        conditions.append(KnowledgeEntryRecord.related_playbook_id.is_not(None))
    if not conditions:
        return category_text == key
    return or_(*conditions)


class KnowledgeDBRepository:
    """
    Knowledge Repository - PostgreSQL 實作

    實作 IKnowledgeRepository Protocol
    """

    def __init__(self, db: AsyncSession):
        self.db = db

    async def create(self, data: KnowledgeEntryCreate) -> KnowledgeEntry:
        """
        建立知識條目。

        P1-1 M3 2026-04-28 ogt + Claude Sonnet 4.6：
          - 若 data.path_type + data.related_incident_id 均非 None，
            使用 INSERT ... ON CONFLICT DO UPDATE（UPSERT）避免重複條目，
            實現 KMWriter 承諾的 (related_incident_id, path_type) 冪等 key。
          - 其餘路徑（無 path_type 或無 incident_id）仍用原始 INSERT。
        """
        # --- UPSERT 路徑（有冪等 key 時）---
        if data.path_type and data.related_incident_id:
            values = {
                "title": data.title,
                "content": data.content,
                "entry_type": _knowledge_enum_db_label(data.entry_type),
                "category": data.category,
                "tags": data.tags,
                "source": _knowledge_enum_db_label(data.source),
                "status": _knowledge_enum_db_value(data.status),
                "related_incident_id": data.related_incident_id,
                "related_playbook_id": data.related_playbook_id,
                "related_approval_id": data.related_approval_id,
                "path_type": data.path_type,
                "symptoms_hash": data.symptoms_hash,
                "created_by": data.created_by,
            }
            # RETURNING id：只取 id 標量（最安全，不依賴 ORM RETURNING 行為）
            stmt = (
                pg_insert(KnowledgeEntryRecord)
                .values(**values)
                .on_conflict_do_update(
                    index_elements=["related_incident_id", "path_type"],
                    # ON CONFLICT condition 需與 partial index 一致（WHERE both NOT NULL）
                    index_where=sa_text(
                        "related_incident_id IS NOT NULL AND path_type IS NOT NULL"
                    ),
                    set_={
                        "title": data.title,
                        "content": data.content,
                        "tags": data.tags,
                        "status": _knowledge_enum_db_value(data.status),
                        "related_approval_id": data.related_approval_id,
                    },
                )
                .returning(KnowledgeEntryRecord.id)
            )
            result = await self.db.execute(stmt)
            entry_id: str = result.scalar_one()
            # Read through the enum-safe projection because production keeps
            # both legacy uppercase labels and the lowercase published label.
            fetch_result = await self.db.execute(
                select(*_knowledge_entry_columns()).where(
                    KnowledgeEntryRecord.id == entry_id
                )
            )
            record = fetch_result.mappings().one()
            logger.info(
                "knowledge_entry_upserted",
                entry_id=record["id"],
                title=record["title"],
                path_type=data.path_type,
                incident_id=data.related_incident_id,
            )
            return self._to_model(record)

        # --- 一般 INSERT 路徑（無冪等 key 時）---
        record = KnowledgeEntryRecord(
            title=data.title,
            content=data.content,
            entry_type=_knowledge_enum_db_label(data.entry_type),
            category=data.category,
            tags=data.tags,
            source=_knowledge_enum_db_label(data.source),
            # 2026-04-04 ogt: Phase 25 P1 — 支援指定 status（ANTI_PATTERN 直接 PUBLISHED）
            status=_knowledge_enum_db_value(data.status),
            related_incident_id=data.related_incident_id,
            related_playbook_id=data.related_playbook_id,
            related_approval_id=data.related_approval_id,
            # P1-1 M3: path_type（此路徑為 None，不觸發冪等）
            path_type=data.path_type,
            # 2026-04-04 ogt: Phase 25 P1 — Anti-Pattern 閉環用症狀 hash
            symptoms_hash=data.symptoms_hash,
            created_by=data.created_by,
        )
        self.db.add(record)
        await self.db.flush()
        entry_id = str(record.id)
        fetch_result = await self.db.execute(
            select(*_knowledge_entry_columns()).where(
                KnowledgeEntryRecord.id == entry_id
            )
        )
        readback = fetch_result.mappings().one()
        logger.info(
            "knowledge_entry_created",
            entry_id=readback["id"],
            title=readback["title"],
        )
        return self._to_model(readback)

    async def get_by_id(self, entry_id: str) -> KnowledgeEntry | None:
        """根據 ID 取得知識條目（排除 archived）"""
        result = await self.db.execute(
            select(*_knowledge_entry_columns()).where(
                KnowledgeEntryRecord.id == entry_id,
                _active_status_filter(),
            )
        )
        record = result.mappings().one_or_none()
        return self._to_model(record) if record else None

    async def update(self, entry_id: str, data: dict) -> KnowledgeEntry | None:
        """更新知識條目"""
        result = await self.db.execute(
            select(KnowledgeEntryRecord).where(KnowledgeEntryRecord.id == entry_id)
        )
        record = result.scalar_one_or_none()
        if not record:
            return None

        for key, value in data.items():
            if value is not None and hasattr(record, key):
                setattr(record, key, value)

        await self.db.flush()
        logger.info("knowledge_entry_updated", entry_id=entry_id)
        return self._to_model(record)

    async def delete(self, entry_id: str) -> bool:
        """軟刪除 → status = archived"""
        result = await self.db.execute(
            update(KnowledgeEntryRecord)
            .where(KnowledgeEntryRecord.id == entry_id)
            .values(status=EntryStatus.ARCHIVED)
        )
        return result.rowcount > 0

    async def list_entries(
        self,
        category: str | None = None,
        entry_type: EntryType | None = None,
        status: EntryStatus | None = None,
        tags: list[str] | None = None,
        q: str | None = None,
        limit: int = 20,
        offset: int = 0,
    ) -> tuple[list[KnowledgeEntry], int]:
        """列出知識條目 (支援篩選)"""
        query = select(*_knowledge_entry_columns()).where(_active_status_filter())
        count_query = select(func.count()).select_from(KnowledgeEntryRecord).where(
            _active_status_filter()
        )

        if category:
            normalized_category = _normalize_category(category)
            query = query.where(_normalized_category_expr() == normalized_category)
            count_query = count_query.where(
                _normalized_category_expr() == normalized_category
            )
        if entry_type:
            query = query.where(
                _enum_text(KnowledgeEntryRecord.entry_type) == entry_type.value
            )
            count_query = count_query.where(
                _enum_text(KnowledgeEntryRecord.entry_type) == entry_type.value
            )
        if status:
            query = query.where(_enum_text(KnowledgeEntryRecord.status) == status.value)
            count_query = count_query.where(_enum_text(KnowledgeEntryRecord.status) == status.value)
        if tags:
            for tag in tags:
                tag_filter = _json_string_array_has_tag(tag)
                query = query.where(tag_filter)
                count_query = count_query.where(tag_filter)
        if q:
            like_q = f"%{q}%"
            filter_cond = or_(
                KnowledgeEntryRecord.title.ilike(like_q),
                KnowledgeEntryRecord.content.ilike(like_q),
            )
            query = query.where(filter_cond)
            count_query = count_query.where(filter_cond)

        total_result = await self.db.execute(count_query)
        total = total_result.scalar() or 0

        query = query.order_by(KnowledgeEntryRecord.updated_at.desc())
        query = query.limit(limit).offset(offset)

        result = await self.db.execute(query)
        records = result.mappings().all()

        return [self._to_model(r) for r in records], total

    async def get_categories(self) -> list[tuple[str, int]]:
        """取得分類統計"""
        category_expr = _normalized_category_expr()
        result = await self.db.execute(
            select(
                category_expr.label("category"),
                func.count().label("cnt"),
            )
            .where(_active_status_filter())
            .group_by(category_expr)
            .order_by(func.count().desc())
        )
        return [(_normalize_category(row.category), row.cnt) for row in result.all()]

    async def get_asset_taxonomy_counts(self) -> list[tuple[str, int]]:
        """取得 AI 自動化資產維度統計。

        這不是替代原始 category，而是讓 UI / Agent 可以用穩定 taxonomy
        將 KM 依專案、產品、網站、服務、套件、工具、Log、Alert、PlayBook、
        RAG、MCP 與排程分群。
        """
        result = await self.db.execute(
            select(
                *(
                    func.count()
                    .filter(_asset_taxonomy_condition(key))
                    .label(key)
                    for key in _ASSET_TAXONOMY_KEYS
                )
            )
            .select_from(KnowledgeEntryRecord)
            .where(_active_status_filter())
        )
        row = result.mappings().one()
        return [(key, int(row.get(key) or 0)) for key in _ASSET_TAXONOMY_KEYS]

    async def search(self, query: str, limit: int = 20) -> list[KnowledgeEntry]:
        """關鍵字搜尋 (title + content + tags)"""
        like_q = f"%{query}%"
        result = await self.db.execute(
            select(*_knowledge_entry_columns())
            .where(
                _active_status_filter(),
                or_(
                    KnowledgeEntryRecord.title.ilike(like_q),
                    KnowledgeEntryRecord.content.ilike(like_q),
                    KnowledgeEntryRecord.tags.cast(String).ilike(like_q),
                ),
            )
            .order_by(KnowledgeEntryRecord.view_count.desc())
            .limit(limit)
        )
        records = result.mappings().all()
        return [self._to_model(r) for r in records]

    async def increment_view_count(self, entry_id: str) -> bool:
        """view_count +1"""
        result = await self.db.execute(
            update(KnowledgeEntryRecord)
            .where(KnowledgeEntryRecord.id == entry_id)
            .values(view_count=KnowledgeEntryRecord.view_count + 1)
        )
        return result.rowcount > 0

    async def list_unembedded_entries(self) -> list[tuple[str, str, str]]:
        """列出尚未產生 embedding 的條目 [(id, title, content)]"""
        from sqlalchemy import text as sa_text
        result = await self.db.execute(
            sa_text(
                "SELECT id, title, content FROM knowledge_entries "
                "WHERE embedding IS NULL "
                "AND LOWER(CAST(status AS TEXT)) != :archived"
            ),
            {"archived": EntryStatus.ARCHIVED.value},
        )
        return [(row.id, row.title, row.content) for row in result.fetchall()]

    async def save_embedding(self, entry_id: str, embedding: list[float]) -> bool:
        """儲存向量 embedding (768 維)

        注意: asyncpg 不支援 :param::type 語法，必須用 CAST(:param AS vector)
        """
        from sqlalchemy import text as sa_text
        result = await self.db.execute(
            sa_text(
                "UPDATE knowledge_entries SET embedding = CAST(:emb AS vector) WHERE id = :id"
            ),
            {"emb": str(embedding), "id": entry_id},
        )
        return result.rowcount > 0

    async def semantic_search(
        self,
        query_embedding: list[float],
        limit: int = 10,
        threshold: float = 0.5,
    ) -> list[tuple[KnowledgeEntry, float]]:
        """
        語意搜尋 — cosine similarity (pgvector)

        Returns:
            list of (entry, similarity_score) 已按分數降序排列
        """
        from sqlalchemy import text as sa_text
        sql = sa_text("""
            SELECT id, 1 - (embedding <=> CAST(:emb AS vector)) AS score
            FROM knowledge_entries
            WHERE LOWER(CAST(status AS TEXT)) != :archived
              AND embedding IS NOT NULL
              AND 1 - (embedding <=> CAST(:emb AS vector)) >= :threshold
            ORDER BY embedding <=> CAST(:emb AS vector)
            LIMIT :limit
        """)
        rows = await self.db.execute(
            sql,
            {
                "emb": str(query_embedding),
                "threshold": threshold,
                "limit": limit,
                "archived": EntryStatus.ARCHIVED.value,
            },
        )
        rows = rows.fetchall()

        if not rows:
            return []

        # 批次取得完整 entry
        ids = [r[0] for r in rows]
        scores = {r[0]: float(r[1]) for r in rows}

        result = await self.db.execute(
            select(*_knowledge_entry_columns()).where(KnowledgeEntryRecord.id.in_(ids))
        )
        records = {r["id"]: r for r in result.mappings().all()}

        return [
            (self._to_model(records[entry_id]), scores[entry_id])
            for entry_id in ids
            if entry_id in records
        ]

    def _to_model(self, record) -> KnowledgeEntry:
        """ORM Record → Pydantic Model"""
        if hasattr(record, "_mapping"):
            record = record._mapping

        def get(name: str):
            if isinstance(record, dict):
                return record.get(name)
            if hasattr(record, "__getitem__") and name in record:
                return record[name]
            return getattr(record, name)

        return KnowledgeEntry(
            id=str(get("id")),
            title=str(get("title") or "Untitled KM entry"),
            content=str(get("content") or ""),
            entry_type=_enum_member(get("entry_type"), EntryType),
            category=_normalize_category(get("category")),
            tags=_normalize_tags(get("tags")),
            source=_enum_member(get("source"), EntrySource),
            status=_enum_member(get("status"), EntryStatus),
            related_incident_id=get("related_incident_id"),
            related_playbook_id=get("related_playbook_id"),
            related_approval_id=get("related_approval_id"),
            # P1-1 M3 2026-04-28 ogt + Claude Sonnet 4.6: 冪等 key
            path_type=get("path_type"),
            symptoms_hash=get("symptoms_hash"),
            view_count=_normalize_int(get("view_count")),
            created_by=get("created_by"),
            created_at=get("created_at") or now_taipei(),
            updated_at=get("updated_at") or get("created_at") or now_taipei(),
        )


def _json_string_array_has_tag(tag: str):
    """建立 JSON/JSONB 皆相容的 tag filter。

    production 的 knowledge_entries.tags 目前是 JSON 欄位，不支援 json @> text。
    這裡改用帶引號的字串比對，避免把 tag 片段誤判成完整 tag。
    """
    escaped = (
        tag
        .replace("\\", "\\\\")
        .replace("%", "\\%")
        .replace("_", "\\_")
    )
    return KnowledgeEntryRecord.tags.cast(String).ilike(f'%"{escaped}"%', escape="\\")