diff --git a/.gitignore b/.gitignore index 288a2c8..645ecdd 100644 --- a/.gitignore +++ b/.gitignore @@ -2,4 +2,4 @@ __pycache__ results rm_cache.py - +.env diff --git a/instaspyder/__main__.py b/instaspyder/__main__.py index cbfe32d..80c5a19 100644 --- a/instaspyder/__main__.py +++ b/instaspyder/__main__.py @@ -1,3 +1,4 @@ +# __main__.py import sys import asyncio import argparse @@ -22,9 +23,15 @@ async def cleanup_on_exit(): save_cumulative_results_for_keyword(kw, current_all_found_matches) await close_async_client() -async def run_search_async(seed=None, keywords=None, depth_arg=None): +async def run_search_async(seed=None, keywords=None, depth_arg=None, face_mode=False, dump_mode=False): global current_visited_users, current_all_found_matches, initial_seed_username_global, search_keywords_global + if face_mode and not keywords and seed: + print(f"\n{Y}[!] Standalone Face Mode: Skipping API search, using cache for @{seed}...{X}") + from instaspyder.core.face_engine import run_face_recognition + await run_face_recognition(seed, face_mode) + return + initialize_search_environment() if seed is None or keywords is None or depth_arg is None: @@ -45,7 +52,9 @@ async def run_search_async(seed=None, keywords=None, depth_arg=None): await init_async_client() try: - print(f"\n{G}Starting search from {C}@{initial_seed_username}{G} (Limit: {depth_arg} depths) for keywords: {Y}{', '.join(search_keywords)}{X}") + mode_str = f" {Y}[Face Mode Active]{G}" if face_mode else "" + dump_str = f" {C}[Dumping Raw Data]{G}" if dump_mode else "" # Visual feedback + print(f"\n{G}Starting search from {C}@{initial_seed_username}{G}{mode_str}{dump_str}...") await recursive_chain_search_async( initial_seed_username, @@ -53,9 +62,17 @@ async def run_search_async(seed=None, keywords=None, depth_arg=None): current_visited_users, current_all_found_matches, depth=0, - depth_limit=int(depth_arg) + depth_limit=int(depth_arg), + dump_mode=dump_mode ) print(f"\n{G}Overall search completed successfully.{X}") + + # Trigger Face Recognition after search if flag was provided + if face_mode: + from instaspyder.core.face_engine import run_face_recognition + print(f"\n{Y}[!] Starting Face Recognition Engine...{X}") + await run_face_recognition(initial_seed_username, face_mode) + except Exception as e: if "Instagram Block" in str(e): print(f"\n{R}Search halted by Instagram security. State NOT saved to prevent corruption.{X}") @@ -73,6 +90,8 @@ def main(): parser.add_argument("-k", "--keywords", help="Comma-separated keywords") parser.add_argument("-d", "--depth", type=int, help="Search depth (overrides config)") parser.add_argument("-c", "--config", action="store_true", help="Opens interactive configuration menu") + parser.add_argument("-f", "--face", type=str, help="Path to target image for face recognition search") + parser.add_argument("--dump", action="store_true", help="Dump raw metadata into hidden cache for later use") args = parser.parse_args() @@ -90,11 +109,9 @@ def main(): print(f"{R}Still no headers found. Exiting...{X}") sys.exit(1) - config = get_config() default_depth = config.get("max_depth", 2) - if args.depth is not None and args.depth > 2: print(f"\n{R} You are using aggressive depth{X} (i.e. {C}{args.depth}{X}) {R}It may flag you session cookies{X}") choice = input(f"\n{Y} Do you want to continue? (y/n): ").lower().strip() @@ -119,10 +136,15 @@ def main(): cli_keywords = [k.strip() for k in args.keywords.split(',') if k.strip()] try: - if args.seed and cli_keywords: - asyncio.run(run_search_async(seed=args.seed, keywords=cli_keywords, depth_arg=args.depth)) + if args.seed and args.face and not cli_keywords: + asyncio.run(run_search_async(seed=args.seed, keywords=None, face_mode=args.face)) + + elif args.seed and cli_keywords: + asyncio.run(run_search_async(seed=args.seed, keywords=cli_keywords, depth_arg=args.depth, face_mode=args.face, dump_mode=args.dump)) + else: - asyncio.run(run_search_async(depth_arg=args.depth)) + asyncio.run(run_search_async(depth_arg=args.depth, face_mode=args.face, dump_mode=args.dump)) + except KeyboardInterrupt: print(f"\n{R}Ctrl+C detected. Exiting gracefully...{X}") sys.exit(0) diff --git a/instaspyder/core/config_manager.py b/instaspyder/core/config_manager.py index f250635..cdc53e0 100644 --- a/instaspyder/core/config_manager.py +++ b/instaspyder/core/config_manager.py @@ -5,16 +5,20 @@ USER_HOME = Path.home() / ".instaspyder" CONFIG_FILE = USER_HOME / "config.json" HEADERS_FILE = USER_HOME / "headers.json" +CACHE_DIR = USER_HOME / ".cache" DEFAULT_CONFIG = { "max_depth": 2, "results_dir": str(USER_HOME / "results"), - "save_state": True + "save_state": True, + "enable_face_recognition": False } + def init_env(): """Ensures the config folder and files exist.""" USER_HOME.mkdir(parents=True, exist_ok=True) + CACHE_DIR.mkdir(parents=True, exist_ok=True) if not CONFIG_FILE.exists(): with open(CONFIG_FILE, 'w') as f: json.dump(DEFAULT_CONFIG, f, indent=4) diff --git a/instaspyder/core/face_engine.py b/instaspyder/core/face_engine.py new file mode 100644 index 0000000..5caebf5 --- /dev/null +++ b/instaspyder/core/face_engine.py @@ -0,0 +1,114 @@ +# instaspyder/core/face_engine.py +import face_recognition +import httpx +import json +import os +import gc +from instaspyder.utils.colors import G, R, Y, C, X +from instaspyder.core.config_manager import CACHE_DIR, USER_HOME + +async def run_face_recognition(seed_username, target_image_path): + cache_file = os.path.join(CACHE_DIR, f"metadata_{seed_username}.json") + progress_file = os.path.join(CACHE_DIR, f"progress_{seed_username}.json") + match_log_file = os.path.join(CACHE_DIR, f"matches_{seed_username}.json") + + temp_dir = USER_HOME / "temp" + temp_dir.mkdir(parents=True, exist_ok=True) + temp_path = str(temp_dir / "current_face.jpg") + + if not os.path.exists(cache_file): + print(f"{R}[-] No cache found for @{seed_username}.{X}") + return + + + try: + target_img = face_recognition.load_image_file(target_image_path) + target_encoding = face_recognition.face_encodings(target_img)[0] + except Exception as e: + print(f"{R}[-] Target Error: {e}{X}") + return + + with open(cache_file, "r") as f: + candidates = json.load(f) + + processed_ids = set() + if os.path.exists(progress_file): + with open(progress_file, "r") as f: + processed_ids = set(json.load(f)) + + all_potential_matches = [] + if os.path.exists(match_log_file): + with open(match_log_file, "r") as f: + all_potential_matches = json.load(f) + + print(f"{G}[+] Resuming scan. Found {len(all_potential_matches)} previous matches:{X}") + for m in all_potential_matches: + print(f" {C}[PREVIOUS MATCH] @{m['username']} ({m['confidence']:.2f}% similarity){X}") + + print(f"\n{Y}[i] Total: {len(candidates)} | Remaining: {len(candidates) - len(processed_ids)}{X}\n") + + async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client: + for i, user in enumerate(candidates, 1): + user_id = str(user.get("pk") or user.get("id")) + if user_id in processed_ids: + continue + + url = user.get("hd_pic_url") or user.get("profile_pic_url") + if not url: continue + + print(f"\r{X}[{i}/{len(candidates)}] Testing @{user['username']}... ", end="", flush=True) + + try: + resp = await client.get(url) + if resp.status_code == 200: + with open(temp_path, "wb") as tmp: + tmp.write(resp.content) + + unknown_img = face_recognition.load_image_file(temp_path) + unknown_encs = face_recognition.face_encodings(unknown_img) + + if unknown_encs: + dist = face_recognition.face_distance([target_encoding], unknown_encs[0])[0] + if dist <= 0.55: + conf = max(0, (1 - dist / 0.6) * 100) + new_match = { + "username": user['username'], + "full_name": user.get('full_name'), + "distance": dist, + "confidence": conf + } + all_potential_matches.append(new_match) + + with open(match_log_file, "w") as f: + json.dump(all_potential_matches, f, indent=4) + + print(f"\n{G}[LIVE MATCH] @{user['username']} ({conf:.2f}% similarity){X}") + + + processed_ids.add(user_id) + with open(progress_file, "w") as f: + json.dump(list(processed_ids), f) + + del unknown_img + gc.collect() + except KeyboardInterrupt: + print(f"\n{Y}[!] Pausing scan. State saved.{X}") + return + except: + continue + + + if os.path.exists(temp_path): os.remove(temp_path) + + print("\n" + "="*60) + print(f"{Y} FINAL REPORT (Sorted by Probability) {X}") + print("="*60) + + sorted_matches = sorted(all_potential_matches, key=lambda x: x['confidence'], reverse=True) + for match in sorted_matches: + color = G if match['confidence'] > 50 else Y + print(f"{color}[{match['confidence']:.2f}%]{X} @{match['username']} - {match['full_name']}") + + + if os.path.exists(progress_file): os.remove(progress_file) + if os.path.exists(match_log_file): os.remove(match_log_file) diff --git a/instaspyder/core/search_logic.py b/instaspyder/core/search_logic.py index 24911d0..17b3ed0 100644 --- a/instaspyder/core/search_logic.py +++ b/instaspyder/core/search_logic.py @@ -1,15 +1,20 @@ -# search_logic.py (Modified for Async - cleaner output) +# search_logic.py (Modified for Async - Deep Metadata Dumping) import asyncio import random from instaspyder.core.instagram_api import fetch_chain_async from instaspyder.core.user_id_fetcher import get_user_id from instaspyder.core.config_manager import get_config +from instaspyder.core.state_manager import silent_cache_metadata from instaspyder.utils.sanitize_text import sanitize_text from instaspyder.utils.colors import C, G, R, Y, X -async def recursive_chain_search_async(username, keywords_to_match, visited_users, all_found_matches, depth=0, known_user_id=None, depth_limit=None): +async def recursive_chain_search_async(username, keywords_to_match, visited_users, all_found_matches, depth=0, known_user_id=None, depth_limit=None, dump_mode=False, master_seed=None): indent = " " * depth + + if master_seed is None: + master_seed = username + # Fallback logic: if no limit was passed from main, check config if depth_limit is None: config = get_config() @@ -34,6 +39,7 @@ async def recursive_chain_search_async(username, keywords_to_match, visited_user users_in_chain = await fetch_chain_async(user_id) + if isinstance(users_in_chain, str) and "_ERROR" in users_in_chain: raise Exception(f"Instagram Block: {users_in_chain}") @@ -41,10 +47,14 @@ async def recursive_chain_search_async(username, keywords_to_match, visited_user print(f"{indent}{Y}No users found in chain for @{username}{X}") return + if dump_mode: + silent_cache_metadata(master_seed, users_in_chain) + + for user_data in users_in_chain: uname = user_data.get("username", "") fname = user_data.get("full_name", "") - uid = user_data.get("id") + uid = user_data.get("pk") or user_data.get("id") if not uname or not uid: continue @@ -74,10 +84,11 @@ async def recursive_chain_search_async(username, keywords_to_match, visited_user print(f" Found via: @{username} (Depth {depth})") print(f"{G}" + "="*60 + f"{X}\n") + if depth < depth_limit: tasks = [] for user_data in users_in_chain: - await asyncio.sleep(random.uniform(0.1, 0.2)) + await asyncio.sleep(random.uniform(0.1, 0.2)) tasks.append( recursive_chain_search_async( user_data["username"], @@ -85,8 +96,10 @@ async def recursive_chain_search_async(username, keywords_to_match, visited_user visited_users, all_found_matches, depth + 1, - known_user_id=user_data.get("id"), - depth_limit=depth_limit # Pass the limit down the chain + known_user_id=user_data.get("pk") or user_data.get("id"), + depth_limit=depth_limit, + dump_mode=dump_mode, + master_seed=master_seed ) ) if tasks: diff --git a/instaspyder/core/state_manager.py b/instaspyder/core/state_manager.py index 2eead21..f589fc5 100644 --- a/instaspyder/core/state_manager.py +++ b/instaspyder/core/state_manager.py @@ -3,10 +3,58 @@ from instaspyder.core.config_manager import get_config, USER_HOME from instaspyder.utils.colors import C, G, R, Y, X + +CACHE_DIR = os.path.join(USER_HOME, ".cache") + def _get_state_filepath(username): safe_username = "".join(c for c in username if c.isalnum()) return os.path.join(USER_HOME, f"search_state_{safe_username}.json") +def silent_cache_metadata(seed_username, users_list): + os.makedirs(CACHE_DIR, exist_ok=True) + cache_file = os.path.join(CACHE_DIR, f"metadata_{seed_username}.json") + + + new_cleaned_entries = [] + for user in users_list: + new_cleaned_entries.append({ + "pk": user.get("pk") or user.get("id"), + "username": user.get("username"), + "full_name": user.get("full_name"), + "profile_pic_url": user.get("profile_pic_url"), + "hd_pic_url": user.get("hd_profile_pic_url_info", {}).get("url") # Added for better face rec if available + }) + + try: + existing_data = [] + if os.path.exists(cache_file): + with open(cache_file, "r", encoding="utf-8") as f: + try: + existing_data = json.load(f) + except json.JSONDecodeError: + existing_data = [] + + # Merge data and avoid duplicates based on user ID (pk) + existing_pks = {str(u.get('pk')) for u in existing_data if u.get('pk')} + + updated_data = existing_data.copy() + added_any = False + + for entry in new_cleaned_entries: + pk_str = str(entry.get('pk')) + if pk_str not in existing_pks: + updated_data.append(entry) + existing_pks.add(pk_str) + added_any = True + + if added_any or not os.path.exists(cache_file): + with open(cache_file, "w", encoding="utf-8") as f: + json.dump(updated_data, f, indent=2, ensure_ascii=False) + + except Exception: + # Fails silently to ensure the main search logic is never interrupted + pass + def load_search_state(initial_username, current_keywords=None): visited_users = set() all_found_matches = [] @@ -15,35 +63,20 @@ def load_search_state(initial_username, current_keywords=None): try: with open(state_filepath, "r", encoding="utf-8") as f: state = json.load(f) - - # Check if this state belongs to the same user if state.get("initial_username") == initial_username: saved_keywords = state.get("keywords", []) - - # If keywords have changed, the old 'visited' list is irrelevant for a new search if current_keywords and set(current_keywords) != set(saved_keywords): - print(f"{Y}[!] Keywords changed from {saved_keywords} to {current_keywords}.{X}") - print(f"{C}[i] Ignoring previous visited history to ensure new keywords are checked.{X}") - # We keep the found_matches but clear visited to allow re-scanning - all_found_matches = state.get("found_matches", []) - return set(), all_found_matches + print(f"{Y}[!] Keywords changed. Ignoring previous visited history.{X}") + return set(), state.get("found_matches", []) visited_users = set(state.get("visited", [])) all_found_matches = state.get("found_matches", []) - print(f"{G}[+] Loaded previous state for '{initial_username}'. Resuming general search...{X}") - print(f" Total visited users: {len(visited_users)}") - print(f" Total matches found so far: {len(all_found_matches)}") + print(f"{G}[+] Loaded state for '{initial_username}'. Resuming...{X}") else: - print(f"{Y}[!] Saved state found at {state_filepath} but for a different initial user. Starting fresh.{X}") + print(f"{Y}[!] State mismatch. Starting fresh.{X}") except FileNotFoundError: - print(f"{C}[i] No previous search state found at {state_filepath}. Starting fresh for '{initial_username}'.{X}") - except json.JSONDecodeError: - print(f"{R}[!] Corrupted state file detected at {state_filepath}. Starting fresh.{X}") - try: - os.remove(state_filepath) - except OSError: - pass + pass except Exception as e: print(f"{R}[-] Error loading state: {e}. Starting fresh.{X}") @@ -55,32 +88,21 @@ def save_search_state(initial_username, visited_users, all_found_matches, keywor "visited": list(visited_users), "found_matches": all_found_matches, "initial_username": initial_username, - "keywords": keywords # Save keywords to detect changes later + "keywords": keywords } try: with open(state_filepath, "w", encoding="utf-8") as f: json.dump(state, f, indent=4, ensure_ascii=False) - print(f"\n{G}[+] Search state saved to {state_filepath}{X}") except Exception as e: - print(f"{R}[-] Error saving state to {state_filepath}: {e}{X}") + print(f"{R}[-] Error saving state: {e}{X}") def save_cumulative_results_for_keyword(keyword, all_matches_data): keyword_matches = [m for m in all_matches_data if m.get("matched_keyword") == keyword] - if not keyword_matches: - print(f"{C}[i] No matches found for keyword '{keyword}'. Skipping saving.{X}") - return - + if not keyword_matches: return config = get_config() - cumulative_results_dir = config.get("results_dir") - - folder = os.path.join(cumulative_results_dir, keyword.lower()) + folder = os.path.join(config.get("results_dir"), keyword.lower()) os.makedirs(folder, exist_ok=True) filepath = os.path.join(folder, f"{keyword.lower()}_matches.json") - - try: - with open(filepath, "w", encoding="utf-8") as f: - json.dump(keyword_matches, f, indent=4, ensure_ascii=False) - print(f"{G}[+] Matches for '{keyword}' saved to {filepath}{X}") - except Exception as e: - print(f"{R}[-] Error saving cumulative results for '{keyword}' to {filepath}: {e}{X}") + with open(filepath, "w", encoding="utf-8") as f: + json.dump(keyword_matches, f, indent=4, ensure_ascii=False)