#!/usr/bin/env python3 import re import math import json import subprocess from collections import Counter from pathlib import Path import pymysql BASE_DIR = Path(__file__).resolve().parent.parent DB_CONFIG = BASE_DIR / "config" / "database.php" SEMANTIC_BRIDGE = BASE_DIR / "engine" / "news_semantic_bridge.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 # Short headline / longer report relationship. # # Example: # "دو انفجار در تنگه هرمز" # versus # "گزارش‌هایی مبنی بر وقوع انفجارهای متعدد # در تنگه هرمز منتشر شده است" # # Important overlap must be high. We deliberately # keep this separate from the general match rules so # generic entity overlap does not loosen everything. if ( score >= 0.40 and cont >= 0.50 and important >= 0.65 and jac >= 0.17 ): 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 # ============================================================ # SEMANTIC DUPLICATE GATE V2.4 # ============================================================ SEMANTIC_MAX_CANDIDATES = 3 SEMANTIC_TIMEOUT_SECONDS = 5 def semantic_candidate_score(result): """ Ranking score only. This does NOT decide whether two reports are duplicates. It only decides which rejected textual candidates are reasonable enough to ask the semantic AI about. """ if not result: return 0.0 primary = result.get("primary") or {} best = result.get("best") or {} primary_score = float( primary.get("score") or 0.0 ) best_score = float( best.get("score") or 0.0 ) primary_important = float( primary.get("important") or 0.0 ) best_important = float( best.get("important") or 0.0 ) return ( max(primary_score, best_score) * 0.70 + max( primary_important, best_important ) * 0.30 ) def is_semantic_candidate(item, story, result): """ Conservative gray-zone selector. AI is NOT called for every recent story. We require at least some lexical/entity/topic anchor. The semantic model then decides SAME vs DIFFERENT. This allows paraphrases such as the Hegseth/Iran example while avoiding comparisons between completely unrelated stories. """ if not result: return False primary = result.get("primary") or {} best = result.get("best") or {} candidates = [ primary, best ] max_score = max( float(x.get("score") or 0.0) for x in candidates ) max_jaccard = max( float(x.get("jaccard") or 0.0) for x in candidates ) max_containment = max( float(x.get("containment") or 0.0) for x in candidates ) max_important = max( float(x.get("important") or 0.0) for x in candidates ) # Completely weak relationships are not worth an AI call. # # Hegseth paraphrase: # score ~0.20 # jaccard ~0.136 # containment ~0.273 # important ~0.20 # # Therefore the lower boundary intentionally includes it. if ( max_score < 0.18 and max_jaccard < 0.10 and max_containment < 0.22 and max_important < 0.18 ): return False # Require at least two meaningful shared tokens. item_tokens = set( item.get("tokens") or [] ) primary_tokens = set( (story.get("primary") or {}).get( "tokens" ) or [] ) shared = ( item_tokens & primary_tokens ) if len(shared) >= 2: return True # Member fallback. for existing in story.get("items") or []: existing_tokens = set( existing.get("tokens") or [] ) if len( item_tokens & existing_tokens ) >= 2: return True return False def semantic_same_story(text_a, text_b): """ Ask the centralized PHP AI gateway. FAIL OPEN: Any error / timeout / malformed output returns False, meaning the new report is NOT suppressed. """ text_a = str( text_a or "" ).strip() text_b = str( text_b or "" ).strip() if not text_a or not text_b: return False, { "ok": False, "answer": "DIFFERENT", "error": "empty-text" } payload = json.dumps( { "text_a": text_a, "text_b": text_b }, ensure_ascii=False ) try: result = subprocess.run( [ "php", str(SEMANTIC_BRIDGE) ], input=payload, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=SEMANTIC_TIMEOUT_SECONDS ) output = ( result.stdout or "" ).strip() if not output: return False, { "ok": False, "answer": "DIFFERENT", "error": ( result.stderr.strip() or "empty-bridge-output" ) } data = json.loads(output) answer = str( data.get("answer") or "" ).strip().upper() return ( answer == "SAME", data ) except subprocess.TimeoutExpired: return False, { "ok": False, "answer": "DIFFERENT", "error": "semantic-timeout" } except Exception as exc: return False, { "ok": False, "answer": "DIFFERENT", "error": ( f"{type(exc).__name__}: {exc}" ) } def find_semantic_story(db, item): """ Called ONLY after the deterministic matcher rejected every normal story match. Returns: story, synthetic_match_result, reason, score, metadata """ stories = load_recent_stories( db, item ) candidates = [] for story in stories: if not can_compare( item, story ): continue result = story_similarity( item, story ) # If deterministic rules accept this story, it should # already have been handled by find_existing_story(). accepted, _ = should_match_story( item, story, result ) if accepted: continue if not is_semantic_candidate( item, story, result ): continue candidates.append( ( semantic_candidate_score( result ), story, result ) ) candidates.sort( key=lambda x: x[0], reverse=True ) candidates = candidates[ :SEMANTIC_MAX_CANDIDATES ] for candidate_score, story, result in candidates: primary = ( story.get("primary") or {} ) same, metadata = semantic_same_story( item.get("text_content"), primary.get("text_content") ) provider = ( metadata.get("provider") or "unknown" ) if same: return ( story, result, f"semantic-same:{provider}", candidate_score, metadata ) return ( None, None, None, 0.0, None ) 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: # ------------------------------------------------ # Semantic gray-zone fallback. # # Deterministic matcher rejected the item. # Before creating a new story, ask AI only about # the strongest plausible semantic candidates. # ------------------------------------------------ ( semantic_story, semantic_result, semantic_reason, semantic_score, semantic_meta ) = find_semantic_story( db, item ) if semantic_story is not None: success = attach_item_to_story( db, semantic_story, item, semantic_result ) if success: db.commit() processed += 1 attached += 1 print( f"[SEMANTIC ATTACHED] " f"feed={item['id']} " f"story={semantic_story['db_story_id']} " f"reason={semantic_reason} " f"score={semantic_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]" )