vent_ai_platform/app/services/knowledge_service.py

from __future__ import annotations

from difflib import SequenceMatcher

from sqlalchemy.orm import Session

from app.db.models.knowledge import KnowledgeItem
from app.db.repositories import knowledge as knowledge_repo
from app.schemas.knowledge import KnowledgeSearchRequest


def _score(item: KnowledgeItem, query: str) -> float:
    normalized = query.strip().lower()
    if not normalized:
        return 0
    title = item.title.lower()
    summary = item.summary.lower()
    content = item.content.lower()
    if normalized == title:
        return 100
    if normalized in title:
        base = 90
    elif normalized in summary:
        base = 78
    elif normalized in content:
        base = 65
    else:
        base = 0
    ratio = max(
        SequenceMatcher(None, normalized, title).ratio(),
        SequenceMatcher(None, normalized, summary[:300]).ratio(),
    )
    return round(min(100, base + ratio * 10), 2)


def search_knowledge(
    db: Session,
    payload: KnowledgeSearchRequest,
) -> list[dict]:
    items, _ = knowledge_repo.list_knowledge(
        db,
        mine_id=payload.mine_id,
        knowledge_type=payload.knowledge_type,
        category=payload.category,
        keyword=payload.query,
        page=1,
        page_size=payload.limit,
    )
    hits = [
        {
            "item": item,
            "score": _score(item, payload.query),
            "highlight": item.summary or item.content[:160],
        }
        for item in items
    ]
    hits.sort(key=lambda item: item["score"], reverse=True)
    return hits