delete orphans on immich server via API

gistfile1.txt

#!/usr/bin/env python3

import json
import requests
import argparse
from datetime import datetime

# Function to prompt the user for API key and Immich server address
def prompt_for_credentials():
    api_key = input('Enter API key: ')
    immich_server = input('Enter full address (including http and port, e.g. http://192.168.0.1:2283) ')
    return api_key, immich_server

# Function to parse command-line arguments
def parse_arguments():
    parser = argparse.ArgumentParser(description='Fetch file report and delete orphaned assets.')
    parser.add_argument('--api_key', help='API key for authentication')
    parser.add_argument('--immich_server', help='Full address including port (e.g. http://192.168.0.1:2283)')
    parser.add_argument('--no_prompt', action='store_true', help='Enable to delete orphaned assets without user confirmation')
    parser.add_argument('--logfile', help='Filename to append deleted objects after successful deletion as json')
    return parser.parse_args()

# Main function
def main():
    # Parse command-line arguments
    args = parse_arguments()

    # If API key and Immich server are not provided as arguments, prompt the user
    if args.api_key and args.immich_server:
        api_key, immich_server = args.api_key, args.immich_server
    else:
        api_key, immich_server = prompt_for_credentials()

    # Construct base URL for Immich API
    base_url = f'{immich_server}/api'

    # Fetching file report from Immich API
    file_report_url = base_url + '/audit/file-report'
    headers = {'x-api-key': api_key}
    response = requests.get(file_report_url, headers=headers)
    response.raise_for_status()

    # Extracting "pathValue" and "entityId" fields from the response
    orphans_data = [{'pathValue': orphan['pathValue'], 'entityId': orphan['entityId']} for orphan in response.json().get('orphans', [])]

    # Counting the number of entries in orphans_data
    num_entries = len(orphans_data)

    # If no orphaned assets to delete and --no_prompt is provided, stop the script
    if num_entries == 0 and args.no_prompt:
        return

    # If --no_prompt is not provided and no orphans to delete, print a message and stop the script
    if num_entries == 0 and not args.no_prompt:
        print('Nothing to delete, stopping.')
        return

    # If --no_prompt is not provided, printing orphans_data
    if not args.no_prompt:
        for data in orphans_data:
            print('Path Value:', data['pathValue'])
            print('Entity ID:', data['entityId'])
            print('---')

    # If --no_prompt is not provided, ask the user if they want to continue
    if not args.no_prompt and num_entries > 0:
        summary = f'There are {num_entries} entries of orphaned data. Do you want to continue and delete the orphaned assets? (yes/no): '
        user_input = input(summary).lower()
        
        if user_input not in ('y', 'yes'):
            print('Script execution aborted.')
            return

    # Deleting orphaned assets
    entity_ids = [data['entityId'] for data in orphans_data]
    asset_url = base_url + '/asset'
    delete_payload = json.dumps({'force': True, 'ids': entity_ids})
    headers = {'Content-Type': 'application/json', 'x-api-key': api_key}
    response = requests.delete(asset_url, headers=headers, data=delete_payload)
    response.raise_for_status()

    print(response.text)

    # Append timestamp, entityId, and pathValue in JSON format to the logfile after successful deletion
    if args.logfile:
        deletiontime = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        for data in orphans_data:
            log_data = {'deletiontime': deletiontime, 'entityId': data['entityId'], 'pathValue': data['pathValue']}
            with open(args.logfile, 'a') as log_file:
                log_file.write(json.dumps(log_data, indent=2) + '\n')

# Entry point of the script
if __name__ == '__main__':
    main()