"""Brain v2 — intelligent matching with memory, semantics, and adaptive learning.""" from __future__ import annotations import json import logging import math import random import re import time from collections import Counter from dataclasses import dataclass, field from typing import Optional from bot.db.core import _exec, _execmany, _fetchone, _fetchall logger = logging.getLogger(__name__) # ═══════════════════════════════════════════════ # Persian Synonym Dictionary # ═══════════════════════════════════════════════ PERSIAN_SYNONYMS: dict[str, set[str]] = { "سلام": {"درود", "سلا", "احوال", "سلام علیکم", "علیک سلام"}, "خوبی": {"خوب", "خوبم", "حالت", "احوال"}, "خوب": {"خوبم", "عالی", "خوش", "بهتر", "باحال", "خوبه"}, "بد": {"بدم", "خراب", "ناخوش", "بده", "بدترین"}, "تشکر": {"مرسی", "ممنون", "سپاس", "ممنونم", "تشکر میکنم", "دستت درد نکنه", "مرس"}, "بله": {"آره", "آری", "بله", "بلی", "اره", "آره", "yeah"}, "نه": {"نخیر", "خیر", "no", "ن", "هرگز"}, "خداحافظ": {"بای", "فعلا", "بابای", "خدا حافظ", "خداحافظی"}, "چطوری": {"چطور", "چونی", "چی", "چطورین", "چه خبر"}, "اسم": {"نام", "لقب", "اسمو"}, "چند": {"چندتا", "چه تعداد", "چندتایی"}, "کجا": {"کجایی", "کدوم جا", "کجاها"}, "کی": {"کی", "چه موقع", "کِی"}, "چی": {"چه", "چیه", "چیست", "چیز"}, "همه": {"تمامی", "همگی", "جملگی", "همشون"}, "میشه": {"می‌شه", "میشود", "می‌شود", "ممکنه"}, "عزیز": {"عزیزم", "جان", "دلبر"}, "دوست": {"رفیق", "یار", "دوستی"}, "بازی": {"گیم", "سرگرمی", "بازی کردن"}, "کار": {"شغل", "کارم"}, "مادر": {"مامان", "مادری", "مادرت"}, "بابا": {"پدر", "بابایی", "پدرت"}, "داداش": {"برادر", "داداشی", "برادرزاده"}, "خواهر": {"خواهری", "خواهرم"}, "فکر": {"فک کنم", "گمان", "نظر"}, "بیا": {"بیاین", "بیائید", "بیایید", "بی"}, "برو": {"برین", "برید"}, "ببین": {"نگاه", "بین", "ببینم", "نگاه کن"}, "گوش": {"گوش بده", "گوش کن", "بشنو"}, "خنده": {"خندیدم", "بخند", "خنده دار"}, "گریه": {"گرییدم", "اشک", "تاسف"}, "هوش": {"هوش مصنوعی", "AI", "هوشمند"}, "ربات": {"بات", "bot"}, "حرف": {"سخن", "گفتار", "حرف زدن"}, "جواب": {"پاسخ", "پاسخت"}, "سوال": {"پرسش", "سوالت", "بپرس"}, "عکس": {"تصویر", "عکسی", "فتو"}, "فیلم": {"ویدیو", "ویدئو", "مووی"}, "آهنگ": {"موزیک", "موسیقی", "ترانه", "آواز"}, "کتاب": {"کتابخونه", "کتابی", "کتابخونه"}, "دانشگاه": {"دانشگا", "آموزش عالی", "فرهنگ"}, "درس": {"تحصیل", "مدرسه"}, "پول": {"فروش", "قیمت", "پولی"}, "زمان": {"وقت", "مدت", "زمانی"}, "امروز": {"امروزه", "این روزا"}, "دیروز": {"دیروزی"}, "فردا": {"فردایی"}, } # ═══════════════════════════════════════════════ # Data Models # ═══════════════════════════════════════════════ @dataclass class Response: id: int = 0 chat_id: int = 0 trigger: str = "" response: str = "" category: str = "general" media_type: str = "text" file_id: str = "" weight: float = 1.0 delay: float = 0.0 tags: list[str] = field(default_factory=list) created_by: int = 0 created_at: int = 0 last_used: int = 0 uses_count: int = 0 def to_dict(self) -> dict: return { "id": self.id, "chat_id": self.chat_id, "trigger": self.trigger, "response": self.response, "category": self.category, "media_type": self.media_type, "file_id": self.file_id, "weight": self.weight, "delay": self.delay, "tags": self.tags, "created_by": self.created_by, "created_at": self.created_at, "last_used": self.last_used, "uses_count": self.uses_count, } @dataclass class MatchResult: response: Response score: float method: str # ═══════════════════════════════════════════════ # Text Normalization # ═══════════════════════════════════════════════ def _norm(text: str) -> str: text = text.strip().lower() text = text.replace("\u200c", " ") text = re.sub(r"[^\w\s\d\u0600-\u06FF]", " ", text) text = re.sub(r"\s+", " ", text).strip() return text def _tokens(text: str) -> set[str]: return set(_norm(text).split()) def _expand_tokens(tokens: set[str], chat_id: int = 0) -> set[str]: """Expand token set with synonyms + learned semantic links.""" expanded = set(tokens) for t in tokens: syns = PERSIAN_SYNONYMS.get(t, set()) expanded.update(syns) for s in syns: deeper = PERSIAN_SYNONYMS.get(s, set()) expanded.update(deeper) return expanded # ═══════════════════════════════════════════════ # Cooldown (in-memory) # ═══════════════════════════════════════════════ _COOLDOWN: dict[str, float] = {} _LAST_CLEANUP = 0.0 def _cleanup_cooldown(): global _LAST_CLEANUP now = time.time() if now - _LAST_CLEANUP < 300: return _LAST_CLEANUP = now expired = [k for k, v in _COOLDOWN.items() if now - v > 600] for k in expired: del _COOLDOWN[k] async def check_cooldown(chat_id: int, key: str, seconds: int = 5) -> bool: _cleanup_cooldown() full_key = f"{chat_id}:{key}" now = time.time() last = _COOLDOWN.get(full_key, 0) if now - last < seconds: return True _COOLDOWN[full_key] = now return False # ═══════════════════════════════════════════════ # Persistent Conversation Memory # ═══════════════════════════════════════════════ async def remember_message( chat_id: int, user_id: int, text: str, role: str = "user", topics: list[str] = None, ) -> None: """Store a message in persistent conversation memory.""" if not text.strip(): return topics_json = json.dumps(topics or [], ensure_ascii=False) await _exec( "INSERT INTO conversation_memory (chat_id, user_id, role, message_text, topics) VALUES (?, ?, ?, ?, ?)", (chat_id, user_id, role, text.strip(), topics_json), ) max_mem = 200 await _exec( "DELETE FROM conversation_memory WHERE id NOT IN (SELECT id FROM conversation_memory WHERE chat_id=? ORDER BY created_at DESC LIMIT ?)", (chat_id, max_mem), ) async def get_recent_context(chat_id: int, n: int = 10) -> list[dict]: """Get last N messages from conversation memory.""" rows = await _fetchall( "SELECT * FROM conversation_memory WHERE chat_id=? ORDER BY created_at DESC LIMIT ?", (chat_id, n), ) result = [] for r in reversed(rows): topics = [] try: if r["topics"]: topics = json.loads(r["topics"]) except (json.JSONDecodeError, TypeError): pass result.append({ "id": r["id"], "user_id": r["user_id"], "role": r["role"], "text": r["message_text"], "topics": topics, "created_at": r["created_at"], }) return result async def extract_topics(text: str, n: int = 5) -> list[str]: """Extract topic keywords from text — nouns and important words.""" tokens = _norm(text).split() if not tokens: return [] stopwords = { "این", "آن", "یه", "یک", "و", "با", "به", "از", "در", "برای", "که", "را", "ها", "های", "است", "هست", "شد", "شده", "دارد", "اما", "اگر", "یا", "تا", "چه", "چی", "چطور", "چرا", "چون", "خیلی", "بسیار", "بعد", "قبل", "حالا", "الان", "میشه", "باشه", } filtered = [t for t in tokens if t not in stopwords and len(t) > 1] counter = Counter(filtered) return [word for word, _ in counter.most_common(n)] async def get_conversation_topics(chat_id: int, n: int = 5) -> set[str]: """Get topic keywords from recent conversation memory.""" rows = await _fetchall( "SELECT message_text FROM conversation_memory WHERE chat_id=? ORDER BY created_at DESC LIMIT 20", (chat_id,), ) all_words = [] for r in rows: all_words.extend(_norm(r["message_text"]).split()) stopwords = { "این", "آن", "یه", "یک", "و", "با", "به", "از", "در", "برای", "که", "را", "ها", "های", "است", "هست", "شد", "شده", "دارد", "اما", "اگر", "یا", "تا", "و", "خوب", "باشه", } filtered = [w for w in all_words if w not in stopwords and len(w) > 1] counter = Counter(filtered) top_n = counter.most_common(n * 3) return set(word for word, _ in top_n) # ═══════════════════════════════════════════════ # Semantic Links (learned associations) # ═══════════════════════════════════════════════ async def add_semantic_link(chat_id: int, word_a: str, word_b: str) -> None: """Learn a semantic association between two words.""" a, b = sorted([word_a.strip().lower(), word_b.strip().lower()]) if a == b or not a or not b: return existing = await _fetchone( "SELECT strength FROM semantic_links WHERE chat_id=? AND word_a=? AND word_b=?", (chat_id, a, b), ) if existing: await _exec( "UPDATE semantic_links SET strength=strength+1 WHERE chat_id=? AND word_a=? AND word_b=?", (chat_id, a, b), ) else: await _exec( "INSERT INTO semantic_links (chat_id, word_a, word_b) VALUES (?, ?, ?)", (chat_id, a, b), ) async def get_semantic_expansions(chat_id: int, word: str) -> set[str]: """Get all semantically linked words for a given word.""" w = word.strip().lower() rows = await _fetchall( "SELECT word_a, word_b FROM semantic_links WHERE chat_id=? AND (word_a=? OR word_b=?)", (chat_id, w, w), ) result = set() for r in rows: if r["word_a"] == w: result.add(r["word_b"]) else: result.add(r["word_a"]) return result async def get_all_semantic_links(chat_id: int, min_strength: int = 1) -> list[dict]: rows = await _fetchall( "SELECT * FROM semantic_links WHERE chat_id=? AND strength>=? ORDER BY strength DESC LIMIT 500", (chat_id, min_strength), ) return [dict(r) for r in rows] # ═══════════════════════════════════════════════ # Learning from Corrections # ═══════════════════════════════════════════════ _CORRECTION_PATTERNS = [ re.compile(r"نه\s+(.*)"), re.compile(r"نخیر\s+(.*)"), re.compile(r"نه منظورم\s+(.*)"), re.compile(r"منظور من\s+(.*)"), re.compile(r"من گفتم\s+(.*)"), re.compile(r"گفتم که\s+(.*)"), ] async def learn_from_correction( chat_id: int, user_id: int, user_message: str, bot_response: str, ) -> bool: """Detect if user is correcting the bot and learn from it.""" user_lower = user_message.strip().lower() bot_lower = bot_response.strip().lower() if bot_response else "" for pattern in _CORRECTION_PATTERNS: m = pattern.search(user_lower) if m: correct_text = m.group(1).strip() if not correct_text or len(correct_text) < 2: continue existing = await _fetchone( "SELECT id, uses_count FROM brain WHERE chat_id=? AND response_text=? ORDER BY uses_count DESC LIMIT 1", (chat_id, bot_lower[:100]), ) if existing and existing["uses_count"] > 1: await _exec( "DELETE FROM brain WHERE id=?", (existing["id"],), ) await add(chat_id, correct_text, bot_lower, category="correction", created_by=user_id) trigger_words = _tokens(user_lower) response_words = _tokens(correct_text) for tw in trigger_words: for rw in response_words: if tw != rw and len(tw) > 1 and len(rw) > 1: await add_semantic_link(chat_id, tw, rw) return True return False # ═══════════════════════════════════════════════ # DB Operations # ═══════════════════════════════════════════════ async def add( chat_id: int, trigger: str, response: str, category: str = "general", media_type: str = "text", file_id: str = "", weight: float = 1.0, delay: float = 0.0, tags: list[str] = None, created_by: int = 0, ) -> Response: trigger = trigger.strip().lower() tags_json = json.dumps(tags or [], ensure_ascii=False) cur = await _exec( """INSERT INTO brain (chat_id, trigger_text, response_text, category, media_type, file_id, weight, delay, tags, created_by, uses_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0)""", (chat_id, trigger, response, category, media_type, file_id, weight, delay, tags_json, created_by), ) count = await _fetchone("SELECT COUNT(*) AS c FROM brain WHERE chat_id=?", (chat_id,)) if count and count["c"] > 2000: await _prune_low_quality(chat_id, count["c"]) return Response( id=cur.lastrowid if cur else 0, chat_id=chat_id, trigger=trigger, response=response, category=category, media_type=media_type, file_id=file_id, weight=weight, delay=delay, tags=tags or [], created_by=created_by, created_at=int(time.time()), ) async def _prune_low_quality(chat_id: int, total: int): """Remove bottom 10% lowest-quality entries when over cap.""" excess = total - 1800 if excess <= 0: return await _exec( "DELETE FROM brain WHERE chat_id=? AND id IN (" "SELECT id FROM brain WHERE chat_id=? ORDER BY uses_count ASC, weight ASC, last_used ASC NULLS FIRST LIMIT ?" ")", (chat_id, chat_id, excess + 100), ) async def remove(chat_id: int, trigger: str, response_id: int = None) -> int: trigger = trigger.strip().lower() if response_id is not None: cur = await _exec("DELETE FROM brain WHERE chat_id=? AND id=?", (chat_id, response_id)) else: cur = await _exec("DELETE FROM brain WHERE chat_id=? AND trigger_text=?", (chat_id, trigger)) return cur.rowcount if cur else 0 async def remove_all(chat_id: int) -> int: cur = await _exec("DELETE FROM brain WHERE chat_id=?", (chat_id,)) return cur.rowcount if cur else 0 async def get_all(chat_id: int, category: str = None) -> list[Response]: if category: rows = await _fetchall( "SELECT * FROM brain WHERE chat_id=? AND category=? ORDER BY uses_count DESC, weight DESC", (chat_id, category), ) else: rows = await _fetchall( "SELECT * FROM brain WHERE chat_id=? ORDER BY uses_count DESC, weight DESC", (chat_id,), ) return [_row_to_response(r) for r in rows] async def search(chat_id: int, query: str) -> list[Response]: q = f"%{query.strip().lower()}%" rows = await _fetchall( "SELECT * FROM brain WHERE chat_id=? AND (trigger_text LIKE ? OR response_text LIKE ? OR category LIKE ?) ORDER BY uses_count DESC", (chat_id, q, q, q), ) return [_row_to_response(r) for r in rows] async def get_categories(chat_id: int) -> list[dict]: rows = await _fetchall( "SELECT category, COUNT(*) as cnt FROM brain WHERE chat_id=? GROUP BY category ORDER BY cnt DESC", (chat_id,), ) return [dict(r) for r in rows] async def get_stats(chat_id: int) -> dict: total = await _fetchone("SELECT COUNT(*) AS c FROM brain WHERE chat_id=?", (chat_id,)) cats = await _fetchone("SELECT COUNT(DISTINCT category) AS c FROM brain WHERE chat_id=?", (chat_id,)) media = await _fetchone("SELECT COUNT(*) AS c FROM brain WHERE chat_id=? AND media_type!='text'", (chat_id,)) most = await _fetchone( "SELECT trigger_text, uses_count FROM brain WHERE chat_id=? ORDER BY uses_count DESC LIMIT 1", (chat_id,), ) conv = await _fetchone("SELECT COUNT(*) AS c FROM conversation_memory WHERE chat_id=?", (chat_id,)) links = await _fetchone("SELECT COUNT(*) AS c FROM semantic_links WHERE chat_id=?", (chat_id,)) return { "total": total["c"] if total else 0, "categories": cats["c"] if cats else 0, "media": media["c"] if media else 0, "most_used": dict(most) if most else None, "memory_size": conv["c"] if conv else 0, "semantic_links": links["c"] if links else 0, } async def increment_uses(chat_id: int, response_id: int): await _exec( "UPDATE brain SET uses_count=uses_count+1, last_used=? WHERE id=? AND chat_id=?", (int(time.time()), response_id, chat_id), ) async def get_by_id(chat_id: int, response_id: int) -> Optional[Response]: row = await _fetchone("SELECT * FROM brain WHERE chat_id=? AND id=?", (chat_id, response_id)) return _row_to_response(row) if row else None # ═══════════════════════════════════════════════ # Matching Engine — 9 Levels # ═══════════════════════════════════════════════ async def match(chat_id: int, text: str) -> Optional[MatchResult]: text_lower = text.strip().lower() text_norm = _norm(text) if not text_lower: return None all_responses = await _fetchall("SELECT * FROM brain WHERE chat_id=?", (chat_id,)) if not all_responses: return None responses = [_row_to_response(r) for r in all_responses] msg_tokens = _tokens(text) msg_expanded = _expand_tokens(msg_tokens, chat_id) # Topic context for boost recent_topics = await get_conversation_topics(chat_id) def _pick(candidates: list[Response], method: str, score: float, topic_boost: bool = False) -> Optional[MatchResult]: if not candidates: return None if len(candidates) == 1: r = candidates[0] if topic_boost and recent_topics: rt = _tokens(r.trigger) if rt & recent_topics: score = min(score + 0.08, 0.99) return MatchResult(response=r, score=score, method=method) weights = [] for r in candidates: w = max(r.weight, 0.1) if topic_boost and recent_topics: rt = _tokens(r.trigger) if rt & recent_topics: w *= 1.5 weights.append(w) total = sum(weights) rnd = random.uniform(0, total) cumulative = 0.0 for i, w in enumerate(weights): cumulative += w if rnd <= cumulative: return MatchResult(response=candidates[i], score=score, method=method) return MatchResult(response=candidates[-1], score=score, method=method) # ── Level 1: Exact match (score: 1.0) ── exact = [r for r in responses if r.trigger == text_lower] if exact: result = _pick(exact, "exact", 1.0, topic_boost=True) await increment_uses(chat_id, result.response.id) return result # ── Level 2: Normalized exact (score: 0.95) ── norm_exact = [r for r in responses if _norm(r.trigger) == text_norm] if norm_exact: result = _pick(norm_exact, "normalized", 0.95, topic_boost=True) await increment_uses(chat_id, result.response.id) return result # ── Level 3: Wildcard with * (score: 0.9) ── wildcard = [] for r in responses: if "*" in r.trigger: pattern = re.escape(r.trigger).replace(r"\*", ".*") if re.fullmatch(pattern, text_lower) or re.fullmatch(pattern, text_norm): wildcard.append(r) if wildcard: result = _pick(wildcard, "wildcard", 0.9, topic_boost=True) await increment_uses(chat_id, result.response.id) return result # ── Level 4: Contains match — longest trigger wins (score: 0.8-0.9) ── contains = [] for r in responses: if len(r.trigger) < 2: continue if r.trigger in text_lower or _norm(r.trigger) in text_norm: contains.append(r) if contains: contains.sort(key=lambda r: len(r.trigger), reverse=True) best_len = len(contains[0].trigger) score = 0.8 + min(best_len / 20, 0.1) result = _pick(contains[:5], "contains", score, topic_boost=True) await increment_uses(chat_id, result.response.id) return result # ── Level 5: All-token match (core words in message, score: 0.7) ── keyword_matches = [] for r in responses: rtokens = set(_norm(r.trigger).split()) if len(rtokens) >= 2: if rtokens.issubset(msg_tokens) or rtokens.issubset(msg_expanded): keyword_matches.append(r) if keyword_matches: result = _pick(keyword_matches, "keyword", 0.7, topic_boost=True) await increment_uses(chat_id, result.response.id) return result # ── Level 6: Synonym-expanded match (score: 0.65-0.75) ── syn_matches = [] for r in responses: rtokens = set(_norm(r.trigger).split()) r_expanded = _expand_tokens(rtokens, chat_id) if len(rtokens) >= 2: overlap = len(rtokens & msg_expanded) ratio = overlap / max(len(rtokens), 1) if ratio >= 0.6: syn_matches.append((r, ratio)) elif len(rtokens) == 1: if rtokens & msg_expanded: syn_matches.append((r, 0.6)) if syn_matches: syn_matches.sort(key=lambda x: -x[1]) best_ratio = syn_matches[0][1] score = 0.65 + best_ratio * 0.1 result = _pick([r for r, _ in syn_matches[:5]], "synonym", score, topic_boost=True) await increment_uses(chat_id, result.response.id) return result # ── Level 7: IDF-weighted semantic similarity (score: 0.5-0.65) ── idf_matches = [] doc_freq: dict[str, int] = {} for r in responses: for w in set(_norm(r.trigger).split()): doc_freq[w] = doc_freq.get(w, 0) + 1 n_docs = len(responses) or 1 def _idf(w: str) -> float: return math.log((1 + n_docs) / (1 + doc_freq.get(w, 0))) + 1 msg_tf = Counter(msg_expanded) for r in responses: rtokens = set(_norm(r.trigger).split()) r_expanded = _expand_tokens(rtokens, chat_id) common = msg_expanded & r_expanded if not common or not rtokens: continue score = 0.0 for w in common: score += _idf(w) * min(msg_tf.get(w, 1), 2) max_possible = sum(_idf(w) * 2 for w in r_expanded) or 1 ratio = score / max_possible if ratio >= 0.35: idf_matches.append((r, ratio)) if idf_matches: idf_matches.sort(key=lambda x: -x[1]) best_ratio = idf_matches[0][1] score = 0.5 + best_ratio * 0.15 result = _pick([r for r, _ in idf_matches[:3]], "semantic", score, topic_boost=True) await increment_uses(chat_id, result.response.id) return result # ── Level 8: Fuzzy Levenshtein (score: 0.4-0.55) ── fuzzy = [] for r in responses: if len(r.trigger) > 2: ratio = _fuzzy_ratio(r.trigger, text_lower) if ratio >= 0.5: fuzzy.append((r, ratio)) norm_r = _norm(r.trigger) ratio = _fuzzy_ratio(norm_r, text_norm) if ratio >= 0.5: fuzzy.append((r, ratio)) seen = set() unique_fuzzy = [] for r, ratio in fuzzy: if r.id not in seen: seen.add(r.id) unique_fuzzy.append((r, ratio)) if unique_fuzzy: unique_fuzzy.sort(key=lambda x: -x[1]) best_ratio = unique_fuzzy[0][1] score = 0.4 + best_ratio * 0.15 result = _pick([r for r, _ in unique_fuzzy[:3]], "fuzzy", score, topic_boost=True) await increment_uses(chat_id, result.response.id) return result return None # ═══════════════════════════════════════════════ # Template Engine # ═══════════════════════════════════════════════ def apply_template(text: str, user=None, chat=None) -> str: now = time.localtime() h = now.tm_hour daypart = "صبح" if 5 <= h < 12 else ("ظهر" if 12 <= h < 14 else ("بعدازظهر" if 14 <= h < 18 else "شب")) days = ["دوشنبه", "سه‌شنبه", "چهارشنبه", "پنجشنبه", "جمعه", "شنبه", "یکشنبه"] emojis = ["👍", "❤️", "🔥", "💪", "😊", "🎉", "✨", "🌸", "💚", "💯"] replacements = { "{user}": user.first_name if user else "", "{name}": user.full_name if user else "", "{username}": f"@{user.username}" if user and user.username else "", "{chat}": chat.title if chat else "", "{id}": str(user.id) if user else "", "{random}": str(random.randint(1, 1000)), "{time}": f"{now.tm_hour:02d}:{now.tm_min:02d}", "{hour}": str(now.tm_hour), "{day}": days[now.tm_wday] if now.tm_wday < 7 else "", "{daypart}": daypart, "{date}": f"{now.tm_year}/{now.tm_mon:02d}/{now.tm_mday:02d}", "{reaction}": random.choice(emojis), } for key, val in replacements.items(): text = text.replace(key, val) return text # ═══════════════════════════════════════════════ # Smart Forgetting v2 # ═══════════════════════════════════════════════ async def smart_forget(chat_id: int = None, days: int = 60) -> int: """Remove unused, low-weight, low-quality responses.""" cutoff = int(time.time()) - days * 86400 removed = 0 if chat_id: cur = await _exec( "DELETE FROM brain WHERE chat_id=? AND uses_count=0 AND weight<=0.5 AND created_at Response: tags = [] try: raw = row["tags"] if raw: tags = json.loads(raw) except (KeyError, json.JSONDecodeError, TypeError): pass return Response( id=row["id"], chat_id=row["chat_id"], trigger=row["trigger_text"], response=row["response_text"], category=row["category"] if "category" in row else "general", media_type=row["media_type"] if "media_type" in row else "text", file_id=row["file_id"] if "file_id" in row else "", weight=row["weight"] if "weight" in row else 1.0, delay=row["delay"] if "delay" in row else 0.0, tags=tags, created_by=row["created_by"] if "created_by" in row else 0, created_at=row["created_at"] if "created_at" in row else 0, last_used=row["last_used"] if "last_used" in row else 0, uses_count=row["uses_count"] if "uses_count" in row else 0, ) def _fuzzy_ratio(a: str, b: str) -> float: if not a or not b: return 0.0 if a == b: return 1.0 return 1.0 - (_levenshtein(a, b) / max(len(a), len(b))) def _levenshtein(a: str, b: str) -> int: if len(a) < len(b): a, b = b, a if not b: return len(a) prev = list(range(len(b) + 1)) for i, ca in enumerate(a): curr = [i + 1] for j, cb in enumerate(b): curr.append(min(curr[j] + 1, prev[j + 1] + 1, prev[j] + (ca != cb))) prev = curr return prev[-1] # ═══════════════════════════════════════════════ # Import / Export # ═══════════════════════════════════════════════ async def export_chat(chat_id: int) -> list[dict]: responses = await get_all(chat_id) return [r.to_dict() for r in responses] async def import_chat(chat_id: int, data: list[dict], created_by: int = 0) -> int: count = 0 for item in data: await add( chat_id, item["trigger"], item["response"], category=item.get("category", "general"), media_type=item.get("media_type", "text"), file_id=item.get("file_id", ""), weight=item.get("weight", 1.0), delay=item.get("delay", 0.0), tags=item.get("tags", []), created_by=created_by, ) count += 1 return count