From 8333295a47c2a184cfb16b71afa27266a3e31780 Mon Sep 17 00:00:00 2001 From: Daniel Date: Sat, 19 Sep 2026 10:06:16 +0000 Subject: [PATCH 1/5] feat(client-workflow): watch-upload/watch-download, check-remote-exists, per-file progress (closes #39) --- README.md | 6 +- immich-admin-cli/SKILL.md | 53 +- internal/commands/assets.go | 1 + internal/commands/assets_check.go | 134 +++++ internal/commands/assets_download.go | 52 +- internal/commands/assets_upload.go | 12 +- internal/commands/clientworkflow.go | 4 + internal/commands/clientworkflow_watch.go | 306 +++++++++++ .../commands/clientworkflow_watch_test.go | 19 + internal/commands/skill_best_practices.md | 7 +- internal/workflows/albumdownload.go | 200 ++++++-- internal/workflows/bulkcheck.go | 140 ++++++ internal/workflows/progress.go | 168 +++++++ internal/workflows/repairassets.go | 10 +- internal/workflows/upload.go | 161 +++--- internal/workflows/watch_test.go | 77 +++ internal/workflows/watchdownload.go | 186 +++++++ internal/workflows/watchupload.go | 474 ++++++++++++++++++ 18 files changed, 1884 insertions(+), 126 deletions(-) create mode 100644 internal/commands/assets_check.go create mode 100644 internal/commands/clientworkflow_watch.go create mode 100644 internal/commands/clientworkflow_watch_test.go create mode 100644 internal/workflows/bulkcheck.go create mode 100644 internal/workflows/progress.go create mode 100644 internal/workflows/watch_test.go create mode 100644 internal/workflows/watchdownload.go create mode 100644 internal/workflows/watchupload.go diff --git a/README.md b/README.md index ea50052..f51d922 100644 --- a/README.md +++ b/README.md @@ -601,7 +601,7 @@ This downloads every asset in the "Rome 2026" album into `Nextcloud\Shares\Rome -**27 of 250 endpoints implemented** (24 deprecated and 2 internal endpoints omitted per project policy). +**28 of 250 endpoints implemented** (24 deprecated and 2 internal endpoints omitted per project policy).
API keys (0/6) @@ -663,13 +663,13 @@ This downloads every asset in the "Rome 2026" album into `Nextcloud\Shares\Rome
-Assets (6/24) +Assets (7/24) | Impl | Method | Path | Operation | State | |:----:|--------|------|-----------|-------| | ✅ | DELETE | `/assets` | `deleteAssets` | Stable | | ✅ | POST | `/assets` | `uploadAsset` | Stable | -| | POST | `/assets/bulk-upload-check` | `checkBulkUpload` | Stable | +| ✅ | POST | `/assets/bulk-upload-check` | `checkBulkUpload` | Stable | | ✅ | PUT | `/assets/copy` | `copyAsset` | Stable | | | POST | `/assets/jobs` | `runAssetJobs` | Stable | | | DELETE | `/assets/metadata` | `deleteBulkAssetMetadata` | Beta | diff --git a/immich-admin-cli/SKILL.md b/immich-admin-cli/SKILL.md index 98dba35..feaa0ac 100644 --- a/immich-admin-cli/SKILL.md +++ b/immich-admin-cli/SKILL.md @@ -41,6 +41,7 @@ Download original asset files (GET /assets/{id}/original) Args: `[ASSET_ID ...]` - `--ids-file` (string): read IDs from `FILE`, one UUID per line ('-' for stdin; '#' or '//' starts a comment) - `--out-dir` (string): directory to save downloaded files into (default: current working directory) +- `--quiet` (bool): disable per-file progress bars on stderr ### assets download-thumbnail @@ -50,6 +51,7 @@ Args: `[ASSET_ID ...]` - `--out-dir` (string): directory to save downloaded files into (default: current working directory) - `--size` (string): media size: fullsize, preview, or thumbnail (see the AssetMediaSize spec enum; 'original' is not accepted here — the spec deprecates size=original on this endpoint, use 'assets download-original' instead) [default: "preview"] - `--edited` (bool): return the edited version of the asset if available +- `--quiet` (bool): disable per-file progress bars on stderr ### assets upload @@ -67,6 +69,16 @@ Args: `[FILE ...]` - `--slug` (string): shared-link slug (query param) - `--checksum` (string): sha1 checksum for duplicate detection before upload (x-immich-checksum header) - `--json` (bool): print results as a JSON array +- `--quiet` (bool): disable per-file progress bars on stderr (--json implies quiet) + +### assets check-remote-exists (alias: check-bulk-upload) + +Check if local files already exist on the server via checksum (POST /assets/bulk-upload-check) +Args: `[FILE|DIR ...]` +- `--json` (bool): print results as a JSON array +- `--duplicates-only` (bool): show only files already on the server +- `--missing-only` (bool): show only files missing on the server +- `--ids-only, -q` (bool): print only duplicate asset IDs, one per line (pipeable into albums add-assets / tags tag) ### assets delete @@ -355,6 +367,7 @@ Download all original files or a smaller variant (preview/thumbnail/fullsize) fr - `--resize-video-preset` (string): re-encode every downloaded VIDEO asset using ffmpeg (path from config tools.ffmpeg_path or IMMICH_FFMPEG_PATH, falling back to PATH); videos are always fetched at --size original for this regardless of --size (many videos have no usable preview/thumbnail rendition, and ffmpeg needs the real stream anyway) — only non-video assets use --size; valid presets: 1080p-web-friendly - `--dry-run` (bool): print the planned downloads/deletions without changing anything - `--yes` (bool): skip the confirmation prompt before deleting local files (--sync only) +- `--quiet` (bool): disable per-file progress bars on stderr ### client-workflow merge-album @@ -365,6 +378,39 @@ Move every asset from one album into another, optionally deleting the emptied so - `--dry-run` (bool): print the merge plan without changing anything - `--yes` (bool): skip the confirmation prompt before moving assets +### client-workflow watch-upload + +Watch a folder and auto-upload new stable files (POST /assets/bulk-upload-check + POST /assets) +- `--watch-dir` (string): local directory to watch [required] +- `--mode` (string): flat or by-subfolder [default: "flat"] +- `--tag-pattern` (string): tag pattern for --mode flat ({yyyy-MM-dd} = upload day) [default: "immich-admin-cli/watch/{yyyy-MM-dd}"] +- `--depth` (int): subfolder depth (v1: only 1 supported) [default: 1] +- `--album-id` (string): opt-in album `ID` to also add uploads to +- `--album-name` (string): opt-in album name to also add uploads to (created unless --dry-run) +- `--interval` (string): poll interval (e.g. 60s); <=0 means run once [default: "60s"] +- `--stable-for` (string): defer files changed within this long (e.g. 30s) [default: "30s"] +- `--once` (bool): run a single scan and exit (for cron) +- `--dry-run` (bool): print what would be uploaded/linked without changing anything +- `--yes` (bool): skip creation prompts for tags/albums +- `--quiet` (bool): disable per-file progress bars on stderr (--json implies quiet) +- `--json` (bool): print per-interval stats as JSON on stdout + +### client-workflow watch-download + +Keep a local folder in sync from an album or tag (loop around download-album --sync) +- `--album-id` (string): album `ID` source (mutually exclusive with --album-name/--tag-*) +- `--album-name` (string): album name source +- `--tag-id` (string): tag `ID` source (mutually exclusive with --tag-value/album-*) +- `--tag-value` (string): tag value (full path) source +- `--target-dir` (string): local directory to sync into (created if missing) [required] +- `--size` (string): media variant: original, fullsize, preview, or thumbnail [default: "original"] +- `--interval` (string): poll interval (e.g. 300s); <=0 means run once [default: "300s"] +- `--once` (bool): run a single sync and exit (for cron) +- `--dry-run` (bool): print the planned sync without changing anything +- `--yes` (bool): skip the deletion confirmation prompt +- `--quiet` (bool): disable per-file progress bars on stderr (--json implies quiet) +- `--json` (bool): print per-interval stats as JSON on stdout + ## immich-workflow Server-side workflow operations @@ -520,4 +566,9 @@ albums separate, or rename instead of merging blindly. Always run `client-workflow tag-delete --include/--exclude` for regex bulk deletion (permanent — dry-run first). - **Upload**: `assets upload FILE...` (timestamps default to file mtime; - `--sidecar`/`--filename` are single-file only). + `--sidecar`/`--filename` are single-file only; per-file byte bar on stderr, `--quiet` disables, `--json` implies quiet). + `assets check-remote-exists FILE|DIR...` (alias `check-bulk-upload`) hashes locally and reports + `uploaded ` / `missing` / `unsupported` without mutating (`--json`, `--ids-only -q` pipes duplicate IDs into `albums add-assets`/`tags tag`). +- **Watch**: `client-workflow watch-upload --watch-dir DIR --mode flat|by-subfolder [--tag-pattern "immich-admin-cli/watch/{yyyy-MM-dd}"] [--album-id|--album-name] --interval 60s --stable-for 30s [--once] [--dry-run] [--yes]` + polls and uploads only stable files (bulk-check first; duplicates only linked). `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` + loops the `.immich-sync.json` manifest sync (album or tag source). `--interval <=0` means run once. Per-interval one-line summary on stderr; `--json` prints JSON stats on stdout. diff --git a/internal/commands/assets.go b/internal/commands/assets.go index e4f963b..53eab71 100644 --- a/internal/commands/assets.go +++ b/internal/commands/assets.go @@ -38,6 +38,7 @@ func Assets() *cli.Command { assetsDownloadCommand(), assetsDownloadThumbnailCommand(), assetsUploadCommand(), + assetsCheckRemoteExistsCommand(), assetsDeleteCommand(), assetsCopyCommand(), }, diff --git a/internal/commands/assets_check.go b/internal/commands/assets_check.go new file mode 100644 index 0000000..fb75b4b --- /dev/null +++ b/internal/commands/assets_check.go @@ -0,0 +1,134 @@ +package commands + +import ( + "context" + "encoding/json" + "fmt" + "os" + + "github.com/urfave/cli/v3" + + "github.com/dhcgn/immich-admin-cli/internal/workflows" +) + +// assetsCheckRemoteExistsCommand exposes POST /assets/bulk-upload-check +// (checkBulkUpload) as `assets check-remote-exists` (alias check-bulk-upload). +func assetsCheckRemoteExistsCommand() *cli.Command { + return &cli.Command{ + Name: "check-remote-exists", + Aliases: []string{"check-bulk-upload"}, + Usage: "Check if local files already exist on the server via checksum (POST /assets/bulk-upload-check)", + Description: "Hashes each local file (base64 SHA1, the format Immich reports " + + "as Checksum) and asks the server which checksums it already has. " + + "Read-only: never uploads, downloads, or mutates anything.", + ArgsUsage: "[FILE|DIR ...]", + Flags: []cli.Flag{ + &cli.BoolFlag{Name: "json", Usage: "print results as a JSON array"}, + &cli.BoolFlag{Name: "duplicates-only", Usage: "show only files already on the server"}, + &cli.BoolFlag{Name: "missing-only", Usage: "show only files missing on the server"}, + &cli.BoolFlag{Name: "ids-only", Aliases: []string{"q"}, Usage: "print only duplicate asset IDs, one per line (pipeable into albums add-assets / tags tag)"}, + }, + Action: assetsCheckRemoteExists, + } +} + +type checkRemoteResult struct { + File string `json:"file"` + Checksum string `json:"checksum"` + Status string `json:"status"` + AssetID string `json:"assetId,omitempty"` + Trashed bool `json:"isTrashed,omitempty"` +} + +func assetsCheckRemoteExists(ctx context.Context, cmd *cli.Command) error { + paths := cmd.Args().Slice() + if len(paths) == 0 { + return fmt.Errorf("no files given: pass [FILE|DIR ...]") + } + files, err := workflows.CollectCheckFiles(paths) + if err != nil { + return err + } + if len(files) == 0 { + return fmt.Errorf("no files found in %v (hidden files/symlinks are skipped)", paths) + } + + c, err := newClient(ctx, cmd) + if err != nil { + return err + } + + // Hash locally first so one unreadable file doesn't abort the server check. + checksums := make(map[string]string, len(files)) + var hashable []string + failures := 0 + for i, f := range files { + fmt.Fprintf(os.Stderr, "[%d/%d] hashing %s\n", i+1, len(files), f) + sum, herr := workflows.FileSHA1Base64(f) + if herr != nil { + fmt.Fprintf(os.Stderr, "Error: file %s: %v\n", f, herr) + failures++ + continue + } + checksums[f] = sum + hashable = append(hashable, f) + } + + entries, err := workflows.CheckBulkUploadChecksums(ctx, c, hashable, checksums) + if err != nil { + return err + } + + dupsOnly := cmd.Bool("duplicates-only") + missOnly := cmd.Bool("missing-only") + idsOnly := cmd.Bool("ids-only") + + var results []checkRemoteResult + for _, e := range entries { + switch { + case idsOnly && e.Status != workflows.BulkCheckUploaded: + continue + case dupsOnly && e.Status != workflows.BulkCheckUploaded: + continue + case missOnly && e.Status != workflows.BulkCheckMissing: + continue + } + r := checkRemoteResult{File: e.File, Checksum: e.Checksum, Status: string(e.Status)} + if e.AssetID != nil { + r.AssetID = e.AssetID.String() + } + r.Trashed = e.IsTrashed + results = append(results, r) + } + + if idsOnly { + for _, r := range results { + fmt.Println(r.AssetID) + } + } else if cmd.Bool("json") { + if results == nil { + results = []checkRemoteResult{} + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + if err := enc.Encode(results); err != nil { + return err + } + } else { + for _, r := range results { + switch r.Status { + case string(workflows.BulkCheckUploaded): + fmt.Printf("%s: uploaded %s\n", r.File, r.AssetID) + case string(workflows.BulkCheckMissing): + fmt.Printf("%s: missing\n", r.File) + default: + fmt.Printf("%s: unsupported\n", r.File) + } + } + } + + if failures > 0 { + return fmt.Errorf("%d of %d files failed", failures, len(files)) + } + return nil +} diff --git a/internal/commands/assets_download.go b/internal/commands/assets_download.go index fe7b0b0..99d32b5 100644 --- a/internal/commands/assets_download.go +++ b/internal/commands/assets_download.go @@ -30,6 +30,7 @@ func assetsDownloadCommand() *cli.Command { Name: "out-dir", Usage: "directory to save downloaded files into (default: current working directory)", }, + &cli.BoolFlag{Name: "quiet", Usage: "disable per-file progress bars on stderr"}, }, Action: assetsDownloadOriginal, } @@ -56,8 +57,10 @@ func assetsDownloadOriginal(ctx context.Context, cmd *cli.Command) error { // Fan out over the single-asset download endpoint: continue on per-ID // errors and report failures at the end (same convention as assetsInfo). failures := 0 - for _, id := range ids { - if err := downloadOneAsset(ctx, c, id, outDir); err != nil { + quiet := cmd.Bool("quiet") + for i, id := range ids { + prog := workflows.NewByteProgress(id.String(), -1, i+1, len(ids), quiet) + if err := downloadOneAsset(ctx, c, id, outDir, prog); err != nil { fmt.Fprintf(os.Stderr, "Error: asset %s: %v\n", id, err) failures++ continue @@ -70,7 +73,7 @@ func assetsDownloadOriginal(ctx context.Context, cmd *cli.Command) error { return nil } -func downloadOneAsset(ctx context.Context, c *client.Client, id openapi_types.UUID, outDir string) error { +func downloadOneAsset(ctx context.Context, c *client.Client, id openapi_types.UUID, outDir string, prog *workflows.ByteProgress) error { infoResp, err := c.API.GetAssetInfoWithResponse(ctx, id, nil) if err == nil { err = client.Check(infoResp, http.StatusOK) @@ -97,13 +100,28 @@ func downloadOneAsset(ctx context.Context, c *client.Client, id openapi_types.UU idStr := id.String() destPath := filepath.Join(outDir, idStr+"_"+originalFileName) + if prog != nil { + prog.Label = originalFileName + prog.Total = resp.ContentLength + if prog.Total <= 0 { + prog.Total = -1 + } + } + f, err := os.Create(destPath) if err != nil { return fmt.Errorf("creating output file %q: %w", destPath, err) } defer f.Close() - written, err := io.Copy(f, resp.Body) + var src io.Reader = resp.Body + if prog != nil { + src = prog.Wrap(resp.Body) + } + written, err := io.Copy(f, src) + if prog != nil { + prog.Finish() + } if err != nil { return fmt.Errorf("writing output file %q: %w", destPath, err) } @@ -149,6 +167,7 @@ func assetsDownloadThumbnailCommand() *cli.Command { Name: "edited", Usage: "return the edited version of the asset if available", }, + &cli.BoolFlag{Name: "quiet", Usage: "disable per-file progress bars on stderr"}, }, Action: assetsDownloadThumbnail, } @@ -187,8 +206,10 @@ func assetsDownloadThumbnail(ctx context.Context, cmd *cli.Command) error { // Fan out over the single-asset endpoint: continue on per-ID errors and // report failures at the end (same convention as assetsDownloadOriginal). failures := 0 - for _, id := range ids { - if err := downloadOneThumbnail(ctx, c, id, outDir, params); err != nil { + quiet := cmd.Bool("quiet") + for i, id := range ids { + prog := workflows.NewByteProgress(id.String(), -1, i+1, len(ids), quiet) + if err := downloadOneThumbnail(ctx, c, id, outDir, params, prog); err != nil { fmt.Fprintf(os.Stderr, "Error: asset %s: %v\n", id, err) failures++ continue @@ -201,7 +222,7 @@ func assetsDownloadThumbnail(ctx context.Context, cmd *cli.Command) error { return nil } -func downloadOneThumbnail(ctx context.Context, c *client.Client, id openapi_types.UUID, outDir string, params *immichapi.ViewAssetParams) error { +func downloadOneThumbnail(ctx context.Context, c *client.Client, id openapi_types.UUID, outDir string, params *immichapi.ViewAssetParams, prog *workflows.ByteProgress) error { infoResp, err := c.API.GetAssetInfoWithResponse(ctx, id, nil) if err == nil { err = client.Check(infoResp, http.StatusOK) @@ -232,13 +253,28 @@ func downloadOneThumbnail(ctx context.Context, c *client.Client, id openapi_type idStr := id.String() destPath := filepath.Join(outDir, idStr+"_"+baseName+ext) + if prog != nil { + prog.Label = baseName + ext + prog.Total = resp.ContentLength + if prog.Total <= 0 { + prog.Total = -1 + } + } + f, err := os.Create(destPath) if err != nil { return fmt.Errorf("creating output file %q: %w", destPath, err) } defer f.Close() - written, err := io.Copy(f, resp.Body) + var src io.Reader = resp.Body + if prog != nil { + src = prog.Wrap(resp.Body) + } + written, err := io.Copy(f, src) + if prog != nil { + prog.Finish() + } if err != nil { return fmt.Errorf("writing output file %q: %w", destPath, err) } diff --git a/internal/commands/assets_upload.go b/internal/commands/assets_upload.go index 7308877..d30893e 100644 --- a/internal/commands/assets_upload.go +++ b/internal/commands/assets_upload.go @@ -37,6 +37,7 @@ func assetsUploadCommand() *cli.Command { &cli.StringFlag{Name: "slug", Usage: "shared-link slug (query param)"}, &cli.StringFlag{Name: "checksum", Usage: "sha1 checksum for duplicate detection before upload (x-immich-checksum header)"}, &cli.BoolFlag{Name: "json", Usage: "print results as a JSON array"}, + &cli.BoolFlag{Name: "quiet", Usage: "disable per-file progress bars on stderr (--json implies quiet)"}, }, Action: assetsUpload, } @@ -72,8 +73,17 @@ func assetsUpload(ctx context.Context, cmd *cli.Command) error { var results []uploadResult failures := 0 - for _, f := range files { + quiet := cmd.Bool("quiet") || cmd.Bool("json") + for i, f := range files { + fi, serr := os.Stat(f) + var total int64 = -1 + if serr == nil && !fi.IsDir() { + total = fi.Size() + } + prog := workflows.NewByteProgress(f, total, i+1, len(files), quiet) + opts.Progress = prog id, err := workflows.UploadAssetFile(ctx, c, f, opts) + prog.Finish() if err != nil { fmt.Fprintf(os.Stderr, "Error: file %s: %v\n", f, err) failures++ diff --git a/internal/commands/clientworkflow.go b/internal/commands/clientworkflow.go index 01ad04e..7f22c8d 100644 --- a/internal/commands/clientworkflow.go +++ b/internal/commands/clientworkflow.go @@ -43,6 +43,8 @@ func ClientWorkflow() *cli.Command { fixAlbumDatesCommand(), downloadAlbumCommand(), mergeAlbumCommand(), + watchUploadCommand(), + watchDownloadCommand(), }, } } @@ -1395,6 +1397,7 @@ func downloadAlbumCommand() *cli.Command { Name: "yes", Usage: "skip the confirmation prompt before deleting local files (--sync only)", }, + &cli.BoolFlag{Name: "quiet", Usage: "disable per-file progress bars on stderr"}, }, Action: clientWorkflowDownloadAlbum, } @@ -1515,6 +1518,7 @@ func clientWorkflowDownloadAlbum(ctx context.Context, cmd *cli.Command) error { Resize: resizeOpts, ResizeVideo: resizeVideoOpts, TimestampPrefix: cmd.Bool("timestamp-prefix"), + Quiet: cmd.Bool("quiet"), } // Printed before any network activity (newClient/ResolveAlbum below), diff --git a/internal/commands/clientworkflow_watch.go b/internal/commands/clientworkflow_watch.go new file mode 100644 index 0000000..d92ef9d --- /dev/null +++ b/internal/commands/clientworkflow_watch.go @@ -0,0 +1,306 @@ +package commands + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/google/uuid" + openapi_types "github.com/oapi-codegen/runtime/types" + "github.com/urfave/cli/v3" + + "github.com/dhcgn/immich-admin-cli/internal/client" + "github.com/dhcgn/immich-admin-cli/internal/immichapi" + "github.com/dhcgn/immich-admin-cli/internal/workflows" +) + +// tagValueByID looks up a tag's full-path value by ID (for display). +func tagValueByID(ctx context.Context, c *client.Client, id openapi_types.UUID) (string, error) { + resp, err := c.API.GetTagByIdWithResponse(ctx, id) + if err == nil { + err = client.Check(resp, http.StatusOK) + } + if err != nil { + return "", err + } + if resp.JSON200 == nil { + return "", fmt.Errorf("tag %s: response had no body", id) + } + return resp.JSON200.Value, nil +} + +func watchUploadCommand() *cli.Command { + return &cli.Command{ + Name: "watch-upload", + Usage: "Watch a folder and auto-upload new stable files (POST /assets/bulk-upload-check + POST /assets)", + Description: "Polls --watch-dir every --interval (default 60s) and uploads new files. " + + "Files whose size/mtime changed within --stable-for (default 30s) are deferred to the next interval. " + + "Duplicates (bulk-upload-check reject/duplicate with assetId) are only linked to the tag/album, never re-uploaded. " + + "--mode flat tags everything with --tag-pattern (default immich-admin-cli/watch/{yyyy-MM-dd}); " + + "--mode by-subfolder uses the raw subfolder name. --album-id/--album-name is opt-in. --once runs a single scan (for cron).", + Flags: []cli.Flag{ + &cli.StringFlag{Name: "watch-dir", Usage: "local directory to watch", Required: true}, + &cli.StringFlag{Name: "mode", Usage: "flat or by-subfolder", Value: "flat"}, + &cli.StringFlag{Name: "tag-pattern", Usage: "tag pattern for --mode flat ({yyyy-MM-dd} = upload day)", Value: workflows.DefaultWatchUploadTagPattern}, + &cli.IntFlag{Name: "depth", Usage: "subfolder depth (v1: only 1 supported)", Value: 1}, + &cli.StringFlag{Name: "album-id", Usage: "opt-in album `ID` to also add uploads to"}, + &cli.StringFlag{Name: "album-name", Usage: "opt-in album name to also add uploads to (created unless --dry-run)"}, + &cli.StringFlag{Name: "interval", Usage: "poll interval (e.g. 60s); <=0 means run once", Value: "60s"}, + &cli.StringFlag{Name: "stable-for", Usage: "defer files changed within this long (e.g. 30s)", Value: "30s"}, + &cli.BoolFlag{Name: "once", Usage: "run a single scan and exit (for cron)"}, + &cli.BoolFlag{Name: "dry-run", Usage: "print what would be uploaded/linked without changing anything"}, + &cli.BoolFlag{Name: "yes", Usage: "skip creation prompts for tags/albums"}, + &cli.BoolFlag{Name: "quiet", Usage: "disable per-file progress bars on stderr (--json implies quiet)"}, + &cli.BoolFlag{Name: "json", Usage: "print per-interval stats as JSON on stdout"}, + }, + Action: clientWorkflowWatchUpload, + } +} + +func watchDownloadCommand() *cli.Command { + return &cli.Command{ + Name: "watch-download", + Usage: "Keep a local folder in sync from an album or tag (loop around download-album --sync)", + Description: "Polls exactly one source (--album-id|--album-name or --tag-id|--tag-value) every --interval " + + "(default 300s) and mirrors it into --target-dir using the .immich-sync.json manifest: skip unchanged, " + + "re-download changed, delete locals whose asset left the source, never touch untracked files. --once runs a single scan.", + Flags: []cli.Flag{ + &cli.StringFlag{Name: "album-id", Usage: "album `ID` source (mutually exclusive with --album-name/--tag-*)"}, + &cli.StringFlag{Name: "album-name", Usage: "album name source"}, + &cli.StringFlag{Name: "tag-id", Usage: "tag `ID` source (mutually exclusive with --tag-value/album-*)"}, + &cli.StringFlag{Name: "tag-value", Usage: "tag value (full path) source"}, + &cli.StringFlag{Name: "target-dir", Usage: "local directory to sync into (created if missing)", Required: true}, + &cli.StringFlag{Name: "size", Usage: "media variant: original, fullsize, preview, or thumbnail", Value: string(immichapi.AssetMediaSizeOriginal)}, + &cli.StringFlag{Name: "interval", Usage: "poll interval (e.g. 300s); <=0 means run once", Value: "300s"}, + &cli.BoolFlag{Name: "once", Usage: "run a single sync and exit (for cron)"}, + &cli.BoolFlag{Name: "dry-run", Usage: "print the planned sync without changing anything"}, + &cli.BoolFlag{Name: "yes", Usage: "skip the deletion confirmation prompt"}, + &cli.BoolFlag{Name: "quiet", Usage: "disable per-file progress bars on stderr (--json implies quiet)"}, + &cli.BoolFlag{Name: "json", Usage: "print per-interval stats as JSON on stdout"}, + }, + Action: clientWorkflowWatchDownload, + } +} + +func parseIntervalOrOnce(raw string, once bool) (time.Duration, bool, error) { + if raw == "" { + return 0, true, nil + } + d, err := time.ParseDuration(raw) + if err != nil { + return 0, false, fmt.Errorf("invalid --interval %q: %w (e.g. 60s, 5m)", raw, err) + } + if d <= 0 { + return 0, true, nil + } + return d, once, nil +} + +func clientWorkflowWatchUpload(ctx context.Context, cmd *cli.Command) error { + if cmd.Args().Len() > 0 { + return fmt.Errorf("watch-upload takes no positional arguments (got %v)", cmd.Args().Slice()) + } + mode := cmd.String("mode") + if mode != "flat" && mode != "by-subfolder" { + return fmt.Errorf("invalid --mode %q: must be flat or by-subfolder", mode) + } + stableFor, err := time.ParseDuration(cmd.String("stable-for")) + if err != nil { + return fmt.Errorf("invalid --stable-for %q: %w", cmd.String("stable-for"), err) + } + if stableFor < 0 { + return fmt.Errorf("invalid --stable-for %q: must not be negative", cmd.String("stable-for")) + } + interval, once, err := parseIntervalOrOnce(cmd.String("interval"), cmd.Bool("once")) + if err != nil { + return err + } + var albumID *openapi_types.UUID + if s := cmd.String("album-id"); s != "" { + id, err := uuid.Parse(s) + if err != nil { + return fmt.Errorf("invalid --album-id %q: %w", s, err) + } + uid := openapi_types.UUID(id) + albumID = &uid + } + if cmd.String("album-id") != "" && cmd.String("album-name") != "" { + return fmt.Errorf("only one of --album-id or --album-name may be given") + } + + c, err := newClient(ctx, cmd) + if err != nil { + return err + } + + opts := workflows.WatchUploadOptions{ + WatchDir: cmd.String("watch-dir"), + Mode: mode, + Depth: int(cmd.Int("depth")), + TagPattern: cmd.String("tag-pattern"), + AlbumID: albumID, + AlbumName: cmd.String("album-name"), + Interval: interval, + StableFor: stableFor, + Once: once, + DryRun: cmd.Bool("dry-run"), + Yes: cmd.Bool("yes"), + Quiet: cmd.Bool("quiet"), + JSON: cmd.Bool("json"), + } + + ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) + defer stop() + + for { + stats, runErr := workflows.RunWatchUploadOnce(ctx, c, opts) + printWatchUploadSummary(stats, opts.JSON) + if runErr != nil && ctx.Err() == nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", runErr) + } + if once { + return runErr + } + select { + case <-ctx.Done(): + fmt.Fprintln(os.Stderr, "Interrupted.") + return nil + case <-time.After(interval): + } + } +} + +func printWatchUploadSummary(stats workflows.WatchUploadStats, asJSON bool) { + if asJSON { + enc := json.NewEncoder(os.Stdout) + _ = enc.Encode(stats) + return + } + fmt.Fprintf(os.Stderr, "uploaded=%d linked-duplicate=%d skipped-unstable=%d failed=%d\n", + stats.Uploaded, stats.LinkedDuplicate, stats.SkippedUnstable, stats.Failed) +} + +func clientWorkflowWatchDownload(ctx context.Context, cmd *cli.Command) error { + if cmd.Args().Len() > 0 { + return fmt.Errorf("watch-download takes no positional arguments (got %v)", cmd.Args().Slice()) + } + set := 0 + for _, v := range []string{cmd.String("album-id"), cmd.String("album-name"), cmd.String("tag-id"), cmd.String("tag-value")} { + if v != "" { + set++ + } + } + if set != 1 { + return fmt.Errorf("exactly one source is required: --album-id | --album-name | --tag-id | --tag-value") + } + size, err := resolveDownloadAlbumSize(cmd.String("size")) + if err != nil { + return err + } + interval, once, err := parseIntervalOrOnce(cmd.String("interval"), cmd.Bool("once")) + if err != nil { + return err + } + + c, err := newClient(ctx, cmd) + if err != nil { + return err + } + + wopts := workflows.WatchDownloadOptions{ + TargetDir: cmd.String("target-dir"), + Interval: int64(interval.Seconds()), + Once: once, + DryRun: cmd.Bool("dry-run"), + Quiet: cmd.Bool("quiet"), + JSON: cmd.Bool("json"), + + Size: size} + wopts.Quiet = cmd.Bool("quiet") || cmd.Bool("json") + + // Resolve source once so the loop prints real names and fails fast. + switch { + case cmd.String("album-id") != "": + id, err := uuid.Parse(cmd.String("album-id")) + if err != nil { + return fmt.Errorf("invalid --album-id %q: %w", cmd.String("album-id"), err) + } + uid := openapi_types.UUID(id) + album, err := workflows.ResolveAlbum(ctx, c, &uid, "") + if err != nil { + return err + } + wopts.Source = workflows.SyncSource{Kind: workflows.SourceKindAlbum, ID: album.Id.String(), Name: album.AlbumName} + wopts.AlbumID = &album.Id + case cmd.String("album-name") != "": + album, err := resolveDownloadAlbum(ctx, c, nil, cmd.String("album-name"), cmd.Bool("yes")) + if err != nil { + return err + } + wopts.Source = workflows.SyncSource{Kind: workflows.SourceKindAlbum, ID: album.Id.String(), Name: album.AlbumName} + uid := album.Id + wopts.AlbumID = &uid + case cmd.String("tag-id") != "": + id, err := uuid.Parse(cmd.String("tag-id")) + if err != nil { + return fmt.Errorf("invalid --tag-id %q: %w", cmd.String("tag-id"), err) + } + uid := openapi_types.UUID(id) + wopts.Source = workflows.SyncSource{Kind: workflows.SourceKindTag, ID: uid.String()} + wopts.TagID = &uid + if name, nerr := tagValueByID(ctx, c, uid); nerr == nil { + wopts.Source.Name = name + } + case cmd.String("tag-value") != "": + tag, err := workflows.ResolveTagByValue(ctx, c, cmd.String("tag-value")) + if err != nil { + return err + } + wopts.Source = workflows.SyncSource{Kind: workflows.SourceKindTag, ID: tag.Id.String(), Name: tag.Value} + tid := tag.Id + wopts.TagID = &tid + } + + ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) + defer stop() + + for { + plan, runErr := workflows.RunWatchDownloadOnce(ctx, c, wopts) + printWatchDownloadSummary(wopts.Source, plan, wopts.JSON) + if runErr != nil && ctx.Err() == nil { + // ApplySync already prints per-file errors; surface summary. + fmt.Fprintf(os.Stderr, "Error: %v\n", runErr) + } + if wopts.DryRun && (len(plan.Additions) == 0 && len(plan.Updates) == 0 && len(plan.Removals) == 0) { + fmt.Println("Already up to date.") + } + if once { + return runErr + } + select { + case <-ctx.Done(): + fmt.Fprintln(os.Stderr, "Interrupted.") + return nil + case <-time.After(interval): + } + } +} + +func printWatchDownloadSummary(source workflows.SyncSource, plan workflows.SyncPlan, asJSON bool) { + if asJSON { + enc := json.NewEncoder(os.Stdout) + _ = enc.Encode(map[string]any{ + "sourceKind": source.Kind, "source": source.Name, + "add": len(plan.Additions), "update": len(plan.Updates), + "unchanged": len(plan.Unchanged), "remove": len(plan.Removals), + }) + return + } + fmt.Fprintf(os.Stderr, "%s %q: add=%d update=%d unchanged=%d remove=%d\n", + source.Kind, source.Name, len(plan.Additions), len(plan.Updates), len(plan.Unchanged), len(plan.Removals)) +} diff --git a/internal/commands/clientworkflow_watch_test.go b/internal/commands/clientworkflow_watch_test.go new file mode 100644 index 0000000..4d07ea9 --- /dev/null +++ b/internal/commands/clientworkflow_watch_test.go @@ -0,0 +1,19 @@ +package commands + +import ( + "testing" + "time" +) + +func TestParseIntervalOrOnce(t *testing.T) { + d, once, err := parseIntervalOrOnce("60s", false) + if err != nil || once || d != 60*time.Second { + t.Fatalf("60s = %v,%v,%v; want 60s,false,nil", d, once, err) + } + if _, once, err := parseIntervalOrOnce("0s", false); err != nil || !once { + t.Fatalf("0s should mean once: %v,%v", once, err) + } + if _, _, err := parseIntervalOrOnce("bogus", false); err == nil { + t.Error("bogus interval expected error") + } +} diff --git a/internal/commands/skill_best_practices.md b/internal/commands/skill_best_practices.md index e3aa7ff..f3ea146 100644 --- a/internal/commands/skill_best_practices.md +++ b/internal/commands/skill_best_practices.md @@ -115,4 +115,9 @@ albums separate, or rename instead of merging blindly. Always run `client-workflow tag-delete --include/--exclude` for regex bulk deletion (permanent — dry-run first). - **Upload**: `assets upload FILE...` (timestamps default to file mtime; - `--sidecar`/`--filename` are single-file only). + `--sidecar`/`--filename` are single-file only; per-file byte bar on stderr, `--quiet` disables, `--json` implies quiet). + `assets check-remote-exists FILE|DIR...` (alias `check-bulk-upload`) hashes locally and reports + `uploaded ` / `missing` / `unsupported` without mutating (`--json`, `--ids-only -q` pipes duplicate IDs into `albums add-assets`/`tags tag`). +- **Watch**: `client-workflow watch-upload --watch-dir DIR --mode flat|by-subfolder [--tag-pattern "immich-admin-cli/watch/{yyyy-MM-dd}"] [--album-id|--album-name] --interval 60s --stable-for 30s [--once] [--dry-run] [--yes]` + polls and uploads only stable files (bulk-check first; duplicates only linked). `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` + loops the `.immich-sync.json` manifest sync (album or tag source). `--interval <=0` means run once. Per-interval one-line summary on stderr; `--json` prints JSON stats on stdout. diff --git a/internal/workflows/albumdownload.go b/internal/workflows/albumdownload.go index afdfadd..156196b 100644 --- a/internal/workflows/albumdownload.go +++ b/internal/workflows/albumdownload.go @@ -29,20 +29,35 @@ import ( // ManifestFileName is the hidden per-target-directory state file written by // --sync mode (see Manifest). It never appears among the downloaded media // files, and no other command reads or writes it. -const ManifestFileName = ".immich-album-sync.json" +const ManifestFileName = ".immich-sync.json" + +// legacyManifestFileName is the pre-watch filename. LoadManifest falls back +// to it when the new file is missing and SaveManifest migrates forward. +const legacyManifestFileName = ".immich-album-sync.json" const manifestVersion = 1 -// Manifest tracks, for one target directory, the album and files +// Sync source kinds for Manifest.SourceKind. +const ( + SourceKindAlbum = "album" + SourceKindTag = "tag" +) + +// Manifest tracks, for one target directory, the album/tag and files // SyncAlbum/ApplyAlbumSync downloaded there, so later runs can detect // changes (re-download) and removals (delete locally) — and, just as // importantly, so files not tracked here (anything the user placed in the -// folder themselves, or files from an unrelated album) are never touched. +// folder themselves, or files from an unrelated source) are never touched. type Manifest struct { - Version int `json:"version"` - AlbumID string `json:"albumId"` - AlbumName string `json:"albumName"` - Size immichapi.AssetMediaSize `json:"size"` + Version int `json:"version"` + AlbumID string `json:"albumId,omitempty"` + AlbumName string `json:"albumName,omitempty"` + // SourceKind is "album" or "tag". Empty means a legacy album manifest + // (written before tag sources existed); see normalizedSource. + SourceKind string `json:"sourceKind,omitempty"` + SourceID string `json:"sourceId,omitempty"` + SourceName string `json:"sourceName,omitempty"` + Size immichapi.AssetMediaSize `json:"size"` // Resize, ResizeVideoPreset, and TimestampPrefix record whether this // target directory was built with --resize / --resize-video-preset / // --timestamp-prefix, mirroring the Size guard: all change the local @@ -58,6 +73,27 @@ type Manifest struct { Assets map[string]ManifestAsset `json:"assets"` } +// normalizedSource returns the canonical (kind, id, name) triple, +// migrating legacy album-only manifests (AlbumID set, SourceKind empty). +func (m Manifest) normalizedSource() (kind, id, name string) { + if m.SourceKind != "" { + return m.SourceKind, m.SourceID, m.SourceName + } + if m.AlbumID != "" { + return SourceKindAlbum, m.AlbumID, m.AlbumName + } + return "", "", "" +} + +// stampSource sets both the new Source* fields and the legacy Album* +// fields (for album sources, so old readers still see them). +func (m *Manifest) stampSource(kind, id, name string) { + m.SourceKind, m.SourceID, m.SourceName = kind, id, name + if kind == SourceKindAlbum { + m.AlbumID, m.AlbumName = id, name + } +} + // ManifestAsset is one asset SyncAlbum is tracking in a target directory. type ManifestAsset struct { // FileName is the file's name (no directory) inside the target @@ -102,6 +138,9 @@ type DownloadAlbumOptions struct { // DryRun previews the planned actions without downloading, deleting, or // writing the manifest. DryRun bool + // Quiet disables per-file byte progress bars on stderr + // (--json implies quiet at the command layer). + Quiet bool } // DefaultResizeQuality is the JPEG quality used by --resize when @@ -360,15 +399,21 @@ func baseNameOf(fileName string) string { return strings.TrimSuffix(fileName, filepath.Ext(fileName)) } -// LoadManifest reads ManifestFileName from targetDir. A missing file is not -// an error: it returns a zero-value Manifest (with an initialized Assets -// map) and existed=false, the normal state for a brand-new target -// directory or the first ever --sync run against it. +// LoadManifest reads ManifestFileName from targetDir, falling back to the +// legacy album filename. A missing file is not an error: it returns a +// zero-value Manifest (with an initialized Assets map) and existed=false, +// the normal state for a brand-new target directory or the first ever +// --sync run against it. func LoadManifest(targetDir string) (m Manifest, existed bool, err error) { path := filepath.Join(targetDir, ManifestFileName) data, err := os.ReadFile(path) if errors.Is(err, os.ErrNotExist) { - return Manifest{Assets: map[string]ManifestAsset{}}, false, nil + legacy := filepath.Join(targetDir, legacyManifestFileName) + data, err = os.ReadFile(legacy) + if errors.Is(err, os.ErrNotExist) { + return Manifest{Assets: map[string]ManifestAsset{}}, false, nil + } + path = legacy } if err != nil { return Manifest{}, false, fmt.Errorf("reading manifest %q: %w", path, err) @@ -379,6 +424,9 @@ func LoadManifest(targetDir string) (m Manifest, existed bool, err error) { if m.Assets == nil { m.Assets = map[string]ManifestAsset{} } + if m.SourceKind == "" && m.AlbumID != "" { + m.SourceKind, m.SourceID, m.SourceName = SourceKindAlbum, m.AlbumID, m.AlbumName + } return m, true, nil } @@ -390,6 +438,11 @@ func LoadManifest(targetDir string) (m Manifest, existed bool, err error) { // never loses more than the one in-flight item's progress. func SaveManifest(targetDir string, m Manifest) error { m.Version = manifestVersion + if m.SourceKind != "" && m.SourceID != "" && m.SourceKind == SourceKindAlbum { + m.AlbumID, m.AlbumName = m.SourceID, m.SourceName + } else if m.SourceKind == "" && m.AlbumID != "" { + m.SourceKind, m.SourceID, m.SourceName = SourceKindAlbum, m.AlbumID, m.AlbumName + } data, err := json.MarshalIndent(m, "", " ") if err != nil { return fmt.Errorf("encoding manifest: %w", err) @@ -413,6 +466,10 @@ func SaveManifest(targetDir string, m Manifest) error { os.Remove(tmpPath) return fmt.Errorf("writing manifest %q: %w", path, err) } + // Best-effort migration: drop the legacy file once the new one lands. + if path != filepath.Join(targetDir, legacyManifestFileName) { + os.Remove(filepath.Join(targetDir, legacyManifestFileName)) + } return nil } @@ -475,20 +532,25 @@ func FetchFilteredAlbumAssets(ctx context.Context, c *client.Client, albumID ope // variant, per size) and returns its body stream (caller must close it) // plus the file extension to use: the original file's own extension for // immichapi.AssetMediaSizeOriginal, or sniffed from the actual response Content-Type for -// every other variant (see ExtensionForContentType). -func fetchAssetStream(ctx context.Context, c *client.Client, a immichapi.AssetResponseDto, size immichapi.AssetMediaSize) (io.ReadCloser, string, error) { +// every other variant (see ExtensionForContentType). Length is the response +// ContentLength, or -1 when unknown (chunked). +func fetchAssetStream(ctx context.Context, c *client.Client, a immichapi.AssetResponseDto, size immichapi.AssetMediaSize) (io.ReadCloser, string, int64, error) { switch size { case immichapi.AssetMediaSizeOriginal: resp, err := c.API.DownloadAsset(ctx, a.Id, nil) if err != nil { - return nil, "", fmt.Errorf("downloading original: %w", err) + return nil, "", -1, fmt.Errorf("downloading original: %w", err) } if resp.StatusCode != http.StatusOK { defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) - return nil, "", fmt.Errorf("server returned %s (expected 200): %s", resp.Status, strings.TrimSpace(string(body))) + return nil, "", -1, fmt.Errorf("server returned %s (expected 200): %s", resp.Status, strings.TrimSpace(string(body))) + } + length := resp.ContentLength + if length <= 0 { + length = -1 } - return resp.Body, filepath.Ext(a.OriginalFileName), nil + return resp.Body, filepath.Ext(a.OriginalFileName), length, nil case immichapi.AssetMediaSizeFullsize, immichapi.AssetMediaSizePreview, immichapi.AssetMediaSizeThumbnail: // GET /assets/{id}/thumbnail?size=fullsize|preview|thumbnail. Note @@ -499,17 +561,21 @@ func fetchAssetStream(ctx context.Context, c *client.Client, a immichapi.AssetRe requestedSize := size resp, err := c.API.ViewAsset(ctx, a.Id, &immichapi.ViewAssetParams{Size: &requestedSize}) if err != nil { - return nil, "", fmt.Errorf("downloading %s: %w", size, err) + return nil, "", -1, fmt.Errorf("downloading %s: %w", size, err) } if resp.StatusCode != http.StatusOK { defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) - return nil, "", fmt.Errorf("server returned %s (expected 200): %s", resp.Status, strings.TrimSpace(string(body))) + return nil, "", -1, fmt.Errorf("server returned %s (expected 200): %s", resp.Status, strings.TrimSpace(string(body))) } - return resp.Body, ExtensionForContentType(resp.Header.Get("Content-Type")), nil + length := resp.ContentLength + if length <= 0 { + length = -1 + } + return resp.Body, ExtensionForContentType(resp.Header.Get("Content-Type")), length, nil default: - return nil, "", fmt.Errorf("unsupported download-album size %q (must be %q, %q, %q, or %q)", size, immichapi.AssetMediaSizeOriginal, immichapi.AssetMediaSizeFullsize, immichapi.AssetMediaSizePreview, immichapi.AssetMediaSizeThumbnail) + return nil, "", -1, fmt.Errorf("unsupported download-album size %q (must be %q, %q, %q, or %q)", size, immichapi.AssetMediaSizeOriginal, immichapi.AssetMediaSizeFullsize, immichapi.AssetMediaSizePreview, immichapi.AssetMediaSizeThumbnail) } } @@ -580,15 +646,23 @@ func effectiveSize(size immichapi.AssetMediaSize, resizeVideo ResizeVideoOptions // Otherwise (both disabled, or this asset/size doesn't qualify — e.g. a // video downloaded as --size original with only --resize set) the file is // saved as-is in its natural format. -func downloadAssetFile(ctx context.Context, c *client.Client, a immichapi.AssetResponseDto, size immichapi.AssetMediaSize, destBasePath string, resize ResizeOptions, resizeVideo ResizeVideoOptions) (string, error) { +func downloadAssetFile(ctx context.Context, c *client.Client, a immichapi.AssetResponseDto, size immichapi.AssetMediaSize, destBasePath string, resize ResizeOptions, resizeVideo ResizeVideoOptions, prog *ByteProgress) (string, error) { size = effectiveSize(size, resizeVideo, a.Type) - body, ext, err := fetchAssetStream(ctx, c, a, size) + body, ext, length, err := fetchAssetStream(ctx, c, a, size) if err != nil { return "", err } defer body.Close() + var src io.Reader = body + if prog != nil { + prog.Label = a.OriginalFileName + prog.Total = length + src = prog.Wrap(body) + defer prog.Finish() + } + resizeImage := shouldResize(resize, size, a.Type) transcodeVideo := shouldResizeVideo(resizeVideo, size, a.Type) @@ -599,7 +673,7 @@ func downloadAssetFile(ctx context.Context, c *client.Client, a immichapi.AssetR // final output path. rawPath = destBasePath + ".download-tmp" + ext } - if err := writeAssetFile(rawPath, body); err != nil { + if err := writeAssetFile(rawPath, src); err != nil { return "", err } @@ -693,12 +767,17 @@ func DownloadAlbum(ctx context.Context, c *client.Client, album immichapi.AlbumR names := AssignLocalNames(assets, opts.TimestampPrefix) progress := NewProgress(len(assets)) + order := make(map[string]int, len(assets)) + for i, a := range assets { + order[a.Id.String()] = i + } return RunBatch(assets, func(a immichapi.AssetResponseDto) string { return a.OriginalFileName }, func(a immichapi.AssetResponseDto) error { progress.Step(a.OriginalFileName) base := filepath.Join(targetDir, names[a.Id.String()]) - dest, err := downloadAssetFile(ctx, c, a, opts.Size, base, opts.Resize, opts.ResizeVideo) + prog := NewByteProgress(a.OriginalFileName, -1, order[a.Id.String()]+1, len(assets), opts.Quiet) + dest, err := downloadAssetFile(ctx, c, a, opts.Size, base, opts.Resize, opts.ResizeVideo, prog) if err != nil { return err } @@ -773,34 +852,26 @@ func ComputeSyncPlan(assets []immichapi.AssetResponseDto, manifest Manifest) Syn // make removal detection unsafe or nonsensical; the caller should point // --target-dir at a fresh folder instead. func PlanAlbumSync(ctx context.Context, c *client.Client, album immichapi.AlbumResponseDto, targetDir string, opts DownloadAlbumOptions) ([]immichapi.AssetResponseDto, SyncPlan, Manifest, error) { - manifest, existed, err := LoadManifest(targetDir) - if err != nil { - return nil, SyncPlan{}, Manifest{}, err + return PlanSync(ctx, c, SyncSource{Kind: SourceKindAlbum, ID: album.Id.String(), Name: album.AlbumName}, targetDir, opts, func(ctx context.Context) ([]immichapi.AssetResponseDto, error) { + return FetchFilteredAlbumAssets(ctx, c, album.Id, opts.IgnoreVideos) + }) +} + +// checkManifestOptions guards against mixing conventions in one folder. +func checkManifestOptions(targetDir string, manifest Manifest, opts DownloadAlbumOptions) error { + if manifest.Size != "" && manifest.Size != opts.Size { + return fmt.Errorf("manifest in %q was created with --size %s, not %s — use a different --target-dir to switch", targetDir, manifest.Size, opts.Size) } - if existed { - if manifest.AlbumID != "" && manifest.AlbumID != album.Id.String() { - return nil, SyncPlan{}, Manifest{}, fmt.Errorf("manifest in %q tracks album %q (%s), not %q (%s) — use a different --target-dir or matching --album-id/--album-name", targetDir, manifest.AlbumName, manifest.AlbumID, album.AlbumName, album.Id.String()) - } - if manifest.Size != "" && manifest.Size != opts.Size { - return nil, SyncPlan{}, Manifest{}, fmt.Errorf("manifest in %q was created with --size %s, not %s — use a different --target-dir to switch", targetDir, manifest.Size, opts.Size) - } - if manifest.Resize != opts.Resize.Enabled { - return nil, SyncPlan{}, Manifest{}, fmt.Errorf("manifest in %q was created with --resize=%t, not %t — use a different --target-dir to switch", targetDir, manifest.Resize, opts.Resize.Enabled) - } - if manifest.ResizeVideoPreset != resizeVideoPresetOf(opts.ResizeVideo) { - return nil, SyncPlan{}, Manifest{}, fmt.Errorf("manifest in %q was created with --resize-video-preset=%q, not %q — use a different --target-dir to switch", targetDir, manifest.ResizeVideoPreset, resizeVideoPresetOf(opts.ResizeVideo)) - } - if manifest.TimestampPrefix != opts.TimestampPrefix { - return nil, SyncPlan{}, Manifest{}, fmt.Errorf("manifest in %q was created with --timestamp-prefix=%t, not %t — use a different --target-dir to switch", targetDir, manifest.TimestampPrefix, opts.TimestampPrefix) - } + if manifest.Resize != opts.Resize.Enabled { + return fmt.Errorf("manifest in %q was created with --resize=%t, not %t — use a different --target-dir to switch", targetDir, manifest.Resize, opts.Resize.Enabled) } - - assets, err := FetchFilteredAlbumAssets(ctx, c, album.Id, opts.IgnoreVideos) - if err != nil { - return nil, SyncPlan{}, Manifest{}, fmt.Errorf("fetching album assets: %w", err) + if manifest.ResizeVideoPreset != resizeVideoPresetOf(opts.ResizeVideo) { + return fmt.Errorf("manifest in %q was created with --resize-video-preset=%q, not %q — use a different --target-dir to switch", targetDir, manifest.ResizeVideoPreset, resizeVideoPresetOf(opts.ResizeVideo)) } - - return assets, ComputeSyncPlan(assets, manifest), manifest, nil + if manifest.TimestampPrefix != opts.TimestampPrefix { + return fmt.Errorf("manifest in %q was created with --timestamp-prefix=%t, not %t — use a different --target-dir to switch", targetDir, manifest.TimestampPrefix, opts.TimestampPrefix) + } + return nil } // ApplyAlbumSync executes plan against targetDir: downloads every asset in @@ -817,6 +888,23 @@ func PlanAlbumSync(ctx context.Context, c *client.Client, album immichapi.AlbumR // successful save is safer than continuing once disk state and the // manifest may have diverged. func ApplyAlbumSync(ctx context.Context, c *client.Client, album immichapi.AlbumResponseDto, targetDir string, allAssets []immichapi.AssetResponseDto, plan SyncPlan, manifest Manifest, opts DownloadAlbumOptions) error { + return ApplySync(ctx, c, SyncSource{Kind: SourceKindAlbum, ID: album.Id.String(), Name: album.AlbumName}, targetDir, allAssets, plan, manifest, opts) +} + +// ApplySync executes plan against targetDir: downloads every asset in +// plan.Additions and plan.Updates, deletes the local file for every +// plan.Removals entry, and persists the updated manifest along the way — +// stamping it with source/opts.Size/etc. so future runs can validate against +// it. The manifest is saved to disk after every single removal and every +// successful download (not just once at the end): interrupting a long sync +// (Ctrl+C, crash, closed terminal) never loses more than the one item that +// was in flight, and every previously downloaded/removed file is correctly +// reflected on the next run. Downloads continue on a per-asset error and +// are summarized at the end (the usual bulk convention), but a local-file +// deletion error aborts immediately — leaving the manifest as of the last +// successful save is safer than continuing once disk state and the +// manifest may have diverged. +func ApplySync(ctx context.Context, c *client.Client, source SyncSource, targetDir string, allAssets []immichapi.AssetResponseDto, plan SyncPlan, manifest Manifest, opts DownloadAlbumOptions) error { if err := os.MkdirAll(targetDir, 0o755); err != nil { return fmt.Errorf("creating target directory %q: %w", targetDir, err) } @@ -824,8 +912,7 @@ func ApplyAlbumSync(ctx context.Context, c *client.Client, album immichapi.Album if manifest.Assets == nil { manifest.Assets = map[string]ManifestAsset{} } - manifest.AlbumID = album.Id.String() - manifest.AlbumName = album.AlbumName + manifest.stampSource(source.Kind, source.ID, source.Name) manifest.Size = opts.Size manifest.Resize = opts.Resize.Enabled manifest.ResizeVideoPreset = resizeVideoPresetOf(opts.ResizeVideo) @@ -849,7 +936,7 @@ func ApplyAlbumSync(ctx context.Context, c *client.Client, album immichapi.Album if err := SaveManifest(targetDir, manifest); err != nil { return err } - fmt.Printf("Removed %s (asset no longer in album)\n", path) + fmt.Printf("Removed %s (asset no longer in %s)\n", path, source.Kind) } names := AssignLocalNames(allAssets, opts.TimestampPrefix) @@ -858,12 +945,17 @@ func ApplyAlbumSync(ctx context.Context, c *client.Client, album immichapi.Album toDownload = append(toDownload, plan.Updates...) downloadProgress := NewProgress(len(toDownload)) + dlOrder := make(map[string]int, len(toDownload)) + for i, a := range toDownload { + dlOrder[a.Id.String()] = i + } return RunBatch(toDownload, func(a immichapi.AssetResponseDto) string { return a.OriginalFileName }, func(a immichapi.AssetResponseDto) error { downloadProgress.Step(a.OriginalFileName) base := filepath.Join(targetDir, names[a.Id.String()]) - dest, err := downloadAssetFile(ctx, c, a, opts.Size, base, opts.Resize, opts.ResizeVideo) + prog := NewByteProgress(a.OriginalFileName, -1, dlOrder[a.Id.String()]+1, len(toDownload), opts.Quiet) + dest, err := downloadAssetFile(ctx, c, a, opts.Size, base, opts.Resize, opts.ResizeVideo, prog) if err != nil { return err } diff --git a/internal/workflows/bulkcheck.go b/internal/workflows/bulkcheck.go new file mode 100644 index 0000000..93dcd8c --- /dev/null +++ b/internal/workflows/bulkcheck.go @@ -0,0 +1,140 @@ +package workflows + +import ( + "context" + "fmt" + "io/fs" + "net/http" + "os" + "path/filepath" + "strings" + + openapi_types "github.com/oapi-codegen/runtime/types" + + "github.com/dhcgn/immich-admin-cli/internal/client" + "github.com/dhcgn/immich-admin-cli/internal/immichapi" +) + +// BulkCheckStatus is the per-file outcome of a bulk-upload-check. +type BulkCheckStatus string + +const ( + BulkCheckUploaded BulkCheckStatus = "uploaded" + BulkCheckMissing BulkCheckStatus = "missing" + BulkCheckUnsupported BulkCheckStatus = "unsupported" +) + +// BulkCheckEntry is one file's bulk-upload-check result. +type BulkCheckEntry struct { + File string + Checksum string + Status BulkCheckStatus + AssetID *openapi_types.UUID + IsTrashed bool +} + +// CollectCheckFiles expands FILE|DIR args into files. Dirs are walked +// recursively; hidden files/dirs (base starting with '.') and symlinks are +// skipped. Pure filesystem, no network. +func CollectCheckFiles(paths []string) ([]string, error) { + var out []string + for _, p := range paths { + fi, err := os.Lstat(p) + if err != nil { + return nil, fmt.Errorf("stat %q: %w", p, err) + } + if fi.Mode()&os.ModeSymlink != 0 { + continue + } + if !fi.IsDir() { + if strings.HasPrefix(filepath.Base(p), ".") { + continue + } + out = append(out, p) + continue + } + err = filepath.WalkDir(p, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + base := filepath.Base(path) + if d.Type()&os.ModeSymlink != 0 { + if d.IsDir() { + return filepath.SkipDir + } + return nil + } + if strings.HasPrefix(base, ".") { + if d.IsDir() && path != p { + return filepath.SkipDir + } + return nil + } + if !d.IsDir() { + out = append(out, path) + } + return nil + }) + if err != nil { + return nil, fmt.Errorf("walking %q: %w", p, err) + } + } + return out, nil +} + +// FileSHA1Base64 returns the base64-standard-encoded SHA1 hash of the file +// at path, matching Immich's Checksum format. Exported for the native +// check command; workflows reuse the same helper as replace-asset verify. +func FileSHA1Base64(path string) (string, error) { + return fileSHA1Base64(path) +} + +// CheckBulkUploadChecksums calls POST /assets/bulk-upload-check for files +// (checksums[file] must be base64 SHA1) in ~500/request chunks and strictly +// classifies each result. Pure network wrapper; hashing happens outside so +// callers can continue-on-error per file. +func CheckBulkUploadChecksums(ctx context.Context, c *client.Client, files []string, checksums map[string]string) ([]BulkCheckEntry, error) { + var out []BulkCheckEntry + for i := 0; i < len(files); i += 500 { + end := min(i+500, len(files)) + chunk := files[i:end] + items := make([]immichapi.AssetBulkUploadCheckItem, 0, len(chunk)) + for _, f := range chunk { + items = append(items, immichapi.AssetBulkUploadCheckItem{Checksum: checksums[f], Id: f}) + } + resp, err := c.API.CheckBulkUploadWithResponse(ctx, immichapi.AssetBulkUploadCheckDto{Assets: items}) + if err == nil { + err = client.Check(resp, http.StatusOK) + } + if err != nil { + return nil, fmt.Errorf("calling POST /assets/bulk-upload-check: %w", err) + } + if resp.JSON200 == nil { + return nil, fmt.Errorf("calling POST /assets/bulk-upload-check: response had no body") + } + for _, r := range resp.JSON200.Results { + out = append(out, ClassifyBulkCheckResult(r.Id, checksums[r.Id], r)) + } + } + return out, nil +} + +// ClassifyBulkCheckResult strictly splits duplicate vs unsupported per +// AssetBulkUploadCheckResult. Pure for testing. +func ClassifyBulkCheckResult(file, checksum string, r immichapi.AssetBulkUploadCheckResult) BulkCheckEntry { + e := BulkCheckEntry{File: file, Checksum: checksum} + switch { + case r.Action == immichapi.Accept: + e.Status = BulkCheckMissing + case r.Action == immichapi.Reject && r.Reason != nil && + *r.Reason == immichapi.AssetRejectReasonDuplicate && r.AssetId != nil: + e.Status = BulkCheckUploaded + e.AssetID = r.AssetId + if r.IsTrashed != nil { + e.IsTrashed = *r.IsTrashed + } + default: + e.Status = BulkCheckUnsupported + } + return e +} diff --git a/internal/workflows/progress.go b/internal/workflows/progress.go new file mode 100644 index 0000000..df54575 --- /dev/null +++ b/internal/workflows/progress.go @@ -0,0 +1,168 @@ +// Package workflows: per-file byte progress for uploads/downloads. +// stdlib-only counting io.Reader wrapper reporting to stderr. +package workflows + +import ( + "fmt" + "io" + "os" + "strings" + "time" +) + +// ByteProgress tracks one file transfer, printing a single-line bar to +// stderr (stdout stays clean for --json/piping). +// Total < 0 means unknown (chunked response): bytes+rate only, no %. +type ByteProgress struct { + Label string + Total int64 + Done int64 + Start time.Time + Pos int + N int + Quiet bool + Out io.Writer + + tty bool + last time.Time + final bool +} + +// NewByteProgress creates a transfer tracker. total < 0 = unknown length. +// quiet disables all output (--quiet, or --json which implies quiet). +func NewByteProgress(label string, total int64, pos, n int, quiet bool) *ByteProgress { + p := &ByteProgress{ + Label: label, + Total: total, + Start: time.Now(), + Pos: pos, + N: n, + Quiet: quiet, + Out: os.Stderr, + } + p.tty = isTerminalWriter(p.Out) + return p +} + +// Wrap returns r wrapped so every Read counts toward the bar. +func (p *ByteProgress) Wrap(r io.Reader) io.Reader { + if p == nil { + return r + } + return &countingReader{r: r, p: p} +} + +// Add counts n bytes and prints a throttled update. +func (p *ByteProgress) Add(n int64) { + if p == nil || p.Quiet || n <= 0 { + if p != nil { + p.Done += max64(0, n) + } + return + } + p.Done += n + now := time.Now() + complete := p.Total >= 0 && p.Done >= p.Total + if !complete && !p.last.IsZero() && now.Sub(p.last) < 200*time.Millisecond { + return + } + p.last = now + p.print(false) +} + +// Finish prints the final state (full bar) and ends the TTY line. +func (p *ByteProgress) Finish() { + if p == nil || p.Quiet || p.final { + return + } + p.final = true + p.print(true) +} + +func (p *ByteProgress) print(final bool) { + if p.Out == nil { + p.Out = os.Stderr + } + line := FormatByteProgressLine(p.Pos, p.N, p.Label, p.Done, p.Total, time.Since(p.Start)) + if p.tty { + // Live in-place single line (TTY only; ANSI clear is fine here). + fmt.Fprintf(p.Out, "\r%s\x1b[K", line) + if final { + fmt.Fprintln(p.Out) + } + return + } + fmt.Fprintln(p.Out, line) +} + +// FormatByteProgressLine renders one progress line. Pure for testing. +func FormatByteProgressLine(pos, n int, label string, done, total int64, elapsed time.Duration) string { + head := fmt.Sprintf("[%d/%d] %s %s", pos, n, label, formatBytesShort(done)) + rate := formatRate(done, elapsed) + if total < 0 { + return fmt.Sprintf("%s [%s]", head, rate) + } + if total == 0 { + return fmt.Sprintf("%s 100%% [%s] %s", head, strings.Repeat("█", 12), rate) + } + pct := float64(done) / float64(total) * 100 + if pct > 100 { + pct = 100 + } + filled := int(pct / 100 * 12) + bar := strings.Repeat("█", filled) + strings.Repeat("-", 12-filled) + return fmt.Sprintf("%s %s %.0f%% [%s] %s", head, formatBytesShort(total), pct, bar, rate) +} + +func formatBytesShort(n int64) string { + const unit = 1024 + if n < unit { + return fmt.Sprintf("%d B", n) + } + div, exp := int64(unit), 0 + for v := n / unit; v >= unit; v /= unit { + div *= unit + exp++ + } + return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp]) +} + +func formatRate(done int64, elapsed time.Duration) string { + if elapsed <= 0 { + return formatBytesShort(done) + "/s" + } + perSec := float64(done) / elapsed.Seconds() + return formatBytesShort(int64(perSec)) + "/s" +} + +func max64(a, b int64) int64 { + if a > b { + return a + } + return b +} + +type countingReader struct { + r io.Reader + p *ByteProgress +} + +func (c *countingReader) Read(b []byte) (int, error) { + n, err := c.r.Read(b) + if n > 0 { + c.p.Add(int64(n)) + } + return n, err +} + +func isTerminalWriter(w io.Writer) bool { + f, ok := w.(*os.File) + if !ok { + return false + } + st, err := f.Stat() + if err != nil { + return false + } + return st.Mode()&os.ModeCharDevice != 0 +} diff --git a/internal/workflows/repairassets.go b/internal/workflows/repairassets.go index 78ce47e..330a72b 100644 --- a/internal/workflows/repairassets.go +++ b/internal/workflows/repairassets.go @@ -449,7 +449,15 @@ func downloadOriginalTo(ctx context.Context, c *client.Client, id openapi_types. } defer f.Close() - if _, err := io.Copy(f, resp.Body); err != nil { + total := resp.ContentLength + if total <= 0 { + total = -1 + } + // ponytail: quiet=false default; repair has no --quiet flag yet, bar is plain lines when piped. + prog := NewByteProgress(id.String(), total, 1, 1, false) + src := prog.Wrap(resp.Body) + defer prog.Finish() + if _, err := io.Copy(f, src); err != nil { return fmt.Errorf("writing temp file %q: %w", destPath, err) } return nil diff --git a/internal/workflows/upload.go b/internal/workflows/upload.go index 54b21ff..471c950 100644 --- a/internal/workflows/upload.go +++ b/internal/workflows/upload.go @@ -1,7 +1,6 @@ package workflows import ( - "bytes" "context" "fmt" "io" @@ -33,6 +32,27 @@ type UploadOptions struct { Key *string Slug *string Checksum *string + // Progress, when non-nil, counts assetData bytes for the per-file bar. + // Total/Label should already be set by the caller (total = file size). + Progress *ByteProgress +} + +// DuplicateUploadError is returned when the server matches the upload +// checksum to an existing asset (200) instead of creating one (201). +// Callers that only link duplicates (watch-upload) can errors.As this to +// get the existing ID; callers that must act on their own upload +// (replace-asset) treat it as a plain aborting error. +type DuplicateUploadError struct { + ExistingID openapi_types.UUID + HasID bool +} + +func (e *DuplicateUploadError) Error() string { + id := "unknown" + if e.HasID { + id = e.ExistingID.String() + } + return fmt.Sprintf("upload was treated as a duplicate of existing asset %s (checksum matches); aborting to avoid acting on the wrong asset", id) } // UploadAssetFile uploads the local file at path as a new asset @@ -41,6 +61,9 @@ type UploadOptions struct { // oapi-codegen only generates a type alias for the multipart body of binary // fields (UploadAssetMultipartRequestBody = AssetMediaCreateDto); it does not // generate a multipart writer, so the request body is built by hand here. +// +// The body streams via io.Pipe (never buffered in RAM) so GB videos don't +// OOM and a counting reader reports real upload bytes. func UploadAssetFile(ctx context.Context, c *client.Client, path string, opts UploadOptions) (openapi_types.UUID, error) { f, err := os.Open(path) if err != nil { @@ -62,78 +85,103 @@ func UploadAssetFile(ctx context.Context, c *client.Client, path string, opts Up modifiedAt = opts.FileModifiedAt.UTC().Format(time.RFC3339) } - var body bytes.Buffer - mw := multipart.NewWriter(&body) - - writeField := func(k, v string) error { - if err := mw.WriteField(k, v); err != nil { - return fmt.Errorf("building upload body: %w", err) - } - return nil - } - if err := writeField("fileCreatedAt", createdAt); err != nil { - return openapi_types.UUID{}, err - } - if err := writeField("fileModifiedAt", modifiedAt); err != nil { - return openapi_types.UUID{}, err + fields := [][2]string{ + {"fileCreatedAt", createdAt}, + {"fileModifiedAt", modifiedAt}, } if opts.Filename != nil { - if err := writeField("filename", *opts.Filename); err != nil { - return openapi_types.UUID{}, err - } + fields = append(fields, [2]string{"filename", *opts.Filename}) } if opts.Duration != nil { - if err := writeField("duration", strconv.Itoa(*opts.Duration)); err != nil { - return openapi_types.UUID{}, err - } + fields = append(fields, [2]string{"duration", strconv.Itoa(*opts.Duration)}) } if opts.IsFavorite != nil { - if err := writeField("isFavorite", strconv.FormatBool(*opts.IsFavorite)); err != nil { - return openapi_types.UUID{}, err - } + fields = append(fields, [2]string{"isFavorite", strconv.FormatBool(*opts.IsFavorite)}) } if opts.Visibility != nil { - if err := writeField("visibility", string(*opts.Visibility)); err != nil { - return openapi_types.UUID{}, err - } + fields = append(fields, [2]string{"visibility", string(*opts.Visibility)}) } if opts.LivePhotoVideoId != nil { - if err := writeField("livePhotoVideoId", opts.LivePhotoVideoId.String()); err != nil { - return openapi_types.UUID{}, err - } - } - part, err := mw.CreateFormFile("assetData", filepath.Base(path)) - if err != nil { - return openapi_types.UUID{}, fmt.Errorf("building upload body: %w", err) - } - if _, err := io.Copy(part, f); err != nil { - return openapi_types.UUID{}, fmt.Errorf("reading %q: %w", path, err) + fields = append(fields, [2]string{"livePhotoVideoId", opts.LivePhotoVideoId.String()}) } - if opts.SidecarPath != "" { - sf, err := os.Open(opts.SidecarPath) - if err != nil { - return openapi_types.UUID{}, fmt.Errorf("opening sidecar %q: %w", opts.SidecarPath, err) + + pr, pw := io.Pipe() + mw := multipart.NewWriter(pw) + contentType := mw.FormDataContentType() + + // Stream the multipart body: fields + file (+ sidecar) without ever + // buffering the whole file in RAM. A counting reader reports real + // upload bytes for the per-file bar. + writeErr := make(chan error, 1) + go func() { + var werr error + defer func() { + cerr := mw.Close() + if werr != nil { + pw.CloseWithError(werr) + } else if cerr != nil { + pw.CloseWithError(fmt.Errorf("building upload body: %w", cerr)) + } else { + pw.Close() + } + writeErr <- werr + }() + for _, kv := range fields { + if err := mw.WriteField(kv[0], kv[1]); err != nil { + werr = fmt.Errorf("building upload body: %w", err) + return + } } - defer sf.Close() - spart, err := mw.CreateFormFile("sidecarData", filepath.Base(opts.SidecarPath)) + part, err := mw.CreateFormFile("assetData", filepath.Base(path)) if err != nil { - return openapi_types.UUID{}, fmt.Errorf("building upload body: %w", err) + werr = fmt.Errorf("building upload body: %w", err) + return } - if _, err := io.Copy(spart, sf); err != nil { - return openapi_types.UUID{}, fmt.Errorf("reading sidecar %q: %w", opts.SidecarPath, err) + src := io.Reader(f) + if opts.Progress != nil { + src = opts.Progress.Wrap(f) } - } - if err := mw.Close(); err != nil { - return openapi_types.UUID{}, fmt.Errorf("building upload body: %w", err) - } + if _, err := io.Copy(part, src); err != nil { + werr = fmt.Errorf("reading %q: %w", path, err) + return + } + if opts.Progress != nil { + opts.Progress.Finish() + } + if opts.SidecarPath != "" { + sf, err := os.Open(opts.SidecarPath) + if err != nil { + werr = fmt.Errorf("opening sidecar %q: %w", opts.SidecarPath, err) + return + } + defer sf.Close() + spart, err := mw.CreateFormFile("sidecarData", filepath.Base(opts.SidecarPath)) + if err != nil { + werr = fmt.Errorf("building upload body: %w", err) + return + } + if _, err := io.Copy(spart, sf); err != nil { + werr = fmt.Errorf("reading sidecar %q: %w", opts.SidecarPath, err) + return + } + } + }() var params *immichapi.UploadAssetParams if opts.Key != nil || opts.Slug != nil || opts.Checksum != nil { params = &immichapi.UploadAssetParams{Key: opts.Key, Slug: opts.Slug, XImmichChecksum: opts.Checksum} } - resp, err := c.API.UploadAssetWithBodyWithResponse(ctx, params, mw.FormDataContentType(), &body) + resp, err := c.API.UploadAssetWithBodyWithResponse(ctx, params, contentType, pr) if err != nil { + // Surface a local pipe-build failure instead of a generic upload error. + select { + case werr := <-writeErr: + if werr != nil { + return openapi_types.UUID{}, werr + } + default: + } return openapi_types.UUID{}, fmt.Errorf("uploading asset: %w", err) } @@ -145,14 +193,13 @@ func UploadAssetFile(ctx context.Context, c *client.Client, path string, opts Up return resp.JSON201.Id, nil case http.StatusOK: // The server matched the file's checksum to an existing asset - // instead of creating a new one. Treating that asset as "our" - // replacement would risk running copy/delete against the wrong - // (or the same) asset, so abort instead. - existingID := "unknown" + // instead of creating a new one. + dup := &DuplicateUploadError{} if resp.JSON200 != nil { - existingID = resp.JSON200.Id.String() + dup.ExistingID = resp.JSON200.Id + dup.HasID = true } - return openapi_types.UUID{}, fmt.Errorf("upload was treated as a duplicate of existing asset %s (checksum matches); aborting to avoid acting on the wrong asset", existingID) + return openapi_types.UUID{}, dup default: return openapi_types.UUID{}, fmt.Errorf("server returned %s (expected 200 or 201): %s", resp.Status(), string(resp.GetBody())) } diff --git a/internal/workflows/watch_test.go b/internal/workflows/watch_test.go new file mode 100644 index 0000000..ba0f53a --- /dev/null +++ b/internal/workflows/watch_test.go @@ -0,0 +1,77 @@ +package workflows + +import ( + "testing" + "time" + + openapi_types "github.com/oapi-codegen/runtime/types" + + "github.com/dhcgn/immich-admin-cli/internal/immichapi" +) + +func TestRenderTagPattern(t *testing.T) { + got, err := RenderTagPattern("immich-admin-cli/watch/{yyyy-MM-dd}", time.Date(2026, 9, 19, 0, 0, 0, 0, time.UTC)) + if err != nil || got != "immich-admin-cli/watch/2026-09-19" { + t.Fatalf("RenderTagPattern = %q, %v; want watch/2026-09-19", got, err) + } + if _, err := RenderTagPattern("x/{yyyy-MM}", time.Now()); err == nil { + t.Error("unknown placeholder expected error, got nil") + } +} + +func TestClassifyBulkCheckResult(t *testing.T) { + dup := immichapi.AssetRejectReasonDuplicate + id := openapi_types.UUID([16]byte{1}) + tr := true + got := ClassifyBulkCheckResult("f", "c", immichapi.AssetBulkUploadCheckResult{Action: immichapi.Reject, Reason: &dup, AssetId: &id, IsTrashed: &tr, Id: "f"}) + if got.Status != BulkCheckUploaded || got.AssetID == nil || !got.IsTrashed { + t.Fatalf("duplicate classify = %+v, want uploaded with id+trashed", got) + } + got = ClassifyBulkCheckResult("f", "c", immichapi.AssetBulkUploadCheckResult{Action: immichapi.Accept, Id: "f"}) + if got.Status != BulkCheckMissing { + t.Fatalf("accept classify = %q, want missing", got.Status) + } + uns := immichapi.AssetRejectReasonUnsupportedFormat + got = ClassifyBulkCheckResult("f", "c", immichapi.AssetBulkUploadCheckResult{Action: immichapi.Reject, Reason: &uns, Id: "f"}) + if got.Status != BulkCheckUnsupported { + t.Fatalf("unsupported classify = %q, want unsupported", got.Status) + } +} + +func TestFormatByteProgressLine(t *testing.T) { + line := FormatByteProgressLine(3, 42, "IMG.jpg", 45, 100, time.Second) + if line == "" || len(line) < 10 { + t.Fatalf("empty progress line: %q", line) + } + // Unknown total: bytes+rate only, no %. + line = FormatByteProgressLine(1, 2, "f", 10, -1, time.Second) + for _, s := range []string{"[1/2]", "f"} { + found := false + for i := 0; i+len(s) <= len(line); i++ { + if line[i:i+len(s)] == s { + found = true + break + } + } + if !found { + t.Fatalf("progress line %q missing %q", line, s) + } + } +} + +func TestStableForGate(t *testing.T) { + st := map[string]watchUploadEntry{} + now := time.Now() + mtime := now.Add(-time.Minute) + if stableForGate(st, "a.jpg", 10, mtime, now, 30*time.Second) { + t.Fatal("first sighting should be unstable") + } + // Same size+mtime 31s later -> stable. + if !stableForGate(st, "a.jpg", 10, mtime, now.Add(31*time.Second), 30*time.Second) { + t.Fatal("unchanged after stable-for should be stable") + } + // Growing file resets. + if stableForGate(st, "a.jpg", 11, now, now.Add(32*time.Second), 30*time.Second) { + t.Fatal("changed size should be unstable") + } +} diff --git a/internal/workflows/watchdownload.go b/internal/workflows/watchdownload.go new file mode 100644 index 0000000..0784d82 --- /dev/null +++ b/internal/workflows/watchdownload.go @@ -0,0 +1,186 @@ +package workflows + +import ( + "context" + "fmt" + "net/http" + + openapi_types "github.com/oapi-codegen/runtime/types" + + "github.com/dhcgn/immich-admin-cli/internal/client" + "github.com/dhcgn/immich-admin-cli/internal/immichapi" +) + +// SyncSource identifies one download mirror source: an album or a tag. +type SyncSource struct { + Kind string + ID string + Name string +} + +// PlanSync is the generic read-only sync planner shared by album and tag +// sources. fetch lists the source's current (already filtered) assets. +func PlanSync(ctx context.Context, c *client.Client, source SyncSource, targetDir string, opts DownloadAlbumOptions, fetch func(ctx context.Context) ([]immichapi.AssetResponseDto, error)) ([]immichapi.AssetResponseDto, SyncPlan, Manifest, error) { + manifest, existed, err := LoadManifest(targetDir) + if err != nil { + return nil, SyncPlan{}, Manifest{}, err + } + if existed { + kind, id, name := manifest.normalizedSource() + if kind != "" && (kind != source.Kind || id != source.ID) { + return nil, SyncPlan{}, Manifest{}, fmt.Errorf("manifest in %q tracks %s %q (%s), not %s %q (%s) — use a different --target-dir", targetDir, kind, name, id, source.Kind, source.Name, source.ID) + } + if err := checkManifestOptions(targetDir, manifest, opts); err != nil { + return nil, SyncPlan{}, Manifest{}, err + } + } + + assets, err := fetch(ctx) + if err != nil { + return nil, SyncPlan{}, Manifest{}, fmt.Errorf("fetching %s assets: %w", source.Kind, err) + } + return assets, ComputeSyncPlan(assets, manifest), manifest, nil +} + +// PlanTagSync plans a tag-source sync (read-only). +func PlanTagSync(ctx context.Context, c *client.Client, tag immichapi.TagResponseDto, targetDir string, opts DownloadAlbumOptions) ([]immichapi.AssetResponseDto, SyncPlan, Manifest, error) { + return PlanSync(ctx, c, SyncSource{Kind: SourceKindTag, ID: tag.Id.String(), Name: tag.Value}, targetDir, opts, func(ctx context.Context) ([]immichapi.AssetResponseDto, error) { + return FetchFilteredTagAssets(ctx, c, tag.Id, opts.IgnoreVideos) + }) +} + +// ResolveTagByValue finds exactly one tag by full-path value. +func ResolveTagByValue(ctx context.Context, c *client.Client, value string) (immichapi.TagResponseDto, error) { + resp, err := c.API.GetAllTagsWithResponse(ctx) + if err == nil { + err = client.Check(resp, http.StatusOK) + } + if err != nil { + return immichapi.TagResponseDto{}, fmt.Errorf("listing tags: %w", err) + } + var matches []immichapi.TagResponseDto + if resp.JSON200 != nil { + for _, t := range *resp.JSON200 { + if t.Value == value { + matches = append(matches, t) + } + } + } + if len(matches) == 0 { + return immichapi.TagResponseDto{}, fmt.Errorf("no tag with value %q found", value) + } + if len(matches) > 1 { + ids := make([]string, len(matches)) + for i, t := range matches { + ids[i] = t.Id.String() + } + return immichapi.TagResponseDto{}, fmt.Errorf("%d tags have value %q — use --tag-id to disambiguate", len(matches), value) + } + return matches[0], nil +} + +// FetchTagAssets returns every asset carrying tagID via tag-scoped metadata +// search (POST /search/metadata, MetadataSearchDto.TagIds), following the +// result cursor until exhausted. +func FetchTagAssets(ctx context.Context, c *client.Client, tagID openapi_types.UUID) ([]immichapi.AssetResponseDto, error) { + var assets []immichapi.AssetResponseDto + var pager SearchPager + size := 250 + for { + body := immichapi.MetadataSearchDto{ + TagIds: &[]openapi_types.UUID{tagID}, + Size: &size, + } + pager.Apply(&body) + + resp, err := c.API.SearchAssetsWithResponse(ctx, &immichapi.SearchAssetsParams{}, body) + if err == nil { + err = client.Check(resp, http.StatusOK) + } + if err != nil { + return nil, fmt.Errorf("searching tag assets: %w", err) + } + if resp.JSON200 == nil { + return nil, fmt.Errorf("searching tag assets: response had no body") + } + assets = append(assets, resp.JSON200.Assets.Items...) + if !pager.Next(resp.JSON200.Assets) { + return assets, nil + } + } +} + +// FetchFilteredTagAssets fetches every asset with tagID and, if +// ignoreVideos, drops VIDEO assets. +func FetchFilteredTagAssets(ctx context.Context, c *client.Client, tagID openapi_types.UUID, ignoreVideos bool) ([]immichapi.AssetResponseDto, error) { + assets, err := FetchTagAssets(ctx, c, tagID) + if err != nil { + return nil, err + } + if ignoreVideos { + assets = FilterOutVideos(assets) + } + return assets, nil +} + +// WatchDownloadOptions controls one watch-download scan / loop. +type WatchDownloadOptions struct { + DownloadAlbumOptions + TargetDir string + Source SyncSource + TagID *openapi_types.UUID + AlbumID *openapi_types.UUID + Interval int64 // seconds; <=0 means run once (per user decision: treat as once) + Once bool + DryRun bool + Quiet bool + JSON bool +} + +// RunWatchDownloadOnce runs a single sync scan from source into targetDir. +func RunWatchDownloadOnce(ctx context.Context, c *client.Client, opts WatchDownloadOptions) (SyncPlan, error) { + dopts := opts.DownloadAlbumOptions + dopts.DryRun = opts.DryRun + dopts.Quiet = opts.Quiet || opts.JSON + + var assets []immichapi.AssetResponseDto + var plan SyncPlan + var manifest Manifest + var err error + switch opts.Source.Kind { + case SourceKindTag: + var tagID openapi_types.UUID + if opts.TagID != nil { + tagID = *opts.TagID + } else { + var tag immichapi.TagResponseDto + tag, err = ResolveTagByValue(ctx, c, opts.Source.Name) + if err != nil { + return SyncPlan{}, err + } + tagID = tag.Id + opts.Source.ID = tag.Id.String() + } + assets, plan, manifest, err = PlanSync(ctx, c, opts.Source, opts.TargetDir, dopts, func(ctx context.Context) ([]immichapi.AssetResponseDto, error) { + return FetchFilteredTagAssets(ctx, c, tagID, dopts.IgnoreVideos) + }) + default: + var albumID openapi_types.UUID + if opts.AlbumID != nil { + albumID = *opts.AlbumID + } + assets, plan, manifest, err = PlanSync(ctx, c, opts.Source, opts.TargetDir, dopts, func(ctx context.Context) ([]immichapi.AssetResponseDto, error) { + return FetchFilteredAlbumAssets(ctx, c, albumID, dopts.IgnoreVideos) + }) + } + if err != nil { + return SyncPlan{}, err + } + if opts.DryRun { + return plan, nil + } + if err := ApplySync(ctx, c, opts.Source, opts.TargetDir, assets, plan, manifest, dopts); err != nil { + return plan, err + } + return plan, nil +} diff --git a/internal/workflows/watchupload.go b/internal/workflows/watchupload.go new file mode 100644 index 0000000..c726030 --- /dev/null +++ b/internal/workflows/watchupload.go @@ -0,0 +1,474 @@ +package workflows + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + openapi_types "github.com/oapi-codegen/runtime/types" + + "github.com/dhcgn/immich-admin-cli/internal/client" + "github.com/dhcgn/immich-admin-cli/internal/immichapi" +) + +// Watch state + upload orchestration (local -> Immich). + +const watchUploadStateFile = ".immich-watch-upload.json" + +// DefaultWatchUploadTagPattern tags flat uploads with the upload day. +const DefaultWatchUploadTagPattern = "immich-admin-cli/watch/{yyyy-MM-dd}" + +// WatchUploadOptions controls watch-upload. +type WatchUploadOptions struct { + WatchDir string + Mode string // flat | by-subfolder + Depth int + TagPattern string + AlbumID *openapi_types.UUID + AlbumName string + Interval time.Duration + StableFor time.Duration + Once bool + DryRun bool + Yes bool + Quiet bool + JSON bool +} + +// WatchUploadStats is the per-interval summary. +type WatchUploadStats struct { + Uploaded int `json:"uploaded"` + LinkedDuplicate int `json:"linkedDuplicate"` + SkippedUnstable int `json:"skippedUnstable"` + Failed int `json:"failed"` + SkippedDone int `json:"-"` +} + +// WatchFile is one candidate local file. +type WatchFile struct { + AbsPath string + Rel string + Subfolder string + Size int64 + ModTime time.Time +} + +type watchUploadEntry struct { + Checksum string `json:"checksum,omitempty"` + AssetID string `json:"assetId,omitempty"` + Size int64 `json:"size"` + MtimeUnix int64 `json:"mtimeUnix"` + FirstSeen int64 `json:"firstSeenUnix"` +} + +// RenderTagPattern replaces {yyyy-MM-dd} with t's date. Only that +// placeholder exists in v1; any other {…} is an error. Pure for testing. +func RenderTagPattern(pattern string, t time.Time) (string, error) { + if pattern == "" { + return "", fmt.Errorf("tag pattern must not be empty") + } + out := strings.ReplaceAll(pattern, "{yyyy-MM-dd}", t.Format("2006-01-02")) + if strings.Contains(out, "{") { + return "", fmt.Errorf("unknown placeholder in --tag-pattern %q: only {yyyy-MM-dd} is supported", pattern) + } + if strings.TrimSpace(out) == "" { + return "", fmt.Errorf("tag pattern %q renders empty", pattern) + } + return out, nil +} + +// ScanWatchDir lists candidate files per mode (depth 1 only in v1). +// flat: top-level files. by-subfolder: immediate subdir files only. +func ScanWatchDir(watchDir, mode string, depth int) ([]WatchFile, error) { + if depth != 1 { + return nil, fmt.Errorf("only --depth 1 is supported in v1 (got %d)", depth) + } + entries, err := os.ReadDir(watchDir) + if err != nil { + return nil, fmt.Errorf("reading watch dir %q: %w", watchDir, err) + } + var out []WatchFile + addFile := func(abs, rel, sub string) { + fi, err := os.Lstat(abs) + if err != nil || fi.Mode()&os.ModeSymlink != 0 || fi.IsDir() { + return + } + base := filepath.Base(abs) + if strings.HasPrefix(base, ".") || base == watchUploadStateFile { + return + } + out = append(out, WatchFile{AbsPath: abs, Rel: rel, Subfolder: sub, Size: fi.Size(), ModTime: fi.ModTime()}) + } + switch mode { + case "flat": + for _, e := range entries { + if e.IsDir() || e.Type()&os.ModeSymlink != 0 { + continue + } + if strings.HasPrefix(e.Name(), ".") { + continue + } + addFile(filepath.Join(watchDir, e.Name()), e.Name(), "") + } + case "by-subfolder": + for _, e := range entries { + if !e.IsDir() || strings.HasPrefix(e.Name(), ".") || e.Type()&os.ModeSymlink != 0 { + continue + } + sub := e.Name() + subEntries, err := os.ReadDir(filepath.Join(watchDir, sub)) + if err != nil { + continue + } + for _, f := range subEntries { + if f.IsDir() || f.Type()&os.ModeSymlink != 0 || strings.HasPrefix(f.Name(), ".") { + continue + } + addFile(filepath.Join(watchDir, sub, f.Name()), filepath.Join(sub, f.Name()), sub) + } + } + default: + return nil, fmt.Errorf("invalid --mode %q: must be flat or by-subfolder", mode) + } + return out, nil +} + +func loadWatchUploadState(watchDir string) (map[string]watchUploadEntry, error) { + path := filepath.Join(watchDir, watchUploadStateFile) + data, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return map[string]watchUploadEntry{}, nil + } + if err != nil { + return nil, fmt.Errorf("reading state %q: %w", path, err) + } + var st map[string]watchUploadEntry + if err := json.Unmarshal(data, &st); err != nil { + return nil, fmt.Errorf("parsing state %q: %w", path, err) + } + if st == nil { + st = map[string]watchUploadEntry{} + } + return st, nil +} + +func saveWatchUploadState(watchDir string, st map[string]watchUploadEntry) error { + data, err := json.MarshalIndent(st, "", " ") + if err != nil { + return fmt.Errorf("encoding watch state: %w", err) + } + path := filepath.Join(watchDir, watchUploadStateFile) + tmp, err := os.CreateTemp(watchDir, watchUploadStateFile+".tmp-*") + if err != nil { + return fmt.Errorf("creating temp state file: %w", err) + } + tmpPath := tmp.Name() + _, werr := tmp.Write(data) + cerr := tmp.Close() + if werr != nil || cerr != nil { + os.Remove(tmpPath) + if werr != nil { + return fmt.Errorf("writing state %q: %w", path, werr) + } + return fmt.Errorf("writing state %q: %w", path, cerr) + } + if err := os.Rename(tmpPath, path); err != nil { + os.Remove(tmpPath) + return fmt.Errorf("writing state %q: %w", path, err) + } + return nil +} + +// stableForGate reports whether the file has been unchanged for stableFor. +// It updates st[rel] in place (first-seen tracking). stableFor <= 0 means +// no gate (always stable). +func stableForGate(st map[string]watchUploadEntry, rel string, size int64, mtime time.Time, now time.Time, stableFor time.Duration) bool { + if stableFor <= 0 { + return true + } + e, ok := st[rel] + mtimeUnix := mtime.Unix() + if !ok || e.Size != size || e.MtimeUnix != mtimeUnix { + st[rel] = watchUploadEntry{Checksum: e.Checksum, AssetID: e.AssetID, Size: size, MtimeUnix: mtimeUnix, FirstSeen: now.Unix()} + return false + } + return now.Unix()-e.FirstSeen >= int64(stableFor.Seconds()) +} + +// ResolveOrCreateAlbumByName finds an album by exact name or creates it. +// dryRun prints instead of creating; yes skips the creation prompt. +func ResolveOrCreateAlbumByName(ctx context.Context, c *client.Client, name string, dryRun, yes bool) (immichapi.AlbumResponseDto, error) { + if album, err := ResolveAlbum(ctx, c, nil, name); err == nil { + return album, nil + } + if dryRun { + fmt.Printf("[dry-run] would create album %q\n", name) + return immichapi.AlbumResponseDto{AlbumName: name}, nil + } + if !yes { + fmt.Printf("Create album %q? [y/N]: ", name) + var answer string + fmt.Scanln(&answer) + answer = strings.ToLower(strings.TrimSpace(answer)) + if answer != "y" && answer != "yes" { + return immichapi.AlbumResponseDto{}, fmt.Errorf("album %q does not exist (creation declined)", name) + } + } + resp, err := c.API.CreateAlbumWithResponse(ctx, immichapi.CreateAlbumDto{AlbumName: name}) + if err != nil { + return immichapi.AlbumResponseDto{}, fmt.Errorf("creating album: %w", err) + } + if resp.StatusCode() != 201 { + return immichapi.AlbumResponseDto{}, fmt.Errorf("creating album: server returned %s", resp.Status()) + } + if resp.JSON201 == nil { + return immichapi.AlbumResponseDto{}, fmt.Errorf("creating album: response had no body") + } + return *resp.JSON201, nil +} + +// ensureWatchTag returns the tag ID for value, creating it unless dryRun. +// yes skips the creation prompt. +func ensureWatchTag(ctx context.Context, c *client.Client, value string, dryRun, yes bool) (openapi_types.UUID, error) { + if t, err := ResolveTagByValue(ctx, c, value); err == nil { + return t.Id, nil + } + if dryRun { + fmt.Printf("[dry-run] would create tag %q\n", value) + return openapi_types.UUID{}, nil + } + if !yes { + fmt.Printf("Create tag %q? [y/N]: ", value) + var answer string + fmt.Scanln(&answer) + answer = strings.ToLower(strings.TrimSpace(answer)) + if answer != "y" && answer != "yes" { + return openapi_types.UUID{}, fmt.Errorf("tag %q does not exist (creation declined)", value) + } + } + return ResolveOrCreateTag(ctx, c, value) +} + +// RunWatchUploadOnce performs one watch-upload scan. +func RunWatchUploadOnce(ctx context.Context, c *client.Client, opts WatchUploadOptions) (WatchUploadStats, error) { + var stats WatchUploadStats + now := time.Now() + + files, err := ScanWatchDir(opts.WatchDir, opts.Mode, opts.Depth) + if err != nil { + return stats, err + } + + st, err := loadWatchUploadState(opts.WatchDir) + if err != nil { + return stats, err + } + stateDirty := false + + // Stability gate first (cheap: size+mtime only). + type pending struct { + f WatchFile + tagValue string + checksum string + } + var stables []pending + tagValues := map[string]bool{} + for _, f := range files { + // Already completed and unchanged: skip without re-hashing. + if e, ok := st[f.Rel]; ok && e.AssetID != "" && e.Checksum != "" && + e.Size == f.Size && e.MtimeUnix == f.ModTime.Unix() { + stats.SkippedDone++ + continue + } + if !stableForGate(st, f.Rel, f.Size, f.ModTime, now, opts.StableFor) { + stats.SkippedUnstable++ + stateDirty = true + continue + } + stateDirty = true + var tv string + if opts.Mode == "flat" { + tv, err = RenderTagPattern(opts.TagPattern, now) + if err != nil { + return stats, err + } + } else { + tv = f.Subfolder + } + tagValues[tv] = true + // Reuse cached checksum when size+mtime match to avoid re-hashing videos. + cs := "" + if e, ok := st[f.Rel]; ok && e.Size == f.Size && e.MtimeUnix == f.ModTime.Unix() { + cs = e.Checksum + } + stables = append(stables, pending{f: f, tagValue: tv, checksum: cs}) + } + + if len(stables) == 0 { + if stateDirty && !opts.DryRun { + _ = saveWatchUploadState(opts.WatchDir, st) + } + return stats, nil + } + + // Hash stable files (unless cached). + checksums := make(map[string]string, len(stables)) + var toCheck []string + var relOf = make(map[string]string) // abs -> rel + for _, p := range stables { + relOf[p.f.AbsPath] = p.f.Rel + if p.checksum != "" { + checksums[p.f.AbsPath] = p.checksum + toCheck = append(toCheck, p.f.AbsPath) + continue + } + sum, herr := fileSHA1Base64(p.f.AbsPath) + if herr != nil { + fmt.Fprintf(os.Stderr, "Error: file %s: %v\n", p.f.AbsPath, herr) + stats.Failed++ + continue + } + checksums[p.f.AbsPath] = sum + toCheck = append(toCheck, p.f.AbsPath) + e := st[p.f.Rel] + e.Checksum, e.Size, e.MtimeUnix = sum, p.f.Size, p.f.ModTime.Unix() + st[p.f.Rel] = e + } + + // Server is source of truth: bulk-upload-check first. + results := map[string]BulkCheckEntry{} + if len(toCheck) > 0 && !opts.DryRun { + entries, err := CheckBulkUploadChecksums(ctx, c, toCheck, checksums) + if err != nil { + return stats, err + } + for _, e := range entries { + results[e.File] = e + } + } + + // Resolve tags (and optional album) unless dry-run only prints. + tagIDs := map[string]openapi_types.UUID{} + var albumID *openapi_types.UUID + if !opts.DryRun { + for tv := range tagValues { + id, err := ensureWatchTag(ctx, c, tv, false, opts.Yes) + if err != nil { + return stats, err + } + tagIDs[tv] = id + } + if opts.AlbumID != nil { + album, err := ResolveAlbum(ctx, c, opts.AlbumID, "") + if err != nil { + return stats, err + } + albumID = &album.Id + } else if opts.AlbumName != "" { + album, err := ResolveOrCreateAlbumByName(ctx, c, opts.AlbumName, false, opts.Yes) + if err != nil { + return stats, err + } + albumID = &album.Id + } + } + + quiet := opts.Quiet || opts.JSON + // Group links per tag/album for fewer requests. + tagGroups := map[string][]openapi_types.UUID{} + var albumIDs []openapi_types.UUID + + for i, p := range stables { + if _, ok := checksums[p.f.AbsPath]; !ok { + continue // hashing failed already counted + } + if opts.DryRun { + tv := p.tagValue + if opts.Mode == "flat" { + tv, _ = RenderTagPattern(opts.TagPattern, now) + } + fmt.Printf("[dry-run] would upload %s (tag %q)\n", p.f.AbsPath, tv) + continue + } + r, ok := results[p.f.AbsPath] + if !ok { + stats.Failed++ + continue + } + var assetID openapi_types.UUID + switch r.Status { + case BulkCheckMissing: + prog := NewByteProgress(p.f.AbsPath, p.f.Size, i+1, len(stables), quiet) + id, uerr := UploadAssetFile(ctx, c, p.f.AbsPath, UploadOptions{Checksum: &r.Checksum, Progress: prog}) + prog.Finish() + if uerr != nil { + var dup *DuplicateUploadError + if errors.As(uerr, &dup) && dup.HasID { + assetID = dup.ExistingID + stats.LinkedDuplicate++ + } else { + fmt.Fprintf(os.Stderr, "Error: file %s: %v\n", p.f.AbsPath, uerr) + stats.Failed++ + continue + } + } else { + assetID = id + stats.Uploaded++ + } + case BulkCheckUploaded: + assetID = *r.AssetID + stats.LinkedDuplicate++ + default: + fmt.Fprintf(os.Stderr, "Error: file %s: unsupported format (server has no assetId)\n", p.f.AbsPath) + stats.Failed++ + continue + } + tagGroups[p.tagValue] = append(tagGroups[p.tagValue], assetID) + if albumID != nil { + albumIDs = append(albumIDs, assetID) + } + e := st[p.f.Rel] + e.Checksum, e.AssetID, e.Size, e.MtimeUnix = checksums[p.f.AbsPath], assetID.String(), p.f.Size, p.f.ModTime.Unix() + st[p.f.Rel] = e + } + + // Link duplicates/uploads to tags + optional album. + for tv, ids := range tagGroups { + if err := TagAssets(ctx, c, ids, tagIDs[tv]); err != nil { + fmt.Fprintf(os.Stderr, "Error: tagging %d asset(s) with %q: %v\n", len(ids), tv, err) + stats.Failed += len(ids) + } + } + if albumID != nil && len(albumIDs) > 0 { + for i := 0; i < len(albumIDs); i += 500 { + end := min(i+500, len(albumIDs)) + resp, err := c.API.AddAssetsToAlbumWithResponse(ctx, *albumID, immichapi.BulkIdsDto{Ids: albumIDs[i:end]}) + if err != nil { + fmt.Fprintf(os.Stderr, "Error: adding %d asset(s) to album: %v\n", end-i, err) + stats.Failed += end - i + } else if resp.JSON200 != nil { + for _, r := range *resp.JSON200 { + if !r.Success { + fmt.Fprintf(os.Stderr, "Error: asset %s: %v\n", r.Id, r.Error) + stats.Failed++ + } + } + } + } + } + + if !opts.DryRun { + if err := saveWatchUploadState(opts.WatchDir, st); err != nil { + return stats, err + } + } + if stats.Failed > 0 { + return stats, fmt.Errorf("%d file(s) failed", stats.Failed) + } + return stats, nil +} From ab2fba47cd1577f7e2d327cc9073f0ef3909588f Mon Sep 17 00:00:00 2001 From: Daniel Date: Sat, 19 Sep 2026 10:21:58 +0000 Subject: [PATCH 2/5] feat(watch-upload): default --stable-for 10s, --once forces 0s (upload immediately) --- immich-admin-cli/SKILL.md | 8 ++--- internal/commands/clientworkflow_watch.go | 35 +++++++++++++------ .../commands/clientworkflow_watch_test.go | 16 +++++++++ internal/commands/skill_best_practices.md | 4 +-- 4 files changed, 47 insertions(+), 16 deletions(-) diff --git a/immich-admin-cli/SKILL.md b/immich-admin-cli/SKILL.md index feaa0ac..18cde5a 100644 --- a/immich-admin-cli/SKILL.md +++ b/immich-admin-cli/SKILL.md @@ -388,8 +388,8 @@ Watch a folder and auto-upload new stable files (POST /assets/bulk-upload-check - `--album-id` (string): opt-in album `ID` to also add uploads to - `--album-name` (string): opt-in album name to also add uploads to (created unless --dry-run) - `--interval` (string): poll interval (e.g. 60s); <=0 means run once [default: "60s"] -- `--stable-for` (string): defer files changed within this long (e.g. 30s) [default: "30s"] -- `--once` (bool): run a single scan and exit (for cron) +- `--stable-for` (string): defer files changed within this long (default 10s; forced to 0s with --once) [default: "10s"] +- `--once` (bool): run a single scan and exit (for cron); uploads immediately (--stable-for is forced to 0s) - `--dry-run` (bool): print what would be uploaded/linked without changing anything - `--yes` (bool): skip creation prompts for tags/albums - `--quiet` (bool): disable per-file progress bars on stderr (--json implies quiet) @@ -569,6 +569,6 @@ albums separate, or rename instead of merging blindly. Always run `--sidecar`/`--filename` are single-file only; per-file byte bar on stderr, `--quiet` disables, `--json` implies quiet). `assets check-remote-exists FILE|DIR...` (alias `check-bulk-upload`) hashes locally and reports `uploaded ` / `missing` / `unsupported` without mutating (`--json`, `--ids-only -q` pipes duplicate IDs into `albums add-assets`/`tags tag`). -- **Watch**: `client-workflow watch-upload --watch-dir DIR --mode flat|by-subfolder [--tag-pattern "immich-admin-cli/watch/{yyyy-MM-dd}"] [--album-id|--album-name] --interval 60s --stable-for 30s [--once] [--dry-run] [--yes]` - polls and uploads only stable files (bulk-check first; duplicates only linked). `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` +- **Watch**: `client-workflow watch-upload --watch-dir DIR --mode flat|by-subfolder [--tag-pattern "immich-admin-cli/watch/{yyyy-MM-dd}"] [--album-id|--album-name] --interval 60s --stable-for 10s [--once] [--dry-run] [--yes]` + polls and uploads only stable files (bulk-check first; duplicates only linked; `--once` forces `--stable-for 0s`, i.e. uploads immediately). `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` loops the `.immich-sync.json` manifest sync (album or tag source). `--interval <=0` means run once. Per-interval one-line summary on stderr; `--json` prints JSON stats on stdout. diff --git a/internal/commands/clientworkflow_watch.go b/internal/commands/clientworkflow_watch.go index d92ef9d..a24360e 100644 --- a/internal/commands/clientworkflow_watch.go +++ b/internal/commands/clientworkflow_watch.go @@ -39,10 +39,11 @@ func watchUploadCommand() *cli.Command { Name: "watch-upload", Usage: "Watch a folder and auto-upload new stable files (POST /assets/bulk-upload-check + POST /assets)", Description: "Polls --watch-dir every --interval (default 60s) and uploads new files. " + - "Files whose size/mtime changed within --stable-for (default 30s) are deferred to the next interval. " + + "Files whose size/mtime changed within --stable-for (default 10s) are deferred to the next interval. " + "Duplicates (bulk-upload-check reject/duplicate with assetId) are only linked to the tag/album, never re-uploaded. " + "--mode flat tags everything with --tag-pattern (default immich-admin-cli/watch/{yyyy-MM-dd}); " + - "--mode by-subfolder uses the raw subfolder name. --album-id/--album-name is opt-in. --once runs a single scan (for cron).", + "--mode by-subfolder uses the raw subfolder name. --album-id/--album-name is opt-in. --once runs a single scan (for cron) " + + "and uploads immediately (--stable-for is forced to 0s).", Flags: []cli.Flag{ &cli.StringFlag{Name: "watch-dir", Usage: "local directory to watch", Required: true}, &cli.StringFlag{Name: "mode", Usage: "flat or by-subfolder", Value: "flat"}, @@ -51,8 +52,8 @@ func watchUploadCommand() *cli.Command { &cli.StringFlag{Name: "album-id", Usage: "opt-in album `ID` to also add uploads to"}, &cli.StringFlag{Name: "album-name", Usage: "opt-in album name to also add uploads to (created unless --dry-run)"}, &cli.StringFlag{Name: "interval", Usage: "poll interval (e.g. 60s); <=0 means run once", Value: "60s"}, - &cli.StringFlag{Name: "stable-for", Usage: "defer files changed within this long (e.g. 30s)", Value: "30s"}, - &cli.BoolFlag{Name: "once", Usage: "run a single scan and exit (for cron)"}, + &cli.StringFlag{Name: "stable-for", Usage: "defer files changed within this long (default 10s; forced to 0s with --once)", Value: "10s"}, + &cli.BoolFlag{Name: "once", Usage: "run a single scan and exit (for cron); uploads immediately (--stable-for is forced to 0s)"}, &cli.BoolFlag{Name: "dry-run", Usage: "print what would be uploaded/linked without changing anything"}, &cli.BoolFlag{Name: "yes", Usage: "skip creation prompts for tags/albums"}, &cli.BoolFlag{Name: "quiet", Usage: "disable per-file progress bars on stderr (--json implies quiet)"}, @@ -101,6 +102,23 @@ func parseIntervalOrOnce(raw string, once bool) (time.Duration, bool, error) { return d, once, nil } +// effectiveStableFor parses --stable-for, forcing 0 when --once is set: +// a one-shot run uploads immediately without waiting for stability. +// Pure for testing. +func effectiveStableFor(raw string, once bool) (time.Duration, error) { + if once { + return 0, nil + } + d, err := time.ParseDuration(raw) + if err != nil { + return 0, fmt.Errorf("invalid --stable-for %q: %w", raw, err) + } + if d < 0 { + return 0, fmt.Errorf("invalid --stable-for %q: must not be negative", raw) + } + return d, nil +} + func clientWorkflowWatchUpload(ctx context.Context, cmd *cli.Command) error { if cmd.Args().Len() > 0 { return fmt.Errorf("watch-upload takes no positional arguments (got %v)", cmd.Args().Slice()) @@ -109,14 +127,11 @@ func clientWorkflowWatchUpload(ctx context.Context, cmd *cli.Command) error { if mode != "flat" && mode != "by-subfolder" { return fmt.Errorf("invalid --mode %q: must be flat or by-subfolder", mode) } - stableFor, err := time.ParseDuration(cmd.String("stable-for")) + interval, once, err := parseIntervalOrOnce(cmd.String("interval"), cmd.Bool("once")) if err != nil { - return fmt.Errorf("invalid --stable-for %q: %w", cmd.String("stable-for"), err) - } - if stableFor < 0 { - return fmt.Errorf("invalid --stable-for %q: must not be negative", cmd.String("stable-for")) + return err } - interval, once, err := parseIntervalOrOnce(cmd.String("interval"), cmd.Bool("once")) + stableFor, err := effectiveStableFor(cmd.String("stable-for"), once) if err != nil { return err } diff --git a/internal/commands/clientworkflow_watch_test.go b/internal/commands/clientworkflow_watch_test.go index 4d07ea9..f6dc4af 100644 --- a/internal/commands/clientworkflow_watch_test.go +++ b/internal/commands/clientworkflow_watch_test.go @@ -17,3 +17,19 @@ func TestParseIntervalOrOnce(t *testing.T) { t.Error("bogus interval expected error") } } + +func TestEffectiveStableFor(t *testing.T) { + if d, err := effectiveStableFor("10s", false); err != nil || d != 10*time.Second { + t.Fatalf("10s,false = %v,%v; want 10s,nil", d, err) + } + // --once forces immediate upload regardless of the flag value. + if d, err := effectiveStableFor("10s", true); err != nil || d != 0 { + t.Fatalf("10s,true = %v,%v; want 0,nil", d, err) + } + if _, err := effectiveStableFor("bogus", false); err == nil { + t.Error("bogus stable-for expected error") + } + if _, err := effectiveStableFor("-1s", false); err == nil { + t.Error("negative stable-for expected error") + } +} diff --git a/internal/commands/skill_best_practices.md b/internal/commands/skill_best_practices.md index f3ea146..bb22a64 100644 --- a/internal/commands/skill_best_practices.md +++ b/internal/commands/skill_best_practices.md @@ -118,6 +118,6 @@ albums separate, or rename instead of merging blindly. Always run `--sidecar`/`--filename` are single-file only; per-file byte bar on stderr, `--quiet` disables, `--json` implies quiet). `assets check-remote-exists FILE|DIR...` (alias `check-bulk-upload`) hashes locally and reports `uploaded ` / `missing` / `unsupported` without mutating (`--json`, `--ids-only -q` pipes duplicate IDs into `albums add-assets`/`tags tag`). -- **Watch**: `client-workflow watch-upload --watch-dir DIR --mode flat|by-subfolder [--tag-pattern "immich-admin-cli/watch/{yyyy-MM-dd}"] [--album-id|--album-name] --interval 60s --stable-for 30s [--once] [--dry-run] [--yes]` - polls and uploads only stable files (bulk-check first; duplicates only linked). `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` +- **Watch**: `client-workflow watch-upload --watch-dir DIR --mode flat|by-subfolder [--tag-pattern "immich-admin-cli/watch/{yyyy-MM-dd}"] [--album-id|--album-name] --interval 60s --stable-for 10s [--once] [--dry-run] [--yes]` + polls and uploads only stable files (bulk-check first; duplicates only linked; `--once` forces `--stable-for 0s`, i.e. uploads immediately). `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` loops the `.immich-sync.json` manifest sync (album or tag source). `--interval <=0` means run once. Per-interval one-line summary on stderr; `--json` prints JSON stats on stdout. From a0c60c585532ea935e7d5000d5a5c55d937eb397 Mon Sep 17 00:00:00 2001 From: Daniel Date: Sat, 19 Sep 2026 10:26:43 +0000 Subject: [PATCH 3/5] feat(watch-upload): report already-done count, warn on empty scan --- internal/commands/clientworkflow_watch.go | 4 ++-- internal/workflows/watchupload.go | 12 +++++++++++- 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/internal/commands/clientworkflow_watch.go b/internal/commands/clientworkflow_watch.go index a24360e..fd31718 100644 --- a/internal/commands/clientworkflow_watch.go +++ b/internal/commands/clientworkflow_watch.go @@ -196,8 +196,8 @@ func printWatchUploadSummary(stats workflows.WatchUploadStats, asJSON bool) { _ = enc.Encode(stats) return } - fmt.Fprintf(os.Stderr, "uploaded=%d linked-duplicate=%d skipped-unstable=%d failed=%d\n", - stats.Uploaded, stats.LinkedDuplicate, stats.SkippedUnstable, stats.Failed) + fmt.Fprintf(os.Stderr, "uploaded=%d linked-duplicate=%d skipped-unstable=%d already-done=%d failed=%d\n", + stats.Uploaded, stats.LinkedDuplicate, stats.SkippedUnstable, stats.SkippedDone, stats.Failed) } func clientWorkflowWatchDownload(ctx context.Context, cmd *cli.Command) error { diff --git a/internal/workflows/watchupload.go b/internal/workflows/watchupload.go index c726030..3e812e1 100644 --- a/internal/workflows/watchupload.go +++ b/internal/workflows/watchupload.go @@ -46,7 +46,10 @@ type WatchUploadStats struct { LinkedDuplicate int `json:"linkedDuplicate"` SkippedUnstable int `json:"skippedUnstable"` Failed int `json:"failed"` - SkippedDone int `json:"-"` + // SkippedDone counts files already uploaded+linked in a previous run + // (recognized via the state file, no re-hash, no API call). Reported + // so re-runs show what happened instead of a silent all-zeros line. + SkippedDone int `json:"alreadyDone"` } // WatchFile is one candidate local file. @@ -263,6 +266,13 @@ func RunWatchUploadOnce(ctx context.Context, c *client.Client, opts WatchUploadO if err != nil { return stats, err } + if len(files) == 0 { + scope := "top-level files" + if opts.Mode == "by-subfolder" { + scope = "files directly inside immediate subfolders" + } + fmt.Fprintf(os.Stderr, "warning: no files found in %q (--mode %s scans %s only; hidden files and symlinks are skipped)\n", opts.WatchDir, opts.Mode, scope) + } st, err := loadWatchUploadState(opts.WatchDir) if err != nil { From 833069bae334d8872c95c2cf3de28ffea8103f79 Mon Sep 17 00:00:00 2001 From: Daniel Date: Sat, 19 Sep 2026 10:28:47 +0000 Subject: [PATCH 4/5] docs(skill): --once suffices for settled watch-upload folders --- immich-admin-cli/SKILL.md | 4 +++- internal/commands/skill_best_practices.md | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/immich-admin-cli/SKILL.md b/immich-admin-cli/SKILL.md index 18cde5a..8e62cd6 100644 --- a/immich-admin-cli/SKILL.md +++ b/immich-admin-cli/SKILL.md @@ -570,5 +570,7 @@ albums separate, or rename instead of merging blindly. Always run `assets check-remote-exists FILE|DIR...` (alias `check-bulk-upload`) hashes locally and reports `uploaded ` / `missing` / `unsupported` without mutating (`--json`, `--ids-only -q` pipes duplicate IDs into `albums add-assets`/`tags tag`). - **Watch**: `client-workflow watch-upload --watch-dir DIR --mode flat|by-subfolder [--tag-pattern "immich-admin-cli/watch/{yyyy-MM-dd}"] [--album-id|--album-name] --interval 60s --stable-for 10s [--once] [--dry-run] [--yes]` - polls and uploads only stable files (bulk-check first; duplicates only linked; `--once` forces `--stable-for 0s`, i.e. uploads immediately). `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` + polls and uploads only stable files (bulk-check first; duplicates only linked; `--once` forces `--stable-for 0s`, i.e. uploads immediately). + A finished folder that nothing is still writing into needs just `--once`; run the loop (no `--once`) only while files are still arriving. + `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` loops the `.immich-sync.json` manifest sync (album or tag source). `--interval <=0` means run once. Per-interval one-line summary on stderr; `--json` prints JSON stats on stdout. diff --git a/internal/commands/skill_best_practices.md b/internal/commands/skill_best_practices.md index bb22a64..62fbfac 100644 --- a/internal/commands/skill_best_practices.md +++ b/internal/commands/skill_best_practices.md @@ -119,5 +119,7 @@ albums separate, or rename instead of merging blindly. Always run `assets check-remote-exists FILE|DIR...` (alias `check-bulk-upload`) hashes locally and reports `uploaded ` / `missing` / `unsupported` without mutating (`--json`, `--ids-only -q` pipes duplicate IDs into `albums add-assets`/`tags tag`). - **Watch**: `client-workflow watch-upload --watch-dir DIR --mode flat|by-subfolder [--tag-pattern "immich-admin-cli/watch/{yyyy-MM-dd}"] [--album-id|--album-name] --interval 60s --stable-for 10s [--once] [--dry-run] [--yes]` - polls and uploads only stable files (bulk-check first; duplicates only linked; `--once` forces `--stable-for 0s`, i.e. uploads immediately). `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` + polls and uploads only stable files (bulk-check first; duplicates only linked; `--once` forces `--stable-for 0s`, i.e. uploads immediately). + A finished folder that nothing is still writing into needs just `--once`; run the loop (no `--once`) only while files are still arriving. + `client-workflow watch-download (--album-id|--album-name|--tag-id|--tag-value) --target-dir DIR --size original --interval 300s [--once]` loops the `.immich-sync.json` manifest sync (album or tag source). `--interval <=0` means run once. Per-interval one-line summary on stderr; `--json` prints JSON stats on stdout. From bd2791a879f9104155b19c356583ac279d20e266 Mon Sep 17 00:00:00 2001 From: Daniel Date: Sat, 19 Sep 2026 10:32:21 +0000 Subject: [PATCH 5/5] fix(watch-upload): readable album-add errors, duplicate already-in-album is benign --- internal/workflows/watch_test.go | 19 +++++++++++++++++++ internal/workflows/watchupload.go | 27 +++++++++++++++++++++++++-- 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/internal/workflows/watch_test.go b/internal/workflows/watch_test.go index ba0f53a..cd571c0 100644 --- a/internal/workflows/watch_test.go +++ b/internal/workflows/watch_test.go @@ -59,6 +59,25 @@ func TestFormatByteProgressLine(t *testing.T) { } } +func TestBulkIDFailure(t *testing.T) { + id := openapi_types.UUID([16]byte{2}) + if got := bulkIDFailure(immichapi.BulkIdResponseDto{Id: id, Success: true}); got != "" { + t.Fatalf("success = %q, want empty", got) + } + dup := immichapi.BulkIdErrorReasonDuplicate + if got := bulkIDFailure(immichapi.BulkIdResponseDto{Id: id, Error: &dup}); got != "" { + t.Fatalf("duplicate = %q, want empty (already in album)", got) + } + perm := immichapi.BulkIdErrorReasonNoPermission + msg := "denied" + if got := bulkIDFailure(immichapi.BulkIdResponseDto{Id: id, Error: &perm, ErrorMessage: &msg}); got != "denied" { + t.Fatalf("no_permission = %q, want %q", got, msg) + } + if got := bulkIDFailure(immichapi.BulkIdResponseDto{Id: id, Error: &perm}); got != "no_permission" { + t.Fatalf("no_permission bare = %q, want the reason value, not a pointer", got) + } +} + func TestStableForGate(t *testing.T) { st := map[string]watchUploadEntry{} now := time.Now() diff --git a/internal/workflows/watchupload.go b/internal/workflows/watchupload.go index 3e812e1..d0b1911 100644 --- a/internal/workflows/watchupload.go +++ b/internal/workflows/watchupload.go @@ -203,6 +203,29 @@ func stableForGate(st map[string]watchUploadEntry, rel string, size int64, mtime return now.Unix()-e.FirstSeen >= int64(stableFor.Seconds()) } +// bulkIDFailure renders a per-ID album-add failure, or "" when the outcome +// is benign: success, duplicate (already in the album — the normal case on +// re-runs after the state file was deleted), or not-found. Same buckets as +// the `albums add-assets` tally and merge-album. Pure for testing. +func bulkIDFailure(r immichapi.BulkIdResponseDto) string { + if r.Success { + return "" + } + if r.Error != nil { + switch *r.Error { + case immichapi.BulkIdErrorReasonDuplicate, immichapi.BulkIdErrorReasonNotFound: + return "" + } + } + if r.ErrorMessage != nil && *r.ErrorMessage != "" { + return *r.ErrorMessage + } + if r.Error != nil && string(*r.Error) != "" { + return string(*r.Error) + } + return "server reported failure" +} + // ResolveOrCreateAlbumByName finds an album by exact name or creates it. // dryRun prints instead of creating; yes skips the creation prompt. func ResolveOrCreateAlbumByName(ctx context.Context, c *client.Client, name string, dryRun, yes bool) (immichapi.AlbumResponseDto, error) { @@ -463,8 +486,8 @@ func RunWatchUploadOnce(ctx context.Context, c *client.Client, opts WatchUploadO stats.Failed += end - i } else if resp.JSON200 != nil { for _, r := range *resp.JSON200 { - if !r.Success { - fmt.Fprintf(os.Stderr, "Error: asset %s: %v\n", r.Id, r.Error) + if msg := bulkIDFailure(r); msg != "" { + fmt.Fprintf(os.Stderr, "Error: asset %s: %s\n", r.Id, msg) stats.Failed++ } }