Created
December 7, 2025 15:27
-
-
Save bertwagner/74996188a65733ecb4cdc42ae9c66f59 to your computer and use it in GitHub Desktop.
Python script to check metadata for Rating=5 and mark those as a "Favorite" in immesh
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| import requests | |
| import os | |
| import time | |
| from concurrent.futures import ThreadPoolExecutor, as_completed | |
| from requests.adapters import HTTPAdapter | |
| from urllib3.util.retry import Retry | |
| # Read .env file manually | |
| def load_env(): | |
| env_path = os.path.join(os.path.dirname(__file__), '.env') | |
| if os.path.exists(env_path): | |
| with open(env_path, 'r') as f: | |
| for line in f: | |
| line = line.strip() | |
| if line and not line.startswith('#') and '=' in line: | |
| key, value = line.split('=', 1) | |
| value = value.strip().strip('"').strip("'") | |
| os.environ[key.strip()] = value | |
| load_env() | |
| # Configuration | |
| IMMICH_URL = os.getenv('API_URL') | |
| API_KEY = os.getenv('API_KEY') | |
| if not API_KEY: | |
| print("ERROR: API_KEY not found in .env file!") | |
| exit(1) | |
| # Create a session with connection pooling and retries | |
| def create_session(): | |
| session = requests.Session() | |
| retry = Retry( | |
| total=3, | |
| backoff_factor=1, | |
| status_forcelist=[429, 500, 502, 503, 504] | |
| ) | |
| adapter = HTTPAdapter( | |
| max_retries=retry, | |
| pool_connections=5, | |
| pool_maxsize=5 | |
| ) | |
| session.mount('http://', adapter) | |
| session.mount('https://', adapter) | |
| session.headers.update({ | |
| 'x-api-key': API_KEY, | |
| 'Content-Type': 'application/json', | |
| 'Accept': 'application/json' | |
| }) | |
| return session | |
| def get_all_asset_ids(): | |
| """Get all asset IDs using search/metadata""" | |
| session = create_session() | |
| all_ids = [] | |
| page = 1 | |
| size = 1000 | |
| while True: | |
| print(f"Fetching asset IDs page {page}...", flush=True) | |
| response = session.post( | |
| f"{IMMICH_URL}/search/metadata", | |
| json={"page": page, "size": size, "withArchived": True} | |
| ) | |
| response.raise_for_status() | |
| assets = response.json().get('assets', {}).get('items', []) | |
| if not assets: | |
| break | |
| all_ids.extend([asset['id'] for asset in assets]) | |
| print(f" Got {len(assets)} asset IDs (total: {len(all_ids)})", flush=True) | |
| if len(assets) < size: | |
| break | |
| page += 1 | |
| session.close() | |
| return all_ids | |
| def get_asset_details(asset_id, session): | |
| """Get full asset details including rating""" | |
| try: | |
| response = session.get(f"{IMMICH_URL}/assets/{asset_id}") | |
| response.raise_for_status() | |
| return response.json() | |
| except Exception as e: | |
| return None | |
| def update_favorite_single(asset_id): | |
| """Mark a single asset as favorite using PUT endpoint""" | |
| session = create_session() | |
| try: | |
| response = session.put( | |
| f"{IMMICH_URL}/assets/{asset_id}", | |
| json={'isFavorite': True} | |
| ) | |
| # 200 or 204 = success | |
| if response.status_code in [200, 204]: | |
| session.close() | |
| return True | |
| else: | |
| response.raise_for_status() | |
| session.close() | |
| return False | |
| except Exception as e: | |
| session.close() | |
| return False | |
| def main(): | |
| print("Step 1: Fetching all asset IDs...") | |
| asset_ids = get_all_asset_ids() | |
| print(f"\nFound {len(asset_ids)} total assets\n") | |
| print("Step 2: Checking ratings (this will take a while)...") | |
| five_star_ids = [] | |
| five_star_names = [] | |
| # Reduced workers from 10 to 3 to avoid port exhaustion | |
| num_workers = 3 | |
| print(f"Using {num_workers} parallel workers\n") | |
| # Create sessions for workers | |
| sessions = [create_session() for _ in range(num_workers)] | |
| session_index = 0 | |
| with ThreadPoolExecutor(max_workers=num_workers) as executor: | |
| # Submit jobs with session rotation | |
| futures = {} | |
| for aid in asset_ids: | |
| session = sessions[session_index % num_workers] | |
| futures[executor.submit(get_asset_details, aid, session)] = aid | |
| session_index += 1 | |
| # Process results | |
| for i, future in enumerate(as_completed(futures)): | |
| if i % 100 == 0: | |
| print(f" Checked {i}/{len(asset_ids)} assets...", flush=True) | |
| asset = future.result() | |
| if asset: | |
| rating = asset.get('exifInfo', {}).get('rating') | |
| if rating == 5 and not asset.get('isFavorite', False): | |
| filename = asset.get('originalFileName', asset['id']) | |
| five_star_ids.append(asset['id']) | |
| five_star_names.append(filename) | |
| # Close all sessions | |
| for session in sessions: | |
| session.close() | |
| print(f"\nStep 3: Found {len(five_star_ids)} photos with 5-star ratings") | |
| if five_star_ids: | |
| print(f"\nSample of 5-star photos:") | |
| for name in five_star_names[:10]: | |
| print(f" • {name}") | |
| if len(five_star_names) > 10: | |
| print(f" ... and {len(five_star_names) - 10} more") | |
| print(f"\nMarking {len(five_star_ids)} photos as favorites (one at a time)...") | |
| print("This will take a few minutes...\n") | |
| success_count = 0 | |
| for i, asset_id in enumerate(five_star_ids): | |
| if update_favorite_single(asset_id): | |
| success_count += 1 | |
| # Progress update every 50 photos | |
| if (i + 1) % 50 == 0: | |
| print(f" Progress: {i + 1}/{len(five_star_ids)} ({success_count} successful)") | |
| time.sleep(0.02) # Small delay to avoid overwhelming the server | |
| print(f"\n{'='*60}") | |
| print(f"✓ Done! Marked {success_count}/{len(five_star_ids)} photos as favorites.") | |
| print(f"{'='*60}") | |
| else: | |
| print("\nNo 5-star rated photos found that aren't already favorites.") | |
| if __name__ == "__main__": | |
| try: | |
| main() | |
| except KeyboardInterrupt: | |
| print("\n\nScript interrupted by user.") | |
| exit(0) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment