From d823a3749aebf00688aab79fd4226bcb3584e7f4 Mon Sep 17 00:00:00 2001 From: mamad Date: Thu, 27 Aug 2026 19:53:34 +0330 Subject: [PATCH] feat(dedup): implement two-stage duplicate detection and content rewriter --- core/dedup.py | 35 ++++++++++ core/llm.py | 91 ++++++++++++++++++++++++++ services/ai_processor.py | 138 +++++++++++++++++++++++++++++++++++++++ tests/test_dedup.py | 22 +++++++ 4 files changed, 286 insertions(+) create mode 100644 core/dedup.py create mode 100644 core/llm.py create mode 100644 services/ai_processor.py create mode 100644 tests/test_dedup.py diff --git a/core/dedup.py b/core/dedup.py new file mode 100644 index 0000000..e488acb --- /dev/null +++ b/core/dedup.py @@ -0,0 +1,35 @@ +import hashlib +import re +from typing import Optional + +def normalize_text(text: Optional[str]) -> str: + """Normalize text by removing URLs, telegram handles, hashtags, excessive punctuation/whitespace, and lowercasing.""" + if not text: + return "" + # Remove URLs + text = re.sub(r'https?://\S+|www\.\S+', '', text) + # Remove Telegram @mentions / hashtags + text = re.sub(r'[@#]\w+', '', text) + # Normalize punctuation and whitespace + text = re.sub(r'[^\w\s]', ' ', text) + text = re.sub(r'\s+', ' ', text) + return text.strip().lower() + +def compute_content_hash(text: Optional[str], media_hash: Optional[str] = None) -> Optional[str]: + """Generate SHA256 hash from normalized text and/or media hash.""" + norm_text = normalize_text(text) + if not norm_text and not media_hash: + return None + raw_key = f"{norm_text}|{media_hash or ''}" + return hashlib.sha256(raw_key.encode('utf-8')).hexdigest() + +def compute_file_hash(file_path: str) -> Optional[str]: + """Generate SHA256 hash of a media file.""" + try: + hasher = hashlib.sha256() + with open(file_path, 'rb') as f: + while chunk := f.read(65536): + hasher.update(chunk) + return hasher.hexdigest() + except Exception: + return None diff --git a/core/llm.py b/core/llm.py new file mode 100644 index 0000000..0329d1d --- /dev/null +++ b/core/llm.py @@ -0,0 +1,91 @@ +import os +import json +import time +import httpx +import logging +from typing import Dict, Any, Optional +from core.metrics import AI_REQUESTS_TOTAL, AI_LATENCY_SECONDS + +logger = logging.getLogger(__name__) + +class LLMClient: + def __init__( + self, + provider: Optional[str] = None, + api_key: Optional[str] = None, + model: Optional[str] = None, + base_url: Optional[str] = None, + ): + self.provider = provider or os.getenv("AI_PROVIDER", "gemini").lower() + self.api_key = api_key or os.getenv("AI_API_KEY", "") + self.model = model or os.getenv("AI_MODEL", "gemini-1.5-flash" if self.provider == "gemini" else "gpt-4o-mini") + self.base_url = base_url or os.getenv("AI_BASE_URL") + + async def generate_json(self, prompt: str, system_prompt: Optional[str] = None, action_name: str = "general") -> Dict[str, Any]: + """Send prompt to LLM and parse JSON response.""" + start_time = time.time() + status = "error" + try: + if self.provider == "gemini": + result = await self._call_gemini(prompt, system_prompt) + else: + result = await self._call_openai(prompt, system_prompt) + status = "success" + return result + except Exception as e: + logger.error(f"LLM generation failed ({self.provider}/{self.model}): {e}") + raise + finally: + duration = time.time() - start_time + AI_LATENCY_SECONDS.labels(action=action_name).observe(duration) + AI_REQUESTS_TOTAL.labels(action=action_name, status=status).inc() + + async def _call_gemini(self, prompt: str, system_prompt: Optional[str] = None) -> Dict[str, Any]: + url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent?key={self.api_key}" + payload: Dict[str, Any] = { + "contents": [ + { + "parts": [{"text": prompt}] + } + ], + "generationConfig": { + "responseMimeType": "application/json", + "temperature": 0.2, + } + } + if system_prompt: + payload["systemInstruction"] = { + "parts": [{"text": system_prompt}] + } + + async with httpx.AsyncClient(timeout=60.0) as client: + resp = await client.post(url, json=payload) + resp.raise_for_status() + data = resp.json() + raw_text = data["candidates"][0]["content"]["parts"][0]["text"] + return json.loads(raw_text) + + async def _call_openai(self, prompt: str, system_prompt: Optional[str] = None) -> Dict[str, Any]: + url = self.base_url or "https://api.openai.com/v1/chat/completions" + headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json" + } + messages = [] + if system_prompt: + messages.append({"role": "system", "content": system_prompt}) + messages.append({"role": "user", "content": prompt}) + + payload = { + "model": self.model, + "messages": messages, + "response_format": {"type": "json_object"}, + "temperature": 0.2, + } + + async with httpx.AsyncClient(timeout=60.0) as client: + resp = await client.post(url, headers=headers, json=payload) + resp.raise_for_status() + data = resp.json() + content = data["choices"][0]["message"]["content"] + return json.loads(content) diff --git a/services/ai_processor.py b/services/ai_processor.py new file mode 100644 index 0000000..3544a20 --- /dev/null +++ b/services/ai_processor.py @@ -0,0 +1,138 @@ +import logging +from typing import List, Optional, Dict, Any +from db.models import Post, TargetChannel +from db.repository import Repository +from core.llm import LLMClient +from core.dedup import compute_content_hash +from core.metrics import DUPLICATES_DETECTED_TOTAL + +logger = logging.getLogger(__name__) + +TAG_EXTRACTION_SYSTEM_PROMPT = """ +You are an AI news analyst and classifier. +Given a social media/channel post, extract: +1. "subject": A brief, specific headline/subject (3-8 words). +2. "tags": A JSON array of 3 to 6 lowercase keywords/topics/entities (e.g. ["ai", "nvidia", "gpus", "hardware"]). +Respond ONLY in JSON format: +{ + "subject": "...", + "tags": ["tag1", "tag2", "tag3"] +} +""" + +DUPLICATE_CHECK_SYSTEM_PROMPT = """ +You are an expert news editor checking for duplicate news stories. +Given a NEW POST and a list of PREVIOUS POSTS, determine if the NEW POST is covering the same exact event, news item, or story as any of the previous posts. + +Respond ONLY in JSON format: +{ + "is_duplicate": true/false, + "duplicate_of_id": , + "similarity_reason": "" +} +""" + +POST_REWRITE_SYSTEM_PROMPT = """ +You are an expert Telegram content creator and copywriter. +Rewrite the provided post to make it engaging, well-formatted, professional, and clear. +Use appropriate emojis, clear paragraphs, and markdown formatting. +Remove any original promotional links, author credits, or watermarks. + +Respond ONLY in JSON format: +{ + "ai_text": "...", + "suggested_target_id": +} +""" + +class AIProcessor: + def __init__(self, repo: Repository, llm: Optional[LLMClient] = None): + self.repo = repo + self.llm = llm or LLMClient() + + async def process_post(self, post_id: int) -> Optional[Post]: + post = await self.repo.get_post_by_id(post_id) + if not post or not post.raw_text: + return post + + raw_text = post.raw_text + is_dup = False + dup_of_id = None + sim_reason = None + + # 1. Exact hash duplicate check + content_hash = post.content_hash or compute_content_hash(raw_text) + if content_hash: + exact_dup = await self.repo.find_duplicate_post_by_hash(content_hash) + if exact_dup and exact_dup.id != post.id: + is_dup = True + dup_of_id = exact_dup.id + sim_reason = "Exact match on normalized text/media hash" + DUPLICATES_DETECTED_TOTAL.labels(method="hash").inc() + + # 2. Extract Tags and Subject via AI + tags = [] + subject = "General News" + try: + tag_res = await self.llm.generate_json( + prompt=f"Post content:\n\n{raw_text}", + system_prompt=TAG_EXTRACTION_SYSTEM_PROMPT, + action_name="extract_tags" + ) + subject = tag_res.get("subject", subject) + tags = [t.lower().strip() for t in tag_res.get("tags", []) if isinstance(t, str)] + await self.repo.update_post_tags(post.id, tags, subject) + except Exception as e: + logger.error(f"Tag extraction failed for post {post.id}: {e}") + + # 3. Candidate search & Semantic AI Deduplication check (if not already exact dup) + if not is_dup and tags: + candidates = await self.repo.find_candidate_posts_by_tags(tags, exclude_post_id=post.id, hours_lookback=72, limit=5) + if candidates: + cand_texts = "\n---\n".join([f"ID {c.id} (Subject: {c.subject}):\n{c.raw_text}" for c in candidates if c.raw_text]) + prompt = f"NEW POST:\n{raw_text}\n\nPREVIOUS CANDIDATE POSTS:\n{cand_texts}" + try: + dup_res = await self.llm.generate_json( + prompt=prompt, + system_prompt=DUPLICATE_CHECK_SYSTEM_PROMPT, + action_name="check_duplicate" + ) + if dup_res.get("is_duplicate"): + is_dup = True + dup_of_id = dup_res.get("duplicate_of_id") + sim_reason = dup_res.get("similarity_reason", "AI detected duplicate news topic") + DUPLICATES_DETECTED_TOTAL.labels(method="ai_semantic").inc() + except Exception as e: + logger.error(f"Semantic duplicate check failed for post {post.id}: {e}") + + # 4. Rewrite post for our channels + ai_text = raw_text + suggested_target_id = None + targets = await self.repo.get_active_targets() + target_info = "\n".join([f"Target ID {t.id}: {t.title} (@{t.username or 'none'})" for t in targets]) + rewrite_prompt = f"TARGET CHANNELS AVAILABLE:\n{target_info or 'None'}\n\nORIGINAL POST:\n{raw_text}" + + try: + rewrite_res = await self.llm.generate_json( + prompt=rewrite_prompt, + system_prompt=POST_REWRITE_SYSTEM_PROMPT, + action_name="rewrite_post" + ) + ai_text = rewrite_res.get("ai_text", raw_text) + suggested_target_id = rewrite_res.get("suggested_target_id") + except Exception as e: + logger.error(f"Post rewrite failed for post {post.id}: {e}") + + # 5. Save AI results into database + await self.repo.update_ai_result( + post_id=post.id, + subject=subject, + ai_text=ai_text, + tags=tags, + suggested_target_id=suggested_target_id, + is_duplicate=is_dup, + duplicate_of_id=dup_of_id, + similarity_reason=sim_reason, + ) + + return await self.repo.get_post_by_id(post.id) diff --git a/tests/test_dedup.py b/tests/test_dedup.py new file mode 100644 index 0000000..ee9c47b --- /dev/null +++ b/tests/test_dedup.py @@ -0,0 +1,22 @@ +from core.dedup import normalize_text, compute_content_hash + +def test_normalize_text(): + raw = " Check out this link: https://t.me/example! @admin #tech news... " + norm = normalize_text(raw) + assert "https" not in norm + assert "admin" not in norm + assert "tech" not in norm + assert norm == "check out this link news" + +def test_content_hash(): + text1 = "Breaking News: Bitcoin hits $100k! Check https://example.com" + text2 = "Breaking News: Bitcoin hits $100k! Check https://other.com" + hash1 = compute_content_hash(text1) + hash2 = compute_content_hash(text2) + # Both normalize to the same text after link removal + assert hash1 == hash2 + +if __name__ == "__main__": + test_normalize_text() + test_content_hash() + print("All deduplication unit tests passed!")