Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
245 lines
9.3 KiB
Python
245 lines
9.3 KiB
Python
"""Unified Search plugin class and manifest."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from app.plugins.base import BasePlugin
|
|
from app.plugins.manifest import FrontendPageRoute, PluginManifest, PluginRouteDef
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class UnifiedSearchPlugin(BasePlugin):
|
|
"""Hybrid full-text and semantic search across all CRM entities."""
|
|
|
|
manifest = PluginManifest(
|
|
name="unified_search",
|
|
version="1.0.0",
|
|
display_name="Unified Search",
|
|
description=(
|
|
"Hybrid full-text (PostgreSQL FTS) and semantic (pgvector) search "
|
|
"with KI query understanding and RRF rank fusion across all CRM data."
|
|
),
|
|
dependencies=[],
|
|
routes=[
|
|
PluginRouteDef(
|
|
path="/api/v1/search",
|
|
module="app.plugins.builtins.unified_search.routes",
|
|
router_attr="router",
|
|
),
|
|
],
|
|
events=[
|
|
"mail.synced",
|
|
"file.uploaded",
|
|
"contact.created",
|
|
"contact.updated",
|
|
"contact.deleted",
|
|
"file.deleted",
|
|
"mail.deleted",
|
|
"event.deleted",
|
|
"calendar.entry.updated",
|
|
"calendar.entry.created",
|
|
"entity.deleted",
|
|
"entity.restored",
|
|
"entity.corrected",
|
|
],
|
|
migrations=["0001_initial.sql", "0002_embeddings.sql", "0003_add_deleted_at.sql", "0004_document_chunks.sql", "0005_add_indexed_at.sql"],
|
|
permissions=["search:read", "search:admin"],
|
|
is_core=True,
|
|
page_routes=[
|
|
FrontendPageRoute(path='/search', component='@/pages/GlobalSearchResults', protected=True),
|
|
],
|
|
author="LeoCRM Team",
|
|
min_app_version="1.0.0",
|
|
contract_version="1.0.0",
|
|
)
|
|
|
|
def get_job_modules(self) -> list[str]:
|
|
return [
|
|
"app.plugins.builtins.unified_search.jobs",
|
|
]
|
|
|
|
async def on_activate(self, db, service_container, event_bus) -> None:
|
|
"""Register search providers and AI tool on activation."""
|
|
await super().on_activate(db, service_container, event_bus)
|
|
try:
|
|
from app.plugins.builtins.unified_search.provider_registry import (
|
|
auto_register_providers,
|
|
)
|
|
await auto_register_providers(db)
|
|
logger.info("Unified Search providers auto-registered")
|
|
except Exception:
|
|
logger.exception("Failed to auto-register search providers")
|
|
|
|
# Register the unified_search AI tool via contract
|
|
try:
|
|
from app.plugins.builtins.contracts import get_contract
|
|
ai_contract = get_contract("ai_assistant")
|
|
if ai_contract is not None:
|
|
from app.plugins.builtins.unified_search.ai_tool import register_unified_search_tool
|
|
register_unified_search_tool(ai_contract.get_tool_registry())
|
|
logger.info("Unified Search AI tool registered")
|
|
except Exception:
|
|
logger.exception("Failed to register unified_search AI tool")
|
|
|
|
async def on_deactivate(self, db, service_container, event_bus) -> None:
|
|
"""Clear provider registry on deactivation."""
|
|
# Contract abmelden
|
|
from app.plugins.builtins.contracts import get_contract_registry
|
|
get_contract_registry().unregister(self.manifest.name)
|
|
from app.plugins.builtins.unified_search.provider_registry import (
|
|
get_search_registry,
|
|
)
|
|
registry = get_search_registry()
|
|
registry.clear()
|
|
await super().on_deactivate(db, service_container, event_bus)
|
|
|
|
# ─── Event Handlers ───
|
|
|
|
async def on_mail_synced(self, payload: dict[str, Any]) -> None:
|
|
"""Enqueue embedding jobs for synced mails."""
|
|
from app.core.jobs import enqueue_job
|
|
|
|
mail_ids = payload.get("mail_ids", [])
|
|
if mail_ids:
|
|
await enqueue_job("index_mails", mail_ids)
|
|
|
|
async def on_file_uploaded(self, payload: dict[str, Any]) -> None:
|
|
"""Enqueue file indexing job."""
|
|
from app.core.jobs import enqueue_job
|
|
|
|
file_id = payload.get("file_id")
|
|
if file_id:
|
|
await enqueue_job("index_file", file_id)
|
|
|
|
async def on_contact_created(self, payload: dict[str, Any]) -> None:
|
|
from app.core.jobs import enqueue_job
|
|
|
|
contact_id = payload.get("contact_id")
|
|
if contact_id:
|
|
await enqueue_job("index_contact", contact_id)
|
|
|
|
async def on_contact_updated(self, payload: dict[str, Any]) -> None:
|
|
from app.core.jobs import enqueue_job
|
|
|
|
contact_id = payload.get("contact_id")
|
|
if contact_id:
|
|
await enqueue_job("index_contact", contact_id)
|
|
|
|
async def on_calendar_entry_created(self, payload: dict[str, Any]) -> None:
|
|
from app.core.jobs import enqueue_job
|
|
|
|
entry_id = payload.get("entry_id")
|
|
if entry_id:
|
|
await enqueue_job("index_event", entry_id)
|
|
|
|
async def on_calendar_entry_updated(self, payload: dict[str, Any]) -> None:
|
|
"""Enqueue re-index job for updated calendar entry."""
|
|
from app.core.jobs import enqueue_job
|
|
|
|
entry_id = payload.get("entry_id")
|
|
if entry_id:
|
|
await enqueue_job("index_event", entry_id)
|
|
|
|
async def on_contact_deleted(self, payload: dict[str, Any]) -> None:
|
|
"""Enqueue job to remove embedding for deleted contact."""
|
|
from app.core.jobs import enqueue_job
|
|
|
|
contact_id = payload.get("contact_id")
|
|
if contact_id:
|
|
await enqueue_job("delete_entity_index", "contact", contact_id)
|
|
|
|
async def on_file_deleted(self, payload: dict[str, Any]) -> None:
|
|
"""Enqueue job to remove file embedding + delete document_chunks."""
|
|
from app.core.jobs import enqueue_job
|
|
|
|
file_id = payload.get("file_id")
|
|
if file_id:
|
|
await enqueue_job("delete_entity_index", "file", file_id)
|
|
await enqueue_job("delete_file_chunks", file_id)
|
|
|
|
async def on_mail_deleted(self, payload: dict[str, Any]) -> None:
|
|
"""Enqueue job to remove mail embedding."""
|
|
from app.core.jobs import enqueue_job
|
|
|
|
mail_id = payload.get("mail_id")
|
|
if mail_id:
|
|
await enqueue_job("delete_entity_index", "mail", mail_id)
|
|
|
|
async def on_event_deleted(self, payload: dict[str, Any]) -> None:
|
|
"""Enqueue job to remove event embedding."""
|
|
from app.core.jobs import enqueue_job
|
|
|
|
event_id = payload.get("event_id")
|
|
if event_id:
|
|
await enqueue_job("delete_entity_index", "event", event_id)
|
|
|
|
async def on_entity_deleted(self, payload: dict[str, Any]) -> None:
|
|
"""Handle entity.deleted lifecycle event — remove from search index."""
|
|
from app.core.db import get_session_factory
|
|
from app.plugins.builtins.unified_search.lifecycle import handle_entity_delete
|
|
|
|
entity_type = payload.get("entity_type")
|
|
entity_id = payload.get("entity_id")
|
|
tenant_id = payload.get("tenant_id")
|
|
if not all([entity_type, entity_id, tenant_id]):
|
|
return
|
|
factory = get_session_factory()
|
|
async with factory() as db:
|
|
await handle_entity_delete(
|
|
db, entity_type, uuid.UUID(str(entity_id)), uuid.UUID(str(tenant_id))
|
|
)
|
|
|
|
async def on_entity_restored(self, payload: dict[str, Any]) -> None:
|
|
"""Handle entity.restored lifecycle event — rebuild search index."""
|
|
from app.core.db import get_session_factory
|
|
from app.plugins.builtins.unified_search.lifecycle import handle_entity_restore
|
|
|
|
entity_type = payload.get("entity_type")
|
|
entity_id = payload.get("entity_id")
|
|
tenant_id = payload.get("tenant_id")
|
|
if not all([entity_type, entity_id, tenant_id]):
|
|
return
|
|
factory = get_session_factory()
|
|
async with factory() as db:
|
|
await handle_entity_restore(
|
|
db, entity_type, uuid.UUID(str(entity_id)), uuid.UUID(str(tenant_id))
|
|
)
|
|
|
|
async def on_entity_corrected(self, payload: dict[str, Any]) -> None:
|
|
"""Handle entity.corrected lifecycle event — rebuild search index."""
|
|
from app.core.db import get_session_factory
|
|
from app.plugins.builtins.unified_search.lifecycle import handle_entity_correction
|
|
|
|
entity_type = payload.get("entity_type")
|
|
entity_id = payload.get("entity_id")
|
|
tenant_id = payload.get("tenant_id")
|
|
if not all([entity_type, entity_id, tenant_id]):
|
|
return
|
|
factory = get_session_factory()
|
|
async with factory() as db:
|
|
await handle_entity_correction(
|
|
db, entity_type, uuid.UUID(str(entity_id)), uuid.UUID(str(tenant_id))
|
|
)
|
|
|
|
def get_notification_types(self) -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"type_key": "search_error",
|
|
"category": "search",
|
|
"label": "Suchfehler",
|
|
"description": "Fehler bei der Suchausführung",
|
|
"is_enabled_by_default": True,
|
|
},
|
|
{
|
|
"type_key": "search_reindex_complete",
|
|
"category": "search",
|
|
"label": "Reindex abgeschlossen",
|
|
"description": "Neuindizierung abgeschlossen",
|
|
"is_enabled_by_default": False,
|
|
},
|
|
]
|