# scraper/crawler.py import os import time import random import logging from datetime import datetime, timezone from instagrapi import Client from instagrapi.exceptions import LoginRequired, RateLimitError from config.loader import load_config, get_active_crawler_config, get_active_hashtags from database.repository import ( log_snapshot, calculate_cycle_bucket, upsert_discovered_tag, promote_pending_tags, get_active_discovered_tags, save_leaderboard, get_rank_changes, get_alerted_tags, mark_trend_alerted, get_meta, set_meta, ) from analytics.engine import calculate_top_velocity_regression, detect_stagnant_tags, detect_rising_tags from notifications import send_trend_alerts def run_discovery(client: Client, tracked_tags: list, crawler_cfg: dict, is_dev: bool): """ Tag lifecycle (C): each cycle, probe topsearch for a rotating subset of tracked tags; newly seen related tags land in discovered_tags, and tags seen often enough get auto-promoted into the tracked set. """ subset_size = crawler_cfg.get("discovery_tags_per_cycle", 3) subset = random.sample(tracked_tags, min(subset_size, len(tracked_tags))) min_media = crawler_cfg.get("discovery_min_media_count", 10000) known = set(t.lower() for t in tracked_tags) found = 0 for tag in subset: try: related = client.web_search_topsearch_hashtags(tag) for h in related: name = (h.name or "").lower().strip("#") if not name or name in known: continue if (h.media_count or 0) < min_media: continue upsert_discovered_tag(name, f"topsearch:{tag}", h.media_count or 0) found += 1 time.sleep(random.uniform(3, 6)) except Exception as e: logging.warning(f"Discovery probe failed for #{tag}: {e}") promoted = promote_pending_tags(crawler_cfg.get("discovery_promote_mentions", 2)) if promoted: logging.info(f"Auto-promoted {len(promoted)} discovered tags into tracking: {promoted}") if found: logging.info(f"Discovery pass: {found} new candidate tag sightings from {len(subset)} probes.") def harvest_viral_hashtags(client: Client, is_dev: bool): """ Cross-project synergy (G): read captions of posts tracked by the sibling viral-posts project, extract hashtags competitor posts actually use, and feed them into discovery. No API calls — pure DB read. """ viral_db = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "..", "2_viral-posts", "data", "viral_tracker.db") if not os.path.exists(viral_db): return 0 try: import sqlite3 conn = sqlite3.connect(viral_db) rows = conn.execute( "SELECT username, caption_text, like_count FROM tracked_posts WHERE caption_text IS NOT NULL" ).fetchall() conn.close() except Exception as e: logging.warning(f"Could not read sibling viral-posts DB: {e}") return 0 found = 0 for username, caption, likes in rows: tags = [t.strip("#").lower() for t in caption.split() if t.startswith("#") and len(t) > 2] for t in tags[:5]: if t.isalnum(): upsert_discovered_tag(t, f"viral_post:{username}", int(likes or 0)) found += 1 if found: logging.info(f"Harvested {found} hashtag sightings from sibling viral-posts project.") return found def _announce_trends_deprecated(leaderboard, is_dev: bool): """Deprecated: replaced by notifications.send_trend_alerts. (E) when a tag NEWLY enters the top-k or accelerates hard.""" if leaderboard.empty: return already = get_alerted_tags() new_rows, accel_rows = [], [] for _, r in leaderboard.iterrows(): tag = str(r["hashtag"]).lstrip("#") if tag not in already: new_rows.append((tag, f"new top-{len(leaderboard)} entry", f"weekly_velocity={r['weekly_velocity']}")) acc = r.get("acceleration") if acc is not None and not pd_isna(acc) and acc >= 2.0: accel_rows.append((tag, "acceleration", f"recent pace {acc}x its 7-day trend")) payloads = new_rows + [a for a in accel_rows if a[0] not in [n[0] for n in new_rows]] if not payloads: return import pandas as pd alerts_df = pd.DataFrame(payloads, columns=["tag", "reason", "detail"]) if send_trend_alerts(alerts_df): for tag, reason, detail in payloads: mark_trend_alerted(tag, reason, detail) def pd_isna(v): return v != v # NaN check without importing pandas here def run_collection_cycle(client: Client): cfg = load_config() crawler_cfg = get_active_crawler_config(cfg) hashtags = get_active_hashtags(cfg) storage_cfg = cfg["storage"] is_dev = cfg.get("develop_mode", False) total_tags = len(hashtags) logging.info( f"Starting crawl cycle [Mode: {'DEVELOP' if is_dev else 'PRODUCTION'}]: " f"{total_tags} hashtags (Cycle window: {crawler_cfg.get('cycle_hours')}h)..." ) tag_list = list(hashtags) if crawler_cfg.get("shuffle_hashtags", True): random.shuffle(tag_list) batch_pause_every = crawler_cfg.get("batch_pause_every", 10) batch_pause_seconds = crawler_cfg.get("batch_pause_seconds", 35.0) for idx, tag in enumerate(tag_list, start=1): clean_tag = tag.strip("#").lower() success = False retries = crawler_cfg.get("max_retries", 2) while retries >= 0 and not success: try: tag_info = client.hashtag_info(clean_tag) tag_data = tag_info.model_dump() log_snapshot( tag_data, hours_per_cycle=crawler_cfg.get("cycle_hours", 3), db_path=storage_cfg["db_file"] ) media_count = tag_data.get("media_count", 0) logging.info(f"[{idx}/{total_tags}] #{clean_tag}: {media_count:,} posts (Saved)") success = True except RateLimitError: cooldown_sec = random.uniform(600.0, 900.0) logging.warning( f"429 RateLimitError on #{clean_tag}! Entering long cooldown: {cooldown_sec / 60.0:.1f} minutes..." ) time.sleep(cooldown_sec) retries -= 1 except LoginRequired: logging.error("Session expired during crawl. Re-authenticating from session settings...") client.load_settings(storage_cfg["session_settings_file"]) retries -= 1 time.sleep(15) except Exception as e: err_msg = str(e) if "429" in err_msg or "Too Many Requests" in err_msg: cooldown_sec = random.uniform(600.0, 900.0) logging.warning( f"HTTP 429 detected in #{clean_tag}: {e}. Entering cooldown: {cooldown_sec / 60.0:.1f} minutes..." ) time.sleep(cooldown_sec) else: err_delay = random.uniform( crawler_cfg.get("error_delay_min_seconds", 5.0 if is_dev else 45.0), crawler_cfg.get("error_delay_max_seconds", 10.0 if is_dev else 90.0) ) logging.error(f"Error crawling #{clean_tag}: {e} (Retries left: {retries}). Waiting {err_delay:.1f}s...") time.sleep(err_delay) retries -= 1 if idx != total_tags: if not is_dev and idx % batch_pause_every == 0: extra_rest = batch_pause_seconds + random.uniform(-5.0, 10.0) logging.info(f"Batch checkpoint reached ({idx}/{total_tags}). Resting {extra_rest:.1f}s...") time.sleep(extra_rest) else: sleep_duration = random.uniform( crawler_cfg.get("min_delay_seconds", 5.0 if is_dev else 12.0), crawler_cfg.get("max_delay_seconds", 10.0 if is_dev else 22.0) ) logging.info(f"Pacing delay: resting {sleep_duration:.1f}s before next hashtag...") time.sleep(sleep_duration) logging.info("Crawl cycle finished. Next cycle scheduled by interval.") # --- Tag discovery (C) + cross-project synergy (G), throttled to once per day --- today_iso = datetime.now(timezone.utc).strftime("%Y-%m-%d") if get_meta("last_discovery") != today_iso: set_meta("last_discovery", today_iso) try: run_discovery(client, list(hashtags), crawler_cfg, is_dev) harvest_viral_hashtags(client, is_dev) except Exception as e: logging.warning(f"Discovery pass failed (non-fatal): {e}") # --- Leaderboard (A/B), history, rank changes, and alerts (E) --- min_days_threshold = 0.0 if is_dev else cfg["analytics"]["default_min_days"] top_k = cfg["analytics"].get("default_top_k", 10) leaderboard = calculate_top_velocity_regression(min_days=min_days_threshold, top_k=top_k) if not leaderboard.empty: bucket = calculate_cycle_bucket(crawler_cfg.get("cycle_hours", 3)) save_leaderboard(leaderboard, bucket) # Compute rank changes vs previous leaderboard current_tags = [str(r["hashtag"]).lstrip("#") for _, r in leaderboard.iterrows()] rank_changes = get_rank_changes(current_tags) if not rank_changes.empty: rank_changes["rank_change"] = rank_changes["prev_rank"] - rank_changes["rank"] rank_changes = rank_changes[rank_changes["rank_change"] != 0] # Add rank to leaderboard for display leaderboard["rank"] = range(1, len(leaderboard) + 1) # Send trend alerts (Telegram/Discord) — config-driven, deduped if not is_dev: send_trend_alerts(leaderboard, rank_changes, reason="top_k") cols = ["rank", "hashtag", "current_count", "weekly_velocity", "growth_rate_pct", "acceleration", "trend_score", "baseline_ratio", "r_squared"] cols = [c for c in cols if c in leaderboard.columns] logging.info("\n" + leaderboard[cols].to_string(index=False)) # --- Lifecycle management (H): prune stagnant, promote rising --- lifecycle_cfg = cfg.get("lifecycle", {}) if lifecycle_cfg.get("enabled", True) and not is_dev: try: # Detect stagnant tags (low velocity for extended period) stagnant = detect_stagnant_tags( min_days=lifecycle_cfg.get("stagnant_min_days", 14), stagnant_threshold=lifecycle_cfg.get("stagnant_velocity_threshold", 100.0) ) if stagnant: logging.info(f"Lifecycle: {len(stagnant)} stagnant tag(s) detected for review: {stagnant}") # Detect rising tags (high acceleration) rising = detect_rising_tags( min_acceleration=lifecycle_cfg.get("rising_min_acceleration", 1.5), min_snapshots=lifecycle_cfg.get("rising_min_snapshots", 5) ) if rising: rising_names = [t for t, _ in rising] logging.info(f"Lifecycle: {len(rising)} rising tag(s) detected: {rising}") except Exception as e: logging.warning(f"Lifecycle check failed (non-fatal): {e}")