#!/usr/bin/env python3 import re import math from collections import Counter from pathlib import Path import pymysql BASE_DIR = Path(__file__).resolve().parent.parent DB_CONFIG = BASE_DIR / "config" / "database.php" # ============================================================ # SETTINGS # ============================================================ MAX_ITEMS = 500 # Only compare news reasonably close in time. MAX_TIME_DISTANCE_HOURS = 12 # Similarity needed to join an existing story. MATCH_THRESHOLD = 0.34 # Strong overlap can compensate for short texts. STRONG_TOKEN_THRESHOLD = 0.48 # ============================================================ # PERSIAN NORMALIZATION # ============================================================ STOPWORDS = { "و", "در", "به", "از", "که", "این", "آن", "را", "با", "برای", "یک", "است", "بود", "شد", "شده", "می", "های", "هم", "تا", "بر", "یا", "اما", "اگر", "ما", "من", "او", "آنها", "آن‌ها", "خود", "کرد", "کرده", "کند", "گفت", "میگوید", "می‌گوید", "درباره", "پس", "بعد", "قبل", "امروز", "دیروز", "همچنین", "دیگر", "دارد", "دارند", "خواهد", "خواهند", "نیز", "فقط", "خیلی", "بسیار", "شما", "هر", "روی", "همین", "هنوز", "شوند", "شود", "کردند", "کنند", "کنیم", "کردیم", "نیست", "هست", "هستند" } def normalize_text(text): text = text or "" # Remove Telegram usernames text = re.sub( r"@[A-Za-z0-9_]+", " ", text ) # Remove URLs text = re.sub( r"https?://\S+|www\.\S+", " ", text ) # Persian/Arabic character normalization text = ( text .replace("ي", "ی") .replace("ى", "ی") .replace("ك", "ک") .replace("ۀ", "ه") .replace("ة", "ه") .replace("\u200c", " ") ) text = text.lower() # Keep Persian/English letters + numbers text = re.sub( r"[^آ-یa-z0-9۰-۹\s]", " ", text ) text = re.sub( r"\s+", " ", text ).strip() return text def tokenize(text): normalized = normalize_text(text) tokens = [] for token in normalized.split(): token = token.strip() if len(token) < 2: continue if token in STOPWORDS: continue tokens.append(token) return tokens # ============================================================ # SIMILARITY # ============================================================ def weighted_jaccard(tokens_a, tokens_b): if not tokens_a or not tokens_b: return 0.0 a = Counter(tokens_a) b = Counter(tokens_b) common = set(a) & set(b) all_tokens = set(a) | set(b) numerator = sum( min(a[x], b[x]) for x in common ) denominator = sum( max(a[x], b[x]) for x in all_tokens ) if denominator == 0: return 0.0 return numerator / denominator def containment(tokens_a, tokens_b): a = set(tokens_a) b = set(tokens_b) if not a or not b: return 0.0 common = len(a & b) return common / min( len(a), len(b) ) def important_overlap(tokens_a, tokens_b): a = { x for x in tokens_a if len(x) >= 4 } b = { x for x in tokens_b if len(x) >= 4 } if not a or not b: return 0.0 common = a & b return len(common) / min( len(a), len(b) ) def similarity(item_a, item_b): ta = item_a["tokens"] tb = item_b["tokens"] jac = weighted_jaccard( ta, tb ) cont = containment( ta, tb ) important = important_overlap( ta, tb ) # Weighted score score = ( jac * 0.40 + cont * 0.35 + important * 0.25 ) return { "score": score, "jaccard": jac, "containment": cont, "important": important } # ============================================================ # DB CONFIG # ============================================================ def load_db_config(): content = DB_CONFIG.read_text( encoding="utf-8", errors="ignore" ) patterns = { "host": r"\$dbHost\s*=\s*['\"]([^'\"]+)['\"]", "database": r"\$dbName\s*=\s*['\"]([^'\"]+)['\"]", "user": r"\$dbUser\s*=\s*['\"]([^'\"]+)['\"]", "password": r"\$dbPass\s*=\s*['\"]([^'\"]*)['\"]", } result = {} for key, pattern in patterns.items(): match = re.search( pattern, content ) if not match: raise RuntimeError( f"Could not read DB config: {key}" ) result[key] = match.group(1) return { "host": result["host"], "user": result["user"], "password": result["password"], "database": result["database"], "charset": "utf8mb4", "cursorclass": pymysql.cursors.DictCursor } # ============================================================ # LOAD RAW ITEMS # ============================================================ def load_items(db): with db.cursor() as cur: cur.execute( """ SELECT nfi.id, nfi.news_source_id, nfi.source_id, nfi.source_message_id, nfi.text_content, nfi.content_type, nfi.media_type, nfi.published_at, s.username FROM news_feed_items nfi LEFT JOIN sources s ON s.id = nfi.source_id WHERE nfi.feed_status = 'raw' ORDER BY nfi.published_at ASC, nfi.id ASC LIMIT %s """, ( MAX_ITEMS, ) ) rows = cur.fetchall() result = [] for row in rows: row["tokens"] = tokenize( row.get("text_content") ) result.append(row) return result # ============================================================ # TIME CHECK # ============================================================ def time_distance_hours(a, b): da = a.get("published_at") db = b.get("published_at") if not da or not db: return 0 return abs( (da - db).total_seconds() ) / 3600 # ============================================================ # STORY MATCHER # ============================================================ def can_compare(item, story): primary = story["primary"] hours = time_distance_hours( item, primary ) return hours <= MAX_TIME_DISTANCE_HOURS def exactish_duplicate(result): """ Very strong textual match. Intended for the same news copied/reposted with only minor formatting/source differences. """ if not result: return False if result["score"] >= 0.72: return True if ( result["containment"] >= 0.88 and result["important"] >= 0.70 ): return True return False def normal_story_match(result): """ V2.2 normal story match. Prevents generic entity/topic overlap from merging different news events while keeping genuine reports of the same event together. """ if not result: return False score = result["score"] cont = result["containment"] important = result["important"] jac = result["jaccard"] # Strong overall textual relationship. if ( score >= 0.56 and cont >= 0.62 and important >= 0.52 ): return True # Strong containment path. # Useful when one report is a longer version of another. if ( cont >= 0.78 and important >= 0.64 and jac >= 0.30 ): return True # Balanced high-confidence relationship. if ( score >= 0.50 and cont >= 0.70 and important >= 0.60 and jac >= 0.32 ): return True return False def story_similarity(item, story): """ V2 anti-chain matcher. A new item is evaluated against: 1) the primary story item 2) every existing member But matching one random member is no longer automatically enough to join a large story. """ primary_result = similarity( item, story["primary"] ) member_results = [] for existing in story["items"]: result = similarity( item, existing ) member_results.append({ "item_id": existing["id"], **result }) member_results.sort( key=lambda x: x["score"], reverse=True ) best = ( member_results[0] if member_results else None ) second = ( member_results[1] if len(member_results) >= 2 else None ) return { "primary": primary_result, "best": best, "second": second, "members": member_results } def should_match_story(item, story, result): """ Decide whether ITEM really belongs to STORY. V2.1: - anti-chain protection - short-text protection - exact duplicates still accepted - short generic updates cannot bridge into a larger story """ if not result: return False, "no-result" primary = result["primary"] best = result["best"] second = result["second"] # -------------------------------------------------------- # Candidate text size # -------------------------------------------------------- item_tokens = item.get("tokens") or [] token_count = len(item_tokens) # Short Telegram headlines such as: # "دو انفجار در تنگه هرمز" # must be treated much more strictly. short_item = token_count <= 7 # -------------------------------------------------------- # A) Near duplicate of primary # # Exact / near-exact copies are safe even when short. # -------------------------------------------------------- if exactish_duplicate(primary): return True, "primary-exact" # -------------------------------------------------------- # B) Proper match against primary # -------------------------------------------------------- if normal_story_match(primary): if short_item: # A short item may join only when its relationship # with the primary is exceptionally strong. # # This blocks generic topical overlap such as: # LNG shipment through Hormuz # "two explosions in Hormuz" # # while still allowing near-identical short reports. if ( primary["score"] >= 0.78 and primary["containment"] >= 0.88 and primary["important"] >= 0.72 ): return True, "primary-short-strong" else: return True, "primary-match" # -------------------------------------------------------- # C) Story currently has only one item # -------------------------------------------------------- if len(story["items"]) == 1: if best and normal_story_match(best): if short_item: if ( best["score"] >= 0.78 and best["containment"] >= 0.88 and best["important"] >= 0.72 ): return True, "single-short-strong" return False, "single-short-reject" return True, "single-story-match" return False, "single-story-reject" # -------------------------------------------------------- # D1) Extremely strong duplicate of existing member # -------------------------------------------------------- if best and exactish_duplicate(best): if ( best["score"] >= 0.82 or ( best["containment"] >= 0.95 and best["important"] >= 0.82 ) ): return True, "member-exact" # -------------------------------------------------------- # D2) Short items must NOT use normal member consensus. # # Short headlines have too few semantic anchors and are # especially dangerous for chain merging. # -------------------------------------------------------- if short_item: return False, "short-anti-chain-reject" # -------------------------------------------------------- # D3) Consensus support for normal-length items # -------------------------------------------------------- if best and second: best_ok = normal_story_match(best) second_support = ( second["score"] >= 0.36 and second["containment"] >= 0.46 and second["important"] >= 0.36 ) if best_ok and second_support: return True, "member-consensus" return False, "anti-chain-reject" def build_stories(items): stories = [] for item in items: best_story = None best_result = None best_reason = None best_score = -1.0 for story in stories: if not can_compare( item, story ): continue result = story_similarity( item, story ) accepted, reason = should_match_story( item, story, result ) if not accepted: continue primary_score = ( result["primary"]["score"] if result.get("primary") else 0.0 ) member_score = ( result["best"]["score"] if result.get("best") else 0.0 ) # Prefer stories whose PRIMARY also agrees. # This makes accidental member bridges less attractive. candidate_score = ( primary_score * 0.65 + member_score * 0.35 ) if candidate_score > best_score: best_score = candidate_score best_story = story best_result = result best_reason = reason if best_story is None: stories.append({ "primary": item, "items": [item], "matches": [] }) else: best_story["items"].append( item ) best = best_result.get("best") or {} primary = best_result.get("primary") or {} best_story["matches"].append({ "item_id": item["id"], "score": best.get( "score", primary.get("score", 0.0) ), "jaccard": best.get( "jaccard", primary.get("jaccard", 0.0) ), "containment": best.get( "containment", primary.get("containment", 0.0) ), "important": best.get( "important", primary.get("important", 0.0) ), "primary_score": primary.get( "score", 0.0 ), "matched_to": best.get( "item_id" ), "reason": best_reason }) return stories # ============================================================ # DISPLAY # ============================================================ def short_text(text, limit=110): text = normalize_text(text) if len(text) <= limit: return text return text[:limit] + "..." # ============================================================ # PRODUCTION STORY ENGINE # ============================================================ import hashlib import sys def make_story_key(primary): """ Stable unique key based on the first feed item of a story. """ raw = ( f"news-story:" f"{primary['source_id']}:" f"{primary['source_message_id']}:" f"{primary['id']}" ) return hashlib.sha256( raw.encode("utf-8") ).hexdigest() def make_headline(item): """ Temporary machine headline. Later the generation layer can rewrite this according to channel tone. Story matching itself must remain independent from generated wording. """ text = normalize_text( item.get("text_content") or item.get("text") or "" ) if not text: return f"News Story #{item['id']}" return text[:500] def contribution_score(primary, item): if int(primary["id"]) == int(item["id"]): return 100.00 result = similarity( item, primary ) score = float( result.get("score", 0.0) ) * 100.0 return round( min(100.0, max(0.0, score)), 2 ) def write_story(db, story): primary = story["primary"] items = story["items"] source_ids = { int(item["source_id"]) for item in items if item.get("source_id") is not None } first_seen = min( item["published_at"] for item in items if item.get("published_at") is not None ) last_seen = max( item["published_at"] for item in items if item.get("published_at") is not None ) story_key = make_story_key( primary ) headline = make_headline( primary ) # Confidence starts conservatively. # More independent sources = more confidence. confidence = min( 100.0, 45.0 + ( max(0, len(source_ids) - 1) * 15.0 ) ) with db.cursor() as cursor: cursor.execute( """ INSERT INTO news_stories ( story_key, headline, story_type, importance_score, confidence_score, source_count, status, first_seen_at, last_seen_at ) VALUES ( %s, %s, 'normal', 0.00, %s, %s, 'collecting', %s, %s ) """, ( story_key, headline, confidence, len(source_ids), first_seen, last_seen ) ) story_id = cursor.lastrowid for item in items: is_primary = ( 1 if int(item["id"]) == int(primary["id"]) else 0 ) score = contribution_score( primary, item ) cursor.execute( """ INSERT INTO news_story_items ( story_id, feed_item_id, source_id, is_primary, contribution_score ) VALUES ( %s, %s, %s, %s, %s ) """, ( story_id, # IMPORTANT: # post_id currently stores news_feed_items.id int(item["id"]), int(item["source_id"]), is_primary, score ) ) cursor.execute( """ UPDATE news_feed_items SET story_id = %s, feed_status = 'story', last_error = NULL WHERE id = %s AND feed_status = 'raw' """, ( story_id, int(item["id"]) ) ) return story_id def production_bootstrap(): db_config = load_db_config() # load_db_config() already contains charset/cursor settings. # Override only what production bootstrap specifically needs. db_config["cursorclass"] = pymysql.cursors.DictCursor db_config["autocommit"] = False db = pymysql.connect( **db_config ) try: # ---------------------------------------------------- # Safety: bootstrap is only for an empty story DB. # ---------------------------------------------------- with db.cursor() as cursor: cursor.execute( """ SELECT COUNT(*) AS total FROM news_stories """ ) existing = int( cursor.fetchone()["total"] ) if existing > 0: print( "ABORTED: news_stories is not empty." ) print( "Bootstrap will not run twice." ) return # ---------------------------------------------------- # Load ALL current RAW items # ---------------------------------------------------- with db.cursor() as cursor: cursor.execute( """ SELECT nfi.id, nfi.news_source_id, nfi.source_id, nfi.source_message_id, nfi.telegram_grouped_id, nfi.published_at, nfi.content_type, nfi.text_content, nfi.media_type, nfi.source_views, nfi.source_forwards, s.username FROM news_feed_items nfi INNER JOIN sources s ON s.id = nfi.source_id WHERE nfi.feed_status = 'raw' ORDER BY nfi.published_at ASC, nfi.id ASC """ ) items = cursor.fetchall() for item in items: item["tokens"] = tokenize( item.get("text_content") or "" ) print("") print( "NEWS STORY ENGINE — PRODUCTION BOOTSTRAP" ) print( "==============================================" ) print( f"RAW Items: {len(items)}" ) if not items: print( "Nothing to process." ) return # ---------------------------------------------------- # V2.1 matcher # ---------------------------------------------------- stories = build_stories( items ) print( f"Detected Stories: {len(stories)}" ) multi = sum( 1 for story in stories if len(story["items"]) > 1 ) print( f"Multi-item Stories: {multi}" ) # ---------------------------------------------------- # Write transaction # ---------------------------------------------------- written_stories = 0 written_items = 0 for story in stories: write_story( db, story ) written_stories += 1 written_items += len( story["items"] ) # ---------------------------------------------------- # Integrity check BEFORE commit # ---------------------------------------------------- with db.cursor() as cursor: cursor.execute( """ SELECT COUNT(*) AS total FROM news_feed_items WHERE feed_status = 'raw' """ ) raw_left = int( cursor.fetchone()["total"] ) cursor.execute( """ SELECT COUNT(*) AS total FROM news_feed_items WHERE feed_status = 'story' AND story_id IS NULL """ ) orphaned = int( cursor.fetchone()["total"] ) if raw_left != 0: raise RuntimeError( f"Safety check failed: " f"{raw_left} RAW items remain" ) if orphaned != 0: raise RuntimeError( f"Safety check failed: " f"{orphaned} story items have no story_id" ) db.commit() print("") print( "==============================================" ) print( "BOOTSTRAP COMPLETE ✅" ) print( f"Stories Written: {written_stories}" ) print( f"Feed Items Linked: {written_items}" ) print( "Transaction: COMMITTED" ) print( "==============================================" ) except Exception as exc: db.rollback() print("") print( "BOOTSTRAP FAILED ❌" ) print( "Transaction: ROLLED BACK" ) print( f"Error: {exc}" ) raise finally: db.close() # ============================================================ # INCREMENTAL REAL-TIME ENGINE # ============================================================ def prepare_item(item): item["tokens"] = tokenize( item.get("text_content") or "" ) return item def load_raw_items(db, limit=100): """ Load only NEW feed items. Old story items are never reprocessed. """ with db.cursor() as cursor: cursor.execute( """ SELECT nfi.id, nfi.news_source_id, nfi.source_id, nfi.source_message_id, nfi.telegram_grouped_id, nfi.published_at, nfi.content_type, nfi.text_content, nfi.media_type, nfi.source_views, nfi.source_forwards, s.username FROM news_feed_items nfi INNER JOIN sources s ON s.id = nfi.source_id WHERE nfi.feed_status = 'raw' AND nfi.story_id IS NULL ORDER BY nfi.published_at ASC, nfi.id ASC LIMIT %s """, (int(limit),) ) rows = cursor.fetchall() return [ prepare_item(row) for row in rows ] def load_recent_stories(db, item): """ Build the exact in-memory story structure expected by story_similarity() / should_match_story(). We use the configured duplicate window. """ published_at = item.get("published_at") if published_at is None: return [] with db.cursor() as cursor: cursor.execute( """ SELECT setting_value FROM news_settings WHERE setting_key = 'duplicate_window_hours' LIMIT 1 """ ) row = cursor.fetchone() try: window_hours = int( (row or {}).get( "setting_value", MAX_TIME_DISTANCE_HOURS ) ) except Exception: window_hours = MAX_TIME_DISTANCE_HOURS # Matcher itself currently has a 12-hour can_compare() # guard. Never load a larger candidate window than the # matcher can actually use. window_hours = min( window_hours, MAX_TIME_DISTANCE_HOURS ) cursor.execute( """ SELECT ns.id AS story_id, ns.first_seen_at, ns.last_seen_at, nfi.id, nfi.news_source_id, nfi.source_id, nfi.source_message_id, nfi.telegram_grouped_id, nfi.published_at, nfi.content_type, nfi.text_content, nfi.media_type, nfi.source_views, nfi.source_forwards, nsi.is_primary, s.username FROM news_stories ns INNER JOIN news_story_items nsi ON nsi.story_id = ns.id INNER JOIN news_feed_items nfi ON nfi.id = nsi.feed_item_id INNER JOIN sources s ON s.id = nfi.source_id WHERE ns.status IN ( 'collecting', 'ready' ) AND ns.last_seen_at >= DATE_SUB(%s, INTERVAL %s HOUR) AND ns.first_seen_at <= DATE_ADD(%s, INTERVAL %s HOUR) ORDER BY ns.id ASC, nsi.is_primary DESC, nfi.published_at ASC, nfi.id ASC """, ( published_at, window_hours, published_at, window_hours ) ) rows = cursor.fetchall() stories_by_id = {} for row in rows: story_id = int( row["story_id"] ) item_row = { "id": row["id"], "news_source_id": row["news_source_id"], "source_id": row["source_id"], "source_message_id": row["source_message_id"], "telegram_grouped_id": row["telegram_grouped_id"], "published_at": row["published_at"], "content_type": row["content_type"], "text_content": row["text_content"], "media_type": row["media_type"], "source_views": row["source_views"], "source_forwards": row["source_forwards"], "username": row["username"], } prepare_item( item_row ) if story_id not in stories_by_id: stories_by_id[story_id] = { "db_story_id": story_id, "primary": None, "items": [], "matches": [] } story = stories_by_id[ story_id ] story["items"].append( item_row ) if int(row["is_primary"] or 0) == 1: story["primary"] = item_row result = [] for story in stories_by_id.values(): # Safety fallback for legacy/malformed story. if story["primary"] is None: if not story["items"]: continue story["primary"] = story[ "items" ][0] result.append( story ) return result def find_existing_story(db, item): stories = load_recent_stories( db, item ) best_story = None best_result = None best_reason = None best_score = -1.0 for story in stories: if not can_compare( item, story ): continue result = story_similarity( item, story ) accepted, reason = should_match_story( item, story, result ) if not accepted: continue primary_score = ( result["primary"]["score"] if result.get("primary") else 0.0 ) member_score = ( result["best"]["score"] if result.get("best") else 0.0 ) candidate_score = ( primary_score * 0.65 + member_score * 0.35 ) if candidate_score > best_score: best_score = candidate_score best_story = story best_result = result best_reason = reason return ( best_story, best_result, best_reason, best_score ) def attach_item_to_story( db, story, item, match_result ): story_id = int( story["db_story_id"] ) primary = story["primary"] score = contribution_score( primary, item ) with db.cursor() as cursor: # Lock RAW item so two workers cannot process # the same feed item simultaneously. cursor.execute( """ SELECT feed_status, story_id FROM news_feed_items WHERE id = %s FOR UPDATE """, (int(item["id"]),) ) current = cursor.fetchone() if not current: raise RuntimeError( f"Feed item #{item['id']} disappeared" ) if ( current["feed_status"] != "raw" or current["story_id"] is not None ): return False cursor.execute( """ INSERT INTO news_story_items ( story_id, feed_item_id, source_id, is_primary, contribution_score ) VALUES ( %s, %s, %s, 0, %s ) """, ( story_id, int(item["id"]), int(item["source_id"]), score ) ) cursor.execute( """ UPDATE news_feed_items SET story_id = %s, feed_status = 'story', last_error = NULL WHERE id = %s """, ( story_id, int(item["id"]) ) ) cursor.execute( """ UPDATE news_stories ns SET ns.last_seen_at = GREATEST( COALESCE( ns.last_seen_at, %s ), %s ), ns.first_seen_at = LEAST( COALESCE( ns.first_seen_at, %s ), %s ), ns.source_count = ( SELECT COUNT( DISTINCT nsi.source_id ) FROM news_story_items nsi WHERE nsi.story_id = ns.id ), ns.confidence_score = LEAST( 100.00, 45.00 + ( GREATEST( 0, ( SELECT COUNT( DISTINCT nsi2.source_id ) FROM news_story_items nsi2 WHERE nsi2.story_id = ns.id ) - 1 ) * 15.00 ) ) WHERE ns.id = %s """, ( item["published_at"], item["published_at"], item["published_at"], item["published_at"], story_id ) ) return True def create_incremental_story( db, item ): story = { "primary": item, "items": [item], "matches": [] } return write_story( db, story ) def process_incremental_batch( limit=100 ): db_config = load_db_config() db_config["cursorclass"] = ( pymysql.cursors.DictCursor ) db_config["autocommit"] = False db = pymysql.connect( **db_config ) processed = 0 attached = 0 created = 0 skipped = 0 failed = 0 try: raw_items = load_raw_items( db, limit ) print("") print( "NEWS STORY ENGINE — INCREMENTAL" ) print( "==============================================" ) print( f"RAW Items: {len(raw_items)}" ) for item in raw_items: try: story, result, reason, score = ( find_existing_story( db, item ) ) if story is not None: success = attach_item_to_story( db, story, item, result ) if success: db.commit() processed += 1 attached += 1 print( f"[ATTACHED] " f"feed={item['id']} " f"story={story['db_story_id']} " f"reason={reason} " f"score={score:.3f}" ) else: db.rollback() skipped += 1 else: story_id = ( create_incremental_story( db, item ) ) db.commit() processed += 1 created += 1 print( f"[NEW] " f"feed={item['id']} " f"story={story_id}" ) except Exception as exc: db.rollback() failed += 1 print( f"[FAILED] " f"feed={item['id']} " f"{exc}" ) print("") print( "==============================================" ) print( "INCREMENTAL COMPLETE" ) print( f"Processed: {processed}" ) print( f"Attached: {attached}" ) print( f"Created: {created}" ) print( f"Skipped: {skipped}" ) print( f"Failed: {failed}" ) print( "==============================================" ) finally: db.close() if __name__ == "__main__": if ( len(sys.argv) >= 2 and sys.argv[1] == "bootstrap" ): production_bootstrap() elif ( len(sys.argv) >= 2 and sys.argv[1] == "incremental" ): limit = 100 if len(sys.argv) >= 3: limit = int( sys.argv[2] ) process_incremental_batch( limit=limit ) else: print("Usage:") print( " python3 engine/news_story_engine.py bootstrap" ) print( " python3 engine/news_story_engine.py incremental [limit]" )