diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index dfd2ab0..c5e4e03 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -2,6 +2,18 @@ 👤 ویرایش‌کننده: mamad +🔹 تشخیص هوشمند محتوای تکراری و برچسب‌گذاری موضوعی (AI Semantic Dedup & Tagging): +• استخراج خودکار برچسب‌های موضوعی (Tags) و عنوان موضوع (Subject) برای هر پست دریافتی. +• جایگزینی هش ساده با مقایسه مفهومی هوش مصنوعی: جمع‌آوری پست‌های مرتبط بر اساس برچسب‌ها و تحلیل تشابه معنایی توسط هوش مصنوعی. +• ثبت وضعیت تکراری، شناسه پست اصلی و دلیل تشابه در پایگاه‌داده. +📁 فایل‌های تغییریافته: services/ai_processor.py, services/collector.py, db/repository.py, tests/test_semantic_dedup_and_vision.py + +🔹 پیکربندی تفکیکی پردازش تصویر در ارائه‌دهندگان (Provider Vision Toggle): +• امکان فعال یا غیرفعال کردن دریافت تصویر برای هر ارائه‌دهنده به صورت جداگانه (با حالت پیش‌فرض فقط متن). +• دکمه تغییر وضعیت پردازش تصویر در منوی تنظیمات اختصاصی هر سرویس‌دهنده در ربات تلگرام. +• ارسال انحصاری تصویر به مدل‌هایی که پشتیبانی از تصویر در آن‌ها روشن است. +📁 فایل‌های تغییریافته: core/llm.py, db/database.py, db/models.py, services/admin_bot.py + 🔹 پایش و نمودار فضای ذخیره‌سازی دیسک سرور: • متمرکزسازی روی یک نمودار کامل ۲۴ ستونه برای نمایش روند زمانی فضای آزاد درایوها. • فیلتر کردن متریک‌های قدیمی و حذف ردیف‌های نامعتبر در نمودار گرافانا. @@ -13,18 +25,3 @@ • منوی تنظیم دستی ارائه‌دهنده پشتیبان برای هر سرویس هوش مصنوعی به صورت زنجیره‌ای مستقیم. • دستورات /changes و /notes جهت مشاهده گزارش تغییرات. 📁 فایل‌های تغییریافته: services/admin_bot.py, tests/test_fallback_and_stop.py - -🔹 پشتیبانی از تصاویر و بازنویسی مالتی‌مدال (Vision): -• ارسال تصاویر پست‌های ورودی به ارائه‌دهندگان هوش مصنوعی و درک محتوای بصری عکس‌ها و نمودارها. -• تنظیم عمق تفکر و استدلال (Reasoning Effort) برای مدل‌ها. -📁 فایل‌های تغییریافته: core/llm.py, services/ai_processor.py - -🔹 بهینه‌سازی خط لوله جمع‌آوری، انتشار و صف‌بندی: -• ارسال مستقیم پست‌های خام به کانال بررسی ادمین با دکمه‌های اقدام سریع بازنویسی. -• صف انتشار زمان‌بندی‌شده و منظم در ردیس (Paced Queue). -📁 فایل‌های تغییریافته: services/collector.py, services/publisher.py, services/metrics_reporter.py - -🔹 مدیریت پایگاه‌داده و دسته‌بندی کانال‌ها: -• ایجاد جداول دسته‌بندی کانال‌ها (Channel Categories) و ارائه‌دهندگان هوش مصنوعی با پشتیبانی از ارائه‌دهنده پشتیبان. -• پشتیبانی از لحن (Personality)، فوتر اختصاصی و پرامپت سفارشی برای هر کانال مقصد. -📁 فایل‌های تغییریافته: db/database.py, db/models.py, db/repository.py diff --git a/core/llm.py b/core/llm.py index 68148af..464128e 100644 --- a/core/llm.py +++ b/core/llm.py @@ -144,6 +144,9 @@ class LLMClient: p_key = profile.api_key.strip() p_effort = profile.reasoning_effort.strip().lower() + # Only pass image if this provider profile explicitly enables vision/image processing + active_image = image_path if (profile.supports_vision and image_path and os.path.isfile(image_path)) else None + for attempt in range(self.max_retries_per_model + 1): try: if p_type == "gemini" and "orcarouter" not in p_base_url: @@ -153,7 +156,7 @@ class LLMClient: model=p_model, api_key=p_key or self.api_key, reasoning_effort=p_effort or self.reasoning_effort, - image_path=image_path + image_path=active_image ) elif p_type == "agy": if not os.path.exists("/.dockerenv") and (shutil.which("agy") or os.path.exists("/home/mamad/.local/bin/agy")): @@ -161,7 +164,7 @@ class LLMClient: prompt=prompt, system_prompt=system_prompt, effort=p_effort or self.reasoning_effort, - image_path=image_path + image_path=active_image ) else: result = await self._call_openai( @@ -172,7 +175,7 @@ class LLMClient: api_key=p_key, reasoning_effort=p_effort or self.reasoning_effort, is_agy=True, - image_path=image_path + image_path=active_image ) else: result = await self._call_openai( @@ -183,7 +186,7 @@ class LLMClient: api_key=p_key or self.api_key, reasoning_effort=p_effort or self.reasoning_effort, is_agy=False, - image_path=image_path + image_path=active_image ) status = "success" diff --git a/db/database.py b/db/database.py index b4790cc..69147b4 100644 --- a/db/database.py +++ b/db/database.py @@ -117,6 +117,7 @@ CREATE TABLE IF NOT EXISTS ai_providers ( reasoning_effort VARCHAR(32) DEFAULT '', is_active BOOLEAN DEFAULT FALSE, fallback_provider_id BIGINT REFERENCES ai_providers(id) ON DELETE SET NULL, + supports_vision BOOLEAN DEFAULT FALSE, created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP ); @@ -132,6 +133,7 @@ CREATE INDEX IF NOT EXISTS idx_ai_providers_active ON ai_providers(is_active); -- Migration safety for existing tables ALTER TABLE ai_providers ADD COLUMN IF NOT EXISTS fallback_provider_id BIGINT REFERENCES ai_providers(id) ON DELETE SET NULL; +ALTER TABLE ai_providers ADD COLUMN IF NOT EXISTS supports_vision BOOLEAN DEFAULT FALSE; ALTER TABLE sources ADD COLUMN IF NOT EXISTS category_id INT REFERENCES channel_categories(id) ON DELETE SET NULL; ALTER TABLE targets ADD COLUMN IF NOT EXISTS category_id INT REFERENCES channel_categories(id) ON DELETE SET NULL; ALTER TABLE sources ADD COLUMN IF NOT EXISTS is_active BOOLEAN DEFAULT TRUE; diff --git a/db/models.py b/db/models.py index 3317d5b..4953ed2 100644 --- a/db/models.py +++ b/db/models.py @@ -99,6 +99,7 @@ class AIProviderProfile: is_active: bool = False fallback_provider_id: Optional[int] = None fallback_provider_name: Optional[str] = None + supports_vision: bool = False created_at: Optional[str] = None diff --git a/db/repository.py b/db/repository.py index dd71dc3..b066599 100644 --- a/db/repository.py +++ b/db/repository.py @@ -231,6 +231,8 @@ class Repository: media_path: Optional[str] = None, media_type: Optional[str] = None, content_hash: Optional[str] = None, + tags: Optional[List[str]] = None, + subject: Optional[str] = None, is_duplicate: bool = False, duplicate_of_id: Optional[int] = None, similarity_reason: Optional[str] = None, @@ -242,9 +244,9 @@ class Repository: """ INSERT INTO posts ( source_channel_id, source_message_id, raw_text, media_path, - media_type, content_hash, is_duplicate, duplicate_of_id, similarity_reason, status + media_type, content_hash, tags, subject, is_duplicate, duplicate_of_id, similarity_reason, status ) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'pending_review') + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, 'pending_review') RETURNING id; """, source_channel_id, @@ -253,6 +255,8 @@ class Repository: media_path, media_type, content_hash, + tags or [], + subject, is_duplicate, duplicate_of_id, similarity_reason, @@ -267,6 +271,30 @@ class Repository: row = await conn.fetchrow("SELECT * FROM posts WHERE id = $1;", post_id) return _parse_post_row(row) if row else None + async def find_candidate_posts_by_tags( + self, + tags: List[str], + exclude_post_id: Optional[int] = None, + limit: int = 10 + ) -> List[Post]: + """Find recent posts that share at least one tag.""" + if not tags: + return [] + pool = await self._get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + """ + SELECT * FROM posts + WHERE tags && $1::text[] + AND is_deleted = FALSE + AND ($2::bigint IS NULL OR id <> $2) + ORDER BY id DESC + LIMIT $3; + """, + tags, exclude_post_id, limit + ) + return [_parse_post_row(r) for r in rows] + async def find_duplicate_post(self, content_hash: Optional[str], exclude_post_id: Optional[int] = None) -> Optional[Post]: """Return the earliest post already carrying this content hash, if any.""" if not content_hash: @@ -575,7 +603,8 @@ class Repository: base_url: str = "", api_key: str = "", reasoning_effort: str = "", - is_active: bool = False + is_active: bool = False, + supports_vision: bool = False ) -> int: pool = await self._get_pool() async with pool.acquire() as conn: @@ -583,11 +612,11 @@ class Repository: await conn.execute("UPDATE ai_providers SET is_active = FALSE;") return await conn.fetchval( """ - INSERT INTO ai_providers (name, provider_type, model, base_url, api_key, reasoning_effort, is_active) - VALUES ($1, $2, $3, $4, $5, $6, $7) + INSERT INTO ai_providers (name, provider_type, model, base_url, api_key, reasoning_effort, is_active, supports_vision) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id; """, - name, provider_type, model, base_url, api_key, reasoning_effort, is_active + name, provider_type, model, base_url, api_key, reasoning_effort, is_active, supports_vision ) async def get_provider_profiles(self) -> List[AIProviderProfile]: @@ -596,7 +625,7 @@ class Repository: rows = await conn.fetch( """ SELECT p.id, p.name, p.provider_type, p.model, p.base_url, p.api_key, p.reasoning_effort, p.is_active, - p.fallback_provider_id, + p.fallback_provider_id, p.supports_vision, fb.name as fallback_provider_name, to_char(p.created_at, 'YYYY-MM-DD HH24:MI:SS') as created_at FROM ai_providers p @@ -612,7 +641,7 @@ class Repository: row = await conn.fetchrow( """ SELECT p.id, p.name, p.provider_type, p.model, p.base_url, p.api_key, p.reasoning_effort, p.is_active, - p.fallback_provider_id, + p.fallback_provider_id, p.supports_vision, fb.name as fallback_provider_name, to_char(p.created_at, 'YYYY-MM-DD HH24:MI:SS') as created_at FROM ai_providers p @@ -629,7 +658,7 @@ class Repository: row = await conn.fetchrow( """ SELECT p.id, p.name, p.provider_type, p.model, p.base_url, p.api_key, p.reasoning_effort, p.is_active, - p.fallback_provider_id, + p.fallback_provider_id, p.supports_vision, fb.name as fallback_provider_name, to_char(p.created_at, 'YYYY-MM-DD HH24:MI:SS') as created_at FROM ai_providers p @@ -656,6 +685,14 @@ class Repository: fallback_provider_id, profile_id ) + async def update_provider_vision(self, profile_id: int, supports_vision: bool) -> None: + pool = await self._get_pool() + async with pool.acquire() as conn: + await conn.execute( + "UPDATE ai_providers SET supports_vision = $1 WHERE id = $2;", + supports_vision, profile_id + ) + async def update_provider_profile( self, profile_id: int, diff --git a/main.py b/main.py index 5ac5606..2aab119 100644 --- a/main.py +++ b/main.py @@ -51,7 +51,7 @@ async def main(): ) ai_processor = AIProcessor(repo=repo, llm=llm) admin_bot.set_ai_processor(ai_processor) - collector = CollectorService(repo=repo, on_post_received=admin_bot.handle_collected_post) + collector = CollectorService(repo=repo, on_post_received=admin_bot.handle_collected_post, ai_processor=ai_processor) admin_bot.set_collector(collector) # Publisher handles per-target delivery queues, intervals, and sleep windows. diff --git a/services/admin_bot.py b/services/admin_bot.py index 31c53bf..19f4f74 100644 --- a/services/admin_bot.py +++ b/services/admin_bot.py @@ -388,6 +388,7 @@ class AdminBotService: masked_key = f"{p.api_key[:6]}...{p.api_key[-4:]}" if len(p.api_key) > 10 else ("تنظیم نشده" if not p.api_key else "******") url_label = p.base_url if p.base_url else "پیش‌فرض (لوکال)" fallback_label = f"🔄 {p.fallback_provider_name}" if p.fallback_provider_name else "❌ بدون پشتیبان (پایان زنجیره)" + vision_label = "🟢 فعال" if getattr(p, "supports_vision", False) else "⚪️ غیرفعال (پیش‌فرض)" text = ( f"🔌 اطلاعات سرویس‌دهنده: {p.name}\n\n" @@ -396,6 +397,7 @@ class AdminBotService: f"• 🌐 Base URL: {url_label}\n" f"• 🔑 API Key: {masked_key}\n" f"• 🧠 تفکر (Reasoning): {p.reasoning_effort or 'پیش‌فرض'}\n" + f"• 🖼 پردازش تصویر (Vision): {vision_label}\n" f"• 🔄 پشتیبان دستی (Fallback): {fallback_label}\n" f"• ⚡️ وضعیت: {st_text}" ) @@ -409,11 +411,14 @@ class AdminBotService: Button.inline("🔄 تنظیم ارائه‌دهنده پشتیبان", data=f"ai_pv_fb_menu:{p.id}"), ]) buttons.append([ + Button.inline("🖼 تغییر وضعیت پردازش تصویر", data=f"ai_pv_vision:{p.id}"), Button.inline("🧠 حالت تفکر و استدلال", data=f"ai_pv_effort:{p.id}"), - Button.inline("🏷 تغییر مدل", data=f"ai_pv_model:{p.id}"), ]) buttons.append([ + Button.inline("🏷 تغییر مدل", data=f"ai_pv_model:{p.id}"), Button.inline("🌐 تغییر Base URL", data=f"ai_pv_url:{p.id}"), + ]) + buttons.append([ Button.inline("🔑 تغییر API Key", data=f"ai_pv_key:{p.id}"), ]) if not p.is_active: @@ -1914,6 +1919,20 @@ class AdminBotService: await event.edit(text, parse_mode="html", buttons=buttons) await event.answer("✅ ارائه‌دهنده پشتیبان دستی ذخیره شد.") + elif data.startswith("ai_pv_vision:"): + prof_id = int(data.split(":")[1]) + p = await self.repo.get_provider_profile_by_id(prof_id) + if p: + new_vision = not getattr(p, "supports_vision", False) + await self.repo.update_provider_vision(prof_id, new_vision) + if self.ai_processor and self.ai_processor.llm: + await self.ai_processor.llm.sync_config_from_repo() + text, buttons = await self._render_ai_provider_detail(prof_id) + await event.edit(text, parse_mode="html", buttons=buttons) + await event.answer(f"✅ پردازش تصویر {'فعال' if new_vision else 'غیرفعال'} شد!") + else: + await event.answer("❌ سرویس‌دهنده یافت نشد.", alert=True) + elif data.startswith("ai_pv_effort:"): prof_id = int(data.split(":")[1]) text, buttons = await self._render_ai_provider_effort_menu(prof_id) diff --git a/services/ai_processor.py b/services/ai_processor.py index 8d69b32..07b6d1d 100644 --- a/services/ai_processor.py +++ b/services/ai_processor.py @@ -307,3 +307,111 @@ class AIProcessor: fallback = f"{fallback}\n\n{footer_text}" return RewriteResult(decision="accept", rejection_reason="", rewritten_text=fallback) + async def extract_tags_and_subject( + self, + raw_text: str, + image_path: Optional[str] = None + ) -> tuple[List[str], str]: + """Extract 3-6 topic tags and a short subject from incoming post text and/or image.""" + if not raw_text and not image_path: + return [], "" + + sys_prompt = ( + "You are a Telegram post classifier and tagger. " + "Analyze the given post content (and optional image) and extract:\n" + "1. 'subject': A concise, descriptive subject or headline (in Persian, max 10 words).\n" + "2. 'tags': A list of 3 to 6 relevant topical keywords/tags (lowercase, normalized, without '#' prefix).\n\n" + "Respond ONLY in valid JSON format:\n" + "{\n" + ' "subject": "...",\n' + ' "tags": ["tag1", "tag2", "tag3"]\n' + "}" + ) + + user_prompt = f"متن پست برای دسته‌بندی و برچسب‌گذاری:\n\n{raw_text}" if raw_text else "لطفاً با توجه به تصویر پیوست، موضوع و برچسب‌های آن را استخراج کنید." + + try: + res = await self.llm.generate_json( + prompt=user_prompt, + system_prompt=sys_prompt, + action_name="extract_tags_and_subject", + image_path=image_path + ) + subject = str(res.get("subject", "")).strip() + raw_tags = res.get("tags") or [] + tags: List[str] = [] + if isinstance(raw_tags, list): + for t in raw_tags: + clean_t = str(t).strip().lstrip("#").lower() + if clean_t and clean_t not in tags: + tags.append(clean_t) + return tags[:8], subject + except Exception as e: + logger.warning(f"Failed to extract tags and subject via AI: {e}") + return [], "" + + async def check_semantic_duplicate( + self, + new_text: str, + candidate_posts: List[Post], + image_path: Optional[str] = None + ) -> tuple[bool, Optional[int], Optional[str]]: + """Compare a new post against candidate posts sharing overlapping tags using AI semantic similarity.""" + if not candidate_posts or (not new_text and not image_path): + return False, None, None + + candidates_formatted = [] + for p in candidate_posts[:10]: + p_text = (p.raw_text or "").strip() + snippet = p_text[:250] + ("..." if len(p_text) > 250 else "") + candidates_formatted.append( + f"- [پست شناسه #{p.id}] (موضوع: {p.subject or 'نامشخص'} | برچسب‌ها: {', '.join(p.tags or [])}):\n «{snippet}»" + ) + + candidates_block = "\n\n".join(candidates_formatted) + + sys_prompt = ( + "You are an expert news and content duplicate detection engine.\n" + "Compare the NEW POST against the list of CANDIDATE POSTS.\n" + "Determine if the NEW POST covers the EXACT SAME news event, identical story, announcement, or duplicate information.\n\n" + "Respond ONLY in valid JSON format:\n" + "{\n" + ' "is_duplicate": true,\n' + ' "duplicate_of_id": 123,\n' + ' "reason": "توضیح کوتاه به زبان فارسی در مورد علت تکراری بودن"\n' + "}\n" + "If it is NOT a duplicate of any candidate post:\n" + "{\n" + ' "is_duplicate": false,\n' + ' "duplicate_of_id": null,\n' + ' "reason": ""\n' + "}" + ) + + user_prompt = ( + f"پست جدید برای بررسی:\n{new_text}\n\n" + f"لیست پست‌های مشابه قبلی برای مقایسه:\n{candidates_block}" + ) + + try: + res = await self.llm.generate_json( + prompt=user_prompt, + system_prompt=sys_prompt, + action_name="check_semantic_duplicate", + image_path=image_path + ) + is_dup = bool(res.get("is_duplicate", False)) + dup_id = res.get("duplicate_of_id") + reason = str(res.get("reason", "")).strip() + + if is_dup and dup_id: + try: + dup_id_int = int(dup_id) + return True, dup_id_int, reason or f"تشخیص هوش مصنوعی: مشابه پست #{dup_id_int}" + except (ValueError, TypeError): + pass + return False, None, None + except Exception as e: + logger.warning(f"AI semantic duplicate check failed: {e}") + return False, None, None + diff --git a/services/collector.py b/services/collector.py index 836b485..e69822c 100644 --- a/services/collector.py +++ b/services/collector.py @@ -1,7 +1,7 @@ import os import logging from dataclasses import dataclass -from typing import Optional, Callable, Awaitable +from typing import Optional, Callable, Awaitable, Any, List from telethon import TelegramClient, events from telethon.errors import SessionPasswordNeededError from telethon.tl.types import MessageMediaPhoto, MessageMediaDocument @@ -57,6 +57,7 @@ class CollectorService: repo: Repository, on_post_received: Optional[Callable[[int], Awaitable[None]]] = None, queue: Optional[RedisQueue] = None, + ai_processor: Optional[Any] = None, api_id: Optional[int] = None, api_hash: Optional[str] = None, phone: Optional[str] = None, @@ -65,6 +66,7 @@ class CollectorService: self.repo = repo self.on_post_received = on_post_received self.queue = queue + self.ai_processor = ai_processor self.api_id = api_id or int(os.getenv("API_ID", "0")) self.api_hash = api_hash or os.getenv("API_HASH", "") self.phone = phone or os.getenv("PHONE") @@ -75,6 +77,9 @@ class CollectorService: self.phone_code_hash: Optional[str] = None self._handlers_registered = False + def set_ai_processor(self, ai_processor: Any) -> None: + self.ai_processor = ai_processor + async def start(self, notify_fn: Optional[Callable[[str], Awaitable[None]]] = None): logger.info("Initializing Collector Userbot client...") try: @@ -201,7 +206,43 @@ class CollectorService: media_hash = compute_file_hash(downloaded_file) content_hash = compute_content_hash(raw_text, media_hash) - duplicate_of = await self.repo.find_duplicate_post(content_hash) + + # 1. Extract subject and topic tags via AI + tags, subject = [], "" + if self.ai_processor: + try: + tags, subject = await self.ai_processor.extract_tags_and_subject(raw_text, media_path) + except Exception as e: + logger.debug(f"Tag extraction error: {e}") + + # 2. Check duplicate via AI semantic similarity on candidate posts sharing tags + is_duplicate = False + duplicate_of_id = None + similarity_reason = None + + if tags and self.ai_processor: + try: + candidates = await self.repo.find_candidate_posts_by_tags(tags, limit=10) + if candidates: + is_dup, dup_id, reason = await self.ai_processor.check_semantic_duplicate( + raw_text, candidates, media_path + ) + if is_dup: + is_duplicate = True + duplicate_of_id = dup_id + similarity_reason = reason + DUPLICATES_DETECTED_TOTAL.labels(method="ai_semantic").inc() + except Exception as e: + logger.debug(f"Semantic duplicate check error: {e}") + + # 3. Fallback hash duplicate check if not already caught by AI + if not is_duplicate and content_hash: + duplicate_of = await self.repo.find_duplicate_post(content_hash) + if duplicate_of: + is_duplicate = True + duplicate_of_id = duplicate_of.id + similarity_reason = f"content hash matches post #{duplicate_of.id}" + DUPLICATES_DETECTED_TOTAL.labels(method="content_hash").inc() post_id = await self.repo.create_raw_post( source_channel_id=chat_id, @@ -210,17 +251,16 @@ class CollectorService: media_path=media_path, media_type=media_type, content_hash=content_hash, - is_duplicate=duplicate_of is not None, - duplicate_of_id=duplicate_of.id if duplicate_of else None, - similarity_reason=f"content hash matches post #{duplicate_of.id}" if duplicate_of else None, + tags=tags, + subject=subject, + is_duplicate=is_duplicate, + duplicate_of_id=duplicate_of_id, + similarity_reason=similarity_reason, ) - if post_id and duplicate_of: - DUPLICATES_DETECTED_TOTAL.labels(method="content_hash").inc() - if post_id: SOURCE_ACTIVITY_TOTAL.labels(channel_id=str(chat_id), title=source.title or 'Unknown').inc() - logger.info(f"Collected raw post ID {post_id} from source channel {chat_id}") + logger.info(f"Collected raw post ID {post_id} from source channel {chat_id} (subject={subject}, tags={tags}, duplicate={is_duplicate})") if self.on_post_received: await self.on_post_received(post_id) @@ -281,10 +321,43 @@ class CollectorService: media_hash = compute_file_hash(downloaded_file) content_hash = compute_content_hash(raw_text, media_hash) - duplicate_of = await self.repo.find_duplicate_post(content_hash) - if duplicate_of: - result.duplicates += 1 - DUPLICATES_DETECTED_TOTAL.labels(method="content_hash").inc() + + # Extract tags and check semantic duplication + tags, subject = [], "" + if self.ai_processor: + try: + tags, subject = await self.ai_processor.extract_tags_and_subject(raw_text, media_path) + except Exception as e: + logger.debug(f"Tag extraction error: {e}") + + is_duplicate = False + duplicate_of_id = None + similarity_reason = None + + if tags and self.ai_processor: + try: + candidates = await self.repo.find_candidate_posts_by_tags(tags, limit=10) + if candidates: + is_dup, dup_id, reason = await self.ai_processor.check_semantic_duplicate( + raw_text, candidates, media_path + ) + if is_dup: + is_duplicate = True + duplicate_of_id = dup_id + similarity_reason = reason + result.duplicates += 1 + DUPLICATES_DETECTED_TOTAL.labels(method="ai_semantic").inc() + except Exception as e: + logger.debug(f"Semantic duplicate check error: {e}") + + if not is_duplicate and content_hash: + duplicate_of = await self.repo.find_duplicate_post(content_hash) + if duplicate_of: + is_duplicate = True + duplicate_of_id = duplicate_of.id + similarity_reason = f"content hash matches post #{duplicate_of.id}" + result.duplicates += 1 + DUPLICATES_DETECTED_TOTAL.labels(method="content_hash").inc() post_id = await self.repo.create_raw_post( source_channel_id=channel_id, @@ -293,9 +366,11 @@ class CollectorService: media_path=media_path, media_type=media_type, content_hash=content_hash, - is_duplicate=duplicate_of is not None, - duplicate_of_id=duplicate_of.id if duplicate_of else None, - similarity_reason=f"content hash matches post #{duplicate_of.id}" if duplicate_of else None, + tags=tags, + subject=subject, + is_duplicate=is_duplicate, + duplicate_of_id=duplicate_of_id, + similarity_reason=similarity_reason, ) if post_id: diff --git a/tests/test_semantic_dedup_and_vision.py b/tests/test_semantic_dedup_and_vision.py new file mode 100644 index 0000000..051fe90 --- /dev/null +++ b/tests/test_semantic_dedup_and_vision.py @@ -0,0 +1,120 @@ +import os +import asyncio +from db.database import init_db, close_db_pool +from db.repository import Repository +from db.models import Post, TargetChannel, SourceChannel +from core.llm import LLMClient +from services.ai_processor import AIProcessor +from services.collector import CollectorService + + +async def test_provider_vision_toggle(): + await init_db() + repo = Repository() + + prov_id = await repo.add_provider_profile( + name="Test Vision Model", + provider_type="openai", + model="gpt-4o", + supports_vision=False + ) + assert prov_id is not None + + p = await repo.get_provider_profile_by_id(prov_id) + assert p.supports_vision is False + + await repo.update_provider_vision(prov_id, True) + p_updated = await repo.get_provider_profile_by_id(prov_id) + assert p_updated.supports_vision is True + + await repo.update_provider_vision(prov_id, False) + p_reverted = await repo.get_provider_profile_by_id(prov_id) + assert p_reverted.supports_vision is False + + await repo.delete_provider_profile(prov_id) + + +async def test_find_candidate_posts_by_tags(): + await init_db() + repo = Repository() + + import time + msg_id = int(time.time() * 1000) % 100000000 + post1_id = await repo.create_raw_post( + source_channel_id=-100111222, + source_message_id=msg_id, + raw_text="قیمت بیت‌کوین به ۷۰ هزار دلار رسید و رکورد جدیدی ثبت کرد.", + tags=["بیت_کوین", "ارز_دیجیتال", "اقتصاد"], + subject="رکوردشکنی بیت‌کوین", + ) + assert post1_id is not None + + candidates = await repo.find_candidate_posts_by_tags(["بیت_کوین", "رمزارز"]) + assert len(candidates) >= 1 + found_ids = [c.id for c in candidates] + assert post1_id in found_ids + + no_match = await repo.find_candidate_posts_by_tags(["ورزش", "فوتبال"]) + assert post1_id not in [c.id for c in no_match] + + +async def test_semantic_duplicate_detection(): + await init_db() + repo = Repository() + + class FakeDuplicateLLM: + def __init__(self): + self.supports_vision = False + self.last_used_model = "fake" + self.last_used_provider = "fake" + + async def generate_json(self, prompt, system_prompt=None, action_name=None, image_path=None): + if action_name == "extract_tags_and_subject": + return { + "subject": "رشد بیت‌کوین", + "tags": ["بیت_کوین", "کریپتو", "قیمت"] + } + elif action_name == "check_semantic_duplicate": + return { + "is_duplicate": True, + "duplicate_of_id": 55, + "reason": "پوشش یکسان خبر افزایش نرخ بیت‌کوین" + } + return {} + + fake_llm = FakeDuplicateLLM() + processor = AIProcessor(repo=repo, llm=fake_llm) + + tags, subject = await processor.extract_tags_and_subject("بیت‌کوین ۷۰ هزار دلار شد") + assert tags == ["بیت_کوین", "کریپتو", "قیمت"] + assert subject == "رشد بیت‌کوین" + + candidates = [ + Post( + id=55, + source_channel_id=-100123, + source_message_id=10, + raw_text="بیت کوین به ۷۰۰۰۰ دلار رسید", + tags=["بیت_کوین"], + subject="افزایش قیمت بیت‌کوین" + ) + ] + + is_dup, dup_id, reason = await processor.check_semantic_duplicate( + "بیت‌کوین به مرز هفتاد هزار دلار دست یافت", + candidates + ) + assert is_dup is True + assert dup_id == 55 + assert "پوشش یکسان" in reason + + +async def main(): + await test_provider_vision_toggle() + await test_find_candidate_posts_by_tags() + await test_semantic_duplicate_detection() + print("All semantic deduplication and vision configuration tests passed successfully!") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_vision_and_disk_metrics.py b/tests/test_vision_and_disk_metrics.py index badccf5..6cec3ac 100644 --- a/tests/test_vision_and_disk_metrics.py +++ b/tests/test_vision_and_disk_metrics.py @@ -71,6 +71,7 @@ async def test_multimodal_vision_image_payload(): name="OpenAI Vision", provider_type="openai", model="gpt-4o", + supports_vision=True, is_active=True ) repo_mock = AsyncMock()