201 lines
8.4 KiB
Python
201 lines
8.4 KiB
Python
import requests
|
|
import os
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
import datetime
|
|
from collections import defaultdict
|
|
import urllib3
|
|
|
|
|
|
parser = argparse.ArgumentParser(description="Create Immich Albums from an external library path based on the top level folders", formatter_class=argparse.ArgumentDefaultsHelpFormatter)
|
|
parser.add_argument("root_path", action='append', help="The external libarary's root path in Immich")
|
|
parser.add_argument("api_url", help="The root API URL of immich, e.g. https://immich.mydomain.com/api/")
|
|
parser.add_argument("api_key", help="The Immich API Key to use")
|
|
parser.add_argument("-r", "--root-path", action="append", help="Additional external libarary root path in Immich; May be specified multiple times for multiple import paths or external libraries.")
|
|
parser.add_argument("-u", "--unattended", action="store_true", help="Do not ask for user confirmation after identifying albums. Set this flag to run script as a cronjob.")
|
|
parser.add_argument("-a", "--album-levels", default=1, type=int, help="Number of sub-folders below the root path used for album name creation. Positive numbers start from top of the folder structure, negative numbers from the bottom. Cannot be 0.")
|
|
parser.add_argument("-s", "--album-separator", default=" ", type=str, help="Separator string to use for compound album names created from nested folders. Only effective if -a is set to a value > 1")
|
|
parser.add_argument("-c", "--chunk-size", default=2000, type=int, help="Maximum number of assets to add to an album with a single API call")
|
|
parser.add_argument("-C", "--fetch-chunk-size", default=5000, type=int, help="Maximum number of assets to fetch with a single API call")
|
|
parser.add_argument("-l", "--log-level", default="INFO", choices=['CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG'], help="Log level to use")
|
|
parser.add_argument("-i", "--ignore-ssl", action="store_true", help="Set to true to ignore SSL verification")
|
|
args = vars(parser.parse_args())
|
|
# set up logger to log in logfmt format
|
|
logging.basicConfig(level=args["log_level"], stream=sys.stdout, format='time=%(asctime)s level=%(levelname)s msg=%(message)s')
|
|
logging.Formatter.formatTime = (lambda self, record, datefmt=None: datetime.datetime.fromtimestamp(record.created, datetime.timezone.utc).astimezone().isoformat(sep="T",timespec="milliseconds"))
|
|
|
|
root_paths = args["root_path"]
|
|
root_url = args["api_url"]
|
|
api_key = args["api_key"]
|
|
number_of_images_per_request = args["chunk_size"]
|
|
number_of_assets_to_fetch_per_request = args["fetch_chunk_size"]
|
|
unattended = args["unattended"]
|
|
album_levels = args["album_levels"]
|
|
album_level_separator = args["album_separator"]
|
|
ignore_ssl = args["ignore_ssl"]
|
|
logging.debug("root_path = %s", root_paths)
|
|
logging.debug("root_url = %s", root_url)
|
|
logging.debug("api_key = %s", api_key)
|
|
logging.debug("number_of_images_per_request = %d", number_of_images_per_request)
|
|
logging.debug("number_of_assets_to_fetch_per_request = %d", number_of_assets_to_fetch_per_request)
|
|
logging.debug("unattended = %s", unattended)
|
|
logging.debug("album_levels = %d", album_levels)
|
|
logging.debug("album_level_separator = %s", album_level_separator)
|
|
logging.debug("ignore_ssl = %s", ignore_ssl)
|
|
|
|
# Verify album levels
|
|
if album_levels == 0:
|
|
parser.print_help()
|
|
exit(1)
|
|
|
|
if ignore_ssl:
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
# Yield successive n-sized
|
|
# chunks from l.
|
|
def divide_chunks(l, n):
|
|
|
|
# looping till length l
|
|
for i in range(0, len(l), n):
|
|
yield l[i:i + n]
|
|
|
|
|
|
requests_kwargs = {
|
|
'headers' : {
|
|
'x-api-key': api_key,
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json'
|
|
},
|
|
'verify' : not ignore_ssl
|
|
}
|
|
|
|
# append trailing slash to all root paths
|
|
for i in range(len(root_paths)):
|
|
if root_paths[i][-1] != '/':
|
|
root_paths[i] = root_paths[i] + '/'
|
|
# append trailing slash to root URL
|
|
if root_url[-1] != '/':
|
|
root_url = root_url + '/'
|
|
|
|
logging.info("Requesting all assets")
|
|
assets = []
|
|
# Initial API call, let's fetch our first chunk
|
|
r = requests.get(root_url+'asset?take='+str(number_of_assets_to_fetch_per_request), **requests_kwargs)
|
|
assert r.status_code == 200
|
|
logging.debug("Received %s assets with chunk 1", len(r.json()))
|
|
assets = assets + r.json()
|
|
|
|
# If we got a full chunk size back, let's perfrom subsequent calls until we get less than a full chunk size
|
|
skip = 0
|
|
while len(r.json()) == number_of_assets_to_fetch_per_request:
|
|
skip += number_of_assets_to_fetch_per_request
|
|
r = requests.get(root_url+'asset?take='+str(number_of_assets_to_fetch_per_request)+'&skip='+str(skip), **requests_kwargs)
|
|
if skip == number_of_assets_to_fetch_per_request and assets == r.json():
|
|
logging.info("Non-chunked Immich API detected, stopping fetching assets since we already got all in our first call")
|
|
break
|
|
assert r.status_code == 200
|
|
logging.debug("Received %s assets with chunk", len(r.json()))
|
|
assets = assets + r.json()
|
|
logging.info("%d photos found", len(assets))
|
|
|
|
|
|
|
|
logging.info("Sorting assets to corresponding albums using folder name")
|
|
album_to_assets = defaultdict(list)
|
|
for asset in assets:
|
|
asset_path = asset['originalPath']
|
|
for root_path in root_paths:
|
|
if root_path not in asset_path:
|
|
continue
|
|
# Chunks of the asset's path below root_path
|
|
path_chunks = asset_path.replace(root_path, '').split('/')
|
|
# A single chunk means it's just the image file in no sub folder, ignore
|
|
if len(path_chunks) == 1:
|
|
continue
|
|
|
|
# remove last item from path chunks, which is the file name
|
|
del path_chunks[-1]
|
|
album_name_chunks = ()
|
|
# either use as many path chunks as we have,
|
|
# or the specified album levels
|
|
album_name_chunk_size = min(len(path_chunks), album_levels)
|
|
if album_levels < 0:
|
|
album_name_chunk_size = min(len(path_chunks), abs(album_levels))*-1
|
|
|
|
# Copy album name chunks from the path to use as album name
|
|
album_name_chunks = path_chunks[:album_name_chunk_size]
|
|
if album_name_chunk_size < 0:
|
|
album_name_chunks = path_chunks[album_name_chunk_size:]
|
|
|
|
album_name = album_level_separator.join(album_name_chunks)
|
|
# Check that the extracted album name is not actually a file name in root_path
|
|
album_to_assets[album_name].append(asset['id'])
|
|
|
|
album_to_assets = {k:v for k, v in sorted(album_to_assets.items(), key=(lambda item: item[0]))}
|
|
|
|
logging.info("%d albums identified", len(album_to_assets))
|
|
logging.info("Album list: %s", list(album_to_assets.keys()))
|
|
if not unattended:
|
|
print("Press Enter to continue, Ctrl+C to abort")
|
|
input()
|
|
|
|
|
|
album_to_id = {}
|
|
|
|
logging.info("Listing existing albums on immich")
|
|
r = requests.get(root_url+'album', **requests_kwargs)
|
|
assert r.status_code == 200
|
|
albums = r.json()
|
|
album_to_id = {album['albumName']:album['id'] for album in albums }
|
|
logging.info("%d existing albums identified", len(albums))
|
|
|
|
|
|
logging.info("Creating albums if needed")
|
|
cpt = 0
|
|
for album in album_to_assets:
|
|
if album in album_to_id:
|
|
continue
|
|
data = {
|
|
'albumName': album,
|
|
'description': album
|
|
}
|
|
r = requests.post(root_url+'album', json=data, **requests_kwargs)
|
|
assert r.status_code in [200, 201]
|
|
album_to_id[album] = r.json()['id']
|
|
logging.info('Album %s added!', album)
|
|
cpt += 1
|
|
logging.info("%d albums created", cpt)
|
|
|
|
|
|
logging.info("Adding assets to albums")
|
|
# Note: Immich manages duplicates without problem,
|
|
# so we can each time ad all assets to same album, no photo will be duplicated
|
|
for album, assets in album_to_assets.items():
|
|
id = album_to_id[album]
|
|
|
|
# Divide our assets into chunks of number_of_images_per_request,
|
|
# So the API can cope
|
|
assets_chunked = list(divide_chunks(assets, number_of_images_per_request))
|
|
for assets_chunk in assets_chunked:
|
|
data = {'ids':assets_chunk}
|
|
r = requests.put(root_url+f'album/{id}/assets', json=data, **requests_kwargs)
|
|
if r.status_code not in [200, 201]:
|
|
print(album)
|
|
print(r.json())
|
|
print(data)
|
|
continue
|
|
assert r.status_code in [200, 201]
|
|
response = r.json()
|
|
|
|
cpt = 0
|
|
for res in response:
|
|
if not res['success']:
|
|
if res['error'] != 'duplicate':
|
|
logging.warning("Error adding an asset to an album: %s", res['error'])
|
|
else:
|
|
cpt += 1
|
|
if cpt > 0:
|
|
logging.info("%d new assets added to %s", cpt, album)
|
|
|
|
logging.info("Done!")
|