feat(dedup): implement two-stage duplicate detection and content rewriter
This commit is contained in:
mamad
@@ -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
|
||||
+91
@@ -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)
|
||||
@@ -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": <id of matched previous post or null>,
|
||||
"similarity_reason": "<short explanation of why it is or is not a duplicate>"
|
||||
}
|
||||
"""
|
||||
|
||||
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": <optional id of best matching target channel or null>
|
||||
}
|
||||
"""
|
||||
|
||||
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)
|
||||
@@ -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!")
|
||||
Reference in New Issue
Block a user