feat: initial homeflow implementation
Go-based file automation daemon (~10MB RAM idle) with FSEvents-native file watching, configurable rule/action pipelines, and integrations for Paperless-ngx, Immich, Calibre Web, and n8n webhooks. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,334 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"os/exec"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"syscall"
|
||||
"text/tabwriter"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/state"
|
||||
"github.com/alexandrev/homeflow/internal/watcher"
|
||||
)
|
||||
|
||||
var configPath string
|
||||
|
||||
var rootCmd = &cobra.Command{
|
||||
Use: "homeflow",
|
||||
Short: "File automation daemon with homelab integrations",
|
||||
}
|
||||
|
||||
func init() {
|
||||
defaultCfg := filepath.Join(mustHomeDir(), ".homeflow", "config.yaml")
|
||||
rootCmd.PersistentFlags().StringVarP(&configPath, "config", "c", defaultCfg, "config file path")
|
||||
|
||||
rootCmd.AddCommand(
|
||||
watchCmd,
|
||||
runOnceCmd,
|
||||
statusCmd,
|
||||
configTestCmd,
|
||||
installCmd,
|
||||
uninstallCmd,
|
||||
logsCmd,
|
||||
listRulesCmd,
|
||||
)
|
||||
}
|
||||
|
||||
// ── watch ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
var watchCmd = &cobra.Command{
|
||||
Use: "watch",
|
||||
Short: "Start daemon in foreground",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
cfg, st, err := loadAll()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer st.Close()
|
||||
|
||||
w, err := watcher.New(cfg, st)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
stop := make(chan struct{})
|
||||
sig := make(chan os.Signal, 1)
|
||||
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
|
||||
go func() {
|
||||
<-sig
|
||||
log.Println("shutting down…")
|
||||
close(stop)
|
||||
}()
|
||||
|
||||
log.Printf("homeflow started (pid %d)", os.Getpid())
|
||||
return w.Start(stop)
|
||||
},
|
||||
}
|
||||
|
||||
// ── run-once ──────────────────────────────────────────────────────────────────
|
||||
|
||||
var runOnceCmd = &cobra.Command{
|
||||
Use: "run-once",
|
||||
Short: "Process existing files in watched directories once",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
cfg, st, err := loadAll()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer st.Close()
|
||||
w, err := watcher.New(cfg, st)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
w.RunOnce()
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
// ── status ────────────────────────────────────────────────────────────────────
|
||||
|
||||
var statusCmd = &cobra.Command{
|
||||
Use: "status",
|
||||
Short: "Show recent processing history and daemon status",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
// LaunchAgent status
|
||||
out, _ := exec.Command("launchctl", "list", "com.alexandrev.homeflow").Output()
|
||||
if len(out) > 0 {
|
||||
fmt.Printf("LaunchAgent: running\n%s\n", out)
|
||||
} else {
|
||||
fmt.Println("LaunchAgent: not loaded")
|
||||
}
|
||||
|
||||
cfg, err := config.Load(configPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
st, err := state.Open(cfg.Settings.StateDB)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer st.Close()
|
||||
|
||||
records, err := st.Recent(20)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(records) == 0 {
|
||||
fmt.Println("No records yet.")
|
||||
return nil
|
||||
}
|
||||
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
|
||||
fmt.Fprintln(tw, "STATUS\tRULE\tFILE\tTIME")
|
||||
for _, r := range records {
|
||||
ago := time.Since(r.ProcessedAt).Round(time.Second)
|
||||
file := filepath.Base(r.Path)
|
||||
line := fmt.Sprintf("%s\t%s\t%s\t%s ago", r.Status, r.Rule, file, ago)
|
||||
if r.ErrorMsg != "" {
|
||||
line += "\t— " + r.ErrorMsg
|
||||
}
|
||||
fmt.Fprintln(tw, line)
|
||||
}
|
||||
tw.Flush()
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
// ── config-test ───────────────────────────────────────────────────────────────
|
||||
|
||||
var configTestCmd = &cobra.Command{
|
||||
Use: "config-test",
|
||||
Short: "Validate config file and resolve credentials",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
skipCreds, _ := cmd.Flags().GetBool("skip-credentials")
|
||||
var (
|
||||
cfg *config.Config
|
||||
err error
|
||||
)
|
||||
if skipCreds {
|
||||
cfg, err = config.LoadDry(configPath)
|
||||
} else {
|
||||
cfg, err = config.Load(configPath)
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("config error: %w", err)
|
||||
}
|
||||
fmt.Printf("Config OK — %d watch(es), %d service(s)\n",
|
||||
len(cfg.Watches), len(cfg.Services))
|
||||
for _, w := range cfg.Watches {
|
||||
fmt.Printf(" watch: %s (%d rules)\n", w.Path, len(w.Rules))
|
||||
}
|
||||
for name := range cfg.Services {
|
||||
fmt.Printf(" service: %s\n", name)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
func init() {
|
||||
configTestCmd.Flags().Bool("skip-credentials", false, "skip !pass resolution (validate structure only)")
|
||||
}
|
||||
|
||||
// ── install ───────────────────────────────────────────────────────────────────
|
||||
|
||||
var installCmd = &cobra.Command{
|
||||
Use: "install",
|
||||
Short: "Install and load the macOS LaunchAgent",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
binary, err := os.Executable()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
binary, err = filepath.EvalSymlinks(binary)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
home := mustHomeDir()
|
||||
logPath := filepath.Join(home, "Library", "Logs", "homeflow.log")
|
||||
plistDir := filepath.Join(home, "Library", "LaunchAgents")
|
||||
plistPath := filepath.Join(plistDir, "com.alexandrev.homeflow.plist")
|
||||
|
||||
os.MkdirAll(plistDir, 0o755)
|
||||
|
||||
plist := fmt.Sprintf(`<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
|
||||
<plist version="1.0">
|
||||
<dict>
|
||||
<key>Label</key>
|
||||
<string>com.alexandrev.homeflow</string>
|
||||
<key>ProgramArguments</key>
|
||||
<array>
|
||||
<string>%s</string>
|
||||
<string>watch</string>
|
||||
<string>--config</string>
|
||||
<string>%s</string>
|
||||
</array>
|
||||
<key>RunAtLoad</key>
|
||||
<true/>
|
||||
<key>KeepAlive</key>
|
||||
<true/>
|
||||
<key>StandardOutPath</key>
|
||||
<string>%s</string>
|
||||
<key>StandardErrorPath</key>
|
||||
<string>%s</string>
|
||||
<key>EnvironmentVariables</key>
|
||||
<dict>
|
||||
<key>PATH</key>
|
||||
<string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin</string>
|
||||
</dict>
|
||||
</dict>
|
||||
</plist>
|
||||
`, binary, configPath, logPath, logPath)
|
||||
|
||||
if err := os.WriteFile(plistPath, []byte(plist), 0o644); err != nil {
|
||||
return fmt.Errorf("writing plist: %w", err)
|
||||
}
|
||||
|
||||
out, err := exec.Command("launchctl", "load", plistPath).CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("launchctl load: %w\n%s", err, out)
|
||||
}
|
||||
fmt.Printf("LaunchAgent installed and loaded.\nBinary: %s\nConfig: %s\nLog: %s\n",
|
||||
binary, configPath, logPath)
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
// ── uninstall ─────────────────────────────────────────────────────────────────
|
||||
|
||||
var uninstallCmd = &cobra.Command{
|
||||
Use: "uninstall",
|
||||
Short: "Unload and remove the macOS LaunchAgent",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
home := mustHomeDir()
|
||||
plistPath := filepath.Join(home, "Library", "LaunchAgents", "com.alexandrev.homeflow.plist")
|
||||
exec.Command("launchctl", "unload", plistPath).Run()
|
||||
if err := os.Remove(plistPath); err != nil && !os.IsNotExist(err) {
|
||||
return err
|
||||
}
|
||||
fmt.Println("LaunchAgent unloaded and removed.")
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
// ── logs ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
var logsCmd = &cobra.Command{
|
||||
Use: "logs",
|
||||
Short: "Tail the homeflow log file",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
logPath := filepath.Join(mustHomeDir(), "Library", "Logs", "homeflow.log")
|
||||
c := exec.Command("tail", "-f", "-n", "50", logPath)
|
||||
c.Stdout = os.Stdout
|
||||
c.Stderr = os.Stderr
|
||||
return c.Run()
|
||||
},
|
||||
}
|
||||
|
||||
// ── list-rules ────────────────────────────────────────────────────────────────
|
||||
|
||||
var listRulesCmd = &cobra.Command{
|
||||
Use: "list-rules",
|
||||
Short: "List all active rules",
|
||||
RunE: func(cmd *cobra.Command, args []string) error {
|
||||
cfg, err := config.LoadDry(configPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
|
||||
fmt.Fprintln(tw, "WATCH\tRULE\tMATCH\tACTIONS")
|
||||
for _, w := range cfg.Watches {
|
||||
for _, r := range w.Rules {
|
||||
match := fmt.Sprintf("ext=%v", r.Match.Extensions)
|
||||
if r.Match.NamePattern != "" {
|
||||
match += " pattern=" + r.Match.NamePattern
|
||||
}
|
||||
if r.Match.NameRegex != "" {
|
||||
match += " regex=" + r.Match.NameRegex
|
||||
}
|
||||
acts := make([]string, len(r.Actions))
|
||||
for i, a := range r.Actions {
|
||||
acts[i] = a.Type
|
||||
}
|
||||
fmt.Fprintf(tw, "%s\t%s\t%s\t%v\n", w.Path, r.Name, match, acts)
|
||||
}
|
||||
}
|
||||
tw.Flush()
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
// ── helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
func loadAll() (*config.Config, *state.DB, error) {
|
||||
cfg, err := config.Load(configPath)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("loading config: %w", err)
|
||||
}
|
||||
st, err := state.Open(cfg.Settings.StateDB)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("opening state: %w", err)
|
||||
}
|
||||
return cfg, st, nil
|
||||
}
|
||||
|
||||
func mustHomeDir() string {
|
||||
h, err := os.UserHomeDir()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
func main() {
|
||||
if err := rootCmd.Execute(); err != nil {
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
settings:
|
||||
state_db: ~/.homeflow/state.db
|
||||
log_level: INFO
|
||||
file_stability_secs: 2 # segundos sin actividad antes de procesar
|
||||
rule_match: first # first | all
|
||||
|
||||
watches:
|
||||
- path: ~/Downloads
|
||||
recursive: false
|
||||
rules:
|
||||
|
||||
# ── Archivos comprimidos con contraseña ──────────────────────────────
|
||||
- name: "Archivos cifrados a extraer"
|
||||
match:
|
||||
extensions: [.zip, .7z, .rar]
|
||||
name_pattern: "*_enc*"
|
||||
actions:
|
||||
- type: decompress
|
||||
dest: ~/Downloads/extracted/
|
||||
password: !pass homelab/archive-password
|
||||
remove_archive: true
|
||||
- type: macos_notification
|
||||
title: "homeflow"
|
||||
body: "{filename} extraído"
|
||||
|
||||
# ── PDFs → Paperless ─────────────────────────────────────────────────
|
||||
- name: "PDFs a Paperless"
|
||||
match:
|
||||
extensions: [.pdf]
|
||||
actions:
|
||||
- type: paperless_upload
|
||||
service: paperless
|
||||
tags: ["inbox"]
|
||||
- type: delete
|
||||
|
||||
# ── Facturas → renombrar + Paperless + n8n ────────────────────────────
|
||||
- name: "Facturas a Paperless"
|
||||
match:
|
||||
extensions: [.pdf]
|
||||
name_pattern: "factura*"
|
||||
actions:
|
||||
- type: rename
|
||||
template: "factura_{date:2006-01-02}_{seq:04d}_{stem}{ext}"
|
||||
- type: paperless_upload
|
||||
service: paperless
|
||||
tags: ["facturas"]
|
||||
title: "{stem}"
|
||||
- type: n8n_webhook
|
||||
service: n8n
|
||||
event: "new_invoice"
|
||||
payload:
|
||||
filename: "{filename}"
|
||||
date: "{date:2006-01-02}"
|
||||
- type: delete
|
||||
|
||||
# ── Libros → Calibre ──────────────────────────────────────────────────
|
||||
- name: "Libros a Calibre"
|
||||
match:
|
||||
extensions: [.epub, .mobi, .cbz, .cbr, .fb2, .azw3]
|
||||
actions:
|
||||
- type: calibre_copy
|
||||
service: calibre
|
||||
- type: macos_tag
|
||||
tags: ["Green"]
|
||||
mode: add
|
||||
- type: delete
|
||||
|
||||
# ── Fotos → Immich ────────────────────────────────────────────────────
|
||||
- name: "Fotos a Immich"
|
||||
match:
|
||||
extensions: [.jpg, .jpeg, .png, .heic, .gif, .webp]
|
||||
actions:
|
||||
- type: immich_upload
|
||||
service: immich
|
||||
- type: move
|
||||
dest: ~/Pictures/Processed/
|
||||
|
||||
# ── Vídeos → Immich ───────────────────────────────────────────────────
|
||||
- name: "Vídeos a Immich"
|
||||
match:
|
||||
extensions: [.mp4, .mov, .avi, .mkv, .m4v]
|
||||
actions:
|
||||
- type: immich_upload
|
||||
service: immich
|
||||
- type: delete
|
||||
|
||||
# ── Scripts / procesado custom ────────────────────────────────────────
|
||||
- name: "Ejecutar script custom"
|
||||
match:
|
||||
extensions: [.csv]
|
||||
name_pattern: "export_*"
|
||||
actions:
|
||||
- type: run_script
|
||||
script: ~/scripts/process_export.sh
|
||||
continue_on_error: true
|
||||
- type: move
|
||||
dest: ~/Documents/exports/processed/
|
||||
|
||||
# ── Archivos sospechosos → cuarentena ────────────────────────────────
|
||||
- name: "Ejecutables a cuarentena"
|
||||
match:
|
||||
extensions: [.exe, .dmg, .pkg]
|
||||
actions:
|
||||
- type: checksum
|
||||
algo: sha256
|
||||
save_to_sidecar: true
|
||||
- type: quarantine
|
||||
dest: ~/quarantine/
|
||||
|
||||
- path: ~/Desktop
|
||||
recursive: false
|
||||
rules:
|
||||
|
||||
# ── PDFs en desktop → Paperless ───────────────────────────────────────
|
||||
- name: "PDFs desktop a Paperless"
|
||||
match:
|
||||
extensions: [.pdf]
|
||||
min_size_kb: 10
|
||||
actions:
|
||||
- type: paperless_upload
|
||||
service: paperless
|
||||
tags: ["desktop", "inbox"]
|
||||
- type: delete
|
||||
|
||||
services:
|
||||
paperless:
|
||||
url: !pass services/paperless
|
||||
token: !pass homelab/paperless-token
|
||||
|
||||
immich:
|
||||
url: !pass services/immich
|
||||
api_key: !pass homelab/immich-api-key
|
||||
|
||||
calibre:
|
||||
inbox_path: ~/calibre-inbox/
|
||||
|
||||
n8n:
|
||||
webhook_url: !pass services/n8n
|
||||
@@ -0,0 +1,22 @@
|
||||
module github.com/alexandrev/homeflow
|
||||
|
||||
go 1.25.0
|
||||
|
||||
require (
|
||||
github.com/dustin/go-humanize v1.0.1 // indirect
|
||||
github.com/fsnotify/fsnotify v1.10.1 // 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/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/sys v0.42.0 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
modernc.org/libc v1.72.3 // indirect
|
||||
modernc.org/mathutil v1.7.1 // indirect
|
||||
modernc.org/memory v1.11.0 // indirect
|
||||
modernc.org/sqlite v1.50.1 // indirect
|
||||
)
|
||||
@@ -0,0 +1,37 @@
|
||||
github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
|
||||
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/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/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/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/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=
|
||||
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
|
||||
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/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=
|
||||
modernc.org/libc v1.72.3 h1:ZnDF4tXn4NBXFutMMQC4vtbTFSXhhKzR73fv0beZEAU=
|
||||
modernc.org/libc v1.72.3/go.mod h1:dn0dZNnnn1clLyvRxLxYExxiKRZIRENOfqQ8XEeg4Qs=
|
||||
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
|
||||
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
|
||||
modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
|
||||
modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
|
||||
modernc.org/sqlite v1.50.1 h1:l+cQvn0sd0zJJtfygGHuQJ5AjlrwXmWPw4KP3ZMwr9w=
|
||||
modernc.org/sqlite v1.50.1/go.mod h1:tcNzv5p84E0skkmJn038y+hWJbLQXQqEnQfeh5r2JLM=
|
||||
@@ -0,0 +1,109 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/engine"
|
||||
)
|
||||
|
||||
// Action is the interface every action must implement.
|
||||
type Action interface {
|
||||
Execute(ctx *engine.ActionContext) error
|
||||
}
|
||||
|
||||
// Factory creates an Action from its raw params map.
|
||||
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
|
||||
"macos_tag": newMacOSTagAction,
|
||||
"macos_notification": newMacOSNotificationAction,
|
||||
"macos_open": newMacOSOpenAction,
|
||||
// homelab
|
||||
"paperless_upload": newPaperlessAction,
|
||||
"immich_upload": newImmichAction,
|
||||
"calibre_copy": newCalibreAction,
|
||||
"n8n_webhook": newN8nAction,
|
||||
}
|
||||
|
||||
// Build constructs the action pipeline for a rule.
|
||||
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)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("action %q: %w", def.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)
|
||||
continue
|
||||
}
|
||||
return fmt.Errorf("action %q: %w", defs[i].Type, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// paramString extracts a string param, returns def if missing.
|
||||
func paramString(p map[string]any, key, def string) string {
|
||||
if v, ok := p[key]; ok {
|
||||
if s, ok := v.(string); ok {
|
||||
return s
|
||||
}
|
||||
}
|
||||
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 {
|
||||
return b
|
||||
}
|
||||
}
|
||||
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) {
|
||||
case []any:
|
||||
out := make([]string, 0, len(t))
|
||||
for _, item := range t {
|
||||
if s, ok := item.(string); ok {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out
|
||||
case []string:
|
||||
return t
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/engine"
|
||||
)
|
||||
|
||||
type calibreAction struct {
|
||||
serviceKey string
|
||||
useDB bool // if true, use calibredb CLI instead of copy
|
||||
}
|
||||
|
||||
func newCalibreAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
return &calibreAction{
|
||||
serviceKey: paramString(p, "service", "calibre"),
|
||||
useDB: paramBool(p, "use_calibredb", false),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *calibreAction) Execute(ctx *engine.ActionContext) error {
|
||||
svc, ok := ctx.Services[a.serviceKey]
|
||||
if !ok {
|
||||
return fmt.Errorf("calibre: service %q not configured", a.serviceKey)
|
||||
}
|
||||
|
||||
if a.useDB {
|
||||
return calibreDBAdd(ctx.CurrentPath, svc)
|
||||
}
|
||||
|
||||
// Default: copy to inbox folder
|
||||
inboxPath := svc.InboxPath
|
||||
if inboxPath == "" {
|
||||
return fmt.Errorf("calibre: service %q missing inbox_path", a.serviceKey)
|
||||
}
|
||||
inboxPath = engine.RenderTemplate(inboxPath, ctx)
|
||||
if err := os.MkdirAll(inboxPath, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
target := filepath.Join(inboxPath, filepath.Base(ctx.CurrentPath))
|
||||
return copyFile(ctx.CurrentPath, target)
|
||||
}
|
||||
|
||||
func calibreDBAdd(src string, svc config.ServiceConfig) error {
|
||||
args := []string{"add", src}
|
||||
if svc.URL != "" {
|
||||
// URL can be used as the library path for calibredb
|
||||
args = append([]string{"--with-library", svc.URL}, args...)
|
||||
}
|
||||
out, err := exec.Command("calibredb", args...).CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("calibredb add: %w\n%s", err, out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"mime/multipart"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/engine"
|
||||
)
|
||||
|
||||
type immichAction struct {
|
||||
serviceKey string
|
||||
album string
|
||||
deviceID string
|
||||
}
|
||||
|
||||
func newImmichAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
return &immichAction{
|
||||
serviceKey: paramString(p, "service", "immich"),
|
||||
album: paramString(p, "album", ""),
|
||||
deviceID: paramString(p, "device_id", "homeflow"),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *immichAction) Execute(ctx *engine.ActionContext) error {
|
||||
svc, ok := ctx.Services[a.serviceKey]
|
||||
if !ok {
|
||||
return fmt.Errorf("immich: service %q not configured", a.serviceKey)
|
||||
}
|
||||
if svc.URL == "" || svc.APIKey == "" {
|
||||
return fmt.Errorf("immich: service %q missing url or api_key", a.serviceKey)
|
||||
}
|
||||
|
||||
f, err := os.Open(ctx.CurrentPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
info, _ := f.Stat()
|
||||
|
||||
var body bytes.Buffer
|
||||
mw := multipart.NewWriter(&body)
|
||||
|
||||
part, err := mw.CreateFormFile("assetData", filepath.Base(ctx.CurrentPath))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := io.Copy(part, f); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
now := time.Now().UTC().Format(time.RFC3339)
|
||||
mtime := now
|
||||
if info != nil {
|
||||
mtime = info.ModTime().UTC().Format(time.RFC3339)
|
||||
}
|
||||
mw.WriteField("deviceAssetId", filepath.Base(ctx.CurrentPath)+"-"+now)
|
||||
mw.WriteField("deviceId", a.deviceID)
|
||||
mw.WriteField("fileCreatedAt", mtime)
|
||||
mw.WriteField("fileModifiedAt", mtime)
|
||||
mw.Close()
|
||||
|
||||
baseURL := strings.TrimRight(svc.URL, "/")
|
||||
req, err := http.NewRequest(http.MethodPost, baseURL+"/api/assets", &body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("x-api-key", svc.APIKey)
|
||||
req.Header.Set("Content-Type", mw.FormDataContentType())
|
||||
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("immich upload: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
respBody, _ := io.ReadAll(resp.Body)
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
return fmt.Errorf("immich upload: HTTP %d: %s", resp.StatusCode, respBody)
|
||||
}
|
||||
|
||||
// Add to album if configured
|
||||
if a.album != "" {
|
||||
var result struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if err := json.Unmarshal(respBody, &result); err == nil && result.ID != "" {
|
||||
a.addToAlbum(baseURL, svc.APIKey, result.ID, a.album)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *immichAction) addToAlbum(baseURL, apiKey, assetID, albumName string) {
|
||||
// Find album by name
|
||||
req, _ := http.NewRequest(http.MethodGet, baseURL+"/api/albums", nil)
|
||||
req.Header.Set("x-api-key", apiKey)
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
var albums []struct {
|
||||
ID string `json:"id"`
|
||||
AlbumName string `json:"albumName"`
|
||||
}
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
json.Unmarshal(body, &albums)
|
||||
|
||||
for _, al := range albums {
|
||||
if al.AlbumName == albumName {
|
||||
payload, _ := json.Marshal(map[string]any{"ids": []string{assetID}})
|
||||
req2, _ := http.NewRequest(http.MethodPut, baseURL+"/api/albums/"+al.ID+"/assets", bytes.NewReader(payload))
|
||||
req2.Header.Set("x-api-key", apiKey)
|
||||
req2.Header.Set("Content-Type", "application/json")
|
||||
resp2, _ := http.DefaultClient.Do(req2)
|
||||
if resp2 != nil {
|
||||
resp2.Body.Close()
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,544 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"archive/zip"
|
||||
"compress/bzip2"
|
||||
"compress/gzip"
|
||||
"crypto/md5"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/engine"
|
||||
)
|
||||
|
||||
// ── move ────────────────────────────────────────────────────────────────────
|
||||
|
||||
type moveAction struct{ dest string }
|
||||
|
||||
func newMoveAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
dest := paramString(p, "dest", "")
|
||||
if dest == "" {
|
||||
return nil, fmt.Errorf("move: dest required")
|
||||
}
|
||||
return &moveAction{dest: config.ExpandPath(dest)}, nil
|
||||
}
|
||||
|
||||
func (a *moveAction) Execute(ctx *engine.ActionContext) error {
|
||||
dest := engine.RenderTemplate(a.dest, ctx)
|
||||
if err := os.MkdirAll(dest, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
target := filepath.Join(dest, filepath.Base(ctx.CurrentPath))
|
||||
if err := os.Rename(ctx.CurrentPath, target); err != nil {
|
||||
// cross-device: copy + remove
|
||||
if err2 := copyFile(ctx.CurrentPath, target); err2 != nil {
|
||||
return err2
|
||||
}
|
||||
if err2 := os.Remove(ctx.CurrentPath); err2 != nil {
|
||||
return err2
|
||||
}
|
||||
}
|
||||
ctx.CurrentPath = target
|
||||
return nil
|
||||
}
|
||||
|
||||
// ── copy ────────────────────────────────────────────────────────────────────
|
||||
|
||||
type copyAction struct{ dest string }
|
||||
|
||||
func newCopyAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
dest := paramString(p, "dest", "")
|
||||
if dest == "" {
|
||||
return nil, fmt.Errorf("copy: dest required")
|
||||
}
|
||||
return ©Action{dest: config.ExpandPath(dest)}, nil
|
||||
}
|
||||
|
||||
func (a *copyAction) Execute(ctx *engine.ActionContext) error {
|
||||
dest := engine.RenderTemplate(a.dest, ctx)
|
||||
if err := os.MkdirAll(dest, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
target := filepath.Join(dest, filepath.Base(ctx.CurrentPath))
|
||||
return copyFile(ctx.CurrentPath, target)
|
||||
}
|
||||
|
||||
// ── delete ──────────────────────────────────────────────────────────────────
|
||||
|
||||
type deleteAction struct{}
|
||||
|
||||
func newDeleteAction(_ map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
return &deleteAction{}, nil
|
||||
}
|
||||
|
||||
func (a *deleteAction) Execute(ctx *engine.ActionContext) error {
|
||||
return os.Remove(ctx.CurrentPath)
|
||||
}
|
||||
|
||||
// ── rename ──────────────────────────────────────────────────────────────────
|
||||
|
||||
type renameAction struct{ tmpl string }
|
||||
|
||||
func newRenameAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
tmpl := paramString(p, "template", "")
|
||||
if tmpl == "" {
|
||||
return nil, fmt.Errorf("rename: template required")
|
||||
}
|
||||
return &renameAction{tmpl: tmpl}, nil
|
||||
}
|
||||
|
||||
func (a *renameAction) Execute(ctx *engine.ActionContext) error {
|
||||
newName := engine.RenderTemplate(a.tmpl, ctx)
|
||||
dir := filepath.Dir(ctx.CurrentPath)
|
||||
target := filepath.Join(dir, newName)
|
||||
if err := os.Rename(ctx.CurrentPath, target); err != nil {
|
||||
return err
|
||||
}
|
||||
ctx.CurrentPath = target
|
||||
return nil
|
||||
}
|
||||
|
||||
// ── decompress ──────────────────────────────────────────────────────────────
|
||||
|
||||
type decompressAction struct {
|
||||
dest string
|
||||
password string
|
||||
removeArchive bool
|
||||
}
|
||||
|
||||
func newDecompressAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
dest := paramString(p, "dest", "")
|
||||
if dest == "" {
|
||||
return nil, fmt.Errorf("decompress: dest required")
|
||||
}
|
||||
return &decompressAction{
|
||||
dest: config.ExpandPath(dest),
|
||||
password: paramString(p, "password", ""),
|
||||
removeArchive: paramBool(p, "remove_archive", true),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *decompressAction) Execute(ctx *engine.ActionContext) error {
|
||||
dest := engine.RenderTemplate(a.dest, ctx)
|
||||
if err := os.MkdirAll(dest, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
ext := strings.ToLower(filepath.Ext(ctx.CurrentPath))
|
||||
var err error
|
||||
switch ext {
|
||||
case ".zip":
|
||||
err = extractZip(ctx.CurrentPath, dest, a.password)
|
||||
case ".gz":
|
||||
if strings.HasSuffix(strings.ToLower(ctx.CurrentPath), ".tar.gz") {
|
||||
err = extractTarGz(ctx.CurrentPath, dest)
|
||||
} else {
|
||||
err = extractGz(ctx.CurrentPath, dest)
|
||||
}
|
||||
case ".bz2":
|
||||
err = extractTarBz2(ctx.CurrentPath, dest)
|
||||
case ".tar":
|
||||
err = extractTar(ctx.CurrentPath, dest)
|
||||
case ".rar", ".7z":
|
||||
err = extractExternal(ctx.CurrentPath, dest, a.password)
|
||||
default:
|
||||
return fmt.Errorf("unsupported archive format: %s", ext)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if a.removeArchive {
|
||||
return os.Remove(ctx.CurrentPath)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func extractZip(src, dest, password string) error {
|
||||
// If password provided, delegate to unzip CLI (stdlib doesn't support encrypted zip)
|
||||
if password != "" {
|
||||
out, err := exec.Command("unzip", "-P", password, "-o", src, "-d", dest).CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("unzip: %w\n%s", err, out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
r, err := zip.OpenReader(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer r.Close()
|
||||
for _, f := range r.File {
|
||||
target := filepath.Join(dest, filepath.Clean(f.Name))
|
||||
if !strings.HasPrefix(target, filepath.Clean(dest)+string(os.PathSeparator)) {
|
||||
return fmt.Errorf("zip: invalid path %s", f.Name)
|
||||
}
|
||||
if f.FileInfo().IsDir() {
|
||||
os.MkdirAll(target, f.Mode())
|
||||
continue
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
rc, err := f.Open()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out, err := os.Create(target)
|
||||
if err != nil {
|
||||
rc.Close()
|
||||
return err
|
||||
}
|
||||
_, copyErr := io.Copy(out, rc)
|
||||
out.Close()
|
||||
rc.Close()
|
||||
if copyErr != nil {
|
||||
return copyErr
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func extractTarGz(src, dest string) error {
|
||||
f, err := os.Open(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
gr, err := gzip.NewReader(f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer gr.Close()
|
||||
return extractTarReader(tar.NewReader(gr), dest)
|
||||
}
|
||||
|
||||
func extractTarBz2(src, dest string) error {
|
||||
f, err := os.Open(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
return extractTarReader(tar.NewReader(bzip2.NewReader(f)), dest)
|
||||
}
|
||||
|
||||
func extractTar(src, dest string) error {
|
||||
f, err := os.Open(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
return extractTarReader(tar.NewReader(f), dest)
|
||||
}
|
||||
|
||||
func extractTarReader(tr *tar.Reader, dest string) error {
|
||||
for {
|
||||
hdr, err := tr.Next()
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
target := filepath.Join(dest, filepath.Clean(hdr.Name))
|
||||
if !strings.HasPrefix(target, filepath.Clean(dest)+string(os.PathSeparator)) {
|
||||
return fmt.Errorf("tar: invalid path %s", hdr.Name)
|
||||
}
|
||||
switch hdr.Typeflag {
|
||||
case tar.TypeDir:
|
||||
os.MkdirAll(target, os.FileMode(hdr.Mode))
|
||||
case tar.TypeReg:
|
||||
os.MkdirAll(filepath.Dir(target), 0o755)
|
||||
out, err := os.Create(target)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, copyErr := io.Copy(out, tr)
|
||||
out.Close()
|
||||
if copyErr != nil {
|
||||
return copyErr
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func extractGz(src, dest string) error {
|
||||
f, err := os.Open(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
gr, err := gzip.NewReader(f)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer gr.Close()
|
||||
name := strings.TrimSuffix(filepath.Base(src), ".gz")
|
||||
out, err := os.Create(filepath.Join(dest, name))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer out.Close()
|
||||
_, err = io.Copy(out, gr)
|
||||
return err
|
||||
}
|
||||
|
||||
func extractExternal(src, dest, password string) error {
|
||||
ext := strings.ToLower(filepath.Ext(src))
|
||||
var cmd *exec.Cmd
|
||||
switch ext {
|
||||
case ".rar":
|
||||
if password != "" {
|
||||
cmd = exec.Command("unrar", "x", "-p"+password, src, dest+"/")
|
||||
} else {
|
||||
cmd = exec.Command("unrar", "x", src, dest+"/")
|
||||
}
|
||||
case ".7z":
|
||||
if password != "" {
|
||||
cmd = exec.Command("7z", "x", "-p"+password, "-o"+dest, src)
|
||||
} else {
|
||||
cmd = exec.Command("7z", "x", "-o"+dest, src)
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("no external extractor for %s", ext)
|
||||
}
|
||||
out, err := cmd.CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s: %w\n%s", ext, err, out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ── compress ─────────────────────────────────────────────────────────────────
|
||||
|
||||
type compressAction struct {
|
||||
format string
|
||||
dest string
|
||||
password string
|
||||
}
|
||||
|
||||
func newCompressAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
return &compressAction{
|
||||
format: paramString(p, "format", "zip"),
|
||||
dest: config.ExpandPath(paramString(p, "dest", "")),
|
||||
password: paramString(p, "password", ""),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *compressAction) Execute(ctx *engine.ActionContext) error {
|
||||
dest := engine.RenderTemplate(a.dest, ctx)
|
||||
if dest == "" {
|
||||
dest = filepath.Dir(ctx.CurrentPath)
|
||||
}
|
||||
os.MkdirAll(dest, 0o755)
|
||||
|
||||
base := filepath.Base(ctx.CurrentPath)
|
||||
switch a.format {
|
||||
case "zip":
|
||||
archivePath := filepath.Join(dest, base+".zip")
|
||||
if err := createZip(ctx.CurrentPath, archivePath); err != nil {
|
||||
return err
|
||||
}
|
||||
ctx.CurrentPath = archivePath
|
||||
case "tar.gz":
|
||||
archivePath := filepath.Join(dest, base+".tar.gz")
|
||||
if err := createTarGz(ctx.CurrentPath, archivePath); err != nil {
|
||||
return err
|
||||
}
|
||||
ctx.CurrentPath = archivePath
|
||||
default:
|
||||
return fmt.Errorf("unsupported compress format: %s", a.format)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func createZip(src, archivePath string) error {
|
||||
out, err := os.Create(archivePath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer out.Close()
|
||||
zw := zip.NewWriter(out)
|
||||
defer zw.Close()
|
||||
f, err := os.Open(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
hdr, err := zip.FileInfoHeader(info)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
hdr.Method = zip.Deflate
|
||||
w, err := zw.CreateHeader(hdr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = io.Copy(w, f)
|
||||
return err
|
||||
}
|
||||
|
||||
func createTarGz(src, archivePath string) error {
|
||||
out, err := os.Create(archivePath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer out.Close()
|
||||
gw := gzip.NewWriter(out)
|
||||
defer gw.Close()
|
||||
tw := tar.NewWriter(gw)
|
||||
defer tw.Close()
|
||||
f, err := os.Open(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
hdr, err := tar.FileInfoHeader(info, "")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tw.WriteHeader(hdr); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = io.Copy(tw, f)
|
||||
return err
|
||||
}
|
||||
|
||||
// ── run_script ───────────────────────────────────────────────────────────────
|
||||
|
||||
type runScriptAction struct {
|
||||
script string
|
||||
}
|
||||
|
||||
func newRunScriptAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
script := paramString(p, "script", "")
|
||||
if script == "" {
|
||||
return nil, fmt.Errorf("run_script: script required")
|
||||
}
|
||||
return &runScriptAction{script: script}, nil
|
||||
}
|
||||
|
||||
func (a *runScriptAction) Execute(ctx *engine.ActionContext) error {
|
||||
script := engine.RenderTemplate(a.script, ctx)
|
||||
var cmd *exec.Cmd
|
||||
// inline script if it contains newlines or starts with shebang/spaces
|
||||
if strings.Contains(script, "\n") || strings.HasPrefix(script, "#!") {
|
||||
cmd = exec.Command("bash", "-c", script)
|
||||
} else {
|
||||
cmd = exec.Command("bash", "-c", script)
|
||||
}
|
||||
cmd.Env = append(os.Environ(),
|
||||
"HOMEFLOW_FILE="+ctx.CurrentPath,
|
||||
"HOMEFLOW_ORIGINAL="+ctx.OriginalPath,
|
||||
"HOMEFLOW_RULE="+ctx.RuleName,
|
||||
)
|
||||
out, err := cmd.CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("script failed: %w\n%s", err, out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ── checksum ─────────────────────────────────────────────────────────────────
|
||||
|
||||
type checksumAction struct {
|
||||
algo string
|
||||
saveToSidecar bool
|
||||
}
|
||||
|
||||
func newChecksumAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
return &checksumAction{
|
||||
algo: paramString(p, "algo", "sha256"),
|
||||
saveToSidecar: paramBool(p, "save_to_sidecar", false),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *checksumAction) Execute(ctx *engine.ActionContext) error {
|
||||
f, err := os.Open(ctx.CurrentPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
var sum string
|
||||
switch a.algo {
|
||||
case "sha256":
|
||||
h := sha256.New()
|
||||
if _, err := io.Copy(h, f); err != nil {
|
||||
return err
|
||||
}
|
||||
sum = hex.EncodeToString(h.Sum(nil))
|
||||
case "md5":
|
||||
h := md5.New()
|
||||
if _, err := io.Copy(h, f); err != nil {
|
||||
return err
|
||||
}
|
||||
sum = hex.EncodeToString(h.Sum(nil))
|
||||
default:
|
||||
return fmt.Errorf("unsupported checksum algo: %s", a.algo)
|
||||
}
|
||||
|
||||
if a.saveToSidecar {
|
||||
sidecar := ctx.CurrentPath + "." + a.algo
|
||||
return os.WriteFile(sidecar, []byte(sum+" "+filepath.Base(ctx.CurrentPath)+"\n"), 0o644)
|
||||
}
|
||||
fmt.Printf("[checksum] %s %s\n", sum, filepath.Base(ctx.CurrentPath))
|
||||
return nil
|
||||
}
|
||||
|
||||
// ── quarantine ───────────────────────────────────────────────────────────────
|
||||
|
||||
type quarantineAction struct{ dest string }
|
||||
|
||||
func newQuarantineAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
dest := paramString(p, "dest", "~/homeflow/quarantine")
|
||||
return &quarantineAction{dest: config.ExpandPath(dest)}, nil
|
||||
}
|
||||
|
||||
func (a *quarantineAction) Execute(ctx *engine.ActionContext) error {
|
||||
dest := engine.RenderTemplate(a.dest, ctx)
|
||||
if err := os.MkdirAll(dest, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
target := filepath.Join(dest, filepath.Base(ctx.CurrentPath))
|
||||
if err := os.Rename(ctx.CurrentPath, target); err != nil {
|
||||
if err2 := copyFile(ctx.CurrentPath, target); err2 != nil {
|
||||
return err2
|
||||
}
|
||||
os.Remove(ctx.CurrentPath)
|
||||
}
|
||||
ctx.CurrentPath = target
|
||||
return nil
|
||||
}
|
||||
|
||||
// ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
func copyFile(src, dst string) error {
|
||||
in, err := os.Open(src)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer in.Close()
|
||||
out, err := os.Create(dst)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer out.Close()
|
||||
_, err = io.Copy(out, in)
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"strings"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/engine"
|
||||
)
|
||||
|
||||
// ── macos_tag ────────────────────────────────────────────────────────────────
|
||||
|
||||
type macOSTagAction struct {
|
||||
tags []string
|
||||
mode string // add | set | remove
|
||||
}
|
||||
|
||||
func newMacOSTagAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
tags := paramStrings(p, "tags")
|
||||
if len(tags) == 0 {
|
||||
return nil, fmt.Errorf("macos_tag: tags required")
|
||||
}
|
||||
return &macOSTagAction{
|
||||
tags: tags,
|
||||
mode: paramString(p, "mode", "add"),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *macOSTagAction) Execute(ctx *engine.ActionContext) error {
|
||||
// Use the `tag` CLI tool if available, otherwise fall back to osascript.
|
||||
if path, err := exec.LookPath("tag"); err == nil {
|
||||
var flag string
|
||||
switch a.mode {
|
||||
case "set":
|
||||
flag = "--set"
|
||||
case "remove":
|
||||
flag = "--remove"
|
||||
default:
|
||||
flag = "--add"
|
||||
}
|
||||
args := []string{flag, strings.Join(a.tags, ","), ctx.CurrentPath}
|
||||
out, err := exec.Command(path, args...).CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("tag: %w\n%s", err, out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Fall back: use xattr via osascript/Finder (set only)
|
||||
tagList := make([]string, len(a.tags))
|
||||
for i, t := range a.tags {
|
||||
tagList[i] = fmt.Sprintf("%q", t)
|
||||
}
|
||||
script := fmt.Sprintf(
|
||||
`tell application "Finder" to set label of (POSIX file %q as alias) to 1`,
|
||||
ctx.CurrentPath,
|
||||
)
|
||||
out, err := exec.Command("osascript", "-e", script).CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("osascript tag: %w\n%s", err, out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ── macos_notification ───────────────────────────────────────────────────────
|
||||
|
||||
type macOSNotificationAction struct {
|
||||
title string
|
||||
body string
|
||||
sound string
|
||||
}
|
||||
|
||||
func newMacOSNotificationAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
return &macOSNotificationAction{
|
||||
title: paramString(p, "title", "homeflow"),
|
||||
body: paramString(p, "body", ""),
|
||||
sound: paramString(p, "sound", ""),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *macOSNotificationAction) Execute(ctx *engine.ActionContext) error {
|
||||
title := engine.RenderTemplate(a.title, ctx)
|
||||
body := engine.RenderTemplate(a.body, ctx)
|
||||
|
||||
var script string
|
||||
if a.sound != "" {
|
||||
script = fmt.Sprintf(
|
||||
`display notification %q with title %q sound name %q`,
|
||||
body, title, a.sound,
|
||||
)
|
||||
} else {
|
||||
script = fmt.Sprintf(`display notification %q with title %q`, body, title)
|
||||
}
|
||||
out, err := exec.Command("osascript", "-e", script).CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("notification: %w\n%s", err, out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ── macos_open ───────────────────────────────────────────────────────────────
|
||||
|
||||
type macOSOpenAction struct{ app string }
|
||||
|
||||
func newMacOSOpenAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
return &macOSOpenAction{app: paramString(p, "app", "")}, nil
|
||||
}
|
||||
|
||||
func (a *macOSOpenAction) Execute(ctx *engine.ActionContext) error {
|
||||
var args []string
|
||||
if a.app != "" {
|
||||
args = []string{"-a", a.app, ctx.CurrentPath}
|
||||
} else {
|
||||
args = []string{ctx.CurrentPath}
|
||||
}
|
||||
return exec.Command("open", args...).Run()
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/engine"
|
||||
)
|
||||
|
||||
type n8nAction struct {
|
||||
serviceKey string
|
||||
event string
|
||||
payload map[string]any
|
||||
}
|
||||
|
||||
func newN8nAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
var payload map[string]any
|
||||
if v, ok := p["payload"]; ok {
|
||||
if m, ok := v.(map[string]any); ok {
|
||||
payload = m
|
||||
}
|
||||
}
|
||||
return &n8nAction{
|
||||
serviceKey: paramString(p, "service", "n8n"),
|
||||
event: paramString(p, "event", "homeflow_event"),
|
||||
payload: payload,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *n8nAction) Execute(ctx *engine.ActionContext) error {
|
||||
svc, ok := ctx.Services[a.serviceKey]
|
||||
if !ok {
|
||||
return fmt.Errorf("n8n: service %q not configured", a.serviceKey)
|
||||
}
|
||||
webhookURL := svc.WebhookURL
|
||||
if webhookURL == "" {
|
||||
webhookURL = svc.URL
|
||||
}
|
||||
if webhookURL == "" {
|
||||
return fmt.Errorf("n8n: service %q missing webhook_url", a.serviceKey)
|
||||
}
|
||||
|
||||
// Build payload with rendered templates
|
||||
body := map[string]any{
|
||||
"event": a.event,
|
||||
"file": ctx.CurrentPath,
|
||||
"original": ctx.OriginalPath,
|
||||
"rule": ctx.RuleName,
|
||||
}
|
||||
for k, v := range a.payload {
|
||||
if s, ok := v.(string); ok {
|
||||
body[k] = engine.RenderTemplate(s, ctx)
|
||||
} else {
|
||||
body[k] = v
|
||||
}
|
||||
}
|
||||
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
url := strings.TrimRight(webhookURL, "/")
|
||||
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(data))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if svc.Token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+svc.Token)
|
||||
}
|
||||
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("n8n webhook: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
respBody, _ := io.ReadAll(resp.Body)
|
||||
return fmt.Errorf("n8n webhook: HTTP %d: %s", resp.StatusCode, respBody)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
package actions
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io"
|
||||
"mime/multipart"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/engine"
|
||||
)
|
||||
|
||||
type paperlessAction struct {
|
||||
serviceKey string
|
||||
tags []string
|
||||
correspondent string
|
||||
documentType string
|
||||
titleTmpl string
|
||||
}
|
||||
|
||||
func newPaperlessAction(p map[string]any, _ map[string]config.ServiceConfig) (Action, error) {
|
||||
return &paperlessAction{
|
||||
serviceKey: paramString(p, "service", "paperless"),
|
||||
tags: paramStrings(p, "tags"),
|
||||
correspondent: paramString(p, "correspondent", ""),
|
||||
documentType: paramString(p, "document_type", ""),
|
||||
titleTmpl: paramString(p, "title", ""),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (a *paperlessAction) Execute(ctx *engine.ActionContext) error {
|
||||
svc, ok := ctx.Services[a.serviceKey]
|
||||
if !ok {
|
||||
return fmt.Errorf("paperless: service %q not configured", a.serviceKey)
|
||||
}
|
||||
if svc.URL == "" || svc.Token == "" {
|
||||
return fmt.Errorf("paperless: service %q missing url or token", a.serviceKey)
|
||||
}
|
||||
|
||||
f, err := os.Open(ctx.CurrentPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
var body bytes.Buffer
|
||||
mw := multipart.NewWriter(&body)
|
||||
|
||||
part, err := mw.CreateFormFile("document", filepath.Base(ctx.CurrentPath))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := io.Copy(part, f); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if a.titleTmpl != "" {
|
||||
mw.WriteField("title", engine.RenderTemplate(a.titleTmpl, ctx))
|
||||
}
|
||||
if a.correspondent != "" {
|
||||
mw.WriteField("correspondent", engine.RenderTemplate(a.correspondent, ctx))
|
||||
}
|
||||
if a.documentType != "" {
|
||||
mw.WriteField("document_type", engine.RenderTemplate(a.documentType, ctx))
|
||||
}
|
||||
for _, tag := range a.tags {
|
||||
mw.WriteField("tags", engine.RenderTemplate(tag, ctx))
|
||||
}
|
||||
mw.Close()
|
||||
|
||||
url := strings.TrimRight(svc.URL, "/") + "/api/documents/post_document/"
|
||||
req, err := http.NewRequest(http.MethodPost, url, &body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Authorization", "Token "+svc.Token)
|
||||
req.Header.Set("Content-Type", mw.FormDataContentType())
|
||||
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("paperless upload: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
respBody, _ := io.ReadAll(resp.Body)
|
||||
return fmt.Errorf("paperless upload: HTTP %d: %s", resp.StatusCode, respBody)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,182 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Settings SettingsConfig `yaml:"settings"`
|
||||
Watches []WatchConfig `yaml:"watches"`
|
||||
Services map[string]ServiceConfig `yaml:"services"`
|
||||
}
|
||||
|
||||
type SettingsConfig struct {
|
||||
StateDB string `yaml:"state_db"`
|
||||
LogLevel string `yaml:"log_level"`
|
||||
StabilitySecs int `yaml:"file_stability_secs"`
|
||||
RuleMatch string `yaml:"rule_match"` // first | all
|
||||
}
|
||||
|
||||
type WatchConfig struct {
|
||||
Path string `yaml:"path"`
|
||||
Recursive bool `yaml:"recursive"`
|
||||
Rules []RuleConfig `yaml:"rules"`
|
||||
}
|
||||
|
||||
type RuleConfig struct {
|
||||
Name string `yaml:"name"`
|
||||
Match MatchConfig `yaml:"match"`
|
||||
Actions []ActionDef `yaml:"actions"`
|
||||
}
|
||||
|
||||
type MatchConfig struct {
|
||||
Extensions []string `yaml:"extensions"`
|
||||
NamePattern string `yaml:"name_pattern"`
|
||||
NameRegex string `yaml:"name_regex"`
|
||||
MinSizeKB int64 `yaml:"min_size_kb"`
|
||||
MaxSizeKB int64 `yaml:"max_size_kb"`
|
||||
MimeTypes []string `yaml:"mime_type"`
|
||||
AgeSecsMin int64 `yaml:"age_secs_min"`
|
||||
AgeSecsMax int64 `yaml:"age_secs_max"`
|
||||
}
|
||||
|
||||
// ActionDef parses the action type and carries remaining fields as Params.
|
||||
type ActionDef struct {
|
||||
Type string
|
||||
ContinueOnError bool
|
||||
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 t, ok := m["type"].(string); ok {
|
||||
a.Type = t
|
||||
delete(m, "type")
|
||||
}
|
||||
if c, ok := m["continue_on_error"].(bool); ok {
|
||||
a.ContinueOnError = c
|
||||
delete(m, "continue_on_error")
|
||||
}
|
||||
a.Params = m
|
||||
return nil
|
||||
}
|
||||
|
||||
type ServiceConfig struct {
|
||||
URL string `yaml:"url"`
|
||||
Token string `yaml:"token"`
|
||||
APIKey string `yaml:"api_key"`
|
||||
WebhookURL string `yaml:"webhook_url"`
|
||||
InboxPath string `yaml:"inbox_path"`
|
||||
}
|
||||
|
||||
func Load(path string) (*Config, error) {
|
||||
return load(path, true)
|
||||
}
|
||||
|
||||
// LoadDry parses and validates the config without resolving !pass credentials.
|
||||
func LoadDry(path string) (*Config, error) {
|
||||
return load(path, false)
|
||||
}
|
||||
|
||||
func load(path string, resolvePass bool) (*Config, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading config: %w", err)
|
||||
}
|
||||
|
||||
var root yaml.Node
|
||||
if err := yaml.Unmarshal(data, &root); err != nil {
|
||||
return nil, fmt.Errorf("parsing config: %w", err)
|
||||
}
|
||||
if resolvePass && root.Kind == yaml.DocumentNode && len(root.Content) > 0 {
|
||||
if err := resolvePassNodes(root.Content[0]); err != nil {
|
||||
return nil, fmt.Errorf("resolving credentials: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
var cfg Config
|
||||
if err := root.Decode(&cfg); err != nil {
|
||||
return nil, fmt.Errorf("decoding config: %w", err)
|
||||
}
|
||||
|
||||
applyDefaults(&cfg)
|
||||
expandPaths(&cfg)
|
||||
return &cfg, nil
|
||||
}
|
||||
|
||||
// resolvePassNodes walks the YAML tree and resolves any !pass tagged scalars.
|
||||
func resolvePassNodes(node *yaml.Node) error {
|
||||
if node.Kind == yaml.ScalarNode && node.Tag == "!pass" {
|
||||
val, err := runPass(node.Value)
|
||||
if err != nil {
|
||||
return fmt.Errorf("pass show %s: %w", node.Value, err)
|
||||
}
|
||||
node.Tag = "!!str"
|
||||
node.Value = strings.TrimSpace(val)
|
||||
return nil
|
||||
}
|
||||
for _, child := range node.Content {
|
||||
if err := resolvePassNodes(child); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func runPass(key string) (string, error) {
|
||||
out, err := exec.Command("pass", "show", key).Output()
|
||||
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
|
||||
}
|
||||
|
||||
func applyDefaults(cfg *Config) {
|
||||
if cfg.Settings.StateDB == "" {
|
||||
cfg.Settings.StateDB = "~/.homeflow/state.db"
|
||||
}
|
||||
if cfg.Settings.LogLevel == "" {
|
||||
cfg.Settings.LogLevel = "INFO"
|
||||
}
|
||||
if cfg.Settings.StabilitySecs <= 0 {
|
||||
cfg.Settings.StabilitySecs = 2
|
||||
}
|
||||
if cfg.Settings.RuleMatch == "" {
|
||||
cfg.Settings.RuleMatch = "first"
|
||||
}
|
||||
}
|
||||
|
||||
func expandPaths(cfg *Config) {
|
||||
cfg.Settings.StateDB = ExpandPath(cfg.Settings.StateDB)
|
||||
for i := range cfg.Watches {
|
||||
cfg.Watches[i].Path = ExpandPath(cfg.Watches[i].Path)
|
||||
}
|
||||
for k, svc := range cfg.Services {
|
||||
svc.InboxPath = ExpandPath(svc.InboxPath)
|
||||
cfg.Services[k] = svc
|
||||
}
|
||||
}
|
||||
|
||||
func ExpandPath(p string) string {
|
||||
if p == "" {
|
||||
return p
|
||||
}
|
||||
if strings.HasPrefix(p, "~/") {
|
||||
home, err := os.UserHomeDir()
|
||||
if err == nil {
|
||||
return filepath.Join(home, p[2:])
|
||||
}
|
||||
}
|
||||
return p
|
||||
}
|
||||
@@ -0,0 +1,272 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/h2non/filetype"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/state"
|
||||
)
|
||||
|
||||
// ActionContext is passed to each action in the pipeline.
|
||||
// Actions that rename or move the file must update CurrentPath.
|
||||
type ActionContext struct {
|
||||
CurrentPath string
|
||||
OriginalPath string
|
||||
RuleName string
|
||||
Vars map[string]string // pre-computed template variables
|
||||
Services map[string]config.ServiceConfig
|
||||
State *state.DB
|
||||
}
|
||||
|
||||
// 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 {
|
||||
var sb strings.Builder
|
||||
i := 0
|
||||
for i < len(tmpl) {
|
||||
if tmpl[i] == '{' {
|
||||
j := strings.Index(tmpl[i:], "}")
|
||||
if j == -1 {
|
||||
sb.WriteByte(tmpl[i])
|
||||
i++
|
||||
continue
|
||||
}
|
||||
token := tmpl[i+1 : i+j]
|
||||
i += j + 1
|
||||
sb.WriteString(resolveToken(token, ctx))
|
||||
} else {
|
||||
sb.WriteByte(tmpl[i])
|
||||
i++
|
||||
}
|
||||
}
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
func resolveToken(token string, ctx *ActionContext) string {
|
||||
key, format, hasFormat := strings.Cut(token, ":")
|
||||
|
||||
switch key {
|
||||
case "filename":
|
||||
return filepath.Base(ctx.CurrentPath)
|
||||
case "stem":
|
||||
base := filepath.Base(ctx.CurrentPath)
|
||||
ext := filepath.Ext(base)
|
||||
return strings.TrimSuffix(base, ext)
|
||||
case "ext":
|
||||
return filepath.Ext(ctx.CurrentPath)
|
||||
case "hash8":
|
||||
if h, ok := ctx.Vars["hash8"]; ok {
|
||||
return h
|
||||
}
|
||||
return ""
|
||||
case "size_kb":
|
||||
if s, ok := ctx.Vars["size_kb"]; ok {
|
||||
return s
|
||||
}
|
||||
return "0"
|
||||
case "seq":
|
||||
if s, ok := ctx.Vars["seq"]; ok {
|
||||
return s
|
||||
}
|
||||
return "0"
|
||||
case "date":
|
||||
if hasFormat {
|
||||
return time.Now().Format(format)
|
||||
}
|
||||
return time.Now().Format("2006-01-02")
|
||||
case "mtime":
|
||||
if ts, ok := ctx.Vars["mtime"]; ok {
|
||||
if hasFormat {
|
||||
t, err := time.Parse(time.RFC3339, ts)
|
||||
if err == nil {
|
||||
return t.Format(format)
|
||||
}
|
||||
}
|
||||
return ts
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// fall back to Vars map
|
||||
if v, ok := ctx.Vars[key]; ok {
|
||||
return v
|
||||
}
|
||||
return "{" + token + "}"
|
||||
}
|
||||
|
||||
// MatchFile checks if a file matches a rule's conditions.
|
||||
func MatchFile(path string, rule *config.RuleConfig) (bool, error) {
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
m := &rule.Match
|
||||
|
||||
// Extension check
|
||||
if len(m.Extensions) > 0 {
|
||||
ext := strings.ToLower(filepath.Ext(path))
|
||||
found := false
|
||||
for _, e := range m.Extensions {
|
||||
if strings.ToLower(e) == ext {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Name pattern (glob)
|
||||
if m.NamePattern != "" {
|
||||
matched, err := filepath.Match(m.NamePattern, filepath.Base(path))
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("invalid name_pattern: %w", err)
|
||||
}
|
||||
if !matched {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Name regex
|
||||
if m.NameRegex != "" {
|
||||
re, err := regexp.Compile(m.NameRegex)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("invalid name_regex: %w", err)
|
||||
}
|
||||
if !re.MatchString(filepath.Base(path)) {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Size checks
|
||||
sizeKB := info.Size() / 1024
|
||||
if m.MinSizeKB > 0 && sizeKB < m.MinSizeKB {
|
||||
return false, nil
|
||||
}
|
||||
if m.MaxSizeKB > 0 && sizeKB > m.MaxSizeKB {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// Age checks
|
||||
ageSecs := int64(time.Since(info.ModTime()).Seconds())
|
||||
if m.AgeSecsMin > 0 && ageSecs < m.AgeSecsMin {
|
||||
return false, nil
|
||||
}
|
||||
if m.AgeSecsMax > 0 && ageSecs > m.AgeSecsMax {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// MIME type check (uses magic bytes — relatively expensive, do last)
|
||||
if len(m.MimeTypes) > 0 {
|
||||
mime, err := detectMIME(path)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
found := false
|
||||
for _, mt := range m.MimeTypes {
|
||||
if strings.EqualFold(mt, mime) {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func detectMIME(path string) (string, error) {
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
buf := make([]byte, 261)
|
||||
n, err := f.Read(buf)
|
||||
if err != nil && err != io.EOF {
|
||||
return "", err
|
||||
}
|
||||
kind, _ := filetype.Match(buf[:n])
|
||||
if kind == filetype.Unknown {
|
||||
return "application/octet-stream", nil
|
||||
}
|
||||
return kind.MIME.Value, nil
|
||||
}
|
||||
|
||||
// BuildContext creates an ActionContext for a matched file + rule.
|
||||
func BuildContext(path string, rule *config.RuleConfig, services map[string]config.ServiceConfig, st *state.DB) (*ActionContext, error) {
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
vars := map[string]string{
|
||||
"mtime": info.ModTime().UTC().Format(time.RFC3339),
|
||||
"size_kb": strconv.FormatInt(info.Size()/1024, 10),
|
||||
}
|
||||
|
||||
// Compute hash8 lazily by doing it once here
|
||||
if h, err := fileHash8(path); err == nil {
|
||||
vars["hash8"] = h
|
||||
}
|
||||
|
||||
// Get seq for this rule
|
||||
var seqStr string
|
||||
if st != nil {
|
||||
if seq, err := st.NextSeq(rule.Name); err == nil {
|
||||
seqStr = strconv.FormatInt(seq, 10)
|
||||
}
|
||||
}
|
||||
vars["seq"] = seqStr
|
||||
|
||||
return &ActionContext{
|
||||
CurrentPath: path,
|
||||
OriginalPath: path,
|
||||
RuleName: rule.Name,
|
||||
Vars: vars,
|
||||
Services: services,
|
||||
State: st,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func fileHash8(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))[:8], nil
|
||||
}
|
||||
|
||||
// FileSHA256 returns the full hex SHA256 of a file (used for dedup in state).
|
||||
func FileSHA256(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
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
package state
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
type Status string
|
||||
|
||||
const (
|
||||
StatusSuccess Status = "success"
|
||||
StatusError Status = "error"
|
||||
StatusSkipped Status = "skipped"
|
||||
)
|
||||
|
||||
type Record struct {
|
||||
ID int64
|
||||
Path string
|
||||
SHA256 string
|
||||
Rule string
|
||||
ProcessedAt time.Time
|
||||
Status Status
|
||||
ErrorMsg string
|
||||
}
|
||||
|
||||
type DB struct {
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
func Open(path string) (*DB, error) {
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
return nil, fmt.Errorf("creating state dir: %w", err)
|
||||
}
|
||||
db, err := sql.Open("sqlite", path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("opening state db: %w", err)
|
||||
}
|
||||
if _, err := db.Exec(`PRAGMA journal_mode=WAL`); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := migrate(db); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &DB{db: db}, nil
|
||||
}
|
||||
|
||||
func migrate(db *sql.DB) error {
|
||||
_, err := db.Exec(`
|
||||
CREATE TABLE IF NOT EXISTS processed (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
path TEXT NOT NULL,
|
||||
sha256 TEXT NOT NULL,
|
||||
rule TEXT NOT NULL,
|
||||
processed_at TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
error_msg TEXT
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_path_sha256 ON processed(path, sha256);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS sequences (
|
||||
rule TEXT PRIMARY KEY,
|
||||
seq INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
`)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *DB) Close() error {
|
||||
return s.db.Close()
|
||||
}
|
||||
|
||||
// IsProcessed returns true if this exact file (path + sha256) was already successfully processed.
|
||||
func (s *DB) IsProcessed(path, sha256 string) (bool, error) {
|
||||
var count int
|
||||
err := s.db.QueryRow(
|
||||
`SELECT COUNT(*) FROM processed WHERE path=? AND sha256=? AND status='success'`,
|
||||
path, sha256,
|
||||
).Scan(&count)
|
||||
return count > 0, err
|
||||
}
|
||||
|
||||
func (s *DB) Insert(path, sha256, rule string, status Status, errMsg string) error {
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO processed (path, sha256, rule, processed_at, status, error_msg) VALUES (?,?,?,?,?,?)`,
|
||||
path, sha256, rule, time.Now().UTC().Format(time.RFC3339), string(status), errMsg,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// NextSeq returns the next autoincrement value for a rule name (used in rename templates).
|
||||
func (s *DB) NextSeq(rule string) (int64, error) {
|
||||
tx, err := s.db.Begin()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
_, err = tx.Exec(
|
||||
`INSERT INTO sequences(rule, seq) VALUES(?,1) ON CONFLICT(rule) DO UPDATE SET seq=seq+1`,
|
||||
rule,
|
||||
)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var seq int64
|
||||
if err := tx.QueryRow(`SELECT seq FROM sequences WHERE rule=?`, rule).Scan(&seq); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return seq, tx.Commit()
|
||||
}
|
||||
|
||||
func (s *DB) Recent(n int) ([]Record, error) {
|
||||
rows, err := s.db.Query(
|
||||
`SELECT id, path, sha256, rule, processed_at, status, COALESCE(error_msg,'')
|
||||
FROM processed ORDER BY id DESC LIMIT ?`, n,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var records []Record
|
||||
for rows.Next() {
|
||||
var r Record
|
||||
var ts string
|
||||
if err := rows.Scan(&r.ID, &r.Path, &r.SHA256, &r.Rule, &ts, &r.Status, &r.ErrorMsg); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r.ProcessedAt, _ = time.Parse(time.RFC3339, ts)
|
||||
records = append(records, r)
|
||||
}
|
||||
return records, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,222 @@
|
||||
package watcher
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/fsnotify/fsnotify"
|
||||
|
||||
"github.com/alexandrev/homeflow/internal/actions"
|
||||
"github.com/alexandrev/homeflow/internal/config"
|
||||
"github.com/alexandrev/homeflow/internal/engine"
|
||||
"github.com/alexandrev/homeflow/internal/state"
|
||||
)
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
func New(cfg *config.Config, st *state.DB) (*Watcher, error) {
|
||||
fsw, err := fsnotify.NewWatcher()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("creating fsnotify watcher: %w", err)
|
||||
}
|
||||
return &Watcher{
|
||||
cfg: cfg,
|
||||
st: st,
|
||||
fsw: fsw,
|
||||
pending: make(map[string]*time.Timer),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Start begins watching all configured directories. Blocks until ctx is done.
|
||||
func (w *Watcher) Start(stop <-chan struct{}) error {
|
||||
for _, watch := range w.cfg.Watches {
|
||||
if err := w.addWatch(watch.Path, watch.Recursive); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("[watch] monitoring %s (recursive=%v)", watch.Path, watch.Recursive)
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return w.fsw.Close()
|
||||
case event, ok := <-w.fsw.Events:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if event.Has(fsnotify.Create) || event.Has(fsnotify.Write) {
|
||||
info, err := os.Stat(event.Name)
|
||||
if err != nil || info.IsDir() {
|
||||
continue
|
||||
}
|
||||
w.scheduleProcess(event.Name)
|
||||
}
|
||||
case err, ok := <-w.fsw.Errors:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
log.Printf("[watch] error: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// RunOnce processes all existing files in watched directories once.
|
||||
func (w *Watcher) RunOnce() {
|
||||
for _, watch := range w.cfg.Watches {
|
||||
w.walkAndProcess(watch.Path, watch.Recursive)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Watcher) addWatch(path string, recursive bool) error {
|
||||
if err := w.fsw.Add(path); err != nil {
|
||||
return fmt.Errorf("watching %s: %w", path, err)
|
||||
}
|
||||
if recursive {
|
||||
entries, err := os.ReadDir(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, e := range entries {
|
||||
if e.IsDir() {
|
||||
subpath := path + "/" + e.Name()
|
||||
if err := w.addWatch(subpath, true); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// scheduleProcess debounces file processing: waits stability_secs after the
|
||||
// last event before processing. Resets the timer if a new event arrives.
|
||||
func (w *Watcher) scheduleProcess(path string) {
|
||||
stability := time.Duration(w.cfg.Settings.StabilitySecs) * time.Second
|
||||
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
|
||||
if t, ok := w.pending[path]; ok {
|
||||
t.Reset(stability)
|
||||
return
|
||||
}
|
||||
w.pending[path] = time.AfterFunc(stability, func() {
|
||||
w.mu.Lock()
|
||||
delete(w.pending, path)
|
||||
w.mu.Unlock()
|
||||
w.processFile(path)
|
||||
})
|
||||
}
|
||||
|
||||
func (w *Watcher) processFile(path string) {
|
||||
// Check file still exists
|
||||
if _, err := os.Stat(path); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
sha, err := engine.FileSHA256(path)
|
||||
if err != nil {
|
||||
log.Printf("[process] %s: hash error: %v", path, err)
|
||||
return
|
||||
}
|
||||
|
||||
// Find which watch config owns this path
|
||||
for i := range w.cfg.Watches {
|
||||
watch := &w.cfg.Watches[i]
|
||||
if !isUnder(path, watch.Path) {
|
||||
continue
|
||||
}
|
||||
|
||||
matchMode := w.cfg.Settings.RuleMatch
|
||||
for ri := range watch.Rules {
|
||||
rule := &watch.Rules[ri]
|
||||
|
||||
matched, err := engine.MatchFile(path, rule)
|
||||
if err != nil {
|
||||
log.Printf("[match] %s rule %q: %v", path, rule.Name, err)
|
||||
continue
|
||||
}
|
||||
if !matched {
|
||||
continue
|
||||
}
|
||||
|
||||
// Dedup check
|
||||
already, err := w.st.IsProcessed(path, sha)
|
||||
if err != nil {
|
||||
log.Printf("[state] %s: %v", path, err)
|
||||
}
|
||||
if already {
|
||||
log.Printf("[skip] %s already processed by rule %q", path, rule.Name)
|
||||
break
|
||||
}
|
||||
|
||||
log.Printf("[match] %s → rule %q", path, rule.Name)
|
||||
w.runRule(path, sha, rule)
|
||||
|
||||
if matchMode == "first" {
|
||||
break
|
||||
}
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Watcher) runRule(path, sha string, rule *config.RuleConfig) {
|
||||
ctx, err := engine.BuildContext(path, rule, w.cfg.Services, w.st)
|
||||
if err != nil {
|
||||
log.Printf("[run] %s: build context: %v", path, err)
|
||||
w.st.Insert(path, sha, rule.Name, state.StatusError, err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
pipeline, err := actions.Build(rule.Actions, w.cfg.Services)
|
||||
if err != nil {
|
||||
log.Printf("[run] %s: build pipeline: %v", path, err)
|
||||
w.st.Insert(path, sha, rule.Name, state.StatusError, err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
if err := actions.RunPipeline(pipeline, rule.Actions, ctx); err != nil {
|
||||
log.Printf("[run] %s: pipeline error: %v", path, err)
|
||||
w.st.Insert(path, sha, rule.Name, state.StatusError, err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("[done] %s → rule %q", path, rule.Name)
|
||||
w.st.Insert(path, sha, rule.Name, state.StatusSuccess, "")
|
||||
}
|
||||
|
||||
func (w *Watcher) walkAndProcess(dir string, recursive bool) {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
log.Printf("[walk] %s: %v", dir, err)
|
||||
return
|
||||
}
|
||||
for _, e := range entries {
|
||||
fullPath := dir + "/" + e.Name()
|
||||
if e.IsDir() {
|
||||
if recursive {
|
||||
w.walkAndProcess(fullPath, true)
|
||||
}
|
||||
continue
|
||||
}
|
||||
w.processFile(fullPath)
|
||||
}
|
||||
}
|
||||
|
||||
func isUnder(path, dir string) bool {
|
||||
if len(path) < len(dir) {
|
||||
return false
|
||||
}
|
||||
return path[:len(dir)] == dir
|
||||
}
|
||||
Reference in New Issue
Block a user