Building maintenance powered by AI

# ri/stickers/signals.py — Signal that triggers automatic regeneration
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Stickers
from ri.buildings.models import AISummaryDebounce
from ri.buildings.tasks import generate_ai_summary_for_building

DEBOUNCE_WINDOW = 30  # seconds

@receiver(post_save, sender=Stickers)
def schedule_ai_summary_on_sticker_change(sender, instance, **kwargs):
    # Only if the sticker has a building and is not removed
    if not instance.building_id or instance.removed:
        return

    # Debounce: prevents multiple API calls in a burst
    if AISummaryDebounce.should_schedule(
        instance.building_id, DEBOUNCE_WINDOW
    ):
        generate_ai_summary_for_building.apply_async(
            args=[instance.building_id],
            countdown=DEBOUNCE_WINDOW,
        )

# ri/core/ai.py — Direct DeepSeek API call without external SDK
def call_deepseek_api(prompt, model='deepseek-v4-flash', max_words=200):
    config = Configuration.get_solo()
    body = {
        "model": model,
        "messages": [
            {"role": "system", "content": "You are an expert assistant in building maintenance."},
            {"role": "user", "content": prompt},
        ],
        "max_tokens": min(int(max_words * 3) + 500, 384000),
        "temperature": 0.3,
    }

    if 'pro' in model:
        body["reasoning_effort"] = "high"

    response = requests.post(
        f"{config.deepseek_base_url}/chat/completions",
        headers={"Authorization": f"Bearer {config.deepseek_api_key}"},
        json=body,
        timeout=60,
    )
    response.raise_for_status()
    return _clean_output(response.json()["choices"][0]["message"]["content"])

# ri/core/chroma_client.py — Hybrid retrieval (semantic + lexical)
def query_edificio(building_id, embedding, n=8, pregunta=None, lexical_weight=0.5):
    # Fetch ALL of the building's docs (exact recall, no HNSW)
    query_terms = content_terms(pregunta)  # no accents/stopwords, singular

    for i, doc in enumerate(docs):
        sim = cosine(embedding, embs[i])      # semantic similarity
        matched = query_terms & doc_terms[i]  # literal matches
        # rare terms (high IDF) weigh more
        lexical = sum(idf(t) for t in matched) / idf_total if matched else 0.0
        combined = sim + lexical_weight * lexical  # hybrid ranking

    # A doc literally mentioning what was asked climbs the ranking
    # even when the embedder misses it. Generic: works for any term.
    return top_n(scored, n)