leocrm/app/plugins/builtins/wiki/services.py

"""Wiki plugin services — CRUD for articles, categories, versioning."""
from __future__ import annotations

import uuid
from datetime import UTC, datetime
from typing import Any

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

from app.plugins.builtins.wiki.models import WikiArticle, WikiArticleVersion, WikiCategory


def _slugify(text: str) -> str:
    return text.lower().strip().replace(" ", "-").replace("ä", "ae").replace("ö", "oe").replace("ü", "ue")[:200]


def _article_to_dict(a: WikiArticle) -> dict[str, Any]:
    return {
        "id": str(a.id), "title": a.title, "slug": a.slug,
        "content": a.content, "content_html": a.content_html,
        "summary": a.summary, "category_id": str(a.category_id) if a.category_id else None,
        "tags": a.tags or [], "status": a.status,
        "entity_links": a.entity_links or [], "version": a.version,
        "published_at": a.published_at.isoformat() if a.published_at else None,
        "owner_id": str(a.owner_id) if a.owner_id else None,
        "created_at": a.created_at.isoformat() if a.created_at else None,
        "updated_at": a.updated_at.isoformat() if a.updated_at else None,
    }


def _category_to_dict(c: WikiCategory) -> dict[str, Any]:
    return {
        "id": str(c.id), "name": c.name, "slug": c.slug,
        "description": c.description, "parent_id": str(c.parent_id) if c.parent_id else None,
        "sort_order": c.sort_order,
    }


async def list_articles(db: AsyncSession, tenant_id: uuid.UUID, *, page: int = 1, page_size: int = 20, category_id: str | None = None, status: str | None = None, search: str | None = None) -> dict[str, Any]:
    query = select(WikiArticle).where(WikiArticle.tenant_id == tenant_id, WikiArticle.deleted_at.is_(None))
    if category_id:
        query = query.where(WikiArticle.category_id == uuid.UUID(category_id))
    if status:
        query = query.where(WikiArticle.status == status)
    if search:
        query = query.where(WikiArticle.title.ilike(f"%{search}%"))
    count_q = select(func.count()).select_from(query.subquery())
    total = (await db.execute(count_q)).scalar() or 0
    query = query.order_by(WikiArticle.updated_at.desc()).offset((page - 1) * page_size).limit(page_size)
    result = await db.execute(query)
    return {"items": [_article_to_dict(a) for a in result.scalars().all()], "total": total, "page": page, "page_size": page_size}


async def get_article(db: AsyncSession, tenant_id: uuid.UUID, article_id: uuid.UUID) -> dict[str, Any] | None:
    result = await db.execute(select(WikiArticle).where(WikiArticle.id == article_id, WikiArticle.tenant_id == tenant_id, WikiArticle.deleted_at.is_(None)))
    a = result.scalar_one_or_none()
    return _article_to_dict(a) if a else None


async def create_article(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict[str, Any]) -> dict[str, Any]:
    article = WikiArticle(
        tenant_id=tenant_id, owner_id=user_id,
        title=data["title"], slug=data.get("slug") or _slugify(data["title"]),
        content=data.get("content", ""), summary=data.get("summary"),
        category_id=uuid.UUID(data["category_id"]) if data.get("category_id") else None,
        tags=data.get("tags", []), status=data.get("status", "draft"),
        entity_links=data.get("entity_links", []), version=1,
        published_at=datetime.now(UTC) if data.get("status") == "published" else None,
    )
    db.add(article)
    await db.flush()
    await db.refresh(article)
    return _article_to_dict(article)


async def update_article(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, article_id: uuid.UUID, data: dict[str, Any]) -> dict[str, Any] | None:
    result = await db.execute(select(WikiArticle).where(WikiArticle.id == article_id, WikiArticle.tenant_id == tenant_id, WikiArticle.deleted_at.is_(None)))
    article = result.scalar_one_or_none()
    if article is None:
        return None
    # Save version history (H-VER)
    version = WikiArticleVersion(
        tenant_id=tenant_id, article_id=article.id, version=article.version,
        title=article.title, content=article.content,
        edited_by=user_id, edit_comment=data.get("edit_comment"),
    )
    db.add(version)
    if "title" in data and data["title"] is not None:
        article.title = data["title"]
    if "content" in data and data["content"] is not None:
        article.content = data["content"]
    if "summary" in data:
        article.summary = data["summary"]
    if "category_id" in data:
        article.category_id = uuid.UUID(data["category_id"]) if data["category_id"] else None
    if "tags" in data and data["tags"] is not None:
        article.tags = data["tags"]
    if "status" in data and data["status"] is not None:
        article.status = data["status"]
        if data["status"] == "published" and article.published_at is None:
            article.published_at = datetime.now(UTC)
    if "entity_links" in data and data["entity_links"] is not None:
        article.entity_links = data["entity_links"]
    article.version += 1
    await db.flush()
    await db.refresh(article)
    return _article_to_dict(article)


async def delete_article(db: AsyncSession, tenant_id: uuid.UUID, article_id: uuid.UUID) -> bool:
    result = await db.execute(select(WikiArticle).where(WikiArticle.id == article_id, WikiArticle.tenant_id == tenant_id, WikiArticle.deleted_at.is_(None)))
    article = result.scalar_one_or_none()
    if article is None:
        return False
    article.deleted_at = datetime.now(UTC)
    await db.flush()
    return True


async def list_versions(db: AsyncSession, tenant_id: uuid.UUID, article_id: uuid.UUID) -> list[dict[str, Any]]:
    result = await db.execute(select(WikiArticleVersion).where(WikiArticleVersion.tenant_id == tenant_id, WikiArticleVersion.article_id == article_id).order_by(WikiArticleVersion.version.desc()))
    return [{"id": str(v.id), "version": v.version, "title": v.title, "content": v.content[:500], "edited_by": str(v.edited_by) if v.edited_by else None, "edit_comment": v.edit_comment, "created_at": v.created_at.isoformat() if v.created_at else None} for v in result.scalars().all()]


async def restore_version(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, article_id: uuid.UUID, version: int) -> dict[str, Any] | None:
    vr = await db.execute(select(WikiArticleVersion).where(WikiArticleVersion.tenant_id == tenant_id, WikiArticleVersion.article_id == article_id, WikiArticleVersion.version == version))
    v = vr.scalar_one_or_none()
    if v is None:
        return None
    return await update_article(db, tenant_id, user_id, article_id, {"title": v.title, "content": v.content, "edit_comment": f"Restored from version {version}"})


async def list_categories(db: AsyncSession, tenant_id: uuid.UUID) -> list[dict[str, Any]]:
    result = await db.execute(select(WikiCategory).where(WikiCategory.tenant_id == tenant_id, WikiCategory.deleted_at.is_(None)).order_by(WikiCategory.sort_order.asc()))
    return [_category_to_dict(c) for c in result.scalars().all()]


async def create_category(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, data: dict[str, Any]) -> dict[str, Any]:
    cat = WikiCategory(
        tenant_id=tenant_id, owner_id=user_id,
        name=data["name"], slug=data.get("slug") or _slugify(data["name"]),
        description=data.get("description"),
        parent_id=uuid.UUID(data["parent_id"]) if data.get("parent_id") else None,
        sort_order=data.get("sort_order", 0),
    )
    db.add(cat)
    await db.flush()
    await db.refresh(cat)
    return _category_to_dict(cat)