Download photos and videos from your authorized Procare parent gallery

README.md

Download photos and videos from your Procare parent gallery

This unofficial Python tool downloads the original photos and videos available through your own Procare parent account. It keeps separate folders for each child, preserves the gallery timestamp in each filename, and writes a manifest so interrupted or later incremental runs do not download the same media again.

Important permission requirement

Procare's June 2026 terms restrict extracting, scraping, or systematically harvesting service data through automated means without Procare's prior written consent. Use this tool only if Procare has given you written permission, and follow your childcare provider's policies. Do not use it to access another account or redistribute other families' media.

This project is independent and is not affiliated with or endorsed by Procare. It uses an undocumented interface that may change without notice.

Requirements

• Python 3.12 or newer
• Your own working Procare parent login
• Enough free disk space for the gallery
• Written permission from Procare for automated extraction
• Optional on macOS: Xcode Command Line Tools if you want to save credentials in Keychain

Download procare-download.py and keychain_store.swift into the same folder. The Swift helper is optional unless you use --save-keychain; it runs with /usr/bin/swift from Xcode Command Line Tools.

Test the latest 30 days

python3 procare-download.py \
  --output "$HOME/Pictures/Procare" \
  --days 30 \
  --save-keychain

The script prompts for your Procare email and password. On macOS, --save-keychain stores the verified credentials in Keychain; they are never written to the archive or manifest. Windows and Linux prompt on every run.

Without --children, every child returned for the account is exported. Use --dry-run to inspect counts without downloading files.

Download older history

Choose the earliest date you need:

python3 procare-download.py \
  --output "$HOME/Pictures/Procare" \
  --since 2024-01-01

Download only new media later

After the first successful export, reuse the same output folder:

python3 procare-download.py \
  --output "$HOME/Pictures/Procare" \
  --incremental

Incremental runs overlap the previous seven days to catch late additions and stop after reaching a complete gallery page already recorded in the manifest. If any download fails, the checkpoint rewinds so a later run retries the full affected date range.

Privacy and safety

• Credentials are prompted for or stored in macOS Keychain.
• Session tokens are sent only to allowlisted HTTPS Procare/Kinderlime hosts.
• Signed media URLs and tokens are never stored in manifest.json.
• New files are written through temporary files, then renamed after validation.
• The manifest records byte sizes and SHA-256 hashes.

Keep the download folder and manifest.json private: filenames and manifest records contain dates, child names, and media identifiers.

Full walkthrough: https://www.leyanlo.com/blog/downloading-photos-from-procare/

keychain_store.swift

import Foundation
import Security

struct Payload: Decodable {
    let service: String
    let account: String
    let label: String
    let secret: String
}

func fail(_ message: String, status: OSStatus? = nil) -> Never {
    if let status {
        let detail = SecCopyErrorMessageString(status, nil) as String? ?? "unknown error"
        FileHandle.standardError.write(Data("\(message): \(detail)\n".utf8))
    } else {
        FileHandle.standardError.write(Data("\(message)\n".utf8))
    }
    exit(1)
}

let input = FileHandle.standardInput.readDataToEndOfFile()
guard let payload = try? JSONDecoder().decode(Payload.self, from: input) else {
    fail("Invalid Keychain payload")
}

let query: [String: Any] = [
    kSecClass as String: kSecClassGenericPassword,
    kSecAttrService as String: payload.service,
    kSecAttrAccount as String: payload.account,
]
let attributes: [String: Any] = [
    kSecAttrLabel as String: payload.label,
    kSecValueData as String: Data(payload.secret.utf8),
]

var status = SecItemUpdate(query as CFDictionary, attributes as CFDictionary)
if status == errSecItemNotFound {
    var newItem = query
    attributes.forEach { newItem[$0.key] = $0.value }
    status = SecItemAdd(newItem as CFDictionary, nil)
}
guard status == errSecSuccess else {
    fail("Could not save Keychain item", status: status)
}

LICENSE

MIT License

Copyright (c) 2026 Leyan Lo

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

procare-download.py

#!/usr/bin/env python3
"""Export recent Procare parent-gallery photos and videos.

Credentials can be prompted for at runtime or read from the user's macOS
Keychain. API bearer tokens are only sent to HTTPS Procare/Kinderlime API
hosts. Signed media URLs are fetched without the bearer token and are never
written to the manifest.

This is an unofficial personal-archive tool. Procare's June 2026 terms restrict
automated extraction without prior written consent. Use it only with Procare's
written permission and only for media available through your own parent account.
"""

from __future__ import annotations

import argparse
import getpass
import hashlib
import json
import os
import re
import shutil
import ssl
import subprocess
import sys
import time
from dataclasses import dataclass, field
from datetime import date, datetime, time as datetime_time, timedelta, timezone
from pathlib import Path
from typing import Any, Iterable
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode, urljoin, urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener


AUTH_URL = "https://online-auth.procareconnect.com/sessions/"
REQUEST_TIMEOUT = 60
MAX_GALLERY_PAGES = 500
RETRIES = 4
POLITE_DELAY = 0.25
API_MIN_INTERVAL = 2.1
INCREMENTAL_OVERLAP_DAYS = 7
USER_AGENT = "Personal-Procare-Archive/1.0"
KEYCHAIN_ACCOUNT_SERVICE = "io.github.leyanlo.procare-downloader.account"
KEYCHAIN_PASSWORD_SERVICE = "io.github.leyanlo.procare-downloader.password"
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".heic", ".heif"}
VIDEO_EXTENSIONS = {".mp4", ".mov", ".m4v", ".webm", ".avi", ".3gp", ".mkv"}
DATE_KEYS = (
    "datetime",
    "captured_at",
    "taken_at",
    "activity_time",
    "created_at",
    "updated_at",
)
URL_KEYS = (
    "original_url",
    "video_file_url",
    "file_url",
    "main_url",
    "url",
    "photo_url",
    "image_url",
)


class ExportError(RuntimeError):
    """A user-facing export failure."""


class SafeRedirectHandler(HTTPRedirectHandler):
    """Allow HTTPS redirects while preventing transport downgrade."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001
        if urlsplit(newurl).scheme.lower() != "https":
            raise ExportError(f"Refusing non-HTTPS redirect (HTTP {code}).")
        return super().redirect_request(req, fp, code, msg, headers, newurl)


HTTPS_OPENER = build_opener(SafeRedirectHandler())
TLS_CONTEXT = ssl.create_default_context()
_last_api_request = 0.0
_media_indexes: dict[Path, dict[str, Path]] = {}


@dataclass
class Child:
    id: str
    name: str


@dataclass
class MediaEntry:
    kind: str
    ident: str
    url: str
    captured_at: datetime
    returned_for: set[str] = field(default_factory=set)
    explicit_children: set[str] = field(default_factory=set)


@dataclass
class DownloadResult:
    path: Path
    paths: list[Path]
    sha256: str
    size: int
    existed: bool


def is_allowed_api_url(url: str) -> bool:
    parts = urlsplit(url)
    host = (parts.hostname or "").lower()
    return parts.scheme == "https" and (
        host == "procareconnect.com"
        or host.endswith(".procareconnect.com")
        or host == "kinderlime.com"
        or host.endswith(".kinderlime.com")
    )


def request_json(
    url: str,
    *,
    method: str = "GET",
    payload: dict[str, Any] | None = None,
    token: str | None = None,
    params: dict[str, Any] | None = None,
) -> Any:
    if params:
        url = f"{url}?{urlencode(params)}"
    if token and not is_allowed_api_url(url):
        raise ExportError("Refusing to send the Procare session token off-domain.")

    data = json.dumps(payload).encode("utf-8") if payload is not None else None
    headers = {"Accept": "application/json", "User-Agent": USER_AGENT}
    if data is not None:
        headers["Content-Type"] = "application/json"
    if token:
        headers["Authorization"] = f"Bearer {token}"

    request = Request(url, data=data, headers=headers, method=method)
    for attempt in range(RETRIES):
        if token:
            pace_api_request()
        try:
            with HTTPS_OPENER.open(request, timeout=REQUEST_TIMEOUT) as response:
                final_url = response.geturl()
                if token and not is_allowed_api_url(final_url):
                    raise ExportError("Procare API redirected the session token off-domain.")
                return json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            retryable = exc.code == 429 or exc.code in {500, 502, 503, 504}
            if retryable and attempt < RETRIES - 1:
                retry_after = exc.headers.get("Retry-After", "")
                if retry_after.isdigit():
                    delay = max(int(retry_after), API_MIN_INTERVAL)
                elif exc.code == 429:
                    delay = 65
                else:
                    delay = 2**attempt
                print(
                    f"\n  Procare returned HTTP {exc.code}; retrying in {delay:g}s...",
                    flush=True,
                )
                time.sleep(delay)
                continue
            detail = ""
            try:
                body = json.loads(exc.read().decode("utf-8"))
                errors = (
                    body.get("errors") or body.get("error")
                    if isinstance(body, dict)
                    else None
                )
                if isinstance(errors, list):
                    detail = ": " + "; ".join(str(value) for value in errors)
                elif errors:
                    detail = f": {errors}"
            except (UnicodeDecodeError, json.JSONDecodeError):
                pass
            raise ExportError(f"Procare returned HTTP {exc.code}{detail}") from exc
        except (URLError, TimeoutError, json.JSONDecodeError) as exc:
            if attempt < RETRIES - 1:
                time.sleep(2**attempt)
                continue
            raise ExportError(f"Could not read a valid response from Procare: {exc}") from exc
    raise ExportError("Could not read a valid response from Procare after retries.")


def pace_api_request() -> None:
    global _last_api_request
    now = time.monotonic()
    remaining = API_MIN_INTERVAL - (now - _last_api_request)
    if remaining > 0:
        time.sleep(remaining)
    _last_api_request = time.monotonic()


def authenticate(email: str, password: str) -> tuple[str, str]:
    response = request_json(
        AUTH_URL,
        method="POST",
        payload={
            "email": email,
            "password": password,
            "role": "carer",
            "platform": "web",
            "preserve_sites": True,
        },
    )
    data = response.get("data", response) if isinstance(response, dict) else {}
    token = data.get("auth_token")
    if not token and isinstance(data.get("user"), dict):
        token = data["user"].get("auth_token")

    sites = data.get("sites") if isinstance(data, dict) else None
    site = None
    if isinstance(sites, list) and sites:
        site = next(
            (item for item in sites if isinstance(item, dict) and item.get("is_default")),
            sites[0],
        )
    base_url = site.get("base_url") if isinstance(site, dict) else None
    if not token or not base_url:
        raise ExportError(
            "Procare did not return a usable parent session. Accounts requiring "
            "SSO or a verification code are not supported by this test exporter."
        )

    api_base = base_url.rstrip("/") + "/api/web/"
    if not is_allowed_api_url(api_base):
        raise ExportError("Procare returned an unexpected API host; stopping safely.")
    return str(token), api_base


def extract_items(payload: Any, preferred_key: str | None = None) -> list[dict[str, Any]]:
    if isinstance(payload, list):
        return [item for item in payload if isinstance(item, dict)]
    if not isinstance(payload, dict):
        return []

    keys = [preferred_key, "kids", "photos", "videos", "items", "data"]
    for key in keys:
        if not key:
            continue
        value = payload.get(key)
        if isinstance(value, list):
            return [item for item in value if isinstance(item, dict)]
        if isinstance(value, dict):
            nested = extract_items(value, preferred_key)
            if nested:
                return nested
    return []


def display_name(item: dict[str, Any]) -> str:
    for key in ("name", "full_name", "display_name"):
        value = item.get(key)
        if isinstance(value, str) and value.strip():
            return value.strip()
    parts = [item.get("first_name"), item.get("last_name")]
    joined = " ".join(str(value).strip() for value in parts if value).strip()
    return joined or "Unknown child"


def fetch_children(token: str, api_base: str) -> list[Child]:
    payload = request_json(urljoin(api_base, "parent/kids/"), token=token)
    children = []
    for item in extract_items(payload, "kids"):
        ident = item.get("id")
        if ident is not None:
            children.append(Child(str(ident), display_name(item)))
    if not children:
        raise ExportError("No children were returned for this parent account.")
    return children


def select_children(available: list[Child], requested: Iterable[str] | None = None) -> list[Child]:
    if requested is None:
        return list(available)

    selected = []
    for wanted in requested:
        normalized = wanted.casefold().strip()
        id_matches = [child for child in available if child.id.casefold() == normalized]
        exact = [child for child in available if child.name.casefold() == normalized]
        matches = id_matches or exact or [
            child
            for child in available
            if normalized in child.name.casefold()
            or normalized in child.name.casefold().split()
        ]
        if len(matches) != 1:
            names = ", ".join(child.name for child in available)
            if not matches:
                raise ExportError(f"Could not find child {wanted!r}. Available: {names}")
            raise ExportError(f"Child name {wanted!r} is ambiguous. Available: {names}")
        if matches[0] not in selected:
            selected.append(matches[0])
    return selected


def parse_datetime(value: Any) -> datetime | None:
    if not isinstance(value, str) or not value.strip():
        return None
    raw = value.strip()
    if raw.endswith("Z"):
        raw = raw[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(raw)
    except ValueError:
        for pattern in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
            try:
                parsed = datetime.strptime(raw, pattern)
                break
            except ValueError:
                continue
        else:
            return None
    if parsed.tzinfo:
        parsed = parsed.astimezone().replace(tzinfo=None)
    return parsed


def find_capture_datetime(item: dict[str, Any]) -> datetime | None:
    for key in DATE_KEYS:
        parsed = parse_datetime(item.get(key))
        if parsed:
            return parsed
    activiable = item.get("activiable")
    if isinstance(activiable, dict):
        return find_capture_datetime(activiable)
    return None


def iter_url_candidates(value: Any, key: str = ""):
    if isinstance(value, dict):
        for child_key, child_value in value.items():
            yield from iter_url_candidates(child_value, child_key)
    elif isinstance(value, list):
        for child_value in value:
            yield from iter_url_candidates(child_value, key)
    elif isinstance(value, str) and value.startswith("https://"):
        yield key.casefold(), value


def url_extension(url: str) -> str:
    return Path(urlsplit(url).path).suffix.casefold()


def find_media_url(item: dict[str, Any], kind: str) -> str | None:
    direct_keys = (
        ("video_file_url", "original_url", "file_url", "url")
        if kind == "video"
        else ("original_url", "main_url", "file_url", "url", "photo_url", "image_url")
    )
    for key in direct_keys:
        value = item.get(key)
        if isinstance(value, str) and value.startswith("https://"):
            extension = url_extension(value)
            if kind == "video" and (extension in VIDEO_EXTENSIONS or "video" in key):
                return value
            if kind == "photo" and (extension in IMAGE_EXTENSIONS or extension == ""):
                return value

    candidates = list(iter_url_candidates(item))
    if kind == "video":
        candidates.sort(key=lambda pair: ("video" not in pair[0], "poster" in pair[0]))
        return next(
            (
                url
                for key, url in candidates
                if url_extension(url) in VIDEO_EXTENSIONS
                or ("video" in key and "poster" not in key and "thumb" not in key)
            ),
            None,
        )
    candidates.sort(
        key=lambda pair: (
            not any(word in pair[0] for word in ("original", "main", "full", "large")),
            any(word in pair[0] for word in ("thumb", "small", "preview")),
        )
    )
    return next((url for _, url in candidates if url_extension(url) in IMAGE_EXTENSIONS), None)


def explicit_child_ids(item: dict[str, Any]) -> set[str]:
    result: set[str] = set()
    for key in ("kid_ids", "student_ids", "child_ids"):
        value = item.get(key)
        if isinstance(value, list):
            result.update(str(ident) for ident in value if ident is not None)
    for key in ("kid_id", "student_id", "child_id"):
        value = item.get(key)
        if value is not None:
            result.add(str(value))
    for key in ("kids", "students", "participants"):
        value = item.get(key)
        if isinstance(value, list):
            result.update(
                str(child["id"])
                for child in value
                if isinstance(child, dict) and child.get("id") is not None
            )
    return result


def stable_ident(item: dict[str, Any], url: str) -> str:
    if item.get("id") is not None:
        return str(item["id"])
    path = urlsplit(url).path
    filename = Path(path).stem
    if filename and filename not in {"open-uri", "original", "main"}:
        return filename
    return hashlib.sha256(path.encode("utf-8")).hexdigest()[:24]


def gallery_params(kind: str, child_id: str, start: date, end: date, page: int) -> dict[str, Any]:
    resource = "photo" if kind == "photo" else "video"
    return {
        "page": page,
        # Procare ignores a top-level ``kid_id`` here. Its gallery sends an
        # array-valued child filter nested under the media resource, including
        # the explicit [] suffix when only one child is selected.
        f"filters[{resource}][kid_ids][]": child_id,
        f"filters[{resource}][datetime_from]": f"{start.isoformat()} 00:00",
        f"filters[{resource}][datetime_to]": f"{end.isoformat()} 23:59",
    }


def month_windows(start: date, end: date):
    cursor = date(start.year, start.month, 1)
    while cursor <= end:
        if cursor.month == 12:
            next_month = date(cursor.year + 1, 1, 1)
        else:
            next_month = date(cursor.year, cursor.month + 1, 1)
        yield max(start, cursor), min(end, next_month - timedelta(days=1))
        cursor = next_month


def collect_gallery(
    token: str,
    api_base: str,
    children: list[Child],
    start: date,
    end: date,
) -> tuple[dict[tuple[str, str], MediaEntry], int]:
    entries: dict[tuple[str, str], MediaEntry] = {}
    missing_dates = 0
    start_dt = datetime.combine(start, datetime_time.min)
    end_dt = datetime.combine(end, datetime_time.max)
    windows = list(month_windows(start, end))
    total_windows = len(children) * 2 * len(windows)
    completed_windows = 0

    for child in children:
        for kind in ("photo", "video"):
            endpoint = urljoin(api_base, f"parent/{kind}s/")
            for window_start, window_end in windows:
                previous_ids: tuple[str, ...] | None = None
                for page in range(1, MAX_GALLERY_PAGES + 1):
                    payload = request_json(
                        endpoint,
                        token=token,
                        params=gallery_params(kind, child.id, window_start, window_end, page),
                    )
                    items = extract_items(payload, f"{kind}s")
                    if not items:
                        break
                    page_ids = tuple(str(item.get("id", "")) for item in items)
                    if page_ids and page_ids == previous_ids:
                        break
                    previous_ids = page_ids

                    for item in items:
                        url = find_media_url(item, kind)
                        captured_at = find_capture_datetime(item)
                        if not url or captured_at is None:
                            if captured_at is None:
                                missing_dates += 1
                            continue
                        if not (start_dt <= captured_at <= end_dt):
                            continue
                        ident = stable_ident(item, url)
                        key = (kind, ident)
                        entry = entries.get(key)
                        if entry is None:
                            entry = MediaEntry(kind, ident, url, captured_at)
                            entries[key] = entry
                        else:
                            entry.url = url
                        entry.returned_for.add(child.id)
                        entry.explicit_children.update(explicit_child_ids(item))
                else:
                    raise ExportError(
                        f"Gallery pagination exceeded {MAX_GALLERY_PAGES} pages "
                        f"for {window_start:%Y-%m}."
                    )
                completed_windows += 1
                print(
                    f"\r  Inventory {completed_windows}/{total_windows}: "
                    f"{child.name} {kind}s {window_start:%Y-%m}",
                    end="",
                    flush=True,
                )
    if total_windows:
        print()
    return entries, missing_dates


def collect_gallery_incremental(
    token: str,
    api_base: str,
    children: list[Child],
    start: date,
    end: date,
    known_keys: set[tuple[str, str]],
) -> tuple[dict[tuple[str, str], MediaEntry], int]:
    """Read newest-first pages until a complete page is already archived."""
    entries: dict[tuple[str, str], MediaEntry] = {}
    missing_dates = 0
    start_dt = datetime.combine(start, datetime_time.min)
    end_dt = datetime.combine(end, datetime_time.max)
    total_streams = len(children) * 2
    completed_streams = 0

    for child in children:
        for kind in ("photo", "video"):
            endpoint = urljoin(api_base, f"parent/{kind}s/")
            previous_ids: tuple[str, ...] | None = None
            stopped_on_known_page = False
            pages_read = 0
            for page in range(1, MAX_GALLERY_PAGES + 1):
                payload = request_json(
                    endpoint,
                    token=token,
                    params=gallery_params(kind, child.id, start, end, page),
                )
                items = extract_items(payload, f"{kind}s")
                if not items:
                    break
                page_ids = tuple(str(item.get("id", "")) for item in items)
                if page_ids and page_ids == previous_ids:
                    break
                previous_ids = page_ids
                pages_read = page
                page_keys: set[tuple[str, str]] = set()

                for item in items:
                    url = find_media_url(item, kind)
                    captured_at = find_capture_datetime(item)
                    if not url or captured_at is None:
                        if captured_at is None:
                            missing_dates += 1
                        continue
                    if not (start_dt <= captured_at <= end_dt):
                        continue
                    ident = stable_ident(item, url)
                    key = (kind, ident)
                    page_keys.add(key)
                    entry = entries.get(key)
                    if entry is None:
                        entry = MediaEntry(kind, ident, url, captured_at)
                        entries[key] = entry
                    else:
                        entry.url = url
                    entry.returned_for.add(child.id)
                    entry.explicit_children.update(explicit_child_ids(item))

                print(
                    f"\r  Incremental {completed_streams + 1}/{total_streams}: "
                    f"{child.name} {kind}s page {page}",
                    end="",
                    flush=True,
                )
                if page_keys and page_keys.issubset(known_keys):
                    stopped_on_known_page = True
                    break
            else:
                raise ExportError(
                    f"Gallery pagination exceeded {MAX_GALLERY_PAGES} pages "
                    f"for {child.name} {kind}s."
                )

            completed_streams += 1
            reason = "known page reached" if stopped_on_known_page else "end reached"
            print(
                f"\r  Incremental {completed_streams}/{total_streams}: "
                f"{child.name} {kind}s, {pages_read} page(s), {reason}"
            )
    return entries, missing_dates


def safe_component(value: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9._ -]+", "-", value).strip(" .-")
    return cleaned or "Unknown"


def child_folder_name(name: str) -> str:
    """Derive a short archive folder from a child's Procare display name."""
    parenthetical = re.search(r"\(([^)]+)\)", name)
    if parenthetical:
        return safe_component(parenthetical.group(1))
    return safe_component(name.split(maxsplit=1)[0])


def parse_child_folder_argument(value: str) -> tuple[str, str]:
    selector, separator, folder = value.partition("=")
    selector = selector.strip()
    folder = folder.strip()
    if not separator or not selector or not folder:
        raise argparse.ArgumentTypeError("expected NAME=FOLDER")
    if safe_component(folder) != folder:
        raise argparse.ArgumentTypeError("folder must be a safe single path component")
    return selector, folder


def validate_unique_child_folders(assignments: Iterable[tuple[str, str]]) -> None:
    claimed: dict[str, str] = {}
    for child_name, folder in assignments:
        normalized = folder.casefold()
        if normalized in claimed:
            raise ExportError(
                f"Child folder {folder!r} would be shared by {claimed[normalized]!r} "
                f"and {child_name!r}. Use --child-folder NAME=FOLDER to assign "
                "unique folders."
            )
        claimed[normalized] = child_name


def resolve_child_folders(
    children: list[Child], requested: Iterable[tuple[str, str]] = ()
) -> dict[str, str]:
    folders = {child.id: child_folder_name(child.name) for child in children}
    for selector, folder in requested:
        child = select_children(children, [selector])[0]
        folders[child.id] = folder
    validate_unique_child_folders((child.name, folders[child.id]) for child in children)
    return folders


def archive_folders(bucket: str, child_names: list[str]) -> list[str]:
    """Map an item to one or more flat child folders.

    Account-wide posts are mirrored into each child's folder because Procare
    returns them for both children without a more specific child assignment.
    """
    assignments = [(name, child_folder_name(name)) for name in child_names]
    validate_unique_child_folders(assignments)
    folders = [folder for _, folder in assignments]
    return folders or [safe_component(bucket)]


def assigned_children(entry: MediaEntry, children: list[Child]) -> list[Child]:
    selected_ids = {child.id for child in children}
    explicit = entry.explicit_children & selected_ids
    assigned = explicit or entry.returned_for
    matches = [child for child in children if child.id in assigned]
    return sorted(matches or children, key=lambda child: (child.name, child.id))


def assigned_bucket(entry: MediaEntry, children: list[Child]) -> tuple[str, list[str]]:
    assigned_names = [child.name for child in assigned_children(entry, children)]
    if len(assigned_names) == 1:
        return safe_component(assigned_names[0]), assigned_names
    return "Shared", assigned_names


def source_child_names(entry: MediaEntry, children: list[Child]) -> list[str]:
    """Return the child gallery queries that produced this media item."""
    return sorted(child.name for child in children if child.id in entry.returned_for)


def sniff_extension(head: bytes, content_type: str, url: str, kind: str) -> str:
    if head.startswith(b"\xff\xd8\xff"):
        return ".jpg"
    if head.startswith(b"\x89PNG\r\n\x1a\n"):
        return ".png"
    if head.startswith((b"GIF87a", b"GIF89a")):
        return ".gif"
    if head.startswith(b"RIFF") and head[8:12] == b"WEBP":
        return ".webp"
    if len(head) >= 12 and head[4:8] == b"ftyp":
        brand = head[8:12]
        if brand in {b"heic", b"heix", b"hevc", b"hevx", b"mif1", b"msf1"}:
            return ".heic"
        return ".mp4"
    content_type = content_type.split(";", 1)[0].strip().casefold()
    type_map = {
        "image/jpeg": ".jpg",
        "image/png": ".png",
        "image/gif": ".gif",
        "image/webp": ".webp",
        "image/heic": ".heic",
        "video/mp4": ".mp4",
        "video/quicktime": ".mov",
        "video/webm": ".webm",
    }
    if content_type in type_map:
        return type_map[content_type]
    extension = url_extension(url)
    allowed = VIDEO_EXTENSIONS if kind == "video" else IMAGE_EXTENSIONS
    if extension in allowed:
        return extension
    return ".mp4" if kind == "video" else ".jpg"


def media_stem(entry: MediaEntry) -> str:
    return (
        f"{entry.captured_at.strftime('%Y-%m-%d_%H%M%S')}_"
        f"{entry.kind}_{safe_component(entry.ident)}"
    )


def existing_media(media_dir: Path, entry: MediaEntry) -> Path | None:
    index = _media_indexes.get(media_dir)
    if index is None:
        index = {
            path.stem: path
            for path in media_dir.iterdir()
            if path.is_file() and path.suffix != ".part"
        }
        _media_indexes[media_dir] = index
    return index.get(media_stem(entry))


def file_digest(path: Path) -> tuple[str, int]:
    digest = hashlib.sha256()
    size = 0
    with path.open("rb") as handle:
        while chunk := handle.read(1024 * 1024):
            digest.update(chunk)
            size += len(chunk)
    return digest.hexdigest(), size


def mirror_media(source: Path, media_dirs: list[Path], digest: str, size: int) -> list[Path]:
    paths = []
    for media_dir in media_dirs:
        target = media_dir / source.name
        if target != source:
            if target.exists():
                target_digest, target_size = file_digest(target)
                if target_digest != digest or target_size != size:
                    raise ExportError(f"Existing mirror does not match: {target}")
            else:
                shutil.copy2(source, target)
                target.chmod(0o600)
                target_digest, target_size = file_digest(target)
                if target_digest != digest or target_size != size:
                    target.unlink(missing_ok=True)
                    raise ExportError(f"Copied mirror failed verification: {target}")
        paths.append(target)
    return paths


def download_media(
    entry: MediaEntry,
    bucket: str,
    child_names: list[str],
    output: Path,
    *,
    folder_names: list[str] | None = None,
) -> DownloadResult:
    if urlsplit(entry.url).scheme != "https":
        raise ExportError("Refusing a non-HTTPS media URL.")
    folders = folder_names if folder_names is not None else archive_folders(bucket, child_names)
    validate_unique_child_folders(zip(child_names, folders, strict=True))
    media_dirs = [output / folder for folder in folders]
    for media_dir in media_dirs:
        media_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
    existing = next(
        (found for media_dir in media_dirs if (found := existing_media(media_dir, entry))),
        None,
    )
    if existing:
        digest, size = file_digest(existing)
        paths = mirror_media(existing, media_dirs, digest, size)
        return DownloadResult(paths[0], paths, digest, size, True)

    stem = media_stem(entry)
    part_path = media_dirs[0] / f"{stem}.part"
    last_error: Exception | None = None

    for attempt in range(RETRIES):
        digest = hashlib.sha256()
        size = 0
        head = b""
        try:
            request = Request(entry.url, headers={"User-Agent": USER_AGENT})
            with HTTPS_OPENER.open(request, timeout=REQUEST_TIMEOUT) as response:
                if urlsplit(response.geturl()).scheme != "https":
                    raise ExportError("Media download redirected to non-HTTPS.")
                content_type = response.headers.get("Content-Type", "")
                expected_raw = response.headers.get("Content-Length", "")
                expected = int(expected_raw) if expected_raw.isdigit() else None
                with part_path.open("wb") as handle:
                    while chunk := response.read(1024 * 1024):
                        if not head:
                            head = chunk[:16]
                        handle.write(chunk)
                        digest.update(chunk)
                        size += len(chunk)
            if not size or (expected is not None and size != expected):
                raise ExportError("Media download was empty or incomplete.")
            if head.lstrip().lower().startswith((b"<!doctype html", b"<html", b"{")):
                raise ExportError("Media URL returned an error document instead of media.")
            extension = sniff_extension(head, content_type, entry.url, entry.kind)
            final_path = media_dirs[0] / f"{stem}{extension}"
            os.replace(part_path, final_path)
            timestamp = entry.captured_at.timestamp()
            os.utime(final_path, (timestamp, timestamp))
            final_path.chmod(0o600)
            _media_indexes.setdefault(media_dirs[0], {})[final_path.stem] = final_path
            paths = mirror_media(final_path, media_dirs, digest.hexdigest(), size)
            return DownloadResult(final_path, paths, digest.hexdigest(), size, False)
        except (HTTPError, URLError, TimeoutError, OSError, ExportError) as exc:
            last_error = exc
            part_path.unlink(missing_ok=True)
            if attempt < RETRIES - 1:
                time.sleep(2**attempt)
    raise ExportError(f"Failed to download {entry.kind} {entry.ident}: {last_error}")


def write_manifest(output: Path, manifest: dict[str, Any]) -> Path:
    output.mkdir(parents=True, exist_ok=True, mode=0o700)
    target = output / "manifest.json"
    temporary = output / "manifest.json.part"
    temporary.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    temporary.chmod(0o600)
    os.replace(temporary, target)
    return target


def load_manifest(output: Path, *, required: bool = False) -> dict[str, Any] | None:
    target = output / "manifest.json"
    if not target.is_file():
        if required:
            raise ExportError(
                f"Incremental mode requires an existing manifest at {target}."
            )
        return None
    try:
        payload = json.loads(target.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ExportError(f"Could not read the existing manifest at {target}.") from exc
    if not isinstance(payload, dict) or not isinstance(payload.get("items"), list):
        raise ExportError(f"The existing manifest at {target} is not valid.")
    return payload


def incremental_query_start(manifest: dict[str, Any], end: date) -> date:
    """Start near the last successful checkpoint, preserving a small overlap."""
    try:
        archive_start = manifest_archive_start(manifest)
        checkpoint = date.fromisoformat(str(manifest.get("date_to")))
    except (TypeError, ValueError, ExportError) as exc:
        raise ExportError(
            "The existing manifest has no valid date_from/date_to checkpoint."
        ) from exc
    checkpoint = min(checkpoint, end)
    return max(archive_start, checkpoint - timedelta(days=INCREMENTAL_OVERLAP_DAYS))


def manifest_checkpoint_end(
    manifest: dict[str, Any] | None,
    start: date,
    end: date,
    failed: int,
) -> str:
    """Advance only complete runs; rewind failed ranges for a later retry."""
    existing_end = str(manifest.get("date_to")) if manifest else ""
    if failed:
        return min(existing_end, start.isoformat()) if existing_end else start.isoformat()
    return max(existing_end, end.isoformat())


def manifest_archive_start(manifest: dict[str, Any]) -> date:
    try:
        return date.fromisoformat(str(manifest.get("date_from")))
    except (TypeError, ValueError) as exc:
        raise ExportError("The existing manifest has no valid date_from checkpoint.") from exc


def manifest_children(manifest: dict[str, Any] | None) -> dict[str, str]:
    if not manifest:
        return {}
    return {
        str(child["id"]): str(child.get("name") or child["id"])
        for child in manifest.get("children", [])
        if isinstance(child, dict) and child.get("id") is not None
    }


def manifest_child_folders(manifest: dict[str, Any] | None) -> dict[str, str]:
    if not manifest:
        return {}
    folders: dict[str, str] = {}
    for child in manifest.get("children", []):
        if not isinstance(child, dict) or child.get("id") is None:
            continue
        child_id = str(child["id"])
        name = str(child.get("name") or child_id)
        folders[child_id] = str(child.get("folder") or child_folder_name(name))
    return folders


def incremental_requires_backfill(
    manifest: dict[str, Any],
    selected: list[Child],
    child_folders: dict[str, str] | None = None,
) -> bool:
    archived = manifest_children(manifest)
    selected_ids = {child.id for child in selected}
    removed = set(archived) - selected_ids
    if removed:
        names = ", ".join(sorted(archived[child_id] for child_id in removed))
        raise ExportError(
            "An incremental export cannot omit children already recorded in the "
            f"manifest ({names}). Use a separate output archive for a smaller selection."
        )
    if selected_ids != set(archived):
        return True
    if child_folders is None:
        return False
    archived_folders = manifest_child_folders(manifest)
    return any(
        archived_folders.get(child.id, "").casefold()
        != child_folders[child.id].casefold()
        for child in selected
    )


def manifest_keys(manifest: dict[str, Any] | None) -> set[tuple[str, str]]:
    if not manifest:
        return set()
    return {
        (str(item["kind"]), str(item["media_id"]))
        for item in manifest.get("items", [])
        if isinstance(item, dict) and item.get("kind") and item.get("media_id") is not None
    }


def merge_manifest_items(
    existing: dict[str, Any] | None,
    current_items: Iterable[dict[str, Any]],
) -> list[dict[str, Any]]:
    merged: dict[tuple[str, str], dict[str, Any]] = {}
    if existing:
        for item in existing.get("items", []):
            if isinstance(item, dict) and item.get("kind") and item.get("media_id") is not None:
                merged[(str(item["kind"]), str(item["media_id"]))] = item
    for item in current_items:
        merged[(str(item["kind"]), str(item["media_id"]))] = item
    return sorted(
        merged.values(),
        key=lambda item: (
            str(item.get("captured_at", "")),
            str(item.get("kind", "")),
            str(item.get("media_id", "")),
        ),
    )


def inventory_report(
    entries: Iterable[MediaEntry],
    children: list[Child],
    start: date,
    end: date,
    missing_dates: int,
) -> dict[str, Any]:
    months: dict[str, dict[str, dict[str, int]]] = {}
    totals: dict[str, dict[str, int]] = {}
    earliest: datetime | None = None
    latest: datetime | None = None
    entry_list = list(entries)
    for entry in entry_list:
        bucket, _ = assigned_bucket(entry, children)
        month = entry.captured_at.strftime("%Y-%m")
        month_counts = months.setdefault(month, {}).setdefault(
            bucket, {"photos": 0, "videos": 0}
        )
        total_counts = totals.setdefault(bucket, {"photos": 0, "videos": 0})
        month_counts[f"{entry.kind}s"] += 1
        total_counts[f"{entry.kind}s"] += 1
        earliest = entry.captured_at if earliest is None else min(earliest, entry.captured_at)
        latest = entry.captured_at if latest is None else max(latest, entry.captured_at)
    return {
        "created_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "query": {"date_from": start.isoformat(), "date_to": end.isoformat()},
        "children": [child.name for child in children],
        "available": {
            "earliest": earliest.isoformat(timespec="seconds") if earliest else None,
            "latest": latest.isoformat(timespec="seconds") if latest else None,
        },
        "totals": totals,
        "months": dict(sorted(months.items())),
        "item_count": len(entry_list),
        "skipped_missing_date": missing_dates,
    }


def write_private_json(target: Path, payload: dict[str, Any]) -> Path:
    target = target.expanduser().resolve()
    target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    temporary = target.with_name(target.name + ".part")
    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    temporary.chmod(0o600)
    os.replace(temporary, target)
    return target


def parse_date_argument(value: str) -> date:
    try:
        return date.fromisoformat(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("expected YYYY-MM-DD") from exc


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--days",
        type=int,
        default=30,
        help="Inclusive number of recent calendar days to export (default: 30).",
    )
    parser.add_argument(
        "--since",
        type=parse_date_argument,
        help="Export on or after YYYY-MM-DD; overrides --days.",
    )
    parser.add_argument(
        "--until",
        type=parse_date_argument,
        help="Export on or before YYYY-MM-DD (default: today).",
    )
    parser.add_argument(
        "--incremental",
        action="store_true",
        help=(
            "Use the existing manifest as a checkpoint and stop each newest-first "
            "gallery stream after a complete page is already archived."
        ),
    )
    parser.add_argument(
        "--children",
        nargs="+",
        metavar="NAME",
        help="Child names to export (default: every child on the account).",
    )
    parser.add_argument(
        "--child-folder",
        action="append",
        default=[],
        type=parse_child_folder_argument,
        metavar="NAME=FOLDER",
        help=(
            "Override a child's archive folder; repeat for multiple children. "
            "Names or Procare child IDs may be used."
        ),
    )
    parser.add_argument(
        "--output",
        type=Path,
        required=True,
        help="Archive directory containing media and manifest.json.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Inspect counts without downloading or writing any files.",
    )
    parser.add_argument(
        "--mac-dialog",
        action="store_true",
        help="Prompt for credentials in native macOS dialogs instead of the terminal.",
    )
    keychain = parser.add_mutually_exclusive_group()
    keychain.add_argument(
        "--save-keychain",
        action="store_true",
        help="Prompt once and save credentials in the user's macOS Keychain.",
    )
    keychain.add_argument(
        "--no-keychain",
        action="store_true",
        help="Ignore saved Keychain credentials and prompt without saving.",
    )
    parser.add_argument(
        "--inventory-report",
        type=Path,
        help="Write a private JSON report containing counts by month, type, and bucket.",
    )
    return parser


def mac_dialog(prompt: str, *, hidden: bool = False) -> str:
    hidden_clause = " with hidden answer" if hidden else ""
    script = (
        f'display dialog {json.dumps(prompt)} default answer ""{hidden_clause} '
        'buttons {"Cancel", "Continue"} default button "Continue" '
        'cancel button "Cancel"'
    )
    try:
        result = subprocess.run(
            ["/usr/bin/osascript", "-e", script],
            check=True,
            capture_output=True,
            text=True,
        )
    except subprocess.CalledProcessError as exc:
        if exc.returncode == 1 and "User canceled" in exc.stderr:
            raise ExportError("Credential entry was canceled.") from exc
        raise ExportError("Could not open the macOS credential dialog.") from exc
    marker = "text returned:"
    if marker not in result.stdout:
        raise ExportError("The macOS credential dialog returned an unexpected response.")
    return result.stdout.split(marker, 1)[1].strip()


def keychain_find(service: str, account: str | None = None) -> str | None:
    if sys.platform != "darwin":
        return None
    command = ["/usr/bin/security", "find-generic-password", "-s", service]
    if account:
        command.extend(["-a", account])
    command.append("-w")
    result = subprocess.run(command, capture_output=True, text=True)
    if result.returncode != 0:
        return None
    value = result.stdout.rstrip("\r\n")
    return value or None


def keychain_store(service: str, account: str, secret: str, label: str) -> None:
    if sys.platform != "darwin":
        raise ExportError("Saving credentials in Keychain requires macOS.")
    helper = Path(__file__).with_name("keychain_store.swift")
    if not helper.is_file():
        raise ExportError("The macOS Keychain helper is missing.")
    payload = json.dumps(
        {"service": service, "account": account, "label": label, "secret": secret}
    )
    try:
        result = subprocess.run(
            ["/usr/bin/swift", str(helper)],
            input=payload,
            capture_output=True,
            text=True,
            timeout=60,
        )
    except subprocess.TimeoutExpired as exc:
        raise ExportError("Timed out while saving credentials in macOS Keychain.") from exc
    if result.returncode != 0:
        raise ExportError("Could not save credentials in macOS Keychain.")


def load_keychain_credentials() -> tuple[str, str] | None:
    email = keychain_find(KEYCHAIN_ACCOUNT_SERVICE, "default")
    if not email:
        return None
    password = keychain_find(KEYCHAIN_PASSWORD_SERVICE, email)
    if not password:
        return None
    return email, password


def save_keychain_credentials(email: str, password: str) -> None:
    keychain_store(
        KEYCHAIN_PASSWORD_SERVICE,
        email,
        password,
        "Procare exporter password",
    )
    keychain_store(
        KEYCHAIN_ACCOUNT_SERVICE,
        "default",
        email,
        "Procare exporter default account",
    )


def prompt_credentials(
    use_mac_dialog: bool,
    *,
    save_keychain: bool = False,
    no_keychain: bool = False,
) -> tuple[str, str]:
    if not save_keychain and not no_keychain:
        saved = load_keychain_credentials()
        if saved:
            print(f"Using Procare credentials from macOS Keychain for {saved[0]}.")
            return saved

    storage_note = "saved in macOS Keychain" if save_keychain else "not saved"
    if use_mac_dialog:
        email = mac_dialog("Procare email:")
        password = mac_dialog(f"Procare password ({storage_note}):", hidden=True)
    else:
        email = input("Procare email: ").strip()
        password = getpass.getpass(f"Procare password ({storage_note}): ")
    if not email:
        raise ExportError("Email is required.")
    if not password:
        raise ExportError("Password is required.")
    email = email.strip()
    return email, password


def main(argv: list[str] | None = None) -> int:
    os.umask(0o077)
    args = build_parser().parse_args(argv)
    if args.days < 1 or args.days > 3660:
        raise ExportError("--days must be between 1 and 3660.")

    output = args.output.expanduser().resolve()
    existing_manifest = load_manifest(output, required=args.incremental)
    end = args.until or date.today()
    if args.incremental and args.since is None:
        start = incremental_query_start(existing_manifest, end)
    else:
        start = args.since or (end - timedelta(days=args.days - 1))
    if start > end:
        raise ExportError("The start date must not be after the end date.")
    mode = "incremental update" if args.incremental else "gallery export"
    print(f"Procare {mode}: {start.isoformat()} through {end.isoformat()} (inclusive)")
    email, password = prompt_credentials(
        args.mac_dialog,
        save_keychain=args.save_keychain,
        no_keychain=args.no_keychain,
    )

    token, api_base = authenticate(email, password)
    if args.save_keychain:
        save_keychain_credentials(email, password)
        print(f"Saved verified Procare credentials in macOS Keychain for {email}.")
    password = ""  # Drop our reference as soon as authentication completes.
    available = fetch_children(token, api_base)
    selected = select_children(available, args.children)
    child_folders = resolve_child_folders(selected, args.child_folder)
    backfill = False
    if args.incremental:
        backfill = incremental_requires_backfill(
            existing_manifest,
            selected,
            child_folders,
        )
        if backfill:
            start = manifest_archive_start(existing_manifest)
            if start > end:
                raise ExportError("The archive start date must not be after the end date.")
            print(
                "Child selection or folder mapping changed; backfilling the complete "
                "archive from "
                f"{start.isoformat()} before resuming incremental updates."
            )
    print("Children: " + ", ".join(child.name for child in selected))
    print("Reading photo and video metadata...")
    known_keys = manifest_keys(existing_manifest)
    if args.incremental and not backfill:
        discovered, missing_dates = collect_gallery_incremental(
            token,
            api_base,
            selected,
            start,
            end,
            known_keys,
        )
        entries = {key: entry for key, entry in discovered.items() if key not in known_keys}
        print(f"New media discovered: {len(entries)}")
    else:
        entries, missing_dates = collect_gallery(token, api_base, selected, start, end)
        if backfill:
            print(f"Media checked during child backfill: {len(entries)}")
    token = ""  # Media downloads use signed URLs and never need the bearer token.

    bucket_counts: dict[str, dict[str, int]] = {}
    assigned: list[tuple[MediaEntry, str, list[str], list[str]]] = []
    for entry in sorted(entries.values(), key=lambda value: (value.captured_at, value.kind, value.ident)):
        bucket, child_names = assigned_bucket(entry, selected)
        entry_children = assigned_children(entry, selected)
        folder_names = [child_folders[child.id] for child in entry_children]
        counts = bucket_counts.setdefault(bucket, {"photos": 0, "videos": 0})
        counts[f"{entry.kind}s"] += 1
        assigned.append((entry, bucket, child_names, folder_names))

    for bucket, counts in sorted(bucket_counts.items()):
        print(f"  {bucket}: {counts['photos']} photos, {counts['videos']} videos")
    if missing_dates:
        print(f"  Skipped {missing_dates} item(s) with no usable capture date.")
    if args.inventory_report:
        report = inventory_report(entries.values(), selected, start, end, missing_dates)
        report_path = write_private_json(args.inventory_report, report)
        print(f"Inventory report: {report_path}")
    if args.dry_run:
        print("Dry run complete; no media files were written.")
        return 0

    output.mkdir(parents=True, exist_ok=True, mode=0o700)
    manifest_items = []
    downloaded = existed = failed = 0
    for index, (entry, bucket, child_names, folder_names) in enumerate(assigned, start=1):
        print(f"[{index}/{len(assigned)}] {bucket}: {entry.captured_at:%Y-%m-%d} {entry.kind}")
        try:
            result = download_media(
                entry,
                bucket,
                child_names,
                output,
                folder_names=folder_names,
            )
        except ExportError as exc:
            failed += 1
            print(f"  Warning: {exc}", file=sys.stderr)
            continue
        downloaded += int(not result.existed)
        existed += int(result.existed)
        manifest_items.append(
            {
                "bucket": bucket,
                "children": child_names,
                "source_children": source_child_names(entry, selected),
                "captured_at": entry.captured_at.isoformat(timespec="seconds"),
                "kind": entry.kind,
                "media_id": entry.ident,
                "path": str(result.path.relative_to(output)),
                "paths": [str(path.relative_to(output)) for path in result.paths],
                "sha256": result.sha256,
                "size": result.size,
            }
        )
        if not result.existed:
            time.sleep(POLITE_DELAY)

    merged_items = merge_manifest_items(existing_manifest, manifest_items)
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    existing_start = (
        str(existing_manifest.get("date_from")) if existing_manifest else start.isoformat()
    )
    archive_photos = sum(item.get("kind") == "photo" for item in merged_items)
    archive_videos = sum(item.get("kind") == "video" for item in merged_items)
    manifest = {
        "layout": "flat-child-folders-v1",
        "created_at": (
            str(existing_manifest.get("created_at")) if existing_manifest else now
        ),
        "updated_at": now,
        "date_from": min(existing_start, start.isoformat()),
        "date_to": manifest_checkpoint_end(existing_manifest, start, end, failed),
        "children": [
            {"id": child.id, "name": child.name, "folder": child_folders[child.id]}
            for child in selected
        ],
        "items": merged_items,
        "summary": {
            "archive_items": len(merged_items),
            "archive_photos": archive_photos,
            "archive_videos": archive_videos,
            "downloaded": downloaded,
            "already_present": existed,
            "failed": failed,
            "skipped_missing_date": missing_dates,
        },
    }
    manifest_path = write_manifest(output, manifest)
    print(
        f"Complete: {downloaded} downloaded, {existed} already present, "
        f"{failed} failed. Manifest: {manifest_path}"
    )
    return 1 if failed else 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except ExportError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        raise SystemExit(2) from exc