# pylint: disable=W0613
"""Django signal receivers for plugin app."""
from typing import Optional, Union
from django.db.models.signals import post_save, pre_delete
from django.dispatch import receiver
from django.forms.models import model_to_dict
from smarter.common.helpers.console_helpers import formatted_json, formatted_text
from smarter.lib import json, logging
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.manifest.broker import AbstractBroker
from smarter.lib.typing import SmarterDictType
from .models import (
PluginDataApi,
PluginDataSql,
PluginDataStatic,
PluginDataWebsearch,
PluginMeta,
PluginPrompt,
PluginSelector,
PluginSelectorHistory,
)
from .plugin.static import PluginBase
from .signals import (
broker_ready,
plugin_called,
plugin_cloned,
plugin_created,
plugin_deleted,
plugin_deleting,
plugin_ready,
plugin_responded,
plugin_selected,
plugin_updated,
search_term_matched,
websearch_failed,
websearch_fetched,
websearch_searched,
)
from .tasks import create_plugin_selector_history
logger = logging.getSmarterLogger(__name__, any_switches=[SmarterWaffleSwitches.RECEIVER_LOGGING])
prefix = "smarter.apps.llmtool.receivers."
[docs]
@receiver(plugin_created, dispatch_uid="plugin_created")
def handle_plugin_created(sender, plugin: PluginBase, **kwargs):
"""Handle plugin created signal."""
logger.info(
"%s - %s - name: %s data: %s",
formatted_text(prefix + "plugin_created"),
plugin.user_profile,
plugin.name,
formatted_json(plugin.data) if plugin.data else None,
)
[docs]
@receiver(plugin_cloned, dispatch_uid="plugin_cloned")
def handle_plugin_cloned(sender, plugin: PluginBase, **kwargs):
"""Handle plugin cloned signal."""
logger.info("%s - %s data: %s", formatted_text(prefix + "plugin_cloned"), plugin.name, plugin.data)
[docs]
@receiver(plugin_updated, dispatch_uid="plugin_updated")
def handle_plugin_updated(sender, plugin: PluginBase, **kwargs):
"""Handle plugin updated signal."""
logger.info(
"%s - %s - name: %s data: %s",
formatted_text(prefix + "plugin_updated"),
plugin.user_profile,
plugin.name,
formatted_json(plugin.data) if plugin.data else None,
)
[docs]
@receiver(plugin_deleting, dispatch_uid=prefix + ".plugin_deleting")
def handle_plugin_deleting(sender, plugin, plugin_meta: PluginMeta, **kwargs):
"""Handle plugin deleting signal."""
logger.info(
"%s %s is being deleted.",
formatted_text("smarter.apps.llmtool.receivers.plugin_deleting"),
plugin_meta.name,
)
[docs]
@receiver(plugin_deleted, dispatch_uid="plugin_deleted")
def handle_plugin_deleted(sender, plugin: PluginBase, plugin_name: str, **kwargs):
"""Handle plugin deleted signal."""
logger.info("%s - %s", formatted_text(prefix + "plugin_deleted"), plugin_name)
[docs]
@receiver(plugin_called, dispatch_uid="plugin_called")
def handle_plugin_called(sender, plugin: PluginBase, **kwargs):
"""Handle plugin called signal."""
inquiry_type: Optional[str] = kwargs.get("inquiry_type")
logger.info("%s - %s inquiry_type: %s", formatted_text(prefix + "plugin_called"), plugin.name, inquiry_type)
[docs]
@receiver(plugin_responded, dispatch_uid="plugin_responded")
def handle_plugin_responded(sender, plugin: PluginBase, **kwargs):
"""Handle plugin responded signal."""
inquiry_type: Optional[str] = kwargs.get("inquiry_type")
# plugins send their response as ``response``.
inquiry_return: Optional[Union[SmarterDictType, list, str]] = kwargs.get("response", kwargs.get("inquiry_return"))
try:
inquiry_return = json.loads(inquiry_return) if isinstance(inquiry_return, str) else inquiry_return
except (TypeError, json.JSONDecodeError):
pass
logger.info(
"%s - %s inquiry_type: %s inquiry_return: %s",
formatted_text(prefix + "plugin_responded"),
plugin.name,
inquiry_type,
formatted_json(inquiry_return) if isinstance(inquiry_return, (dict, list)) else inquiry_return,
)
[docs]
@receiver(plugin_ready, dispatch_uid="plugin_ready")
def handle_plugin_ready(sender, plugin: PluginBase, **kwargs):
"""Handle plugin ready signal."""
logger.info("%s - %s", formatted_text(prefix + "plugin_ready"), plugin.name)
[docs]
@receiver(plugin_selected, dispatch_uid="plugin_selected")
def handle_plugin_selected(sender, *args, **kwargs):
"""Handle plugin selected signal."""
# plugin: PluginBase, user, messages: list[dict], search_term: str, messages: list[dict] = None
input_text: Optional[str] = kwargs.get("input_text")
plugin: Optional[PluginBase] = kwargs.get("plugin")
user = kwargs.get("user")
messages: list[SmarterDictType] = kwargs.get("messages", [])
search_term: str = kwargs.get("search_term", "")
user_id: int = user.id if user else None # type: ignore
prompt = input_text if input_text else formatted_json(messages)
logger.info(
"signal received for %s - %s search_term: %s prompt(s): %s",
formatted_text(prefix + "plugin_selected"),
plugin.name if plugin else "Unknown Plugin",
search_term,
prompt,
)
if plugin is None or plugin.id is None:
logger.warning(
"%s received without a plugin id. Plugin selector history is not recorded.",
formatted_text(prefix + "plugin_selected"),
)
return
create_plugin_selector_history.delay(
plugin_id=plugin.id,
user_id=user_id,
input_text=input_text,
messages=messages,
search_term=search_term,
)
[docs]
@receiver(search_term_matched, dispatch_uid="search_term_matched")
def handle_search_term_matched(sender, prompt: str, search_term: str, method: str, matched_text=None, **kwargs):
"""Handle search term matched signal."""
logger.info(
"signal received for %s - search_term: %s method: %s matched_text: %s prompt: %s",
formatted_text(prefix + "search_term_matched"),
search_term,
method,
matched_text,
prompt,
)
# ------------------------------------------------------------------------------
# Django model receivers.
# ------------------------------------------------------------------------------
[docs]
@receiver(post_save, sender=PluginSelector)
def handle_plugin_selector_saved(sender, instance, created, **kwargs):
"""Handle plugin selector saved signal."""
if created:
logger.info("%s", formatted_text(prefix + "post_save() PluginSelector() record created."))
else:
logger.info(
"%s - %s",
formatted_text(prefix + "post_save() PluginSelector() record updated:"),
formatted_json(model_to_dict(instance)),
)
[docs]
@receiver(post_save, sender=PluginPrompt)
def handle_plugin_prompt_saved(sender, instance, created, **kwargs):
"""Handle plugin prompt saved signal."""
if created:
logger.info("%s", formatted_text(prefix + "post_save() PluginPrompt() record created."))
else:
logger.info(
"%s - %s",
formatted_text(prefix + "post_save() PluginPrompt() record updated:"),
formatted_json(model_to_dict(instance)),
)
[docs]
@receiver(post_save, sender=PluginDataStatic)
def handle_plugin_data_saved(sender, instance, created, **kwargs):
"""Handle plugin data saved signal."""
if created:
logger.info("%s", formatted_text(prefix + "post_save() PluginDataStatic() record created."))
else:
logger.info(
"%s - %s",
formatted_text(prefix + "post_save() PluginDataStatic() record updated:"),
formatted_json(model_to_dict(instance)),
)
[docs]
@receiver(post_save, sender=PluginSelectorHistory)
def handle_plugin_selector_history_saved(sender, instance, created, **kwargs):
"""Handle plugin selector history saved signal."""
if created:
logger.info(
"%s - %s",
formatted_text(prefix + "post_save() PluginSelectorHistory() created"),
formatted_json(model_to_dict(instance)),
)
else:
logger.info(
"%s - %s",
formatted_text(prefix + "post_save() PluginSelectorHistory() updated"),
formatted_json(model_to_dict(instance)),
)
[docs]
@receiver(post_save, sender=PluginDataApi)
def handle_plugin_data_api_saved(sender, instance, created, **kwargs):
"""Handle plugin data API saved signal."""
if created:
logger.info(
"%s - %s",
formatted_text(prefix + "post_save() PluginDataApi() created"),
formatted_json(model_to_dict(instance)),
)
else:
logger.info(
"%s - %s",
formatted_text(prefix + "post_save() PluginDataApi() updated"),
formatted_json(model_to_dict(instance)),
)
[docs]
@receiver(post_save, sender=PluginDataSql)
def handle_plugin_data_sql_saved(sender, instance, created, **kwargs):
"""Handle plugin data SQL saved signal."""
if created:
logger.info(
"%s - %s",
formatted_text(prefix + "post_save() PluginDataSql() created"),
formatted_json(model_to_dict(instance)),
)
else:
logger.info(
"%s - %s",
formatted_text(prefix + "post_save() PluginDataSql() updated"),
formatted_json(model_to_dict(instance)),
)
# ------------------------------------------------------------------------------
# pre_delete signals for
# PluginDataApi,
# PluginDataSql,
# PluginDataStatic,
# PluginMeta,
# PluginPrompt,
# PluginSelector,
# PluginSelectorHistory,
# ------------------------------------------------------------------------------
[docs]
@receiver(pre_delete, sender=PluginDataApi)
def handle_plugin_data_api_pre_delete(sender, instance, **kwargs):
"""Handle pre-delete signal for PluginDataApi."""
logger.info(
"%s - %s deleting.",
formatted_text(prefix + "PluginDataApi().pre_delete()"),
instance,
)
[docs]
@receiver(pre_delete, sender=PluginDataSql)
def handle_plugin_data_sql_pre_delete(sender, instance, **kwargs):
"""Handle pre-delete signal for PluginDataSql."""
logger.info(
"%s - %s deleting.",
formatted_text(prefix + "PluginDataSql().pre_delete()"),
instance,
)
[docs]
@receiver(pre_delete, sender=PluginDataStatic)
def handle_plugin_data_static_pre_delete(sender, instance, **kwargs):
"""Handle pre-delete signal for PluginDataStatic."""
logger.info(
"%s - %s deleting.",
formatted_text(prefix + "PluginDataStatic().pre_delete()"),
instance,
)
[docs]
@receiver(pre_delete, sender=PluginPrompt)
def handle_plugin_prompt_pre_delete(sender, instance, **kwargs):
"""Handle pre-delete signal for PluginPrompt."""
logger.info(
"%s - %s deleting.",
formatted_text(prefix + "PluginPrompt().pre_delete()"),
instance,
)
[docs]
@receiver(pre_delete, sender=PluginSelector)
def handle_plugin_selector_pre_delete(sender, instance, **kwargs):
"""Handle pre-delete signal for PluginSelector."""
logger.info(
"%s - %s deleting.",
formatted_text(prefix + "PluginSelector().pre_delete()"),
instance,
)
[docs]
@receiver(pre_delete, sender=PluginSelectorHistory)
def handle_plugin_selector_history_pre_delete(sender, instance, **kwargs):
"""Handle pre-delete signal for PluginSelectorHistory."""
logger.info(
"%s - %s deleting.",
formatted_text(prefix + "PluginSelectorHistory().pre_delete()"),
instance,
)
[docs]
@receiver(broker_ready, dispatch_uid="broker_ready")
def handle_broker_ready(sender, broker: AbstractBroker, **kwargs):
"""Handle broker ready signal."""
logger.info(
"%s %s %s for %s is ready.",
formatted_text(f"{prefix}broker_ready()"),
broker.kind,
str(broker),
broker.name,
)
[docs]
@receiver(websearch_searched, dispatch_uid="websearch_searched")
def handle_websearch_searched(sender, plugin: PluginBase, query: str, provider: str, result_count: int, **kwargs):
"""Handle websearch searched signal."""
logger.info(
"%s - %s query: %s provider: %s results: %s cached: %s",
formatted_text(prefix + "websearch_searched"),
plugin.name,
query,
provider,
result_count,
kwargs.get("cached"),
)
[docs]
@receiver(websearch_fetched, dispatch_uid="websearch_fetched")
def handle_websearch_fetched(sender, plugin: PluginBase, url: str, **kwargs):
"""Handle websearch fetched signal."""
logger.info(
"%s - %s url: %s final_url: %s characters: %s truncated: %s cached: %s",
formatted_text(prefix + "websearch_fetched"),
plugin.name,
url,
kwargs.get("final_url"),
kwargs.get("characters"),
kwargs.get("truncated"),
kwargs.get("cached"),
)
[docs]
@receiver(websearch_failed, dispatch_uid="websearch_failed")
def handle_websearch_failed(sender, plugin: PluginBase, operation: str, target: str, error: str, **kwargs):
"""Handle websearch failed signal."""
logger.warning(
"%s - %s %s %s failed: %s",
formatted_text(prefix + "websearch_failed"),
plugin.name,
operation,
target,
error,
)
[docs]
@receiver(post_save, sender=PluginDataWebsearch)
def handle_plugin_data_websearch_saved(sender, instance, created, **kwargs):
"""Handle plugin data websearch saved signal."""
logger.info(
"%s - %s",
formatted_text(prefix + f"post_save() PluginDataWebsearch() {'created' if created else 'updated'}"),
formatted_json(instance.data()),
)
[docs]
@receiver(pre_delete, sender=PluginDataWebsearch)
def handle_plugin_data_websearch_pre_delete(sender, instance, **kwargs):
"""Handle pre-delete signal for PluginDataWebsearch."""
logger.info(
"%s - %s deleting.",
formatted_text(prefix + "PluginDataWebsearch().pre_delete()"),
instance,
)