refactor: Shift album management from Immich API to local directories
Co-authored-by: aider (gemini/gemini-2.5-pro) <aider@aider.chat>
This commit is contained in:
131
app/app.py
131
app/app.py
@@ -223,129 +223,14 @@ def read_exif_datetimes(file_bytes: bytes):
|
||||
pass
|
||||
return created, modified
|
||||
|
||||
def immich_headers(request: Optional[Request] = None) -> dict:
|
||||
"""Headers for Immich API calls using either session access token or API key."""
|
||||
headers = {"Accept": "application/json"}
|
||||
token = None
|
||||
try:
|
||||
if request is not None:
|
||||
token = request.session.get("accessToken")
|
||||
except Exception:
|
||||
token = None
|
||||
if token:
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
elif SETTINGS.immich_api_key:
|
||||
headers["x-api-key"] = SETTINGS.immich_api_key
|
||||
return headers
|
||||
|
||||
def get_or_create_album(request: Optional[Request] = None, album_name_override: Optional[str] = None) -> Optional[str]:
|
||||
"""Get existing album by name or create a new one. Returns album ID or None."""
|
||||
global ALBUM_ID
|
||||
album_name = album_name_override if album_name_override is not None else SETTINGS.album_name
|
||||
# Skip if no album name configured
|
||||
if not album_name:
|
||||
return None
|
||||
# Return cached album ID if already fetched and using default settings name
|
||||
if album_name_override is None and ALBUM_ID:
|
||||
return ALBUM_ID
|
||||
try:
|
||||
# First, try to find existing album
|
||||
url = f"{SETTINGS.normalized_base_url}/albums"
|
||||
r = requests.get(url, headers=immich_headers(request), timeout=10)
|
||||
|
||||
if r.status_code == 200:
|
||||
albums = r.json()
|
||||
for album in albums:
|
||||
if album.get("albumName") == album_name:
|
||||
found_id = album.get("id")
|
||||
if album_name_override is None:
|
||||
ALBUM_ID = found_id
|
||||
logger.info(f"Found existing album '%s' with ID: %s", album_name, ALBUM_ID)
|
||||
return ALBUM_ID
|
||||
else:
|
||||
return found_id
|
||||
|
||||
# Album doesn't exist, create it
|
||||
create_url = f"{SETTINGS.normalized_base_url}/albums"
|
||||
payload = {
|
||||
"albumName": album_name,
|
||||
"description": "Auto-created album for Immich Drop uploads"
|
||||
}
|
||||
r = requests.post(create_url, headers={**immich_headers(request), "Content-Type": "application/json"},
|
||||
json=payload, timeout=10)
|
||||
|
||||
if r.status_code in (200, 201):
|
||||
data = r.json()
|
||||
new_id = data.get("id")
|
||||
if album_name_override is None:
|
||||
ALBUM_ID = new_id
|
||||
logger.info("Created new album '%s' with ID: %s", album_name, ALBUM_ID)
|
||||
return ALBUM_ID
|
||||
else:
|
||||
logger.info("Created new album '%s' with ID: %s", album_name, new_id)
|
||||
return new_id
|
||||
else:
|
||||
logger.warning("Failed to create album: %s - %s", r.status_code, r.text)
|
||||
except Exception as e:
|
||||
logger.exception("Error managing album: %s", e)
|
||||
|
||||
return None
|
||||
|
||||
def add_asset_to_album(asset_id: str, request: Optional[Request] = None, album_id_override: Optional[str] = None, album_name_override: Optional[str] = None) -> bool:
|
||||
"""Add an asset to the configured album. Returns True on success."""
|
||||
album_id = album_id_override
|
||||
if not album_id:
|
||||
album_id = get_or_create_album(request=request, album_name_override=album_name_override)
|
||||
if not album_id or not asset_id:
|
||||
return False
|
||||
|
||||
try:
|
||||
url = f"{SETTINGS.normalized_base_url}/albums/{album_id}/assets"
|
||||
payload = {"ids": [asset_id]}
|
||||
r = requests.put(url, headers={**immich_headers(request), "Content-Type": "application/json"},
|
||||
json=payload, timeout=10)
|
||||
|
||||
if r.status_code == 200:
|
||||
results = r.json()
|
||||
# Check if any result indicates success
|
||||
for result in results:
|
||||
if result.get("success"):
|
||||
return True
|
||||
elif result.get("error") == "duplicate":
|
||||
# Asset already in album, consider it success
|
||||
return True
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.exception("Error adding asset to album: %s", e)
|
||||
return False
|
||||
|
||||
def immich_ping() -> bool:
|
||||
"""Best-effort reachability check against a few Immich endpoints."""
|
||||
if not SETTINGS.immich_api_key:
|
||||
return False
|
||||
base = SETTINGS.normalized_base_url
|
||||
for path in ("/server-info", "/server/version", "/users/me"):
|
||||
try:
|
||||
r = requests.get(f"{base}{path}", headers=immich_headers(), timeout=4)
|
||||
if 200 <= r.status_code < 400:
|
||||
return True
|
||||
except Exception:
|
||||
continue
|
||||
return False
|
||||
|
||||
def immich_bulk_check(checks: List[dict]) -> Dict[str, dict]:
|
||||
"""Try Immich bulk upload check; return map id->result (or empty on failure)."""
|
||||
if SETTINGS.local_save_only:
|
||||
return {}
|
||||
try:
|
||||
url = f"{SETTINGS.normalized_base_url}/assets/bulk-upload-check"
|
||||
r = requests.post(url, headers=immich_headers(), json={"assets": checks}, timeout=10)
|
||||
if r.status_code == 200:
|
||||
results = r.json().get("results", [])
|
||||
return {x["id"]: x for x in results}
|
||||
except Exception:
|
||||
pass
|
||||
return {}
|
||||
def get_or_create_album_dir(album_name: str) -> str:
|
||||
"""Get or create a directory for an album. Returns the path."""
|
||||
if not album_name or not isinstance(album_name, str):
|
||||
album_name = "public"
|
||||
safe_album_name = sanitize_filename(album_name)
|
||||
save_dir = os.path.join("/data/uploads", safe_album_name)
|
||||
os.makedirs(save_dir, exist_ok=True)
|
||||
return save_dir
|
||||
|
||||
async def send_progress(session_id: str, item_id: str, status: str, progress: int = 0, message: str = "", response_id: Optional[str] = None) -> None:
|
||||
"""Push a progress update over WebSocket for one queue item."""
|
||||
|
||||
Reference in New Issue
Block a user