immich_delete_orphans.py

#!/usr/bin/env python3

import json
import requests
import argparse
from datetime import datetime

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

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 (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()

def main():
    args = parse_arguments()

    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()

    base_url = f'{immich_server}/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()

    # Check if there are orphaned person assets and print a message
    person_assets = [
        {'pathValue': person['pathValue'], 'entityId': person['entityId'], 'entityType': person['entityType']}
        for person in response.json().get('orphans', []) if person.get('entityType') == 'person'
    ]

    if person_assets and not args.no_prompt:
        print('Found Person assets, run job RECOGNIZE FACES All after this script')

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

    if num_entries == 0 and args.no_prompt:
        return

    if num_entries == 0 and not args.no_prompt:
        print('Nothing to delete, stopping.')
        return

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

    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 in a loop with a separate REST call for each entityId
    for data in orphans_data:
        entity_id = data['entityId']
        asset_url = f'{base_url}/asset'
        delete_payload = json.dumps({'force': True, 'ids': [entity_id]})
        headers = {'Content-Type': 'application/json', 'x-api-key': api_key}
        response = requests.delete(asset_url, headers=headers, data=delete_payload)
        response.raise_for_status()
        
        if not args.no_prompt:
            print(f'Deleting asset with Entity ID {entity_id}')

        # 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')
            log_data = {'deletiontime': deletiontime, 'entityId': entity_id, 'pathValue': data['pathValue']}
            with open(args.logfile, 'a') as log_file:
                log_file.write(json.dumps(log_data, indent=2) + '\n')

if __name__ == '__main__':
    main()