import requests import json import os import shutil import pandas as pd from pathlib import Path import time import webbrowser import urllib.parse from http.server import HTTPServer, BaseHTTPRequestHandler import threading CONFIG = { 'CLIENT_ID': 'x', 'CLIENT_SECRET': 'x', 'BROADCASTER_ID': 'x', 'PHOTOS_DIR': 'x', 'SUBS_FOLDER': 'SUBS', 'OLD_SUBS_FOLDER': 'OLD_SUBS', 'EXCEL_FILENAME': 'subscribers_list.xlsx', 'REDIRECT_URI': 'http://localhost:8080', 'SCOPES': ['channel:read:subscriptions'] } class OAuthHandler(BaseHTTPRequestHandler): """HTTP server to handle OAuth redirect""" def do_GET(self): parsed_url = urllib.parse.urlparse(self.path) query_params = urllib.parse.parse_qs(parsed_url.query) if 'code' in query_params: self.server.auth_code = query_params['code'][0] self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() success_page = """ Authorization Successful

āœ… Authorization Successful!

You can now close this browser window and return to your application.

The subscriber photo organizer will continue automatically.

""" self.wfile.write(success_page.encode()) else: error = query_params.get('error', ['Unknown error'])[0] error_desc = query_params.get('error_description', [''])[0] self.send_response(400) self.send_header('Content-type', 'text/html') self.end_headers() error_page = f""" Authorization Failed

āŒ Authorization Failed

Error: {error}

Description: {error_desc}

Please try again or check your application settings.

""" self.wfile.write(error_page.encode()) def log_message(self, format, *args): pass def get_user_access_token(client_id, client_secret, redirect_uri, scopes): """ Get User Access Token using Authorization Code Flow This method requires browser interaction (one-time setup) """ print("šŸ” Getting User Access Token...") print("āš ļø This requires browser interaction (one-time setup)") scope_string = ' '.join(scopes) state = 'twitch_subscriber_organizer' auth_url = ( f"https://id.twitch.tv/oauth2/authorize" f"?response_type=code" f"&client_id={client_id}" f"&redirect_uri={urllib.parse.quote(redirect_uri)}" f"&scope={urllib.parse.quote(scope_string)}" f"&state={state}" ) print(f"šŸŒ Opening browser for Twitch authorization...") print(f"šŸ“‹ Required scopes: {', '.join(scopes)}") print(f"šŸ”— Authorization URL: {auth_url}") server = HTTPServer(('localhost', 8080), OAuthHandler) server.auth_code = None server_thread = threading.Thread(target=server.serve_forever) server_thread.daemon = True server_thread.start() try: webbrowser.open(auth_url) except Exception as e: print(f"āŒ Could not open browser automatically: {e}") print(f"šŸ“‹ Please manually open this URL in your browser:") print(f" {auth_url}") print("ā³ Waiting for authorization... (complete the process in your browser)") timeout = 300 start_time = time.time() while server.auth_code is None and (time.time() - start_time) < timeout: time.sleep(1) server.shutdown() if server.auth_code is None: raise Exception("āŒ Authorization timed out or failed") print("āœ… Authorization code received!") token_url = "https://id.twitch.tv/oauth2/token" token_data = { 'client_id': client_id, 'client_secret': client_secret, 'code': server.auth_code, 'grant_type': 'authorization_code', 'redirect_uri': redirect_uri } print("šŸ”„ Exchanging authorization code for access token...") response = requests.post(token_url, data=token_data) if response.status_code == 200: token_info = response.json() access_token = token_info['access_token'] refresh_token = token_info.get('refresh_token') expires_in = token_info.get('expires_in') print("āœ… User access token obtained successfully!") print(f"ā° Token expires in: {expires_in} seconds ({expires_in/3600:.1f} hours)") save_tokens(access_token, refresh_token, expires_in) return access_token, refresh_token else: raise Exception(f"āŒ Failed to get access token: {response.text}") def save_tokens(access_token, refresh_token, expires_in): """Save tokens to file for future use""" token_data = { 'access_token': access_token, 'refresh_token': refresh_token, 'expires_at': time.time() + expires_in - 300 } with open('twitch_tokens.json', 'w') as f: json.dump(token_data, f) print("šŸ’¾ Tokens saved to twitch_tokens.json") def load_tokens(): """Load tokens from file""" try: with open('twitch_tokens.json', 'r') as f: return json.load(f) except FileNotFoundError: return None def refresh_access_token(client_id, client_secret, refresh_token): """Refresh an expired access token""" print("šŸ”„ Refreshing access token...") token_url = "https://id.twitch.tv/oauth2/token" token_data = { 'client_id': client_id, 'client_secret': client_secret, 'refresh_token': refresh_token, 'grant_type': 'refresh_token' } response = requests.post(token_url, data=token_data) if response.status_code == 200: token_info = response.json() access_token = token_info['access_token'] new_refresh_token = token_info.get('refresh_token', refresh_token) expires_in = token_info.get('expires_in') print("āœ… Access token refreshed successfully!") save_tokens(access_token, new_refresh_token, expires_in) return access_token, new_refresh_token else: raise Exception(f"āŒ Failed to refresh token: {response.text}") def get_valid_access_token(client_id, client_secret, redirect_uri, scopes): """Get a valid access token (load existing or get new one)""" tokens = load_tokens() if tokens: if time.time() < tokens['expires_at']: print("āœ… Using existing valid access token") return tokens['access_token'], tokens['refresh_token'] else: try: return refresh_access_token(client_id, client_secret, tokens['refresh_token']) except Exception as e: print(f"āŒ Token refresh failed: {e}") print("šŸ”„ Getting new access token...") return get_user_access_token(client_id, client_secret, redirect_uri, scopes) def get_all_subscribers(client_id, access_token, broadcaster_id): """Fetch all subscribers from Twitch API using User Access Token""" print("šŸ“‹ Fetching subscribers from Twitch...") url = "https://api.twitch.tv/helix/subscriptions" headers = { 'Authorization': f'Bearer {access_token}', 'Client-Id': client_id } all_subs = [] cursor = None while True: params = { 'broadcaster_id': broadcaster_id, 'first': 100 } if cursor: params['after'] = cursor response = requests.get(url, headers=headers, params=params) if response.status_code == 401: print("āŒ Unauthorized - Token may be invalid or expired") raise Exception("Token expired or invalid. Please re-run the script to get a new token.") elif response.status_code != 200: print(f"āŒ API Error: {response.status_code} - {response.text}") break data = response.json() subs = data.get('data', []) all_subs.extend(subs) print(f"šŸ“„ Fetched {len(subs)} subscribers (Total: {len(all_subs)})") pagination = data.get('pagination', {}) cursor = pagination.get('cursor') if not cursor: break time.sleep(0.1) print(f"āœ… Total subscribers found: {len(all_subs)}") return all_subs def create_excel_report(subscribers, filename): """Create Excel file with subscriber data""" print(f"šŸ“Š Creating Excel report: {filename}") data = [] for sub in subscribers: data.append({ 'Username': sub.get('user_login', ''), 'Display Name': sub.get('user_name', ''), 'User ID': sub.get('user_id', ''), 'Tier': sub.get('tier', ''), 'Is Gift': 'Yes' if sub.get('is_gift') else 'No', 'Gifter': sub.get('gifter_name', '') if sub.get('is_gift') else 'N/A' }) df = pd.DataFrame(data) df.to_excel(filename, index=False) print(f"āœ… Excel report saved: {filename}") return df def organize_photos(subscribers, photos_dir, subs_folder, old_subs_folder): """Organize photos based on subscription status""" print(f"šŸ“‚ Organizing photos in: {photos_dir}") subscriber_usernames = {sub['user_login'].lower() for sub in subscribers} photos_path = Path(photos_dir) subs_path = photos_path / subs_folder old_subs_path = photos_path / old_subs_folder subs_path.mkdir(exist_ok=True) old_subs_path.mkdir(exist_ok=True) image_exts = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp', '.tiff'} moved_to_subs = 0 moved_to_old = 0 errors = 0 for file_path in photos_path.iterdir(): if file_path.is_file() and file_path.suffix.lower() in image_exts: try: username = file_path.stem.lower() if username in subscriber_usernames: destination = subs_path / file_path.name moved_to_subs += 1 print(f"šŸ“ø ā†’ SUBS: {file_path.name}") else: destination = old_subs_path / file_path.name moved_to_old += 1 print(f"šŸ“ø ā†’ OLD_SUBS: {file_path.name}") shutil.move(str(file_path), str(destination)) except Exception as e: print(f"āŒ Error with {file_path.name}: {e}") errors += 1 print("\n" + "="*50) print("šŸ“Š ORGANIZATION SUMMARY") print("="*50) print(f"āœ… Moved to SUBS: {moved_to_subs}") print(f"šŸ“ Moved to OLD_SUBS: {moved_to_old}") print(f"āš ļø Errors: {errors}") print(f"šŸ“ Total processed: {moved_to_subs + moved_to_old + errors}") return { 'subs': moved_to_subs, 'old_subs': moved_to_old, 'errors': errors } def main(): """Main execution function""" print("šŸŽ® Twitch Subscriber Photo Organizer (User Access Token Version)") print("="*60) if CONFIG['CLIENT_ID'] == 'your_client_id_here': print("āŒ Please update the CONFIG dictionary with your Twitch app details!") print("šŸ“– See the setup instructions for detailed guidance") return if not os.path.exists(CONFIG['PHOTOS_DIR']): print(f"āŒ Photos directory not found: {CONFIG['PHOTOS_DIR']}") print("šŸ“ Please create the directory and add your subscriber photos") return try: access_token, refresh_token = get_valid_access_token( CONFIG['CLIENT_ID'], CONFIG['CLIENT_SECRET'], CONFIG['REDIRECT_URI'], CONFIG['SCOPES'] ) subscribers = get_all_subscribers( CONFIG['CLIENT_ID'], access_token, CONFIG['BROADCASTER_ID'] ) if not subscribers: print("āš ļø No subscribers found!") return create_excel_report(subscribers, CONFIG['EXCEL_FILENAME']) organize_photos( subscribers, CONFIG['PHOTOS_DIR'], CONFIG['SUBS_FOLDER'], CONFIG['OLD_SUBS_FOLDER'] ) print("\nšŸŽ‰ All done! Your subscriber photos have been organized!") print("šŸ’¾ Your access tokens have been saved for future runs") except Exception as e: print(f"āŒ Error: {e}") print("šŸ’” Check your configuration and try again") if __name__ == "__main__": main()