"""Shared utilities used across tool modules.""" from __future__ import annotations import asyncio import logging import re from datetime import date, datetime from difflib import SequenceMatcher logger = logging.getLogger(__name__) _PUNCT_RE = re.compile(r"[^\w\s]") def schedule_embedding(note_id: int, user_id: int, title: str, body: str) -> None: """Fire-and-forget: update the embedding for a note after it's created/modified.""" from fabledassistant.services.embeddings import upsert_note_embedding text = f"{title}\n{body}".strip() if body else (title or "") if text: asyncio.create_task(upsert_note_embedding(note_id, user_id, text)) _PROJECT_QUERY_NOISE = {"project", "projects"} def _normalize(s: str) -> str: """Lowercase and collapse non-alphanumerics to single spaces.""" return re.sub(r"[^a-z0-9]+", " ", s.lower()).strip() def _normalize_query(query: str) -> str: """Normalize plus drop trailing type-nouns ('project' / 'projects') that users add as filler when referring to a project by name.""" tokens = [t for t in _normalize(query).split() if t not in _PROJECT_QUERY_NOISE] return " ".join(tokens) def score_project_match(query: str, project) -> float: """Score how well `query` matches `project`. Range [0.0, 1.0]. Tiered: exact title → 1.0, substring either-way → 0.85, query found in description/summary → 0.70, otherwise SequenceMatcher ratio against the title. Substring tiers exist because LLM-generated colloquial queries (e.g. "famous supply project" for "Famous-Supply Work topics") would otherwise score too low under pure SequenceMatcher and be treated as no match. Filler words like "project" are stripped from the query so the substring check still fires. """ q = _normalize_query(query) if not q: return 0.0 title = _normalize(project.title) description = _normalize(project.description or "") summary = _normalize(project.auto_summary or "") combined = f"{title} {description} {summary}".strip() if q == title: return 1.0 if q in title or title in q: return 0.85 if q in combined: return 0.70 # SequenceMatcher against the title — comparing against `combined` # dilutes the ratio with long description/summary text and produces # uniformly low scores even for plausible matches. return SequenceMatcher(None, q, title).ratio() async def resolve_project(user_id: int, project_name: str): """Exact-then-scored project lookup. Returns the Project or None. Resolution order: 1. Exact title match (case-insensitive via DB query). 2. Highest `score_project_match` across all projects, threshold 0.55. """ from fabledassistant.services.projects import get_project_by_title, list_projects proj = await get_project_by_title(user_id, project_name) if proj is not None: return proj all_p = await list_projects(user_id) best, best_score = None, 0.0 for p in all_p: score = score_project_match(project_name, p) if score >= 0.55 and score > best_score: best, best_score = p, score return best def parse_due_date(value: str | None) -> date | None: """Parse a due date string, returning None on failure.""" if not value: return None try: return datetime.strptime(value, "%Y-%m-%d").date() except (ValueError, TypeError): logger.warning("Invalid due_date format: %s", value) return None def fuzzy_title_match(title: str, candidates, threshold: float = 0.82): """Return (best_match, ratio) if any candidate's title is similar enough. Uses SequenceMatcher ratio. Threshold 0.82 catches near-duplicates like "Game Premise" / "Game Premise Notes" while leaving clearly different titles alone. Returns (None, 0.0) when no candidate meets the threshold. """ needle = title.lower().strip() best, best_r = None, 0.0 for c in candidates: r = SequenceMatcher(None, needle, c.title.lower().strip()).ratio() if r >= threshold and r > best_r: best, best_r = c, r return best, best_r async def check_duplicate( user_id: int, title: str, body: str, is_task: bool, confirmed: bool, ) -> dict | None: """Check for exact, fuzzy, and semantic duplicates. Returns error dict or None.""" from fabledassistant.services.notes import list_notes item_label = "task" if is_task else "note" existing, _ = await list_notes(user_id=user_id, q=title, is_task=is_task, limit=1) exact = next((n for n in existing if n.title.lower() == title.lower()), None) if exact is not None: return { "success": False, "error": f"A {item_label} titled '{title}' already exists (id: {exact.id}). Use update_note to modify it instead of creating a duplicate.", } clean_q = _PUNCT_RE.sub(" ", title).strip() candidates, _ = await list_notes(user_id=user_id, q=clean_q, is_task=is_task, limit=20) near, ratio = fuzzy_title_match(title, candidates) if near is not None: return { "success": False, "requires_confirmation": True, "similar_note": {"id": near.id, "title": near.title}, "error": f"A {item_label} with a very similar title '{near.title}' already exists (similarity: {ratio:.0%}). Ask the user to confirm before creating a separate entry.", } if not confirmed and len(body.strip()) >= 80: from fabledassistant.services.embeddings import semantic_search_notes as _ssn sem_query = f"{title}\n{body}".strip() sem_hits = await _ssn(user_id, sem_query, limit=3, threshold=0.90, is_task=is_task) if sem_hits: best_score, best_note = sem_hits[0] return { "success": False, "requires_confirmation": True, "similar_note": {"id": best_note.id, "title": best_note.title}, "error": f"A {item_label} with very similar content exists: '{best_note.title}' (semantic similarity: {best_score:.0%}). Ask the user to confirm before creating a separate entry.", } return None