sysopsandClaude Sonnet 5
be93614c9f
Täglicher Cron-Job (archivmail reconcile) berechnet pro Tenant/Quelle (SMTP-Journal, IMAP-Konto, POP3-Konto, Datei-Import) archivierte Mail-Zahlen, für IMAP zusätzlich einen Soll/Ist-Vergleich via UID-Tracking. Abweichungen über Schwellenwert erzeugen Audit-Log-Warnung. Neue Admin-Dashboard-Kachel "Vollständigkeits-Check" (letzte 7 Tage, Warn-Badge, CSV-Export). Schließt die "teilweise erfüllt"-Lücke bei Vollständigkeit im GoBD/DSGVO-Compliance-Check (VOI-Grundsatz 2). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
364 lines
10 KiB
Go
364 lines
10 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"flag"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"time"
|
||
|
||
"archivmail/config"
|
||
"archivmail/internal/index"
|
||
"archivmail/internal/storage"
|
||
"archivmail/pkg/mailparser"
|
||
)
|
||
|
||
|
||
type importResult struct {
|
||
Status string `json:"status"`
|
||
Imported int `json:"imported"`
|
||
Skipped int `json:"skipped"`
|
||
Errors int `json:"errors"`
|
||
DurationSec float64 `json:"duration_sec"`
|
||
}
|
||
|
||
func runImport(args []string) {
|
||
fs := flag.NewFlagSet("import", flag.ExitOnError)
|
||
configPath := fs.String("config", "/etc/archivmail/config.yml", "path to config file")
|
||
file := fs.String("file", "", "single EML or MBOX file to import")
|
||
dir := fs.String("dir", "", "directory to import EML/MBOX files from")
|
||
recursive := fs.Bool("recursive", false, "recurse into subdirectories (with --dir)")
|
||
dryRun := fs.Bool("dry-run", false, "simulate import without saving")
|
||
jsonOut := fs.Bool("json", false, "machine-readable JSON output")
|
||
tenantFlag := fs.Int64("tenant", 0, "tenant ID to assign imported mails to (0 = none/global)")
|
||
|
||
fs.Usage = func() {
|
||
fmt.Fprintln(os.Stderr, "Usage: archivmail import [flags]")
|
||
fmt.Fprintln(os.Stderr, "")
|
||
fmt.Fprintln(os.Stderr, "Flags:")
|
||
fs.PrintDefaults()
|
||
}
|
||
fs.Parse(args)
|
||
|
||
if *file == "" && *dir == "" {
|
||
fmt.Fprintln(os.Stderr, "error: --file or --dir required")
|
||
fs.Usage()
|
||
os.Exit(1)
|
||
}
|
||
|
||
start := time.Now()
|
||
|
||
cfg, err := config.Load(*configPath)
|
||
if err != nil {
|
||
fmt.Fprintf(os.Stderr, "error: load config: %v\n", err)
|
||
os.Exit(1)
|
||
}
|
||
|
||
storeCfg := storage.Config{
|
||
Dir: cfg.Storage.StorePath,
|
||
Keyfile: cfg.Storage.Keyfile,
|
||
DSN: cfg.Database.DSN(),
|
||
CompressEnabled: cfg.Storage.Compress,
|
||
}
|
||
mailStore, err := storage.New(storeCfg)
|
||
if err != nil {
|
||
fmt.Fprintf(os.Stderr, "error: storage init: %v\n", err)
|
||
os.Exit(1)
|
||
}
|
||
defer mailStore.Close()
|
||
|
||
batchSize := cfg.Index.BatchSize
|
||
if batchSize <= 0 {
|
||
batchSize = 100
|
||
}
|
||
backend := cfg.Index.Backend
|
||
if backend == "" {
|
||
backend = "manticore"
|
||
}
|
||
|
||
var idxMgr index.TenantIndexer
|
||
if backend == "manticore" {
|
||
dsn := cfg.Index.ManticoreDSN
|
||
if dsn == "" {
|
||
dsn = "manticore@tcp(127.0.0.1:9306)/?charset=utf8mb4"
|
||
}
|
||
m, err := index.NewManticoreTenantManager(dsn)
|
||
if err != nil {
|
||
fmt.Fprintf(os.Stderr, "error: manticore init: %v\n", err)
|
||
os.Exit(1)
|
||
}
|
||
idxMgr = m
|
||
} else {
|
||
m, err := index.NewTenantIndexManager(cfg.Index.Path, batchSize, backend)
|
||
if err != nil {
|
||
fmt.Fprintf(os.Stderr, "error: index init: %v\n", err)
|
||
os.Exit(1)
|
||
}
|
||
idxMgr = m
|
||
}
|
||
defer idxMgr.Close()
|
||
|
||
// Collect files to process
|
||
type fileEntry struct {
|
||
path string
|
||
isMbox bool
|
||
}
|
||
var files []fileEntry
|
||
|
||
if *file != "" {
|
||
isMbox := strings.HasSuffix(strings.ToLower(*file), ".mbox")
|
||
files = append(files, fileEntry{*file, isMbox})
|
||
}
|
||
|
||
if *dir != "" {
|
||
info, err := os.Stat(*dir)
|
||
if err != nil {
|
||
fmt.Fprintf(os.Stderr, "error: cannot access dir %s: %v\n", *dir, err)
|
||
os.Exit(1)
|
||
}
|
||
if !info.IsDir() {
|
||
fmt.Fprintf(os.Stderr, "error: %s is not a directory\n", *dir)
|
||
os.Exit(1)
|
||
}
|
||
|
||
walkFn := func(path string, d os.DirEntry, werr error) error {
|
||
if werr != nil {
|
||
return werr
|
||
}
|
||
if d.IsDir() {
|
||
if !*recursive && path != *dir {
|
||
return filepath.SkipDir
|
||
}
|
||
return nil
|
||
}
|
||
lower := strings.ToLower(d.Name())
|
||
if strings.HasSuffix(lower, ".eml") {
|
||
files = append(files, fileEntry{path, false})
|
||
} else if strings.HasSuffix(lower, ".mbox") {
|
||
files = append(files, fileEntry{path, true})
|
||
}
|
||
return nil
|
||
}
|
||
filepath.WalkDir(*dir, walkFn)
|
||
}
|
||
|
||
if len(files) == 0 {
|
||
if !*jsonOut {
|
||
fmt.Println("No EML or MBOX files found.")
|
||
} else {
|
||
printImportJSON(importResult{Status: "done"}, start)
|
||
}
|
||
os.Exit(0)
|
||
}
|
||
|
||
if !*jsonOut {
|
||
if *dryRun {
|
||
fmt.Printf("Dry run – scanning %d file(s)...\n", len(files))
|
||
} else {
|
||
fmt.Printf("Found %d file(s) to process...\n", len(files))
|
||
}
|
||
}
|
||
|
||
var tenantID *int64
|
||
if *tenantFlag > 0 {
|
||
tid := *tenantFlag
|
||
tenantID = &tid
|
||
}
|
||
|
||
imported := 0
|
||
skipped := 0
|
||
errors := 0
|
||
total := 0
|
||
|
||
for _, fe := range files {
|
||
raw, err := os.ReadFile(fe.path)
|
||
if err != nil {
|
||
fmt.Fprintf(os.Stderr, "error: read %s: %v\n", fe.path, err)
|
||
errors++
|
||
continue
|
||
}
|
||
|
||
var messages [][]byte
|
||
if fe.isMbox {
|
||
messages = mailparser.SplitMbox(raw)
|
||
if len(messages) == 0 {
|
||
continue
|
||
}
|
||
} else {
|
||
messages = [][]byte{raw}
|
||
}
|
||
|
||
for _, msg := range messages {
|
||
total++
|
||
result := importMessage(mailStore, idxMgr, msg, *dryRun, tenantID)
|
||
switch result {
|
||
case "imported":
|
||
imported++
|
||
case "skipped":
|
||
skipped++
|
||
case "error":
|
||
errors++
|
||
}
|
||
if !*jsonOut && total%100 == 0 {
|
||
fmt.Printf("Progress: %d processed (imported: %d, skipped: %d, errors: %d)\n",
|
||
total, imported, skipped, errors)
|
||
}
|
||
}
|
||
}
|
||
|
||
if *jsonOut {
|
||
printImportJSON(importResult{
|
||
Status: "done",
|
||
Imported: imported,
|
||
Skipped: skipped,
|
||
Errors: errors,
|
||
DurationSec: time.Since(start).Seconds(),
|
||
}, start)
|
||
} else {
|
||
fmt.Printf("\nFertig:\n")
|
||
fmt.Printf(" Importiert: %d\n", imported)
|
||
fmt.Printf(" Übersprungen: %d (Duplikate)\n", skipped)
|
||
fmt.Printf(" Fehler: %d\n", errors)
|
||
if *dryRun {
|
||
fmt.Println("\n[dry-run] Keine Daten wurden gespeichert.")
|
||
}
|
||
}
|
||
|
||
if errors > 0 {
|
||
os.Exit(1)
|
||
}
|
||
}
|
||
|
||
// importMessage stores and indexes a single raw message. Returns "imported", "skipped", or "error".
|
||
// tenantID, when non-nil, assigns the imported mail to that tenant (both in
|
||
// storage and in the tenant-specific search index) instead of the
|
||
// global/tenant-less context.
|
||
func importMessage(mailStore *storage.Store, idxMgr index.TenantIndexer, raw []byte, dryRun bool, tenantID *int64) string {
|
||
pm, err := mailparser.Parse(raw)
|
||
if err != nil {
|
||
fmt.Fprintf(os.Stderr, "warning: parse failed: %v\n", err)
|
||
return "error"
|
||
}
|
||
|
||
if dryRun {
|
||
return "imported"
|
||
}
|
||
|
||
id, err := mailStore.Save(context.Background(), raw, pm.Date, tenantID)
|
||
if err != nil {
|
||
fmt.Fprintf(os.Stderr, "warning: save failed: %v\n", err)
|
||
return "error"
|
||
}
|
||
|
||
// PROJ-52: mark bulk-imported mails as source 'import' for reconciliation.
|
||
if err := mailStore.TagSource(context.Background(), id, "import", nil); err != nil {
|
||
fmt.Fprintf(os.Stderr, "warning: tag source failed: %v\n", err)
|
||
}
|
||
|
||
var attachNames []string
|
||
for _, a := range pm.Attachments {
|
||
attachNames = append(attachNames, a.Filename)
|
||
}
|
||
|
||
doc := index.MailDocument{
|
||
ID: id,
|
||
From: pm.From,
|
||
To: strings.Join(pm.To, " "),
|
||
CC: strings.Join(pm.CC, " "),
|
||
Subject: pm.Subject,
|
||
Body: pm.TextBody + " " + pm.HTMLBody,
|
||
AttachNames: strings.Join(attachNames, " "),
|
||
HasAttachment: len(pm.Attachments) > 0,
|
||
Date: pm.Date,
|
||
Size: int64(len(raw)),
|
||
TenantID: tenantID,
|
||
}
|
||
|
||
if err := idxMgr.ForTenant(tenantID).IndexSync(doc); err != nil {
|
||
fmt.Fprintf(os.Stderr, "warning: index failed for %s: %v\n", id, err)
|
||
return "error"
|
||
}
|
||
|
||
return "imported"
|
||
}
|
||
|
||
func printImportJSON(r importResult, start time.Time) {
|
||
r.DurationSec = time.Since(start).Seconds()
|
||
enc := json.NewEncoder(os.Stdout)
|
||
enc.SetIndent("", " ")
|
||
enc.Encode(r)
|
||
}
|
||
|
||
func printHelp() {
|
||
fmt.Printf(`archivmail %s – Mail-Archiv-Daemon und CLI
|
||
|
||
Commands:
|
||
serve Daemon starten (Standard wenn kein Befehl angegeben)
|
||
import E-Mails importieren (EML, MBOX, Verzeichnis)
|
||
import-piler Aus mailpiler migrieren (pilerexport oder direkte Store-Methode)
|
||
export E-Mails exportieren (EML, MBOX)
|
||
reindex Index neu aufbauen (alle oder pro Mandant)
|
||
purge Mails mit abgelaufener Aufbewahrungsfrist löschen (cron-fähig)
|
||
recompress Bestehende Mails nachträglich gzip-komprimieren
|
||
rethread Thread-IDs rückwirkend aus In-Reply-To/References befüllen
|
||
ocr-reprocess OCR für Anhänge nachholen (alle oder pro Mandant/Status)
|
||
index-pending Ungeindexte Mails nachindexieren (cron-fähig, PROJ-58 batch_mode)
|
||
update Auf neueste Version aktualisieren (führt update.sh aus)
|
||
status Healthcheck für DB, Manticore und Storage
|
||
version Version anzeigen
|
||
help Diese Hilfe anzeigen
|
||
|
||
archivmail import [flags]
|
||
--config Pfad zur Konfigurationsdatei (Standard: /etc/archivmail/config.yml)
|
||
--file Einzelne EML- oder MBOX-Datei
|
||
--dir Verzeichnis mit EML/MBOX-Dateien
|
||
--recursive Unterverzeichnisse einschließen (mit --dir)
|
||
--tenant Mandanten-ID für importierte Mails (0 = keine/global)
|
||
--dry-run Simulation ohne Speichern
|
||
--json Maschinenlesbare JSON-Ausgabe
|
||
|
||
archivmail import-piler [flags]
|
||
--config Pfad zur Konfigurationsdatei (Standard: /etc/archivmail/config.yml)
|
||
--method auto | pilerexport | direct (Standard: auto)
|
||
--pilerexport Pfad zum pilerexport Binary (auto-erkennung)
|
||
--export-dir Ausgabeverzeichnis für pilerexport (temp wenn leer)
|
||
--store-dir mailpiler Store-Verzeichnis (Standard: /var/piler/store)
|
||
--key-file mailpiler AES-Schlüsseldatei (Standard: /var/piler/store/piler.key)
|
||
--date-from Export ab Datum YYYY-MM-DD (pilerexport-Methode)
|
||
--date-to Export bis Datum YYYY-MM-DD (pilerexport-Methode)
|
||
--dry-run Simulation ohne Speichern
|
||
--json Maschinenlesbare JSON-Ausgabe
|
||
|
||
archivmail export [flags]
|
||
--config Pfad zur Konfigurationsdatei (Standard: /etc/archivmail/config.yml)
|
||
--out Zielverzeichnis oder Zieldatei (Pflicht)
|
||
--format eml (Standard) oder mbox
|
||
--from Filter nach Absender
|
||
--to Filter nach Empfänger
|
||
--date-from Filter ab Datum (ISO 8601: 2024-01-01)
|
||
--date-to Filter bis Datum (ISO 8601: 2024-12-31)
|
||
--query Volltext-Suche
|
||
--force Vorhandene Dateien überschreiben
|
||
--json Maschinenlesbare JSON-Ausgabe
|
||
|
||
archivmail reindex [flags]
|
||
--config Pfad zur Konfigurationsdatei (Standard: /etc/archivmail/config.yml)
|
||
--tenant Mandanten-ID für partiellen Reindex (0 = alle)
|
||
|
||
archivmail recompress [flags]
|
||
--config Pfad zur Konfigurationsdatei (Standard: /etc/archivmail/config.yml)
|
||
--dry-run Simulation: zeigt wie viel gespart würde, ohne Dateien zu ändern
|
||
|
||
archivmail rethread [flags]
|
||
--config Pfad zur Konfigurationsdatei (Standard: /etc/archivmail/config.yml)
|
||
--dry-run Simulation: zeigt wie viele Mails gethreaded würden, ohne DB zu ändern
|
||
|
||
archivmail status [flags]
|
||
--config Pfad zur Konfigurationsdatei (Standard: /etc/archivmail/config.yml)
|
||
--json Maschinenlesbare JSON-Ausgabe
|
||
`, AppVersion)
|
||
}
|