From ac3416a1d18b191257c28ce5f09de212a6223428 Mon Sep 17 00:00:00 2001 From: sethwv-alt Date: Fri, 17 Oct 2025 12:10:57 -0400 Subject: [PATCH] initial commit --- README.md | 141 +++++++++++++++ requirements.txt | 2 + xtream-search.py | 451 +++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 594 insertions(+) create mode 100644 README.md create mode 100644 requirements.txt create mode 100644 xtream-search.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..8fc44c6 --- /dev/null +++ b/README.md @@ -0,0 +1,141 @@ +# Xtream Codes Channel Search + +A fast, interactive Python tool for searching channels across Xtream Codes IPTV services. Features multithreaded searching with progress bars and a clean, user-friendly interface. + +## Features + +- ๐Ÿ” **Interactive Search Mode** - Connect once, search multiple times +- โšก **Multithreaded** - Fast concurrent searches across categories +- ๐Ÿ“Š **Progress Bars** - Real-time progress with ETA +- ๐ŸŽฏ **Content Type Filtering** - Search Live, VOD, Series separately or all together +- ๐Ÿงน **Clean Output** - Compact, readable results +- ๐Ÿ”ง **Debug Mode** - API connectivity testing and troubleshooting + +## Installation + +1. Clone or download this repository +2. Install dependencies: + ```bash + pip install -r requirements.txt + ``` + +## Usage + +### Interactive Mode (Recommended) +```bash +python xtream-search.py --api-url "http://your-server.com" --username "your_user" --password "your_pass" +``` + +### One-time Search +```bash +python xtream-search.py --api-url "http://your-server.com" --username "your_user" --password "your_pass" --channel "cnn" +``` + +### Advanced Options +```bash +# Search all content types (Live + VOD + Series) +python xtream-search.py --api-url "http://your-server.com" --username "user" --password "pass" --channel "movie" --all-types + +# Increase thread count for faster searching +python xtream-search.py --api-url "http://your-server.com" --username "user" --password "pass" --max-workers 20 + +# Debug mode to test API connectivity +python xtream-search.py --api-url "http://your-server.com" --username "user" --password "pass" --debug +``` + +## Interactive Search Commands + +Once in interactive mode, use these search commands: + +- **`channel_name`** - Search live streams (default) +- **`live:espn`** - Search only live streams +- **`vod:movie`** - Search only VOD content +- **`series:friends`** - Search only series +- **`all:news`** - Search all content types +- **`debug`** - Show API information +- **`quit`** or **`exit`** - Exit program + +## Sample Output + +``` +Connected to: http://your-server.com +Commands: live: | vod: | series: | all: | debug | quit +-------------------------------------------------- + +๐Ÿ” Search: cnn + +๐Ÿ“ก Fetching live categories... +โœ… Found 25 categories to search +๐Ÿ” Searching live categories: 100%|โ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆ| 25/25 [00:03<00:00] + +โœ… Found 3 result(s) for 'cnn': + 1. [Live] News Channels โ†’ CNN International + 2. [Live] US News โ†’ CNN USA + 3. [Live] Breaking News โ†’ CNN Breaking News + +๐Ÿ” Search: vod:action + +๐Ÿ“ก Fetching VOD categories... +โœ… Found 15 VOD categories +๐Ÿ” Searching vod: 100%|โ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆโ–ˆ| 15/15 [00:02<00:00] + +โœ… Found 5 result(s) for 'action': + 1. [VOD] Action Movies โ†’ Die Hard + 2. [VOD] Action Movies โ†’ Mad Max + 3. [VOD] Thriller โ†’ Action Jackson + 4. [VOD] Adventures โ†’ Action Heroes + 5. [VOD] Classics โ†’ Action Classics +``` + +## Command Line Arguments + +| Argument | Required | Description | +|----------|----------|-------------| +| `--api-url` | โœ… | Xtream Codes server URL | +| `--username` | โœ… | Your username | +| `--password` | โœ… | Your password | +| `--channel` | โŒ | Channel to search (one-time mode) | +| `--all-types` | โŒ | Search Live + VOD + Series | +| `--interactive` | โŒ | Force interactive mode | +| `--max-workers` | โŒ | Number of threads (default: 10) | +| `--debug` | โŒ | Show API debug information | + +## Performance Tuning + +- **Default threads (10)**: Good balance for most servers +- **High performance (15-20 threads)**: Faster for powerful servers +- **Conservative (5 threads)**: Better for slower connections or servers with rate limits + +## Requirements + +- Python 3.6+ +- requests >= 2.25.0 +- tqdm >= 4.60.0 + +## Troubleshooting + +### Connection Issues +```bash +# Test API connectivity +python xtream-search.py --api-url "http://your-server.com" --username "user" --password "pass" --debug +``` + +### Slow Performance +```bash +# Reduce thread count if getting timeouts +python xtream-search.py --api-url "http://your-server.com" --username "user" --password "pass" --max-workers 5 +``` + +### No Results Found +- Check your search term spelling +- Try partial matches (e.g., "cnn" instead of "CNN International") +- Use `all:term` to search across all content types +- Verify your credentials with `--debug` + +## License + +MIT License - feel free to modify and distribute. + +## Contributing + +Pull requests welcome! Please ensure code follows the existing style and includes appropriate error handling. \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b20928e --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +requests>=2.25.0 +tqdm>=4.60.0 \ No newline at end of file diff --git a/xtream-search.py b/xtream-search.py new file mode 100644 index 0000000..e749099 --- /dev/null +++ b/xtream-search.py @@ -0,0 +1,451 @@ +import requests +import json +import sys +from urllib.parse import urljoin +import argparse +from concurrent.futures import ThreadPoolExecutor, as_completed +import threading +try: + from tqdm import tqdm +except ImportError: + print("Installing tqdm for progress bars...") + import subprocess + subprocess.check_call([sys.executable, "-m", "pip", "install", "tqdm"]) + from tqdm import tqdm + +def search_category(api_url, username, password, channel_name, category, pbar, pbar_lock): + """ + Search a single category for channels (for multithreading) + """ + category_id = category.get('category_id') + category_name = category.get('category_name', 'Unknown') + + # Create a new session for this thread + session = requests.Session() + categories_url = urljoin(api_url.rstrip('/'), '/player_api.php') + + matches_in_category = [] + + try: + # Get streams for this category + streams_params = { + 'username': username, + 'password': password, + 'action': 'get_live_streams', + 'category_id': category_id + } + + streams_response = session.get(categories_url, params=streams_params) + streams_response.raise_for_status() + streams = streams_response.json() + + # Check if channel exists in this category (fuzzy matching) + for stream in streams: + stream_name = stream.get('name', '').lower() + if channel_name.lower() in stream_name: + matches_in_category.append({ + 'group_name': category_name, + 'group_id': category_id, + 'channel_name': stream.get('name'), + 'channel_id': stream.get('stream_id'), + 'stream_type': stream.get('stream_type'), + 'stream_icon': stream.get('stream_icon'), + 'epg_channel_id': stream.get('epg_channel_id') + }) + + if matches_in_category: + with pbar_lock: + pbar.write(f" โœ… Found {len(matches_in_category)} match(es) in {category_name}") + + except requests.exceptions.RequestException as e: + with pbar_lock: + pbar.write(f" โŒ Error in {category_name}: {str(e)[:50]}...") + + finally: + with pbar_lock: + pbar.update(1) + + return matches_in_category + +def find_channel_in_groups(api_url, username, password, channel_name, max_workers=10): + """ + Find which groups contain a specific channel using Xtream Codes API (multithreaded) + + Args: + api_url (str): Base URL for the Xtream Codes API + username (str): Username for authentication + password (str): Password for authentication + channel_name (str): Channel name to search for (supports fuzzy matching) + max_workers (int): Maximum number of threads to use + + Returns: + list: List of categories containing the channel + """ + session = requests.Session() + + try: + # Get all live categories + print("๐Ÿ“ก Fetching live categories...") + categories_url = urljoin(api_url.rstrip('/'), '/player_api.php') + categories_params = { + 'username': username, + 'password': password, + 'action': 'get_live_categories' + } + + categories_response = session.get(categories_url, params=categories_params) + categories_response.raise_for_status() + categories = categories_response.json() + + print(f"โœ… Found {len(categories)} categories to search") + + matching_groups = [] + pbar_lock = threading.Lock() + + # Search through categories with multithreading + with tqdm(total=len(categories), desc="๐Ÿ” Searching live categories", unit="cat", + bar_format="{desc}: {percentage:3.0f}%|{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]") as pbar: + with ThreadPoolExecutor(max_workers=max_workers) as executor: + # Submit all tasks + future_to_category = { + executor.submit(search_category, api_url, username, password, + channel_name, category, pbar, pbar_lock): category + for category in categories + } + + # Collect results as they complete + for future in as_completed(future_to_category): + try: + category_matches = future.result() + matching_groups.extend(category_matches) + except Exception as e: + category = future_to_category[future] + category_name = category.get('category_name', 'Unknown') + with pbar_lock: + pbar.write(f" โŒ Exception in {category_name}: {str(e)[:50]}...") + + return matching_groups + + except requests.exceptions.RequestException as e: + print(f"โŒ Error connecting to Xtream Codes API: {e}") + return [] + +def get_api_info(api_url, username, password): + """Get basic API information for debugging""" + session = requests.Session() + + try: + # Test basic connectivity and get categories + print("๐Ÿ”ง Testing API connectivity...") + api_endpoint = urljoin(api_url.rstrip('/'), '/player_api.php') + + # Test live categories + categories_params = { + 'username': username, + 'password': password, + 'action': 'get_live_categories' + } + + categories_response = session.get(api_endpoint, params=categories_params) + print(f"Live categories endpoint status: {categories_response.status_code}") + + if categories_response.status_code == 200: + categories = categories_response.json() + print(f"Found {len(categories)} live categories") + if categories: + print("Sample category structure:") + print(json.dumps(categories[0], indent=2)) + + # Test getting streams from first category + if categories_response.status_code == 200 and categories: + print("๐Ÿ”ง Testing streams endpoint...") + first_category_id = categories[0].get('category_id') + streams_params = { + 'username': username, + 'password': password, + 'action': 'get_live_streams', + 'category_id': first_category_id + } + + streams_response = session.get(api_endpoint, params=streams_params) + print(f"Streams endpoint status: {streams_response.status_code}") + + if streams_response.status_code == 200: + streams = streams_response.json() + print(f"Found {len(streams)} streams in first category") + if streams: + print("Sample stream structure:") + print(json.dumps(streams[0], indent=2)) + + except Exception as e: + print(f"โŒ Error testing API: {e}") + +def search_content_type_category(api_url, username, password, search_term, category, content_type, stream_action, pbar, pbar_lock): + """Search a single category within a content type (for multithreading)""" + category_id = category.get('category_id') + category_name = category.get('category_name', 'Unknown') + + session = requests.Session() + api_endpoint = urljoin(api_url.rstrip('/'), '/player_api.php') + + matches_in_category = [] + + try: + # Get streams for this category + stream_params = { + 'username': username, + 'password': password, + 'action': stream_action, + 'category_id': category_id + } + + stream_response = session.get(api_endpoint, params=stream_params) + if stream_response.status_code != 200: + with pbar_lock: + pbar.write(f" โŒ Failed to get {content_type.lower()} from {category_name}") + return matches_in_category + + streams = stream_response.json() + + for stream in streams: + stream_name = stream.get('name', '').lower() + if search_term.lower() in stream_name: + matches_in_category.append({ + 'content_type': content_type, + 'group_name': category_name, + 'group_id': category_id, + 'channel_name': stream.get('name'), + 'channel_id': stream.get('stream_id') or stream.get('series_id'), + 'stream_type': stream.get('stream_type'), + 'stream_icon': stream.get('stream_icon'), + 'epg_channel_id': stream.get('epg_channel_id') + }) + + if matches_in_category: + with pbar_lock: + pbar.write(f" โœ… Found {len(matches_in_category)} match(es) in {category_name}") + + except requests.exceptions.RequestException as e: + with pbar_lock: + pbar.write(f" โŒ Error in {category_name}: {str(e)[:50]}...") + + finally: + with pbar_lock: + pbar.update(1) + + return matches_in_category + +def search_all_content_types(api_url, username, password, search_term, max_workers=15): + """Search across live streams, VOD, and series (multithreaded)""" + session = requests.Session() + api_endpoint = urljoin(api_url.rstrip('/'), '/player_api.php') + all_matches = [] + + content_types = [ + ('Live Streams', 'get_live_categories', 'get_live_streams'), + ('VOD', 'get_vod_categories', 'get_vod_streams'), + ('Series', 'get_series_categories', 'get_series') + ] + + for content_type, cat_action, stream_action in content_types: + try: + print(f"\n๏ฟฝ Fetching {content_type} categories...") + + # Get categories + cat_params = { + 'username': username, + 'password': password, + 'action': cat_action + } + + cat_response = session.get(api_endpoint, params=cat_params) + if cat_response.status_code != 200: + print(f"โŒ Could not get {content_type} categories (status: {cat_response.status_code})") + continue + + categories = cat_response.json() + print(f"โœ… Found {len(categories)} {content_type.lower()} categories") + + if not categories: + continue + + pbar_lock = threading.Lock() + + # Use multithreading for this content type + with tqdm(total=len(categories), desc=f"๐Ÿ” Searching {content_type.lower()}", unit="cat", + bar_format="{desc}: {percentage:3.0f}%|{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]") as pbar: + with ThreadPoolExecutor(max_workers=max_workers) as executor: + # Submit all tasks for this content type + future_to_category = { + executor.submit(search_content_type_category, api_url, username, password, + search_term, category, content_type, stream_action, pbar, pbar_lock): category + for category in categories + } + + # Collect results as they complete + for future in as_completed(future_to_category): + try: + category_matches = future.result() + all_matches.extend(category_matches) + except Exception as e: + category = future_to_category[future] + category_name = category.get('category_name', 'Unknown') + with pbar_lock: + pbar.write(f" โŒ Exception in {category_name}: {str(e)[:50]}...") + + except Exception as e: + print(f"โŒ Error searching {content_type}: {e}") + continue + + return all_matches + +def interactive_search(api_url, username, password, max_workers=10): + """Interactive search mode""" + print(f"Connected to: {api_url}") + print("Commands: live: | vod: | series: | all: | debug | quit") + print("-" * 50) + + while True: + try: + query = input("\n๐Ÿ” Search: ").strip() + + if not query: + continue + + if query.lower() in ['quit', 'exit', 'q']: + print("๐Ÿ‘‹ Goodbye!") + break + + if query.lower() == 'debug': + print("\n=== API Debug Info ===") + get_api_info(api_url, username, password) + print("=====================") + continue + + # Parse search type and term + search_all_types = False + search_term = query + + if ':' in query: + search_type, search_term = query.split(':', 1) + search_type = search_type.lower().strip() + search_term = search_term.strip() + + if search_type == 'all': + search_all_types = True + elif search_type in ['vod', 'series']: + # For now, we'll treat these as all-types searches + # You could implement specific VOD/series-only searches later + search_all_types = True + print(f"๐Ÿ” Searching {search_type.upper()} for '{search_term}'...") + elif search_type == 'live': + print(f"๐Ÿ” Searching Live Streams for '{search_term}'...") + else: + print(f"โ“ Unknown search type '{search_type}', searching live streams for '{query}'...") + search_term = query + else: + print(f"๐Ÿ” Searching Live Streams for '{search_term}'...") + + # Perform search + if search_all_types: + matching_groups = search_all_content_types( + api_url, username, password, search_term, max_workers + ) + else: + matching_groups = find_channel_in_groups( + api_url, username, password, search_term, max_workers + ) + + # Display results + if matching_groups: + print(f"\nโœ… Found {len(matching_groups)} result(s) for '{search_term}':") + for i, match in enumerate(matching_groups, 1): + content_type = match.get('content_type', 'Live Stream') + type_short = content_type.replace(' Streams', '').replace(' ', '') + print(f"{i:2d}. [{type_short}] {match['group_name']} โ†’ {match['channel_name']}") + else: + print(f"โŒ No results found for '{search_term}'") + + except KeyboardInterrupt: + print("\n\n๐Ÿ‘‹ Goodbye!") + break + except Exception as e: + print(f"โŒ Error during search: {e}") + +def main(): + parser = argparse.ArgumentParser(description='Xtream Codes Channel Search Tool') + parser.add_argument('--api-url', required=True, help='Xtream Codes API base URL') + parser.add_argument('--username', required=True, help='Username for authentication') + parser.add_argument('--password', required=True, help='Password for authentication') + + # Optional one-time search arguments + parser.add_argument('--channel', help='Channel name to search for (one-time search)') + parser.add_argument('--debug', action='store_true', help='Show API debugging info') + parser.add_argument('--all-types', action='store_true', help='Search in Live, VOD, and Series (one-time search)') + parser.add_argument('--interactive', '-i', action='store_true', help='Start interactive search mode') + parser.add_argument('--max-workers', type=int, default=10, help='Maximum number of concurrent threads (default: 10)') + + args = parser.parse_args() + + # Test connection first + print("๐Ÿ”ง Testing connection to Xtream Codes API...") + try: + session = requests.Session() + test_url = urljoin(args.api_url.rstrip('/'), '/player_api.php') + test_params = { + 'username': args.username, + 'password': args.password, + 'action': 'get_live_categories' + } + response = session.get(test_url, params=test_params, timeout=10) + if response.status_code == 200: + print("โœ… Connection successful!") + else: + print(f"โŒ Connection failed with status {response.status_code}") + return + except Exception as e: + print(f"โŒ Connection failed: {e}") + return + + if args.debug: + print("\n=== API Debug Info ===") + get_api_info(args.api_url, args.username, args.password) + print("=====================\n") + + # If channel is provided, do one-time search + if args.channel: + print(f"๐Ÿ” Searching for '{args.channel}' in Xtream Codes API...") + + if args.all_types: + matching_groups = search_all_content_types( + args.api_url, + args.username, + args.password, + args.channel, + args.max_workers + ) + else: + matching_groups = find_channel_in_groups( + args.api_url, + args.username, + args.password, + args.channel, + args.max_workers + ) + + print(f"\n{'='*50}") + if matching_groups: + print(f"โœ… Found {len(matching_groups)} result(s) for '{args.channel}':") + for i, match in enumerate(matching_groups, 1): + content_type = match.get('content_type', 'Live Stream') + type_short = content_type.replace(' Streams', '').replace(' ', '') + print(f"{i:2d}. [{type_short}] {match['group_name']} โ†’ {match['channel_name']}") + else: + print(f"โŒ No groups found containing '{args.channel}'") + + # Start interactive mode if requested or no channel provided + if args.interactive or not args.channel: + interactive_search(args.api_url, args.username, args.password, args.max_workers) + +if __name__ == "__main__": + main() \ No newline at end of file