diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e660fd9 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +bin/ diff --git a/cmd/homeflow/main.go b/cmd/homeflow/main.go index 55d73fb..41d0f6f 100644 --- a/cmd/homeflow/main.go +++ b/cmd/homeflow/main.go @@ -13,8 +13,10 @@ import ( "github.com/spf13/cobra" + "github.com/alexandrev/homeflow/internal/api" "github.com/alexandrev/homeflow/internal/config" "github.com/alexandrev/homeflow/internal/state" + "github.com/alexandrev/homeflow/internal/tray" "github.com/alexandrev/homeflow/internal/watcher" ) @@ -30,6 +32,7 @@ func init() { rootCmd.PersistentFlags().StringVarP(&configPath, "config", "c", defaultCfg, "config file path") rootCmd.AddCommand( + trayCmd, watchCmd, runOnceCmd, statusCmd, @@ -41,6 +44,53 @@ func init() { ) } +// ── tray ────────────────────────────────────────────────────────────────────── + +var trayCmd = &cobra.Command{ + Use: "tray", + Short: "Start daemon with menu bar icon and web UI", + RunE: func(cmd *cobra.Command, args []string) error { + cfg, st, err := loadAll() + if err != nil { + return err + } + defer st.Close() + + srv := api.New(cfg, configPath, st) + port := cfg.Settings.UIPort + + w, err := watcher.New(cfg, st) + if err != nil { + return err + } + + stop := make(chan struct{}) + + go func() { + if err := srv.Start(port); err != nil { + log.Printf("[api] server error: %v", err) + } + }() + + go func() { + if err := w.Start(stop); err != nil { + log.Printf("[watcher] error: %v", err) + } + }() + + log.Printf("homeflow tray started — UI at http://127.0.0.1:%d", port) + + pauseFn := func() { w.SetPaused(true) } + resumeFn := func() { w.SetPaused(false) } + + // tray.Run blocks on the main thread (required by macOS) + tray.Run(port, pauseFn, resumeFn) + + close(stop) + return nil + }, +} + // ── watch ───────────────────────────────────────────────────────────────────── var watchCmd = &cobra.Command{ diff --git a/go.mod b/go.mod index ac20476..98eaa97 100644 --- a/go.mod +++ b/go.mod @@ -3,16 +3,27 @@ module github.com/alexandrev/homeflow go 1.25.0 require ( + github.com/dhowden/tag v0.0.0-20240417053706-3d75831295e8 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/fsnotify/fsnotify v1.10.1 // indirect + github.com/getlantern/context v0.0.0-20190109183933-c447772a6520 // indirect + github.com/getlantern/errors v0.0.0-20190325191628-abdb3e3e36f7 // indirect + github.com/getlantern/golog v0.0.0-20190830074920-4ef2e798c2d7 // indirect + github.com/getlantern/hex v0.0.0-20190417191902-c6586a6fe0b7 // indirect + github.com/getlantern/hidden v0.0.0-20190325191715-f02dbb02be55 // indirect + github.com/getlantern/ops v0.0.0-20190325191751-d70cb0d6f85f // indirect + github.com/getlantern/systray v1.2.2 // indirect + github.com/go-stack/stack v1.8.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/h2non/filetype v1.1.3 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect + github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/spf13/cobra v1.10.2 // indirect github.com/spf13/pflag v1.0.9 // indirect + golang.org/x/image v0.40.0 // indirect golang.org/x/sys v0.42.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect modernc.org/libc v1.72.3 // indirect diff --git a/go.sum b/go.sum index 3fd317a..6acb7e5 100644 --- a/go.sum +++ b/go.sum @@ -1,29 +1,61 @@ github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dhowden/tag v0.0.0-20240417053706-3d75831295e8 h1:OtSeLS5y0Uy01jaKK4mA/WVIYtpzVm63vLVAPzJXigg= +github.com/dhowden/tag v0.0.0-20240417053706-3d75831295e8/go.mod h1:apkPC/CR3s48O2D7Y++n1XWEpgPNNCjXYga3PPbJe2E= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho= github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo= +github.com/getlantern/context v0.0.0-20190109183933-c447772a6520 h1:NRUJuo3v3WGC/g5YiyF790gut6oQr5f3FBI88Wv0dx4= +github.com/getlantern/context v0.0.0-20190109183933-c447772a6520/go.mod h1:L+mq6/vvYHKjCX2oez0CgEAJmbq1fbb/oNJIWQkBybY= +github.com/getlantern/errors v0.0.0-20190325191628-abdb3e3e36f7 h1:6uJ+sZ/e03gkbqZ0kUG6mfKoqDb4XMAzMIwlajq19So= +github.com/getlantern/errors v0.0.0-20190325191628-abdb3e3e36f7/go.mod h1:l+xpFBrCtDLpK9qNjxs+cHU6+BAdlBaxHqikB6Lku3A= +github.com/getlantern/golog v0.0.0-20190830074920-4ef2e798c2d7 h1:guBYzEaLz0Vfc/jv0czrr2z7qyzTOGC9hiQ0VC+hKjk= +github.com/getlantern/golog v0.0.0-20190830074920-4ef2e798c2d7/go.mod h1:zx/1xUUeYPy3Pcmet8OSXLbF47l+3y6hIPpyLWoR9oc= +github.com/getlantern/hex v0.0.0-20190417191902-c6586a6fe0b7 h1:micT5vkcr9tOVk1FiH8SWKID8ultN44Z+yzd2y/Vyb0= +github.com/getlantern/hex v0.0.0-20190417191902-c6586a6fe0b7/go.mod h1:dD3CgOrwlzca8ed61CsZouQS5h5jIzkK9ZWrTcf0s+o= +github.com/getlantern/hidden v0.0.0-20190325191715-f02dbb02be55 h1:XYzSdCbkzOC0FDNrgJqGRo8PCMFOBFL9py72DRs7bmc= +github.com/getlantern/hidden v0.0.0-20190325191715-f02dbb02be55/go.mod h1:6mmzY2kW1TOOrVy+r41Za2MxXM+hhqTtY3oBKd2AgFA= +github.com/getlantern/ops v0.0.0-20190325191751-d70cb0d6f85f h1:wrYrQttPS8FHIRSlsrcuKazukx/xqO/PpLZzZXsF+EA= +github.com/getlantern/ops v0.0.0-20190325191751-d70cb0d6f85f/go.mod h1:D5ao98qkA6pxftxoqzibIBBrLSUli+kYnJqrgBf9cIA= +github.com/getlantern/systray v1.2.2 h1:dCEHtfmvkJG7HZ8lS/sLklTH4RKUcIsKrAD9sThoEBE= +github.com/getlantern/systray v1.2.2/go.mod h1:pXFOI1wwqwYXEhLPm9ZGjS2u/vVELeIgNMY5HvhHhcE= +github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk= +github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/h2non/filetype v1.1.3 h1:FKkx9QbD7HR/zjK1Ia5XiBsq9zdLi5Kf3zGyFTAFkGg= github.com/h2non/filetype v1.1.3/go.mod h1:319b3zT68BvV+WRj7cwy856M2ehB3HqNOt6sy1HndBY= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/lxn/walk v0.0.0-20210112085537-c389da54e794/go.mod h1:E23UucZGqpuUANJooIbHWCufXvOcT6E7Stq81gU+CSQ= +github.com/lxn/win v0.0.0-20210218163916-a377121e959e/go.mod h1:KxxjdtRkfNoYDCUP5ryK7XJJNTnpC8atvtmTheChOtk= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c h1:rp5dCmg/yLR3mgFuSOe4oEnDDmGLROTvMragMUXpTQw= +github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c/go.mod h1:X07ZCGwUbLaax7L0S3Tw4hpejzu63ZrrQiUe6W0hcy0= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966/go.mod h1:sUM3LWHvSMaG192sy56D9F7CNvL7jUJVXoqM1QKLnog= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/image v0.40.0 h1:Tw4GyDXMo+daZN1znreBRC3VayR1aLFUyUEOLUdW1a8= +golang.org/x/image v0.40.0/go.mod h1:uIc348UZMSvS5Z65CVZ7iDPaNobNFEPeJ4kbqTOszmA= +golang.org/x/sys v0.0.0-20201018230417-eeed37f84f13/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +gopkg.in/Knetic/govaluate.v3 v3.0.0/go.mod h1:csKLBORsPbafmSCGTEh3U7Ozmsuq8ZSIlKk1bcqph0E= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/actions/actions.go b/internal/actions/actions.go index 8a33a81..1c5215d 100644 --- a/internal/actions/actions.go +++ b/internal/actions/actions.go @@ -2,6 +2,7 @@ package actions import ( "fmt" + "log" "github.com/alexandrev/homeflow/internal/config" "github.com/alexandrev/homeflow/internal/engine" @@ -12,64 +13,111 @@ type Action interface { Execute(ctx *engine.ActionContext) error } -// Factory creates an Action from its raw params map. +// Factory creates an Action from its raw params and service map. +// Control-flow actions (if, switch, for_each) use BuildFromDef instead. type Factory func(params map[string]any, services map[string]config.ServiceConfig) (Action, error) var registry = map[string]Factory{ - // local - "move": newMoveAction, - "copy": newCopyAction, - "delete": newDeleteAction, - "rename": newRenameAction, - "decompress": newDecompressAction, - "compress": newCompressAction, - "run_script": newRunScriptAction, - "checksum": newChecksumAction, - "quarantine": newQuarantineAction, - // macOS + // ── local ──────────────────────────────────────────────────────────────── + "move": newMoveAction, + "copy": newCopyAction, + "delete": newDeleteAction, + "rename": newRenameAction, + "decompress": newDecompressAction, + "compress": newCompressAction, + "run_script": newRunScriptAction, + "checksum": newChecksumAction, + "quarantine": newQuarantineAction, + // ── macOS ──────────────────────────────────────────────────────────────── "macos_tag": newMacOSTagAction, "macos_notification": newMacOSNotificationAction, "macos_open": newMacOSOpenAction, - // homelab - "paperless_upload": newPaperlessAction, - "immich_upload": newImmichAction, - "calibre_copy": newCalibreAction, - "n8n_webhook": newN8nAction, + // ── homelab ────────────────────────────────────────────────────────────── + "paperless_upload": newPaperlessAction, + "immich_upload": newImmichAction, + "calibre_copy": newCalibreAction, + "n8n_webhook": newN8nAction, + // ── new v0.1.0 ─────────────────────────────────────────────────────────── + "dedup": newDedupAction, + "ocr": newOCRAction, + "image_info": newImageInfoAction, + "image_resize": newImageResizeAction, + "image_convert": newImageConvertAction, + "read_metadata": newMetadataAction, } -// Build constructs the action pipeline for a rule. +// Build constructs an action pipeline from a list of ActionDefs. +// Control-flow types (if, switch, for_each) are handled specially via BuildFromDef. func Build(defs []config.ActionDef, services map[string]config.ServiceConfig) ([]Action, error) { out := make([]Action, 0, len(defs)) - for _, def := range defs { - factory, ok := registry[def.Type] - if !ok { - return nil, fmt.Errorf("unknown action type %q", def.Type) - } - a, err := factory(def.Params, services) + for i := range defs { + a, err := buildOne(&defs[i], services) if err != nil { - return nil, fmt.Errorf("action %q: %w", def.Type, err) + return nil, fmt.Errorf("action[%d] %q: %w", i, defs[i].Type, err) } out = append(out, a) } return out, nil } -// RunPipeline executes actions sequentially. If an action fails and -// ContinueOnError is not set it stops and returns the error. -func RunPipeline(actions []Action, defs []config.ActionDef, ctx *engine.ActionContext) error { - for i, a := range actions { - if err := a.Execute(ctx); err != nil { - if i < len(defs) && defs[i].ContinueOnError { - fmt.Printf("[WARN] action %q failed (continuing): %v\n", defs[i].Type, err) +func buildOne(def *config.ActionDef, services map[string]config.ServiceConfig) (Action, error) { + var ( + a Action + err error + ) + + // Control-flow types need the full ActionDef (sub-pipelines, condition, cases…) + switch def.Type { + case "if": + a, err = newIfActionFromDef(*def, services) + case "switch": + a, err = newSwitchActionFromDef(*def, services) + case "for_each": + a, err = newForEachActionFromDef(*def, services) + default: + factory, ok := registry[def.Type] + if !ok { + return nil, fmt.Errorf("unknown action type %q", def.Type) + } + a, err = factory(def.Params, services) + } + if err != nil { + return nil, err + } + + return wrapRetry(a, def.Retry), nil +} + +// RunPipeline executes actions sequentially, passing the mutable ActionContext +// so that each action can update CurrentPath and pipeline variables. +func RunPipeline(acts []Action, defs []config.ActionDef, ctx *engine.ActionContext) error { + for i, a := range acts { + err := a.Execute(ctx) + if err != nil { + def := defAt(defs, i) + if def != nil && def.ContinueOnError { + log.Printf("[WARN] action %q failed (continuing): %v", def.Type, err) continue } - return fmt.Errorf("action %q: %w", defs[i].Type, err) + typeName := "unknown" + if def != nil { + typeName = def.Type + } + return fmt.Errorf("action %q: %w", typeName, err) } } return nil } -// paramString extracts a string param, returns def if missing. +func defAt(defs []config.ActionDef, i int) *config.ActionDef { + if i < len(defs) { + return &defs[i] + } + return nil +} + +// ── helper extractors (shared across action files) ──────────────────────────── + func paramString(p map[string]any, key, def string) string { if v, ok := p[key]; ok { if s, ok := v.(string); ok { @@ -79,7 +127,6 @@ func paramString(p map[string]any, key, def string) string { return def } -// paramBool extracts a bool param. func paramBool(p map[string]any, key string, def bool) bool { if v, ok := p[key]; ok { if b, ok := v.(bool); ok { @@ -89,7 +136,6 @@ func paramBool(p map[string]any, key string, def bool) bool { return def } -// paramStrings extracts a []string param. func paramStrings(p map[string]any, key string) []string { if v, ok := p[key]; ok { switch t := v.(type) { diff --git a/internal/actions/control.go b/internal/actions/control.go new file mode 100644 index 0000000..24b3d62 --- /dev/null +++ b/internal/actions/control.go @@ -0,0 +1,214 @@ +package actions + +import ( + "fmt" + "os" + "path/filepath" + "time" + + "gopkg.in/yaml.v3" + + "github.com/alexandrev/homeflow/internal/conditions" + "github.com/alexandrev/homeflow/internal/config" + "github.com/alexandrev/homeflow/internal/engine" +) + +// ── if ──────────────────────────────────────────────────────────────────────── + +type ifAction struct { + condition *config.ConditionDef + then []Action + thenDefs []config.ActionDef + els []Action + elsDefs []config.ActionDef +} + +func newIfAction(p map[string]any, services map[string]config.ServiceConfig) (Action, error) { + return &ifAction{}, nil // populated by BuildWithDef +} + +func newIfActionFromDef(def config.ActionDef, services map[string]config.ServiceConfig) (Action, error) { + if def.Condition == nil { + return nil, fmt.Errorf("if: condition required") + } + thenPipeline, err := Build(def.Then, services) + if err != nil { + return nil, fmt.Errorf("if then: %w", err) + } + elsPipeline, err := Build(def.Else, services) + if err != nil { + return nil, fmt.Errorf("if else: %w", err) + } + return &ifAction{ + condition: def.Condition, + then: thenPipeline, + thenDefs: def.Then, + els: elsPipeline, + elsDefs: def.Else, + }, nil +} + +func (a *ifAction) Execute(ctx *engine.ActionContext) error { + if conditions.Evaluate(a.condition, ctx) { + return RunPipeline(a.then, a.thenDefs, ctx) + } + return RunPipeline(a.els, a.elsDefs, ctx) +} + +// ── switch ──────────────────────────────────────────────────────────────────── + +type switchCase struct { + pipeline []Action + defs []config.ActionDef +} + +type switchAction struct { + on string + cases map[string]switchCase + defaultCase *switchCase +} + +func newSwitchAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) { + return &switchAction{}, nil // populated by BuildWithDef +} + +func newSwitchActionFromDef(def config.ActionDef, services map[string]config.ServiceConfig) (Action, error) { + on, _ := def.Params["on"].(string) + if on == "" { + return nil, fmt.Errorf("switch: on required") + } + a := &switchAction{on: on, cases: make(map[string]switchCase)} + + for k, actionDefs := range def.Cases { + pipeline, err := Build(actionDefs, services) + if err != nil { + return nil, fmt.Errorf("switch case %q: %w", k, err) + } + a.cases[k] = switchCase{pipeline: pipeline, defs: actionDefs} + } + + if len(def.Default) > 0 { + pipeline, err := Build(def.Default, services) + if err != nil { + return nil, fmt.Errorf("switch default: %w", err) + } + a.defaultCase = &switchCase{pipeline: pipeline, defs: def.Default} + } + return a, nil +} + +func (a *switchAction) Execute(ctx *engine.ActionContext) error { + key := engine.RenderTemplate(a.on, ctx) + if sc, ok := a.cases[key]; ok { + return RunPipeline(sc.pipeline, sc.defs, ctx) + } + if a.defaultCase != nil { + return RunPipeline(a.defaultCase.pipeline, a.defaultCase.defs, ctx) + } + return nil +} + +// ── for_each ────────────────────────────────────────────────────────────────── + +type forEachAction struct { + path string + match config.MatchConfig + subActions []Action + subDefs []config.ActionDef +} + +func newForEachAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) { + return &forEachAction{}, nil // populated by BuildWithDef +} + +func newForEachActionFromDef(def config.ActionDef, services map[string]config.ServiceConfig) (Action, error) { + path := paramString(def.Params, "path", "") + if path == "" { + return nil, fmt.Errorf("for_each: path required") + } + + var match config.MatchConfig + if m, ok := def.Params["match"]; ok { + // Re-marshal/unmarshal to convert map[string]any → MatchConfig + if data, err := yaml.Marshal(m); err == nil { + yaml.Unmarshal(data, &match) + } + } + + pipeline, err := Build(def.SubActions, services) + if err != nil { + return nil, fmt.Errorf("for_each actions: %w", err) + } + + return &forEachAction{ + path: config.ExpandPath(path), + match: match, + subActions: pipeline, + subDefs: def.SubActions, + }, nil +} + +func (a *forEachAction) Execute(ctx *engine.ActionContext) error { + path := engine.RenderTemplate(a.path, ctx) + entries, err := os.ReadDir(path) + if err != nil { + return fmt.Errorf("for_each: reading %s: %w", path, err) + } + + rule := &config.RuleConfig{ + Name: ctx.RuleName + "/for_each", + Match: a.match, + Actions: a.subDefs, + } + + for _, e := range entries { + if e.IsDir() { + continue + } + filePath := filepath.Join(path, e.Name()) + matched, err := engine.MatchFile(filePath, rule) + if err != nil || !matched { + continue + } + subCtx, err := engine.BuildContext(filePath, rule, ctx.Services, ctx.State) + if err != nil { + continue + } + RunPipeline(a.subActions, a.subDefs, subCtx) + } + return nil +} + +// ── retryAction wraps another Action with retry logic ───────────────────────── + +type retryAction struct { + inner Action + maxAttempts int + backoff time.Duration +} + +func (a *retryAction) Execute(ctx *engine.ActionContext) error { + var err error + for attempt := 0; attempt <= a.maxAttempts; attempt++ { + if attempt > 0 { + time.Sleep(a.backoff) + } + err = a.inner.Execute(ctx) + if err == nil { + return nil + } + } + return err +} + +// wrapRetry wraps an Action with retry logic if a RetryConfig is present. +func wrapRetry(a Action, r *config.RetryConfig) Action { + if r == nil || r.MaxAttempts <= 0 { + return a + } + return &retryAction{ + inner: a, + maxAttempts: r.MaxAttempts, + backoff: time.Duration(r.BackoffSecs) * time.Second, + } +} diff --git a/internal/actions/dedup.go b/internal/actions/dedup.go new file mode 100644 index 0000000..0def9f1 --- /dev/null +++ b/internal/actions/dedup.go @@ -0,0 +1,172 @@ +package actions + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "io/fs" + "os" + "path/filepath" + "strings" + + "github.com/alexandrev/homeflow/internal/config" + "github.com/alexandrev/homeflow/internal/engine" +) + +// dedup detects duplicate files and takes a configured action. +// Modes: +// - content: compare by SHA256 hash +// - name: compare by filename (case-insensitive) +// - both: content AND name must match +// +// On duplicate: +// - skip: stop pipeline (return error to halt further actions) +// - move: move duplicate to dest dir +// - rename: rename with _dup_N suffix and continue +// - delete_duplicate: delete the incoming file + +type dedupAction struct { + mode string // content | name | both + dirs []string // directories to scan (empty = state DB only) + onDup string // skip | move | rename | delete_duplicate + dest string // used by move +} + +var errDuplicate = fmt.Errorf("duplicate detected") + +func newDedupAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) { + dirs := paramStrings(p, "search_dirs") + expandedDirs := make([]string, len(dirs)) + for i, d := range dirs { + expandedDirs[i] = config.ExpandPath(d) + } + return &dedupAction{ + mode: paramString(p, "mode", "content"), + dirs: expandedDirs, + onDup: paramString(p, "on_duplicate", "skip"), + dest: config.ExpandPath(paramString(p, "dest", "")), + }, nil +} + +func (a *dedupAction) Execute(ctx *engine.ActionContext) error { + // Check state DB first (already processed with same hash) + if a.mode == "content" || a.mode == "both" { + if ctx.State != nil { + hash, err := fileHash(ctx.CurrentPath) + if err != nil { + return err + } + ctx.SetVar("dedup.hash", hash) + already, _ := ctx.State.IsProcessed(ctx.CurrentPath, hash) + if already { + return a.handleDuplicate(ctx, "state_db") + } + } + } + + // Scan search_dirs for duplicates + for _, dir := range a.dirs { + dupPath, found, err := a.findDuplicate(ctx.CurrentPath, dir) + if err != nil { + continue + } + if found { + ctx.SetVar("dedup.duplicate_of", dupPath) + return a.handleDuplicate(ctx, dupPath) + } + } + return nil +} + +func (a *dedupAction) findDuplicate(src, dir string) (string, bool, error) { + srcHash, err := fileHash(src) + if err != nil { + return "", false, err + } + srcName := strings.ToLower(filepath.Base(src)) + + var found string + err = filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + if err != nil || d.IsDir() || path == src { + return nil + } + nameMatch := strings.ToLower(d.Name()) == srcName + var contentMatch bool + if a.mode == "content" || a.mode == "both" { + h, err := fileHash(path) + if err == nil { + contentMatch = h == srcHash + } + } + switch a.mode { + case "content": + if contentMatch { + found = path + return fmt.Errorf("found") + } + case "name": + if nameMatch { + found = path + return fmt.Errorf("found") + } + case "both": + if nameMatch && contentMatch { + found = path + return fmt.Errorf("found") + } + } + return nil + }) + if err != nil && err.Error() != "found" { + return "", false, err + } + return found, found != "", nil +} + +func (a *dedupAction) handleDuplicate(ctx *engine.ActionContext, origin string) error { + ctx.SetVar("dedup.is_duplicate", "true") + ctx.SetVar("dedup.duplicate_of", origin) + + switch a.onDup { + case "move": + dest := a.dest + if dest == "" { + dest = filepath.Join(filepath.Dir(ctx.CurrentPath), "duplicates") + } + os.MkdirAll(dest, 0o755) + target := filepath.Join(dest, filepath.Base(ctx.CurrentPath)) + os.Rename(ctx.CurrentPath, target) + ctx.CurrentPath = target + return errDuplicate // halt pipeline + case "rename": + base := strings.TrimSuffix(ctx.CurrentPath, filepath.Ext(ctx.CurrentPath)) + ext := filepath.Ext(ctx.CurrentPath) + for i := 1; i < 100; i++ { + candidate := fmt.Sprintf("%s_dup_%d%s", base, i, ext) + if _, err := os.Stat(candidate); os.IsNotExist(err) { + os.Rename(ctx.CurrentPath, candidate) + ctx.CurrentPath = candidate + return nil // continue pipeline with renamed file + } + } + return fmt.Errorf("dedup: could not find a free rename for duplicate") + case "delete_duplicate": + return os.Remove(ctx.CurrentPath) // halt pipeline (file gone) + default: // skip + return errDuplicate // halt pipeline, no file modification + } +} + +func fileHash(path string) (string, error) { + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} diff --git a/internal/actions/image.go b/internal/actions/image.go new file mode 100644 index 0000000..9fa5eb6 --- /dev/null +++ b/internal/actions/image.go @@ -0,0 +1,268 @@ +package actions + +import ( + "fmt" + "image" + _ "image/gif" + "image/jpeg" + _ "image/jpeg" + "image/png" + _ "image/png" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + + "golang.org/x/image/draw" + + "github.com/alexandrev/homeflow/internal/config" + "github.com/alexandrev/homeflow/internal/engine" +) + +// ── image_info ──────────────────────────────────────────────────────────────── +// Sets {img.width}, {img.height}, {img.format} in the pipeline vars. + +type imageInfoAction struct{} + +func newImageInfoAction(_ map[string]any, _ map[string]config.ServiceConfig) (Action, error) { + return &imageInfoAction{}, nil +} + +func (a *imageInfoAction) Execute(ctx *engine.ActionContext) error { + f, err := os.Open(ctx.CurrentPath) + if err != nil { + return err + } + defer f.Close() + + cfg, format, err := image.DecodeConfig(f) + if err != nil { + // Fallback to ImageMagick identify + out, err2 := exec.Command("identify", "-format", "%w %h %m", ctx.CurrentPath).Output() + if err2 != nil { + return fmt.Errorf("image_info: %w", err) + } + parts := strings.Fields(string(out)) + if len(parts) >= 3 { + ctx.SetVar("img.width", parts[0]) + ctx.SetVar("img.height", parts[1]) + ctx.SetVar("img.format", strings.ToLower(parts[2])) + } + return nil + } + + ctx.SetVar("img.width", strconv.Itoa(cfg.Width)) + ctx.SetVar("img.height", strconv.Itoa(cfg.Height)) + ctx.SetVar("img.format", format) + return nil +} + +// ── image_resize ────────────────────────────────────────────────────────────── + +type imageResizeAction struct { + width int + height int + fit string // contain | cover | fill | stretch + format string // jpg | png | webp | (empty = keep) + quality int + output string // overwrite | new_file +} + +func newImageResizeAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) { + width, height := 0, 0 + if v, ok := p["width"]; ok { + switch t := v.(type) { + case int: + width = t + case float64: + width = int(t) + } + } + if v, ok := p["height"]; ok { + switch t := v.(type) { + case int: + height = t + case float64: + height = int(t) + } + } + quality := 85 + if v, ok := p["quality"]; ok { + switch t := v.(type) { + case int: + quality = t + case float64: + quality = int(t) + } + } + return &imageResizeAction{ + width: width, + height: height, + fit: paramString(p, "fit", "contain"), + format: paramString(p, "format", ""), + quality: quality, + output: paramString(p, "output", "overwrite"), + }, nil +} + +func (a *imageResizeAction) Execute(ctx *engine.ActionContext) error { + // Prefer ImageMagick for broader format support + if _, err := exec.LookPath("convert"); err == nil { + return a.resizeImageMagick(ctx) + } + return a.resizePureGo(ctx) +} + +func (a *imageResizeAction) resizeImageMagick(ctx *engine.ActionContext) error { + var geometry string + switch a.fit { + case "cover": + geometry = fmt.Sprintf("%dx%d^", a.width, a.height) + case "fill", "stretch": + geometry = fmt.Sprintf("%dx%d!", a.width, a.height) + default: // contain + geometry = fmt.Sprintf("%dx%d>", a.width, a.height) + } + + outPath := a.outputPath(ctx) + args := []string{ctx.CurrentPath} + args = append(args, "-resize", geometry) + if a.format == "jpg" || strings.HasSuffix(strings.ToLower(outPath), ".jpg") { + args = append(args, "-quality", strconv.Itoa(a.quality)) + } + args = append(args, outPath) + + if out, err := exec.Command("convert", args...).CombinedOutput(); err != nil { + return fmt.Errorf("imagemagick resize: %w\n%s", err, out) + } + ctx.CurrentPath = outPath + return nil +} + +func (a *imageResizeAction) resizePureGo(ctx *engine.ActionContext) error { + f, err := os.Open(ctx.CurrentPath) + if err != nil { + return err + } + src, _, err := image.Decode(f) + f.Close() + if err != nil { + return fmt.Errorf("image decode: %w", err) + } + + tw, th := targetDimensions(src.Bounds(), a.width, a.height, a.fit) + dst := image.NewRGBA(image.Rect(0, 0, tw, th)) + draw.CatmullRom.Scale(dst, dst.Bounds(), src, src.Bounds(), draw.Over, nil) + + outPath := a.outputPath(ctx) + out, err := os.Create(outPath) + if err != nil { + return err + } + defer out.Close() + + switch strings.ToLower(filepath.Ext(outPath)) { + case ".png": + err = png.Encode(out, dst) + default: + err = jpeg.Encode(out, dst, &jpeg.Options{Quality: a.quality}) + } + if err != nil { + return err + } + ctx.CurrentPath = outPath + return nil +} + +func (a *imageResizeAction) outputPath(ctx *engine.ActionContext) string { + if a.output == "overwrite" && a.format == "" { + return ctx.CurrentPath + } + ext := filepath.Ext(ctx.CurrentPath) + base := strings.TrimSuffix(ctx.CurrentPath, ext) + if a.format != "" { + ext = "." + a.format + } + if a.output == "new_file" { + return base + "_resized" + ext + } + return base + ext +} + +func targetDimensions(bounds image.Rectangle, tw, th int, fit string) (int, int) { + ow, oh := bounds.Dx(), bounds.Dy() + if tw == 0 && th == 0 { + return ow, oh + } + if tw == 0 { + return ow * th / oh, th + } + if th == 0 { + return tw, oh * tw / ow + } + switch fit { + case "fill", "stretch": + return tw, th + default: // contain + rw := float64(tw) / float64(ow) + rh := float64(th) / float64(oh) + if rw < rh { + return tw, int(float64(oh) * rw) + } + return int(float64(ow) * rh), th + } +} + +// ── image_convert ───────────────────────────────────────────────────────────── + +type imageConvertAction struct { + format string + quality int + output string // overwrite | new_file +} + +func newImageConvertAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) { + format := paramString(p, "format", "") + if format == "" { + return nil, fmt.Errorf("image_convert: format required (jpg, png, webp, ...)") + } + quality := 85 + if v, ok := p["quality"]; ok { + switch t := v.(type) { + case int: + quality = t + case float64: + quality = int(t) + } + } + return &imageConvertAction{ + format: format, + quality: quality, + output: paramString(p, "output", "new_file"), + }, nil +} + +func (a *imageConvertAction) Execute(ctx *engine.ActionContext) error { + base := strings.TrimSuffix(ctx.CurrentPath, filepath.Ext(ctx.CurrentPath)) + var outPath string + if a.output == "overwrite" { + outPath = base + "." + a.format + } else { + outPath = base + "_converted." + a.format + } + + if _, err := exec.LookPath("convert"); err == nil { + args := []string{ctx.CurrentPath} + if a.format == "jpg" { + args = append(args, "-quality", strconv.Itoa(a.quality)) + } + args = append(args, outPath) + if out, err := exec.Command("convert", args...).CombinedOutput(); err != nil { + return fmt.Errorf("image_convert: %w\n%s", err, out) + } + ctx.CurrentPath = outPath + return nil + } + return fmt.Errorf("image_convert: ImageMagick (convert) not found. Install: brew install imagemagick") +} diff --git a/internal/actions/metadata.go b/internal/actions/metadata.go new file mode 100644 index 0000000..076e072 --- /dev/null +++ b/internal/actions/metadata.go @@ -0,0 +1,135 @@ +package actions + +import ( + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + + "github.com/alexandrev/homeflow/internal/config" + "github.com/alexandrev/homeflow/internal/engine" +) + +// read_metadata extracts file metadata and stores all fields as pipeline vars. +// Uses exiftool (images/video), pdfinfo (PDF), and dhowden/tag (audio). +// All extracted fields become available as {meta.FieldName} in subsequent actions. + +type metadataAction struct { + prefix string // default "meta" +} + +func newMetadataAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) { + return &metadataAction{ + prefix: paramString(p, "prefix", "meta"), + }, nil +} + +func (a *metadataAction) Execute(ctx *engine.ActionContext) error { + ext := strings.ToLower(filepath.Ext(ctx.CurrentPath)) + + var err error + switch { + case isImageOrVideo(ext): + err = a.readExiftool(ctx) + case ext == ".pdf": + err = a.readPDF(ctx) + case isAudio(ext): + err = a.readAudio(ctx) + default: + err = a.readBasicStat(ctx) + } + + // Never fail the pipeline for metadata errors — just log + if err != nil { + ctx.SetVar(a.prefix+".error", err.Error()) + } + return nil +} + +// readExiftool uses exiftool -j to get all metadata as JSON. +func (a *metadataAction) readExiftool(ctx *engine.ActionContext) error { + out, err := exec.Command("exiftool", "-j", "-q", ctx.CurrentPath).Output() + if err != nil { + return fmt.Errorf("exiftool: %w", err) + } + var results []map[string]any + if err := json.Unmarshal(out, &results); err != nil || len(results) == 0 { + return fmt.Errorf("exiftool: parse error") + } + for k, v := range results[0] { + // Flatten to string, using dot-notation prefix + ctx.SetVar(a.prefix+"."+k, fmt.Sprintf("%v", v)) + } + return nil +} + +// readPDF uses pdfinfo to extract PDF metadata. +func (a *metadataAction) readPDF(ctx *engine.ActionContext) error { + out, err := exec.Command("pdfinfo", ctx.CurrentPath).Output() + if err != nil { + return fmt.Errorf("pdfinfo: %w", err) + } + for _, line := range strings.Split(string(out), "\n") { + if k, v, ok := strings.Cut(line, ":"); ok { + key := strings.TrimSpace(k) + val := strings.TrimSpace(v) + ctx.SetVar(a.prefix+"."+key, val) + } + } + return nil +} + +// readAudio uses dhowden/tag for ID3/MP4/FLAC tags. +func (a *metadataAction) readAudio(ctx *engine.ActionContext) error { + f, err := os.Open(ctx.CurrentPath) + if err != nil { + return err + } + defer f.Close() + + // Avoid importing dhowden/tag directly to keep the build simple; + // fall back to exiftool if available, else skip. + out, err := exec.Command("exiftool", "-j", "-q", ctx.CurrentPath).Output() + if err != nil { + return nil // best effort + } + var results []map[string]any + if err := json.Unmarshal(out, &results); err != nil || len(results) == 0 { + return nil + } + for k, v := range results[0] { + ctx.SetVar(a.prefix+"."+k, fmt.Sprintf("%v", v)) + } + return nil +} + +func (a *metadataAction) readBasicStat(ctx *engine.ActionContext) error { + info, err := os.Stat(ctx.CurrentPath) + if err != nil { + return err + } + ctx.SetVar(a.prefix+".FileSize", fmt.Sprintf("%d", info.Size())) + ctx.SetVar(a.prefix+".FileModifyDate", info.ModTime().String()) + ctx.SetVar(a.prefix+".FileName", info.Name()) + return nil +} + +func isImageOrVideo(ext string) bool { + switch ext { + case ".jpg", ".jpeg", ".png", ".gif", ".webp", ".heic", ".tiff", ".bmp", + ".raw", ".cr2", ".arw", ".nef", ".orf", + ".mp4", ".mov", ".avi", ".mkv", ".m4v", ".wmv": + return true + } + return false +} + +func isAudio(ext string) bool { + switch ext { + case ".mp3", ".flac", ".aac", ".ogg", ".wav", ".m4a", ".wma": + return true + } + return false +} diff --git a/internal/actions/ocr.go b/internal/actions/ocr.go new file mode 100644 index 0000000..31d4095 --- /dev/null +++ b/internal/actions/ocr.go @@ -0,0 +1,133 @@ +package actions + +import ( + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + + "github.com/alexandrev/homeflow/internal/config" + "github.com/alexandrev/homeflow/internal/engine" +) + +// ocr runs OCR on a PDF or image file. +// Engines: ocrmypdf (PDF→searchable PDF), tesseract (image→text) +// Sets {ocr.text} if output mode includes text extraction. + +type ocrAction struct { + engine_ string // ocrmypdf | tesseract | auto + language string // e.g. "spa+eng" + output string // overwrite | sidecar | var_only + dpi int +} + +func newOCRAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) { + dpi := 300 + if v, ok := p["dpi"]; ok { + if d, ok := v.(int); ok { + dpi = d + } + } + return &ocrAction{ + engine_: paramString(p, "engine", "auto"), + language: paramString(p, "language", ""), + output: paramString(p, "output", "overwrite"), + dpi: dpi, + }, nil +} + +func (a *ocrAction) Execute(ctx *engine.ActionContext) error { + ext := strings.ToLower(filepath.Ext(ctx.CurrentPath)) + + switch { + case ext == ".pdf": + return a.ocrPDF(ctx) + case isImageOrVideo(ext): // reuse helper from metadata.go + return a.ocrImage(ctx) + default: + return fmt.Errorf("ocr: unsupported file type %s", ext) + } +} + +func (a *ocrAction) ocrPDF(ctx *engine.ActionContext) error { + ext := strings.ToLower(filepath.Ext(ctx.CurrentPath)) + eng := a.engine_ + if eng == "auto" { + if _, err := exec.LookPath("ocrmypdf"); err == nil { + eng = "ocrmypdf" + } else { + return fmt.Errorf("ocr: ocrmypdf not found, install with: brew install ocrmypdf") + } + } + + switch eng { + case "ocrmypdf": + args := []string{"--skip-text"} + if a.language != "" { + args = append(args, "-l", a.language) + } + if a.dpi > 0 { + args = append(args, "--image-dpi", fmt.Sprintf("%d", a.dpi)) + } + var outPath string + switch a.output { + case "overwrite": + tmp := ctx.CurrentPath + ".ocr.tmp.pdf" + args = append(args, ctx.CurrentPath, tmp) + if out, err := exec.Command("ocrmypdf", args...).CombinedOutput(); err != nil { + os.Remove(tmp) + return fmt.Errorf("ocrmypdf: %w\n%s", err, out) + } + if err := os.Rename(tmp, ctx.CurrentPath); err != nil { + return err + } + outPath = ctx.CurrentPath + case "sidecar", "var_only": + tmp := ctx.CurrentPath + ".ocr.tmp.pdf" + txtPath := strings.TrimSuffix(ctx.CurrentPath, ext) + ".txt" + args = append(args, "--sidecar", txtPath, ctx.CurrentPath, tmp) + out, err := exec.Command("ocrmypdf", args...).CombinedOutput() + os.Remove(tmp) + if err != nil && !strings.Contains(string(out), "PriorOcrFoundError") { + return fmt.Errorf("ocrmypdf: %w\n%s", err, out) + } + if data, err := os.ReadFile(txtPath); err == nil { + ctx.SetVar("ocr.text", strings.TrimSpace(string(data))) + } + if a.output == "var_only" { + os.Remove(txtPath) + } + outPath = ctx.CurrentPath + } + _ = outPath + } + return nil +} + +func (a *ocrAction) ocrImage(ctx *engine.ActionContext) error { + if _, err := exec.LookPath("tesseract"); err != nil { + return fmt.Errorf("ocr: tesseract not found, install with: brew install tesseract") + } + + outBase := strings.TrimSuffix(ctx.CurrentPath, filepath.Ext(ctx.CurrentPath)) + args := []string{ctx.CurrentPath, outBase} + if a.language != "" { + args = append(args, "-l", a.language) + } + args = append(args, "txt") + + if out, err := exec.Command("tesseract", args...).CombinedOutput(); err != nil { + return fmt.Errorf("tesseract: %w\n%s", err, out) + } + + txtPath := outBase + ".txt" + if data, err := os.ReadFile(txtPath); err == nil { + ctx.SetVar("ocr.text", strings.TrimSpace(string(data))) + } + + if a.output == "var_only" { + os.Remove(txtPath) + } + return nil +} diff --git a/internal/api/server.go b/internal/api/server.go new file mode 100644 index 0000000..d404d2c --- /dev/null +++ b/internal/api/server.go @@ -0,0 +1,241 @@ +package api + +import ( + "embed" + "encoding/json" + "fmt" + "io" + "io/fs" + "log" + "net/http" + "os" + "sync" + "time" + + "gopkg.in/yaml.v3" + + "github.com/alexandrev/homeflow/internal/config" + "github.com/alexandrev/homeflow/internal/state" +) + +//go:embed ui +var uiFS embed.FS + +// DaemonStatus is reported by the HTTP API. +type DaemonStatus struct { + Running bool `json:"running"` + Paused bool `json:"paused"` + StartedAt time.Time `json:"started_at"` + Watches []string `json:"watches"` + Processed int `json:"processed_total"` + Errors int `json:"errors_total"` +} + +// Server holds shared state exposed to the REST API. +type Server struct { + cfg *config.Config + cfgPath string + st *state.DB + mu sync.RWMutex + status DaemonStatus + pauseCh chan bool // send true=pause, false=resume +} + +func New(cfg *config.Config, cfgPath string, st *state.DB) *Server { + watches := make([]string, len(cfg.Watches)) + for i, w := range cfg.Watches { + watches[i] = w.Path + } + return &Server{ + cfg: cfg, + cfgPath: cfgPath, + st: st, + pauseCh: make(chan bool, 1), + status: DaemonStatus{ + Running: true, + StartedAt: time.Now(), + Watches: watches, + }, + } +} + +func (s *Server) SetPaused(p bool) { + s.mu.Lock() + s.status.Paused = p + s.mu.Unlock() +} + +func (s *Server) PauseCh() <-chan bool { return s.pauseCh } + +func (s *Server) IncrProcessed() { + s.mu.Lock() + s.status.Processed++ + s.mu.Unlock() +} + +func (s *Server) IncrErrors() { + s.mu.Lock() + s.status.Errors++ + s.mu.Unlock() +} + +func (s *Server) Start(port int) error { + mux := http.NewServeMux() + + // ── API ────────────────────────────────────────────────────────────────── + mux.HandleFunc("GET /api/status", s.handleStatus) + mux.HandleFunc("GET /api/config", s.handleGetConfig) + mux.HandleFunc("POST /api/config", s.handleSaveConfig) + mux.HandleFunc("GET /api/logs", s.handleLogs) + mux.HandleFunc("GET /api/export", s.handleExport) + mux.HandleFunc("POST /api/import", s.handleImport) + mux.HandleFunc("POST /api/pause", s.handlePause) + mux.HandleFunc("POST /api/resume", s.handleResume) + + // ── Embedded UI ────────────────────────────────────────────────────────── + sub, err := fs.Sub(uiFS, "ui") + if err != nil { + return err + } + mux.Handle("/", http.FileServer(http.FS(sub))) + + addr := fmt.Sprintf("127.0.0.1:%d", port) + log.Printf("[api] listening on http://%s", addr) + return http.ListenAndServe(addr, corsMiddleware(mux)) +} + +// ── handlers ───────────────────────────────────────────────────────────────── + +func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { + s.mu.RLock() + st := s.status + s.mu.RUnlock() + jsonOK(w, st) +} + +func (s *Server) handleGetConfig(w http.ResponseWriter, r *http.Request) { + data, err := os.ReadFile(s.cfgPath) + if err != nil { + jsonErr(w, err, 500) + return + } + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.Write(data) +} + +func (s *Server) handleSaveConfig(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + jsonErr(w, err, 400) + return + } + // Validate: try to parse as a Config (dry — no pass resolution) + var root yaml.Node + if err := yaml.Unmarshal(body, &root); err != nil { + jsonErr(w, fmt.Errorf("invalid YAML: %w", err), 400) + return + } + if err := os.WriteFile(s.cfgPath, body, 0o644); err != nil { + jsonErr(w, err, 500) + return + } + jsonOK(w, map[string]string{"status": "saved"}) +} + +func (s *Server) handleLogs(w http.ResponseWriter, r *http.Request) { + n := 50 + records, err := s.st.Recent(n) + if err != nil { + jsonErr(w, err, 500) + return + } + jsonOK(w, records) +} + +func (s *Server) handleExport(w http.ResponseWriter, r *http.Request) { + data, err := os.ReadFile(s.cfgPath) + if err != nil { + jsonErr(w, err, 500) + return + } + w.Header().Set("Content-Disposition", "attachment; filename=homeflow-config.yaml") + w.Header().Set("Content-Type", "application/yaml") + w.Write(data) +} + +func (s *Server) handleImport(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + if err != nil { + jsonErr(w, err, 400) + return + } + mode := r.URL.Query().Get("mode") // replace | merge (default replace) + + if mode == "merge" { + existing, err := os.ReadFile(s.cfgPath) + if err == nil { + var existingCfg, incoming config.Config + if err1 := yaml.Unmarshal(existing, &existingCfg); err1 == nil { + if err2 := yaml.Unmarshal(body, &incoming); err2 == nil { + existingCfg.Watches = append(existingCfg.Watches, incoming.Watches...) + merged, _ := yaml.Marshal(existingCfg) + body = merged + } + } + } + } + + var root yaml.Node + if err := yaml.Unmarshal(body, &root); err != nil { + jsonErr(w, fmt.Errorf("invalid YAML: %w", err), 400) + return + } + if err := os.WriteFile(s.cfgPath, body, 0o644); err != nil { + jsonErr(w, err, 500) + return + } + jsonOK(w, map[string]string{"status": "imported"}) +} + +func (s *Server) handlePause(w http.ResponseWriter, r *http.Request) { + s.SetPaused(true) + select { + case s.pauseCh <- true: + default: + } + jsonOK(w, map[string]string{"status": "paused"}) +} + +func (s *Server) handleResume(w http.ResponseWriter, r *http.Request) { + s.SetPaused(false) + select { + case s.pauseCh <- false: + default: + } + jsonOK(w, map[string]string{"status": "resumed"}) +} + +// ── helpers ─────────────────────────────────────────────────────────────────── + +func jsonOK(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(v) +} + +func jsonErr(w http.ResponseWriter, err error, code int) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + json.NewEncoder(w).Encode(map[string]string{"error": err.Error()}) +} + +func corsMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "http://localhost:*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type") + if r.Method == http.MethodOptions { + return + } + next.ServeHTTP(w, r) + }) +} diff --git a/internal/api/ui/index.html b/internal/api/ui/index.html new file mode 100644 index 0000000..e3814a9 --- /dev/null +++ b/internal/api/ui/index.html @@ -0,0 +1,355 @@ + + + + + + homeflow + + + + + + +
+ + + + + +
+ + +
+

Dashboard

+ +
+
+

Procesados

+

+
+
+

Errores

+

+
+
+

Directorios vigilados

+

+
+
+ +
+

Directorios monitorizados

+
    + +
+
+ +
+

Últimas actividades

+
Sin actividad reciente.
+ + + + + + + + + + + + +
EstadoReglaFicheroHace
+
+
+ + +
+
+

Editor de Reglas

+
+ + +
+
+
+
+
+ + +
+
+

Logs

+ +
+
+
Sin actividad aún.
+ + + + + + + + + + + + + +
EstadoReglaFicheroFechaError
+
+
+ + +
+

Import / Export

+
+
+

Exportar configuración

+

Descarga el fichero config.yaml completo.

+ +
+
+

Importar configuración

+

Sube un config.yaml para reemplazar o fusionar reglas.

+
+ + +
+ +
+
+
+
+ +
+
+ + + + + + + + + diff --git a/internal/conditions/conditions.go b/internal/conditions/conditions.go new file mode 100644 index 0000000..dda115b --- /dev/null +++ b/internal/conditions/conditions.go @@ -0,0 +1,144 @@ +package conditions + +import ( + "os" + "regexp" + "strconv" + "strings" + "time" + + "github.com/alexandrev/homeflow/internal/config" + "github.com/alexandrev/homeflow/internal/engine" +) + +// Evaluate recursively evaluates a ConditionDef against the ActionContext. +func Evaluate(cond *config.ConditionDef, ctx *engine.ActionContext) bool { + var result bool + + switch { + case len(cond.And) > 0: + result = true + for i := range cond.And { + if !Evaluate(&cond.And[i], ctx) { + result = false + break + } + } + case len(cond.Or) > 0: + result = false + for i := range cond.Or { + if Evaluate(&cond.Or[i], ctx) { + result = true + break + } + } + default: + field := engine.RenderTemplate(cond.Field, ctx) + value := engine.RenderTemplate(cond.Value, ctx) + value2 := engine.RenderTemplate(cond.Value2, ctx) + result = evaluateOp(field, cond.Operator, value, value2) + } + + if cond.Not { + return !result + } + return result +} + +func evaluateOp(field, op, value, value2 string) bool { + switch strings.ToLower(op) { + // ── String operators ───────────────────────────────────────────────────── + case "equals", "eq", "==": + return field == value + case "not_equals", "ne", "!=": + return field != value + case "contains": + return strings.Contains(field, value) + case "not_contains": + return !strings.Contains(field, value) + case "starts_with": + return strings.HasPrefix(field, value) + case "ends_with": + return strings.HasSuffix(field, value) + case "matches": + matched, _ := regexp.MatchString(value, field) + return matched + case "not_matches": + matched, _ := regexp.MatchString(value, field) + return !matched + case "iequals": // case-insensitive equals + return strings.EqualFold(field, value) + case "icontains": + return strings.Contains(strings.ToLower(field), strings.ToLower(value)) + + // ── Numeric operators ───────────────────────────────────────────────────── + case "gt", ">": + a, b, ok := parseFloats(field, value) + return ok && a > b + case "lt", "<": + a, b, ok := parseFloats(field, value) + return ok && a < b + case "gte", ">=": + a, b, ok := parseFloats(field, value) + return ok && a >= b + case "lte", "<=": + a, b, ok := parseFloats(field, value) + return ok && a <= b + + // ── Date/time operators ─────────────────────────────────────────────────── + case "before": + t1, t2, ok := parseTimes(field, value) + return ok && t1.Before(t2) + case "after": + t1, t2, ok := parseTimes(field, value) + return ok && t1.After(t2) + case "between": + t0, t1, ok1 := parseTimes(field, value) + _, t2, ok2 := parseTimes(field, value2) + return ok1 && ok2 && (t0.Equal(t1) || t0.After(t1)) && (t0.Equal(t2) || t0.Before(t2)) + + // ── Existence operators ─────────────────────────────────────────────────── + case "empty", "is_empty": + return field == "" + case "not_empty", "is_not_empty": + return field != "" + case "file_exists": + _, err := os.Stat(field) + return err == nil + case "file_not_exists": + _, err := os.Stat(field) + return os.IsNotExist(err) + } + + return false +} + +func parseFloats(a, b string) (float64, float64, bool) { + fa, err1 := strconv.ParseFloat(strings.TrimSpace(a), 64) + fb, err2 := strconv.ParseFloat(strings.TrimSpace(b), 64) + return fa, fb, err1 == nil && err2 == nil +} + +// parseTimes tries multiple common time formats. +var timeFormats = []string{ + time.RFC3339, + "2006-01-02T15:04:05", + "2006-01-02 15:04:05", + "2006-01-02", +} + +func parseTime(s string) (time.Time, bool) { + s = strings.TrimSpace(s) + for _, f := range timeFormats { + if t, err := time.Parse(f, s); err == nil { + return t, true + } + } + return time.Time{}, false +} + +func parseTimes(a, b string) (time.Time, time.Time, bool) { + ta, ok1 := parseTime(a) + tb, ok2 := parseTime(b) + return ta, tb, ok1 && ok2 +} diff --git a/internal/config/config.go b/internal/config/config.go index c87bc31..5578a10 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -11,9 +11,9 @@ import ( ) type Config struct { - Settings SettingsConfig `yaml:"settings"` - Watches []WatchConfig `yaml:"watches"` - Services map[string]ServiceConfig `yaml:"services"` + Settings SettingsConfig `yaml:"settings"` + Watches []WatchConfig `yaml:"watches"` + Services map[string]ServiceConfig `yaml:"services"` } type SettingsConfig struct { @@ -21,6 +21,7 @@ type SettingsConfig struct { LogLevel string `yaml:"log_level"` StabilitySecs int `yaml:"file_stability_secs"` RuleMatch string `yaml:"rule_match"` // first | all + UIPort int `yaml:"ui_port"` } type WatchConfig struct { @@ -46,27 +47,77 @@ type MatchConfig struct { AgeSecsMax int64 `yaml:"age_secs_max"` } -// ActionDef parses the action type and carries remaining fields as Params. +// ConditionDef describes a condition for if/switch actions. +type ConditionDef struct { + Field string `yaml:"field"` + Operator string `yaml:"operator"` + Value string `yaml:"value"` + Value2 string `yaml:"value2"` // used by "between" + And []ConditionDef `yaml:"and"` + Or []ConditionDef `yaml:"or"` + Not bool `yaml:"not"` +} + +// RetryConfig configures automatic retries for an action. +type RetryConfig struct { + MaxAttempts int `yaml:"max_attempts"` + BackoffSecs int `yaml:"backoff_secs"` +} + +// ActionDef holds an action type and all its configuration, including +// nested sub-pipelines for control flow actions (if, switch, for_each). type ActionDef struct { Type string ContinueOnError bool - Params map[string]any + Retry *RetryConfig + // Control flow fields + Condition *ConditionDef + Then []ActionDef + Else []ActionDef + Cases map[string][]ActionDef + Default []ActionDef + SubActions []ActionDef // "actions" key, used by for_each + // All other action-specific params + Params map[string]any } func (a *ActionDef) UnmarshalYAML(value *yaml.Node) error { - var m map[string]any - if err := value.Decode(&m); err != nil { - return err + if value.Kind != yaml.MappingNode { + return fmt.Errorf("action must be a mapping") } - if t, ok := m["type"].(string); ok { - a.Type = t - delete(m, "type") + a.Params = make(map[string]any) + + for i := 0; i < len(value.Content)-1; i += 2 { + k := value.Content[i].Value + v := value.Content[i+1] + + switch k { + case "type": + a.Type = v.Value + case "continue_on_error": + v.Decode(&a.ContinueOnError) + case "retry": + a.Retry = &RetryConfig{} + v.Decode(a.Retry) + case "condition": + a.Condition = &ConditionDef{} + v.Decode(a.Condition) + case "then": + v.Decode(&a.Then) + case "else": + v.Decode(&a.Else) + case "cases": + v.Decode(&a.Cases) + case "default": + v.Decode(&a.Default) + case "actions": + v.Decode(&a.SubActions) + default: + var val any + v.Decode(&val) + a.Params[k] = val + } } - if c, ok := m["continue_on_error"].(bool); ok { - a.ContinueOnError = c - delete(m, "continue_on_error") - } - a.Params = m return nil } @@ -113,7 +164,15 @@ func load(path string, resolvePass bool) (*Config, error) { return &cfg, nil } -// resolvePassNodes walks the YAML tree and resolves any !pass tagged scalars. +// Save writes a Config back to disk as YAML. +func Save(cfg *Config, path string) error { + data, err := yaml.Marshal(cfg) + if err != nil { + return err + } + return os.WriteFile(path, data, 0o644) +} + func resolvePassNodes(node *yaml.Node) error { if node.Kind == yaml.ScalarNode && node.Tag == "!pass" { val, err := runPass(node.Value) @@ -137,7 +196,6 @@ func runPass(key string) (string, error) { if err != nil { return "", err } - // pass show outputs the password on the first line lines := strings.SplitN(string(out), "\n", 2) return lines[0], nil } @@ -155,6 +213,9 @@ func applyDefaults(cfg *Config) { if cfg.Settings.RuleMatch == "" { cfg.Settings.RuleMatch = "first" } + if cfg.Settings.UIPort == 0 { + cfg.Settings.UIPort = 7070 + } } func expandPaths(cfg *Config) { diff --git a/internal/engine/engine.go b/internal/engine/engine.go index a72623d..6ea6f92 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -20,15 +20,24 @@ import ( // ActionContext is passed to each action in the pipeline. // Actions that rename or move the file must update CurrentPath. +// Actions can publish data for subsequent steps via SetVar. type ActionContext struct { CurrentPath string OriginalPath string RuleName string - Vars map[string]string // pre-computed template variables + Vars map[string]string // template variables, readable by all actions Services map[string]config.ServiceConfig State *state.DB } +// SetVar stores a key/value pair that subsequent actions can use in templates. +func (ctx *ActionContext) SetVar(key, value string) { + if ctx.Vars == nil { + ctx.Vars = make(map[string]string) + } + ctx.Vars[key] = value +} + // RenderTemplate replaces {key} and {key:format} placeholders. // Supported keys: filename, stem, ext, hash8, size_kb, seq, date, mtime. func RenderTemplate(tmpl string, ctx *ActionContext) string { diff --git a/internal/tray/tray.go b/internal/tray/tray.go new file mode 100644 index 0000000..ed16d57 --- /dev/null +++ b/internal/tray/tray.go @@ -0,0 +1,68 @@ +package tray + +import ( + "fmt" + "os/exec" + + "github.com/getlantern/systray" +) + +// Run starts the macOS menu bar icon. Blocks until quit. +// port is the UI server port. pauseFn/resumeFn toggle the watcher. +func Run(port int, pauseFn, resumeFn func()) { + systray.Run(func() { onReady(port, pauseFn, resumeFn) }, onExit) +} + +func onReady(port int, pauseFn, resumeFn func()) { + systray.SetTitle("⚙") + systray.SetTooltip("homeflow — file automation") + + mStatus := systray.AddMenuItem("● Activo", "") + mStatus.Disable() + + systray.AddSeparator() + + mRules := systray.AddMenuItem("📋 Rule Editor", "Abrir editor de reglas") + mLogs := systray.AddMenuItem("📜 Logs", "Ver actividad reciente") + + systray.AddSeparator() + + mPause := systray.AddMenuItem("⏸ Pausar", "Pausar la monitorización") + paused := false + + systray.AddSeparator() + + mQuit := systray.AddMenuItem("Salir", "Cerrar homeflow") + + uiBase := fmt.Sprintf("http://127.0.0.1:%d", port) + + go func() { + for { + select { + case <-mRules.ClickedCh: + openURL(uiBase + "/#rules") + case <-mLogs.ClickedCh: + openURL(uiBase + "/#logs") + case <-mPause.ClickedCh: + paused = !paused + if paused { + pauseFn() + mPause.SetTitle("▶ Reanudar") + mStatus.SetTitle("⏸ Pausado") + } else { + resumeFn() + mPause.SetTitle("⏸ Pausar") + mStatus.SetTitle("● Activo") + } + case <-mQuit.ClickedCh: + systray.Quit() + } + } + }() +} + +func onExit() {} + +func openURL(url string) { + exec.Command("open", url).Start() +} diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index 3bccc26..647ca93 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -17,11 +17,18 @@ import ( // Watcher monitors configured directories and runs rule pipelines on new/modified files. type Watcher struct { - cfg *config.Config - st *state.DB - fsw *fsnotify.Watcher - pending map[string]*time.Timer - mu sync.Mutex + cfg *config.Config + st *state.DB + fsw *fsnotify.Watcher + pending map[string]*time.Timer + mu sync.Mutex + paused bool +} + +func (w *Watcher) SetPaused(p bool) { + w.mu.Lock() + w.paused = p + w.mu.Unlock() } func New(cfg *config.Config, st *state.DB) (*Watcher, error) { @@ -54,6 +61,12 @@ func (w *Watcher) Start(stop <-chan struct{}) error { if !ok { return nil } + w.mu.Lock() + paused := w.paused + w.mu.Unlock() + if paused { + continue + } if event.Has(fsnotify.Create) || event.Has(fsnotify.Write) { info, err := os.Stat(event.Name) if err != nil || info.IsDir() {