root/copykar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add topic tagging, semantic duplicate detection, and per-provider vision toggle

Browse Source
This commit is contained in:
mamad
2026-08-28 20:45:42 +03:30
parent b12ddc1bfd
commit 6a5971bc08
11 changed files with 409 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/collector.py
+91 -16
View File
@@ -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: