#!/usr/bin/env python3 import asyncio import re import statistics import sys from pathlib import Path from urllib.parse import urlparse import pymysql from telethon import TelegramClient, functions, types from telethon.errors import ( FloodWaitError, UsernameInvalidError, UsernameNotOccupiedError, ChannelPrivateError, ) BASE_DIR = Path(__file__).resolve().parent.parent DB_CONFIG = BASE_DIR / "config" / "database.php" SESSION_DIR = BASE_DIR / "engine" / "sessions" PUBLISHER_ID = 1 SEED_SCAN_LIMIT = 300 CANDIDATE_SAMPLE_SCAN = 250 MAX_RECOMMENDATIONS_PER_SEED = 100 MAX_NEW_CANDIDATES_PER_RUN = 100 GROUP_PENDING_LIMIT = 5000 def fail(message: str): print("") print("❌ ERROR") print(message) sys.exit(1) def read_php_database_config(): content = DB_CONFIG.read_text( encoding="utf-8", errors="ignore" ) patterns = { "host": r"\$dbHost\s*=\s*['\"]([^'\"]+)['\"]", "name": 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: fail( f"Could not read {key} " "from config/database.php" ) result[key] = match.group(1) return result def get_database(): cfg = read_php_database_config() return pymysql.connect( host=cfg["host"], user=cfg["user"], password=cfg["password"], database=cfg["name"], charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=True, ) def get_publisher(db): with db.cursor() as cursor: cursor.execute( """ SELECT * FROM publishers WHERE id = %s AND is_active = 1 LIMIT 1 """, (PUBLISHER_ID,) ) return cursor.fetchone() def build_client(publisher): api_id_raw = str( publisher.get("telegram_api_id") or "" ).strip() api_hash = str( publisher.get("telegram_api_hash") or "" ).strip() session_name = str( publisher.get("session_name") or "" ).strip() if not api_id_raw: fail("Telegram API ID missing.") if not api_hash: fail("Telegram API Hash missing.") api_id = int(api_id_raw) if not session_name: session_name = f"publisher_{PUBLISHER_ID}" session_name = re.sub( r"[^a-zA-Z0-9_-]", "_", session_name ) session_path = SESSION_DIR / session_name return TelegramClient( str(session_path), api_id, api_hash ) def normalize_username(value): if not value: return "" value = str(value).strip() value = re.sub( r"^https?://(?:www\.)?t\.me/", "", value, flags=re.I ) value = re.sub( r"^https?://(?:www\.)?telegram\.me/", "", value, flags=re.I ) value = value.lstrip("@") value = value.split("?")[0] value = value.split("/")[0] if not re.fullmatch( r"[A-Za-z0-9_]{4,64}", value ): return "" return value GENERIC_DISCOVERY_USERNAMES = { "channel", "premium", "telegram", "telegramtips", "botfather", "spam", } def is_generic_discovery_username(username): value = normalize_username(username).lower() if not value: return True return value in GENERIC_DISCOVERY_USERNAMES def extract_tme_usernames(text): if not text: return set() results = set() patterns = [ r"(?:https?://)?t\.me/([A-Za-z0-9_]{4,64})", r"(?:https?://)?telegram\.me/([A-Za-z0-9_]{4,64})", ] for pattern in patterns: for match in re.findall( pattern, text, flags=re.I ): username = normalize_username(match) if username: results.add(username) return results def get_group(db, group_key): with db.cursor() as cursor: cursor.execute( """ SELECT * FROM content_groups WHERE group_key = %s AND is_active = 1 LIMIT 1 """, (group_key,) ) return cursor.fetchone() def get_seeds(db, group_id): with db.cursor() as cursor: cursor.execute( """ SELECT s.id AS source_id, s.username, s.telegram_chat_id, s.name FROM source_group_sources sgs INNER JOIN sources s ON s.id = sgs.source_id WHERE sgs.group_id = %s AND sgs.source_role IN ('seed','approved') AND sgs.is_active = 1 AND s.is_active = 1 AND s.source_kind = 'telegram' ORDER BY s.id ASC """, (group_id,) ) return cursor.fetchall() def get_pending_count( db, group_id ): with db.cursor() as cursor: cursor.execute( """ SELECT COUNT(*) AS total FROM source_discovery_candidates WHERE group_id = %s AND review_status = 'pending' """, (group_id,) ) row = cursor.fetchone() return int( (row or {}).get( "total", 0 ) ) def load_blocked_usernames(db): result = set() with db.cursor() as cursor: cursor.execute( """ SELECT username FROM source_discovery_blacklist WHERE username IS NOT NULL """ ) for row in cursor.fetchall(): username = normalize_username( row["username"] ) if username: result.add(username.lower()) cursor.execute( """ SELECT username FROM sources WHERE source_kind = 'telegram' AND username IS NOT NULL """ ) for row in cursor.fetchall(): username = normalize_username( row["username"] ) if username: result.add(username.lower()) return result def media_type(message): if getattr(message, "photo", None): return "photo" if getattr(message, "video", None): return "video" if getattr(message, "voice", None): return "voice" if getattr(message, "audio", None): return "audio" if getattr(message, "document", None): return "document" if getattr(message, "gif", None): return "gif" return "text" def sample_score(message, median_views): views = int( getattr(message, "views", 0) or 0 ) forwards = int( getattr(message, "forwards", 0) or 0 ) text = ( getattr(message, "message", None) or "" ).strip() if median_views <= 0: relative = 0 else: relative = views / median_views score = relative * 100 score += min( forwards, 1000 ) * 0.20 if text: score += 8 kind = media_type(message) if kind in { "photo", "video", "audio", "voice" }: score += 5 return round(score, 2) async def get_channel_metadata(client, entity): name = ( getattr(entity, "title", None) or getattr(entity, "first_name", None) or "" ) username = normalize_username( getattr(entity, "username", "") ) telegram_chat_id = str( getattr(entity, "id", "") or "" ) description = None member_count = None try: full = await client( functions.channels.GetFullChannelRequest( channel=entity ) ) description = getattr( full.full_chat, "about", None ) member_count = getattr( full.full_chat, "participants_count", None ) except Exception: pass return { "name": name, "username": username, "telegram_chat_id": telegram_chat_id, "description": description, "member_count": member_count, } async def analyze_music_candidate(client, entity): """ Music-specific candidate analysis. Scans recent usable posts BEFORE the candidate is saved. Rules: - At least 20% of usable posts must be Telegram audio. - At least 2 audio posts must exist. - Samples shown in Source Discovery are audio-first. - Returns a music relevance score based on audio ratio. """ messages = [] try: async for message in client.iter_messages( entity, limit=CANDIDATE_SAMPLE_SCAN ): if not message: continue text = ( getattr(message, "message", None) or "" ).strip() kind = media_type(message) # Ignore real empty/service messages. if not text and kind == "text": continue messages.append(message) except FloodWaitError: raise except Exception: return { "accepted": False, "usable_count": 0, "audio_count": 0, "audio_ratio": 0.0, "music_score": 0.0, "samples": [], } usable_count = len(messages) audio_messages = [ message for message in messages if media_type(message) == "audio" ] audio_count = len(audio_messages) audio_ratio = ( audio_count / usable_count if usable_count > 0 else 0.0 ) # A real Music source must have enough actual audio. accepted = ( usable_count > 0 and audio_count >= 2 and audio_ratio >= 0.20 ) if not accepted: return { "accepted": False, "usable_count": usable_count, "audio_count": audio_count, "audio_ratio": audio_ratio, "music_score": 0.0, "samples": [], } view_values = [ int(getattr(m, "views", 0) or 0) for m in messages if int( getattr(m, "views", 0) or 0 ) > 0 ] median_views = ( statistics.median(view_values) if view_values else 0 ) ranked = [] for message in audio_messages: score = sample_score( message, median_views ) ranked.append( ( score, int( getattr( message, "views", 0 ) or 0 ), message ) ) ranked.sort( key=lambda x: ( x[0], x[1] ), reverse=True ) samples = [] for score, _, message in ranked[:2]: text = ( getattr(message, "message", None) or "" ).strip() samples.append( { "telegram_message_id": int(message.id), "message_text": text or None, "media_type": "audio", "views": int( getattr( message, "views", 0 ) or 0 ), "forwards": int( getattr( message, "forwards", 0 ) or 0 ), "posted_at": ( message.date .replace(tzinfo=None) if message.date else None ), "sample_score": score, } ) # 20% audio = 20 score # 100% audio = 100 score music_score = round( min( 100.0, audio_ratio * 100.0 ), 2 ) return { "accepted": True, "usable_count": usable_count, "audio_count": audio_count, "audio_ratio": audio_ratio, "music_score": music_score, "samples": samples, } async def get_best_samples(client, entity): messages = [] try: async for message in client.iter_messages( entity, limit=CANDIDATE_SAMPLE_SCAN ): if not message: continue text = ( getattr(message, "message", None) or "" ).strip() if ( not text and media_type(message) == "text" ): continue messages.append(message) except Exception: return [] if not messages: return [] view_values = [ int(getattr(m, "views", 0) or 0) for m in messages if int( getattr(m, "views", 0) or 0 ) > 0 ] median_views = ( statistics.median(view_values) if view_values else 0 ) ranked = [] for message in messages: score = sample_score( message, median_views ) ranked.append( ( score, int( getattr( message, "views", 0 ) or 0 ), message ) ) ranked.sort( key=lambda x: ( x[0], x[1] ), reverse=True ) chosen = [] seen_texts = set() for score, _, message in ranked: text = ( getattr(message, "message", None) or "" ).strip() normalized = re.sub( r"\s+", " ", text ).strip().lower() if normalized and normalized in seen_texts: continue if normalized: seen_texts.add(normalized) chosen.append( { "telegram_message_id": int(message.id), "message_text": text or None, "media_type": media_type(message), "views": int( getattr( message, "views", 0 ) or 0 ), "forwards": int( getattr( message, "forwards", 0 ) or 0 ), "posted_at": ( message.date .replace(tzinfo=None) if message.date else None ), "sample_score": score, } ) if len(chosen) >= 2: break return chosen async def get_recommendations(client, seed_entity): results = [] try: if not hasattr( functions.channels, "GetChannelRecommendationsRequest" ): return results response = await client( functions.channels .GetChannelRecommendationsRequest( channel=seed_entity ) ) chats = ( getattr(response, "chats", None) or [] ) for chat in chats[ :MAX_RECOMMENDATIONS_PER_SEED ]: if not isinstance( chat, types.Channel ): continue username = normalize_username( getattr( chat, "username", "" ) ) if not username: continue results.append( { "username": username, "entity": chat, "reason": "telegram_recommendation", "score": 100, } ) except FloodWaitError: raise except Exception: pass return results async def get_seed_link_candidates( client, seed_entity ): found = {} try: async for message in client.iter_messages( seed_entity, limit=SEED_SCAN_LIMIT ): text = ( getattr( message, "message", None ) or "" ) for username in extract_tme_usernames( text ): key = username.lower() if key not in found: found[key] = { "username": username, "entity": None, "reason": "seed_post_link", "score": 60, } else: found[key]["score"] += 5 fwd = getattr( message, "fwd_from", None ) if fwd: from_id = getattr( fwd, "from_id", None ) if isinstance( from_id, types.PeerChannel ): try: entity = await client.get_entity( from_id ) username = normalize_username( getattr( entity, "username", "" ) ) if username: key = username.lower() if key not in found: found[key] = { "username": username, "entity": entity, "reason": "forwarded_channel", "score": 75, } else: found[key][ "score" ] += 10 except Exception: pass except FloodWaitError: raise except Exception: pass return list(found.values()) def upsert_candidate( db, group_id, source_id, metadata, relevance_score, reason, discovery_depth=1 ): username = metadata["username"] if not username: return None with db.cursor() as cursor: cursor.execute( """ INSERT INTO source_discovery_candidates ( group_id, discovered_from_source_id, telegram_chat_id, username, name, description, member_count, relevance_score, discovery_reason, discovery_depth, review_status, join_status ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'pending', 'not_queued' ) ON DUPLICATE KEY UPDATE relevance_score = GREATEST( relevance_score, VALUES(relevance_score) ), name = COALESCE( VALUES(name), name ), telegram_chat_id = COALESCE( VALUES(telegram_chat_id), telegram_chat_id ), description = COALESCE( VALUES(description), description ), member_count = COALESCE( VALUES(member_count), member_count ), discovery_depth = LEAST( discovery_depth, VALUES(discovery_depth) ), updated_at = CURRENT_TIMESTAMP """, ( group_id, source_id, metadata[ "telegram_chat_id" ] or None, username, metadata["name"] or None, metadata[ "description" ], metadata[ "member_count" ], relevance_score, reason, discovery_depth, ) ) cursor.execute( """ SELECT id FROM source_discovery_candidates WHERE group_id = %s AND LOWER(username) = LOWER(%s) LIMIT 1 """, ( group_id, username ) ) row = cursor.fetchone() return ( int(row["id"]) if row else None ) def replace_samples( db, candidate_id, samples ): with db.cursor() as cursor: cursor.execute( """ DELETE FROM source_discovery_samples WHERE candidate_id = %s """, (candidate_id,) ) insert_sql = """ INSERT INTO source_discovery_samples ( candidate_id, telegram_message_id, message_text, media_type, views, forwards, posted_at, sample_score ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s ) """ for sample in samples: cursor.execute( insert_sql, ( candidate_id, sample[ "telegram_message_id" ], sample[ "message_text" ], sample[ "media_type" ], sample["views"], sample["forwards"], sample["posted_at"], sample[ "sample_score" ], ) ) async def main(): group_key = ( sys.argv[1].strip().lower() if len(sys.argv) > 1 else "deep" ) db = get_database() publisher = get_publisher(db) if not publisher: fail( f"Active Publisher " f"#{PUBLISHER_ID} not found." ) group = get_group( db, group_key ) if not group: fail( f"Content group '{group_key}' " "not found." ) group_id = int(group["id"]) pending_count = get_pending_count( db, group_id ) available_slots = max( 0, GROUP_PENDING_LIMIT - pending_count ) run_candidate_cap = min( MAX_NEW_CANDIDATES_PER_RUN, available_slots ) if available_slots <= 0: print("") print( "SOURCE DISCOVERY WORKER" ) print( "==============================================" ) print( f"Group: {group_key}" ) print( f"Pending: {pending_count}" f"/{GROUP_PENDING_LIMIT}" ) print( "Discovery: PAUSED ✅" ) print( "Reason: Pending pool is full." ) print( "==============================================" ) db.close() return seeds = get_seeds( db, group_id ) if not seeds: fail( f"No seeds found for " f"group '{group_key}'." ) blocked = load_blocked_usernames( db ) client = build_client( publisher ) await client.connect() if not await client.is_user_authorized(): fail( "Telegram session " "is not authorized." ) print("") print("SOURCE DISCOVERY WORKER V1.0") print( "==============================================" ) print(f"Group: {group_key}") print( f"Pending Pool: " f"{pending_count}/{GROUP_PENDING_LIMIT}" ) print( f"Available: {available_slots}" ) print(f"Seeds: {len(seeds)}") print( f"Candidate cap: " f"{run_candidate_cap}" ) print("Mode: READ/DISCOVER ONLY") print("Join: DISABLED") print( "==============================================" ) processed = 0 saved = 0 sample_count = 0 run_seen = set() try: for seed in seeds: if saved >= run_candidate_cap: break seed_username = normalize_username( seed["username"] ) if not seed_username: continue print("") print( f"[SEED] @{seed_username}" ) try: seed_entity = ( await client.get_entity( seed_username ) ) except Exception as exc: print( " Resolve failed: " + str(exc) ) continue candidates = [] recommendations = ( await get_recommendations( client, seed_entity ) ) candidates.extend( recommendations ) links = ( await get_seed_link_candidates( client, seed_entity ) ) candidates.extend( links ) merged = {} for item in candidates: username = normalize_username( item.get("username") ) if not username: continue if is_generic_discovery_username(username): continue key = username.lower() if ( key == seed_username.lower() or key in blocked ): continue if key not in merged: merged[key] = item else: merged[key]["score"] += ( item.get("score", 0) ) old_reason = ( merged[key] .get("reason", "") ) new_reason = ( item.get( "reason", "" ) ) if ( new_reason and new_reason not in old_reason ): merged[key][ "reason" ] = ( old_reason + "+" + new_reason ).strip("+") print( f" Found: {len(merged)}" ) for item in sorted( merged.values(), key=lambda x: x.get( "score", 0 ), reverse=True ): if ( saved >= MAX_NEW_CANDIDATES_PER_RUN ): break username = normalize_username( item["username"] ) key = username.lower() if key in run_seen: continue run_seen.add(key) try: entity = ( item.get("entity") ) if entity is None: entity = ( await client .get_entity( username ) ) if not isinstance( entity, types.Channel ): continue if not getattr( entity, "broadcast", False ): continue metadata = ( await get_channel_metadata( client, entity ) ) if not metadata[ "username" ]: continue # ------------------------------------------------- # MUSIC GROUP FILTER # ------------------------------------------------- # # Telegram recommendations are not trusted blindly # for Music. Inspect the candidate's recent content # BEFORE saving it into the candidate pool. # music_analysis = None relevance_score = float( item.get( "score", 0 ) ) if group_key == "music": music_analysis = ( await analyze_music_candidate( client, entity ) ) ratio_percent = round( music_analysis[ "audio_ratio" ] * 100, 1 ) if not music_analysis[ "accepted" ]: print( " - @" + metadata["username"] + " | MUSIC SKIP" + " | audio=" + str( music_analysis[ "audio_count" ] ) + "/" + str( music_analysis[ "usable_count" ] ) + " (" + str(ratio_percent) + "%)" ) continue relevance_score = ( music_analysis[ "music_score" ] ) candidate_id = ( upsert_candidate( db, group_id, int( seed[ "source_id" ] ), metadata, relevance_score, item.get( "reason", "discovered" ), ) ) if not candidate_id: continue if ( group_key == "music" and music_analysis is not None ): samples = ( music_analysis[ "samples" ] ) else: samples = ( await get_best_samples( client, entity ) ) replace_samples( db, candidate_id, samples ) processed += 1 saved += 1 sample_count += len( samples ) print( " + @" + metadata[ "username" ] + " | score=" + str( relevance_score ) + " | samples=" + str( len(samples) ) ) except FloodWaitError as exc: print( "" "⚠️ FloodWait " f"{exc.seconds}s" ) print( "Discovery stopped safely." ) return except ( UsernameInvalidError, UsernameNotOccupiedError, ChannelPrivateError, ): continue except Exception as exc: print( " Candidate error @" + username + ": " + str(exc) ) finally: await client.disconnect() db.close() print("") print( "==============================================" ) print("DISCOVERY COMPLETE ✅") print(f"Processed: {processed}") print(f"Candidates: {saved}") print(f"Samples: {sample_count}") print("Joined: 0") print( "==============================================" ) if __name__ == "__main__": asyncio.run(main())