"""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)) def _normalize(s: str) -> str: """Lowercase and collapse non-alphanumerics to single spaces.""" return re.sub(r"[^a-z0-9]+", " ", s.lower()).strip() 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. 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. """ q = _normalize(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 return SequenceMatcher(None, q, combined).ratio() async def resolve_project(user_id: int, project_name: str): """Exact-then-fuzzy project lookup. Returns the Project or None. Resolution order: 1. Exact title match (case-insensitive via DB) 2. project_name is a substring of an existing title 3. Existing title is a substring of project_name 4. SequenceMatcher ratio >= 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 needle = project_name.lower().strip() all_p = await list_projects(user_id) for p in all_p: haystack = p.title.lower().strip() if needle in haystack or haystack in needle: return p best, best_r = None, 0.0 for p in all_p: r = SequenceMatcher(None, needle, p.title.lower().strip()).ratio() if r >= 0.55 and r > best_r: best, best_r = p, r 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