Created
January 10, 2024 19:44
-
-
Save T-One/c857005e58286149914ad38f24a891e1 to your computer and use it in GitHub Desktop.
This file contains 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 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)').rstrip('/') | |
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') | |
args = parser.parse_args() | |
# Manipulate the --immich_server argument to remove trailing slashes | |
if args.immich_server: | |
args.immich_server = args.immich_server.rstrip('/') | |
return 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' | |
] | |
# 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 person_assets and not args.no_prompt: | |
print('Found Person assets, run job RECOGNIZE FACES All after this') | |
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Thanks!