christoph/ai-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

zwischenstand

Browse Source
This commit is contained in:
Christoph K.
2026-03-20 23:24:56 +01:00
parent b1a576f61e
commit 905981cd1e
25 changed files with 3607 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
internal/agents/actions.go
View File

@@ -15,9 +15,22 @@ const (
ActionDone = "done"
ActionDelete = "delete"
// Memory
ActionIngestURL = "url"
ActionIngestPDF = "pdf"
ActionProfile = "profile"
ActionProfileShow = "profile-show"
// Knowledge
ActionKnowledgeList = "list"
ActionKnowledgeDelete = "delete"
// Tool/Email
ActionEmail = "email"
ActionEmail = "email"
ActionEmailSummary = "summary"
ActionEmailUnread = "unread"
ActionEmailRemind = "remind"
ActionEmailIngest = "ingest"
ActionEmailMove = "move"
ActionEmailTriage = "triage"
)

152
internal/agents/tool/agent.go
View File

@@ -3,9 +3,12 @@ package tool
import (
"fmt"
"strings"
"my-brain-importer/internal/agents"
"my-brain-importer/internal/agents/tool/email"
"my-brain-importer/internal/brain"
"my-brain-importer/internal/config"
)
// Agent verteilt Tool-Anfragen an spezialisierte Sub-Agenten.
@@ -41,8 +44,14 @@ func (a *Agent) handleEmail(req agents.Request) agents.Response {
result, err = email.SummarizeUnread()
case agents.ActionEmailRemind:
result, err = email.ExtractReminders()
case agents.ActionEmailIngest:
return a.handleEmailIngest(req)
case agents.ActionEmailMove:
return a.handleEmailMove(req)
case agents.ActionEmailTriage:
return a.handleEmailTriage()
default:
return agents.Response{Text: fmt.Sprintf("❌ Unbekannte Email-Aktion `%s`. Verfügbar: summary, unread, remind", subAction)}
return agents.Response{Text: fmt.Sprintf("❌ Unbekannte Email-Aktion `%s`. Verfügbar: summary, unread, remind, ingest, move, triage", subAction)}
}
if err != nil {
@@ -50,3 +59,144 @@ func (a *Agent) handleEmail(req agents.Request) agents.Response {
}
return agents.Response{Text: "📧 **Email-Analyse:**\n\n" + result}
}
// handleEmailIngest importiert Emails aus einem IMAP-Ordner in Qdrant.
// Args[1] = Ordnername (Standard: "Archiv")
func (a *Agent) handleEmailIngest(req agents.Request) agents.Response {
folder := "Archiv"
if len(req.Args) > 1 && req.Args[1] != "" {
folder = req.Args[1]
}
accounts := config.AllEmailAccounts()
if len(accounts) == 0 {
return agents.Response{Text: "❌ Kein Email-Account konfiguriert."}
}
total := 0
var errs []string
for _, acc := range accounts {
n, err := brain.IngestEmailFolder(acc, folder, 500)
if err != nil {
errs = append(errs, fmt.Sprintf("%s: %v", acc.Name, err))
continue
}
total += n
}
if len(errs) > 0 && total == 0 {
return agents.Response{Text: fmt.Sprintf("❌ Email-Ingest fehlgeschlagen:\n%s", joinLines(errs))}
}
msg := fmt.Sprintf("✅ **Email-Ingest abgeschlossen:** %d Emails aus `%s` in die Wissensdatenbank importiert.", total, folder)
if len(errs) > 0 {
msg += "\n⚠️ Fehler bei einigen Accounts:\n" + joinLines(errs)
}
return agents.Response{Text: msg}
}
// handleEmailMove verschiebt alle ungelesenen Emails in einen konfigurierten Archivordner.
// Args[1] = Zielordner-Name (aus archive_folders in config oder Legacy: 2Jahre/5Jahre/Archiv)
func (a *Agent) handleEmailMove(req agents.Request) agents.Response {
if len(req.Args) < 2 || req.Args[1] == "" {
return agents.Response{Text: "❌ Zielordner fehlt. " + buildMoveFoldersHint()}
}
dest := req.Args[1]
imapFolder, ok := resolveArchiveFolder(dest)
if !ok {
return agents.Response{Text: fmt.Sprintf("❌ Unbekannter Ordner `%s`. %s", dest, buildMoveFoldersHint())}
}
accounts := config.AllEmailAccounts()
if len(accounts) == 0 {
return agents.Response{Text: "❌ Kein Email-Account konfiguriert."}
}
total := 0
var errs []string
for _, acc := range accounts {
n, err := email.MoveUnread(acc, imapFolder)
if err != nil {
errs = append(errs, fmt.Sprintf("%s: %v", acc.Name, err))
continue
}
total += n
}
if len(errs) > 0 && total == 0 {
return agents.Response{Text: fmt.Sprintf("❌ Verschieben fehlgeschlagen:\n%s", joinLines(errs))}
}
if total == 0 {
return agents.Response{Text: fmt.Sprintf("📭 Keine ungelesenen Emails zum Verschieben nach `%s`.", imapFolder)}
}
msg := fmt.Sprintf("✅ **%d Email(s)** nach `%s` verschoben.", total, imapFolder)
if len(errs) > 0 {
msg += "\n⚠️ Fehler:\n" + joinLines(errs)
}
return agents.Response{Text: msg}
}
// handleEmailTriage klassifiziert die letzten 10 Emails aller Accounts und verschiebt sie.
func (a *Agent) handleEmailTriage() agents.Response {
result, err := email.TriageRecentAllAccounts(10)
if err != nil {
return agents.Response{Error: err, Text: fmt.Sprintf("❌ Triage fehlgeschlagen: %v", err)}
}
return agents.Response{Text: "🗂️ **Email-Triage (letzte 10 Emails):**\n\n" + result}
}
// ResolveArchiveFolder ist die exportierte Version von resolveArchiveFolder für den Discord-Layer.
func ResolveArchiveFolder(name string) (imapFolder string, ok bool) {
return resolveArchiveFolder(name)
}
// resolveArchiveFolder sucht den IMAP-Ordnernamen für einen Anzeigenamen aus der Config.
// Fallback: Legacy-Hardcoding für 2Jahre/5Jahre/Archiv wenn keine archive_folders konfiguriert.
func resolveArchiveFolder(name string) (imapFolder string, ok bool) {
for _, acc := range config.AllEmailAccounts() {
for _, af := range acc.ArchiveFolders {
if strings.EqualFold(af.Name, name) || strings.EqualFold(af.IMAPFolder, name) {
return af.IMAPFolder, true
}
}
}
// Legacy-Fallback für Konfigurationen ohne archive_folders
legacy := map[string]string{
"2jahre": "2Jahre",
"5jahre": "5Jahre",
"archiv": "Archiv",
}
if canonical, found := legacy[strings.ToLower(name)]; found {
return canonical, true
}
return "", false
}
// buildMoveFoldersHint gibt eine Hinweis-Nachricht mit verfügbaren Archivordnern zurück.
func buildMoveFoldersHint() string {
seen := map[string]bool{}
var names []string
for _, acc := range config.AllEmailAccounts() {
for _, af := range acc.ArchiveFolders {
key := strings.ToLower(af.Name)
if !seen[key] {
seen[key] = true
names = append(names, fmt.Sprintf("`%s`", af.Name))
}
}
}
if len(names) == 0 {
return "Verfügbar: `2Jahre`, `5Jahre`, `Archiv`"
}
return fmt.Sprintf("Verfügbar: %s", strings.Join(names, ", "))
}
func joinLines(lines []string) string {
result := ""
for _, l := range lines {
result += "• " + l + "\n"
}
return result
}

402
internal/agents/tool/email/client.go
View File

@@ -3,7 +3,12 @@ package email
import (
"crypto/tls"
"encoding/base64"
"fmt"
"log/slog"
"mime/quotedprintable"
"strings"
"time"
imap "github.com/emersion/go-imap/v2"
"github.com/emersion/go-imap/v2/imapclient"
@@ -18,15 +23,43 @@ type Message struct {
Date string
}
// Client wraps die IMAP-Verbindung.
type Client struct {
c *imapclient.Client
// SelectMessage koppelt eine Message mit ihrer IMAP-Sequenznummer für UI-Zwecke.
type SelectMessage struct {
Message
SeqNum uint32
Unread bool // true = \Seen flag nicht gesetzt
}
// Connect öffnet eine IMAP-Verbindung.
// MessageWithBody repräsentiert eine Email mit Text-Inhalt (für Datenbankimport).
type MessageWithBody struct {
Message
Body string
}
// Client wraps die IMAP-Verbindung.
type Client struct {
c *imapclient.Client
folder string // INBOX-Ordner (leer = "INBOX")
}
// Connect öffnet eine IMAP-Verbindung mit dem Legacy-Email-Block aus der Config.
func Connect() (*Client, error) {
cfg := config.Cfg.Email
addr := fmt.Sprintf("%s:%d", cfg.Host, cfg.Port)
acc := config.EmailAccount{
Host: cfg.Host,
Port: cfg.Port,
User: cfg.User,
Password: cfg.Password,
TLS: cfg.TLS,
StartTLS: cfg.StartTLS,
Folder: cfg.Folder,
}
return ConnectAccount(acc)
}
// ConnectAccount öffnet eine IMAP-Verbindung für einen bestimmten EmailAccount.
func ConnectAccount(acc config.EmailAccount) (*Client, error) {
addr := fmt.Sprintf("%s:%d", acc.Host, acc.Port)
var (
c *imapclient.Client
@@ -34,11 +67,11 @@ func Connect() (*Client, error) {
)
switch {
case cfg.TLS:
tlsCfg := &tls.Config{ServerName: cfg.Host}
case acc.TLS:
tlsCfg := &tls.Config{ServerName: acc.Host}
c, err = imapclient.DialTLS(addr, &imapclient.Options{TLSConfig: tlsCfg})
case cfg.StartTLS:
tlsCfg := &tls.Config{ServerName: cfg.Host}
case acc.StartTLS:
tlsCfg := &tls.Config{ServerName: acc.Host}
c, err = imapclient.DialStartTLS(addr, &imapclient.Options{TLSConfig: tlsCfg})
default:
c, err = imapclient.DialInsecure(addr, nil)
@@ -47,12 +80,12 @@ func Connect() (*Client, error) {
return nil, fmt.Errorf("IMAP verbinden: %w", err)
}
if err := c.Login(cfg.User, cfg.Password).Wait(); err != nil {
if err := c.Login(acc.User, acc.Password).Wait(); err != nil {
c.Close()
return nil, fmt.Errorf("IMAP login: %w", err)
}
return &Client{c: c}, nil
return &Client{c: c, folder: acc.Folder}, nil
}
// Close schließt die Verbindung.
@@ -61,9 +94,28 @@ func (cl *Client) Close() {
cl.c.Close()
}
// EnsureFolder legt einen IMAP-Ordner an falls er nicht existiert.
// Strato-kompatibel: ignoriert alle "already exists"-Varianten.
func (cl *Client) EnsureFolder(folder string) error {
err := cl.c.Create(folder, nil).Wait()
if err == nil {
slog.Info("IMAP: Ordner angelegt", "ordner", folder)
return nil
}
errLower := strings.ToLower(err.Error())
if strings.Contains(errLower, "already exists") ||
strings.Contains(errLower, "alreadyexists") ||
strings.Contains(errLower, "mailbox exists") ||
strings.Contains(errLower, "exists") {
return nil // Ordner existiert bereits — kein Fehler
}
slog.Error("IMAP: Ordner anlegen fehlgeschlagen", "ordner", folder, "fehler", err)
return fmt.Errorf("IMAP create folder %s: %w", folder, err)
}
// FetchRecent holt die letzten n Emails (Envelope-Daten, kein Body).
func (cl *Client) FetchRecent(n uint32) ([]Message, error) {
folder := config.Cfg.Email.Folder
folder := cl.folder
if folder == "" {
folder = "INBOX"
}
@@ -94,7 +146,7 @@ func (cl *Client) FetchRecent(n uint32) ([]Message, error) {
// FetchUnread holt ungelesene Emails (Envelope-Daten, kein Body).
func (cl *Client) FetchUnread() ([]Message, error) {
folder := config.Cfg.Email.Folder
folder := cl.folder
if folder == "" {
folder = "INBOX"
}
@@ -129,7 +181,7 @@ func (cl *Client) FetchUnread() ([]Message, error) {
// FetchUnreadSeqNums holt ungelesene Emails und gibt zusätzlich die Sequenznummern zurück.
// Selektiert den Ordner im Lese-Schreib-Modus (für nachfolgendes Verschieben).
func (cl *Client) FetchUnreadSeqNums() ([]Message, []uint32, error) {
folder := config.Cfg.Email.Folder
folder := cl.folder
if folder == "" {
folder = "INBOX"
}
@@ -172,6 +224,328 @@ func (cl *Client) MoveMessages(seqNums []uint32, destFolder string) error {
return nil
}
// FetchUnreadForSelect gibt ungelesene Emails mit ihren Sequenznummern zurück.
// Selektiert den Ordner im Lese-Schreib-Modus (für nachfolgendes Verschieben).
func (cl *Client) FetchUnreadForSelect() ([]SelectMessage, error) {
folder := cl.folder
if folder == "" {
folder = "INBOX"
}
if _, err := cl.c.Select(folder, nil).Wait(); err != nil {
return nil, fmt.Errorf("IMAP select: %w", err)
}
searchData, err := cl.c.Search(&imap.SearchCriteria{
NotFlag: []imap.Flag{imap.FlagSeen},
}, nil).Wait()
if err != nil {
return nil, fmt.Errorf("IMAP search: %w", err)
}
seqNums := searchData.AllSeqNums()
if len(seqNums) == 0 {
return nil, nil
}
var seqSet imap.SeqSet
seqSet.AddNum(seqNums...)
rawMsgs, err := cl.c.Fetch(seqSet, &imap.FetchOptions{Envelope: true}).Collect()
if err != nil {
return nil, fmt.Errorf("IMAP fetch: %w", err)
}
seqToMsg := make(map[uint32]*imapclient.FetchMessageBuffer, len(rawMsgs))
for _, m := range rawMsgs {
if m.Envelope != nil {
seqToMsg[m.SeqNum] = m
}
}
result := make([]SelectMessage, 0, len(seqNums))
for _, sn := range seqNums {
m, ok := seqToMsg[sn]
if !ok {
continue
}
result = append(result, SelectMessage{
Message: parseMessage(m),
SeqNum: sn,
Unread: true,
})
}
return result, nil
}
// FetchRecentForSelect gibt die letzten n Emails mit Sequenznummern und Unread-Status zurück.
// Selektiert den Ordner im Lese-Schreib-Modus (für nachfolgendes Verschieben).
func (cl *Client) FetchRecentForSelect(n uint32) ([]SelectMessage, error) {
folder := cl.folder
if folder == "" {
folder = "INBOX"
}
selectData, err := cl.c.Select(folder, nil).Wait()
if err != nil {
return nil, fmt.Errorf("IMAP select: %w", err)
}
if selectData.NumMessages == 0 {
return nil, nil
}
start := uint32(1)
if selectData.NumMessages > n {
start = selectData.NumMessages - n + 1
}
var seqSet imap.SeqSet
seqSet.AddRange(start, selectData.NumMessages)
rawMsgs, err := cl.c.Fetch(seqSet, &imap.FetchOptions{Envelope: true, Flags: true}).Collect()
if err != nil {
return nil, fmt.Errorf("IMAP fetch: %w", err)
}
result := make([]SelectMessage, 0, len(rawMsgs))
for _, m := range rawMsgs {
if m.Envelope == nil {
continue
}
unread := true
for _, f := range m.Flags {
if f == imap.FlagSeen {
unread = false
break
}
}
result = append(result, SelectMessage{
Message: parseMessage(m),
SeqNum: m.SeqNum,
Unread: unread,
})
}
return result, nil
}
// MoveOldMessages verschiebt alle Emails im Ordner, die älter als olderThanDays Tage sind, nach destFolder.
// Gibt die Anzahl verschobener Nachrichten zurück. olderThanDays <= 0 ist ein No-op.
func (cl *Client) MoveOldMessages(folder, destFolder string, olderThanDays int) (int, error) {
if olderThanDays <= 0 {
return 0, nil
}
if folder == "" {
folder = "INBOX"
}
cutoff := time.Now().AddDate(0, 0, -olderThanDays).Truncate(24 * time.Hour)
if _, err := cl.c.Select(folder, nil).Wait(); err != nil {
return 0, fmt.Errorf("IMAP select %s: %w", folder, err)
}
searchData, err := cl.c.Search(&imap.SearchCriteria{Before: cutoff}, nil).Wait()
if err != nil {
return 0, fmt.Errorf("IMAP search: %w", err)
}
seqNums := searchData.AllSeqNums()
if len(seqNums) == 0 {
return 0, nil
}
var seqSet imap.SeqSet
seqSet.AddNum(seqNums...)
if _, err := cl.c.Move(seqSet, destFolder).Wait(); err != nil {
return 0, fmt.Errorf("IMAP move: %w", err)
}
return len(seqNums), nil
}
// MoveSpecificMessages selektiert den Inbox-Ordner und verschiebt die angegebenen Sequenznummern.
func (cl *Client) MoveSpecificMessages(seqNums []uint32, destFolder string) error {
folder := cl.folder
if folder == "" {
folder = "INBOX"
}
if _, err := cl.c.Select(folder, nil).Wait(); err != nil {
return fmt.Errorf("IMAP select: %w", err)
}
return cl.MoveMessages(seqNums, destFolder)
}
// CleanupOldEmails löscht Emails im Ordner, die älter als retentionDays sind.
// Gibt die Anzahl gelöschter Nachrichten zurück. retentionDays <= 0 ist ein No-op.
func (cl *Client) CleanupOldEmails(folder string, retentionDays int) (int, error) {
if retentionDays <= 0 {
return 0, nil
}
if folder == "" {
folder = "INBOX"
}
cutoff := time.Now().AddDate(0, 0, -retentionDays).Truncate(24 * time.Hour)
if _, err := cl.c.Select(folder, nil).Wait(); err != nil {
return 0, fmt.Errorf("IMAP select %s: %w", folder, err)
}
searchData, err := cl.c.Search(&imap.SearchCriteria{Before: cutoff}, nil).Wait()
if err != nil {
return 0, fmt.Errorf("IMAP search: %w", err)
}
seqNums := searchData.AllSeqNums()
if len(seqNums) == 0 {
return 0, nil
}
var seqSet imap.SeqSet
seqSet.AddNum(seqNums...)
storeFlags := &imap.StoreFlags{
Op: imap.StoreFlagsAdd,
Silent: true,
Flags: []imap.Flag{imap.FlagDeleted},
}
if _, err := cl.c.Store(seqSet, storeFlags, nil).Collect(); err != nil {
return 0, fmt.Errorf("IMAP store flags: %w", err)
}
if _, err := cl.c.Expunge().Collect(); err != nil {
return 0, fmt.Errorf("IMAP expunge: %w", err)
}
return len(seqNums), nil
}
// FetchWithBody holt bis zu n Emails aus dem angegebenen Ordner mit Text-Body.
// Emails werden in Batches von 50 gefetcht um den IMAP-Server nicht zu überlasten.
func (cl *Client) FetchWithBody(folder string, n uint32) ([]MessageWithBody, error) {
if folder == "" {
folder = "INBOX"
}
selectData, err := cl.c.Select(folder, &imap.SelectOptions{ReadOnly: true}).Wait()
if err != nil {
return nil, fmt.Errorf("IMAP select %s: %w", folder, err)
}
if selectData.NumMessages == 0 {
return nil, nil
}
// Letzte n Nachrichten
total := selectData.NumMessages
start := uint32(1)
if total > n {
start = total - n + 1
}
bodySec := &imap.FetchItemBodySection{Specifier: imap.PartSpecifierText}
hdrSec := &imap.FetchItemBodySection{Specifier: imap.PartSpecifierHeader}
var result []MessageWithBody
batchSize := uint32(50)
for i := start; i <= total; i += batchSize {
end := i + batchSize - 1
if end > total {
end = total
}
var seqSet imap.SeqSet
seqSet.AddRange(i, end)
msgs, err := cl.c.Fetch(seqSet, &imap.FetchOptions{
Envelope: true,
BodySection: []*imap.FetchItemBodySection{bodySec, hdrSec},
}).Collect()
if err != nil {
return nil, fmt.Errorf("IMAP fetch batch %d-%d: %w", i, end, err)
}
for _, msg := range msgs {
if msg.Envelope == nil {
continue
}
m := MessageWithBody{Message: parseMessage(msg)}
// Content-Transfer-Encoding aus Header lesen
enc := ""
if hdr := msg.FindBodySection(hdrSec); hdr != nil {
for _, line := range strings.Split(string(hdr), "\n") {
if strings.HasPrefix(strings.ToLower(line), "content-transfer-encoding:") {
enc = strings.TrimSpace(strings.ToLower(strings.SplitN(line, ":", 2)[1]))
}
}
}
if body := msg.FindBodySection(bodySec); body != nil {
m.Body = decodeBody(body, enc)
}
result = append(result, m)
}
}
return result, nil
}
// decodeBody dekodiert einen Email-Body je nach Content-Transfer-Encoding.
func decodeBody(raw []byte, enc string) string {
var text string
switch enc {
case "base64":
cleaned := strings.ReplaceAll(strings.TrimSpace(string(raw)), "\r\n", "")
if decoded, err := base64.StdEncoding.DecodeString(cleaned); err == nil {
text = string(decoded)
} else if decoded, err := base64.RawStdEncoding.DecodeString(cleaned); err == nil {
text = string(decoded)
} else {
text = string(raw) // Fallback: roh
}
case "quoted-printable":
r := quotedprintable.NewReader(strings.NewReader(string(raw)))
if buf := new(strings.Builder); true {
buf.Grow(len(raw))
tmp := make([]byte, 4096)
for {
n, err := r.Read(tmp)
buf.Write(tmp[:n])
if err != nil {
break
}
}
text = buf.String()
}
default:
text = string(raw)
}
// Kürzen auf max 2000 Zeichen
text = strings.TrimSpace(text)
if len(text) > 2000 {
text = text[:2000]
}
return text
}
// parseMessage extrahiert eine Message aus einem FetchMessageBuffer.
func parseMessage(msg *imapclient.FetchMessageBuffer) Message {
m := Message{
Subject: msg.Envelope.Subject,
Date: msg.Envelope.Date.Format("2006-01-02 15:04"),
}
if len(msg.Envelope.From) > 0 {
addr := msg.Envelope.From[0]
if addr.Name != "" {
m.From = fmt.Sprintf("%s <%s@%s>", addr.Name, addr.Mailbox, addr.Host)
} else {
m.From = fmt.Sprintf("%s@%s", addr.Mailbox, addr.Host)
}
}
return m
}
func parseMessages(msgs []*imapclient.FetchMessageBuffer) []Message {
result := make([]Message, 0, len(msgs))
for _, msg := range msgs {

153
internal/agents/tool/email/idle.go Normal file
View File

@@ -0,0 +1,153 @@
// email/idle.go – IMAP IDLE Watcher für Echtzeit-Email-Benachrichtigungen
package email
import (
"context"
"crypto/tls"
"fmt"
"log/slog"
"sync/atomic"
"time"
imap "github.com/emersion/go-imap/v2"
"github.com/emersion/go-imap/v2/imapclient"
"my-brain-importer/internal/config"
)
// IdleWatcher überwacht einen IMAP-Account per IDLE auf neue Nachrichten.
type IdleWatcher struct {
acc config.EmailAccount
onNew func(accountName, summary string)
fetching atomic.Bool
}
// NewIdleWatcher erstellt einen IdleWatcher für einen einzelnen Account.
// onNew wird aufgerufen wenn neue Emails gefunden wurden (mit Account-Name und Zusammenfassung).
func NewIdleWatcher(acc config.EmailAccount, onNew func(accountName, summary string)) *IdleWatcher {
return &IdleWatcher{acc: acc, onNew: onNew}
}
// Run startet die IDLE-Schleife. Blockiert bis ctx abgebrochen wird.
func (w *IdleWatcher) Run(ctx context.Context) {
for {
if ctx.Err() != nil {
return
}
slog.Info("IDLE: Verbinde", "account", accountLabel(w.acc), "host", w.acc.Host)
if err := w.runOnce(ctx); err != nil {
slog.Warn("IDLE: Fehler, Neuverbindung in 60s", "account", accountLabel(w.acc), "fehler", err)
select {
case <-ctx.Done():
return
case <-time.After(60 * time.Second):
}
}
}
}
func (w *IdleWatcher) runOnce(ctx context.Context) error {
// numMsgs wird atomar geschrieben/gelesen: UnilateralDataHandler läuft in einem
// separaten Goroutine (imapclient-intern), IDLE-Loop liest im Hauptgoroutine.
var numMsgs atomic.Uint32
hasNew := make(chan struct{}, 1)
addr := fmt.Sprintf("%s:%d", w.acc.Host, w.acc.Port)
options := &imapclient.Options{
UnilateralDataHandler: &imapclient.UnilateralDataHandler{
Mailbox: func(data *imapclient.UnilateralDataMailbox) {
if data.NumMessages != nil && *data.NumMessages > numMsgs.Load() {
numMsgs.Store(*data.NumMessages)
select {
case hasNew <- struct{}{}:
default:
}
}
},
},
}
var (
c *imapclient.Client
err error
)
switch {
case w.acc.TLS:
tlsCfg := &tls.Config{ServerName: w.acc.Host}
options.TLSConfig = tlsCfg
c, err = imapclient.DialTLS(addr, options)
case w.acc.StartTLS:
tlsCfg := &tls.Config{ServerName: w.acc.Host}
options.TLSConfig = tlsCfg
c, err = imapclient.DialStartTLS(addr, options)
default:
c, err = imapclient.DialInsecure(addr, options)
}
if err != nil {
return fmt.Errorf("verbinden: %w", err)
}
defer func() {
c.Logout().Wait()
c.Close()
}()
if err := c.Login(w.acc.User, w.acc.Password).Wait(); err != nil {
return fmt.Errorf("login: %w", err)
}
folder := w.acc.Folder
if folder == "" {
folder = "INBOX"
}
selectData, err := c.Select(folder, &imap.SelectOptions{ReadOnly: true}).Wait()
if err != nil {
return fmt.Errorf("select: %w", err)
}
numMsgs.Store(selectData.NumMessages)
slog.Info("IDLE: Aktiv", "account", accountLabel(w.acc), "folder", folder, "numMsgs", selectData.NumMessages)
for {
if ctx.Err() != nil {
return nil
}
idleCmd, err := c.Idle()
if err != nil {
return fmt.Errorf("IDLE starten: %w", err)
}
select {
case <-ctx.Done():
idleCmd.Close()
idleCmd.Wait()
return nil
case <-hasNew:
idleCmd.Close()
if err := idleCmd.Wait(); err != nil {
slog.Warn("IDLE Wait Fehler", "account", accountLabel(w.acc), "fehler", err)
}
slog.Info("IDLE: Neue Email erkannt", "account", accountLabel(w.acc))
// Nur einen gleichzeitigen Fetch erlauben
if !w.fetching.Swap(true) {
go func() {
defer w.fetching.Store(false)
w.notifyNewEmail()
}()
}
}
}
}
func (w *IdleWatcher) notifyNewEmail() {
summary, err := SummarizeUnreadAccount(w.acc)
if err != nil {
slog.Error("IDLE: Email-Zusammenfassung fehlgeschlagen", "account", accountLabel(w.acc), "fehler", err)
return
}
if summary == "📭 Keine ungelesenen Emails." {
return
}
w.onNew(accountLabel(w.acc), summary)
}

514
internal/agents/tool/email/summary.go
View File

@@ -11,60 +11,510 @@ import (
openai "github.com/sashabaranov/go-openai"
"my-brain-importer/internal/config"
"my-brain-importer/internal/triage"
)
// Summarize verbindet sich mit IMAP, holt die letzten 20 Emails und fasst sie per LLM zusammen.
func Summarize() (string, error) {
return fetchAndSummarize(20, "Fasse diese Emails kurz zusammen und hebe wichtige oder dringende hervor.")
accounts := config.AllEmailAccounts()
if len(accounts) == 0 {
return "", fmt.Errorf("Kein Email-Account konfiguriert")
}
if len(accounts) == 1 {
return fetchAndSummarizeAccount(accounts[0], 20, "Fasse diese Emails kurz zusammen und hebe wichtige oder dringende hervor.")
}
var parts []string
for _, acc := range accounts {
result, err := fetchAndSummarizeAccount(acc, 20, "Fasse diese Emails kurz zusammen und hebe wichtige oder dringende hervor.")
if err != nil {
slog.Error("Email-Fehler", "account", acc.Name, "fehler", err)
parts = append(parts, fmt.Sprintf("❌ **%s:** %v", accountLabel(acc), err))
continue
}
parts = append(parts, fmt.Sprintf("**%s:**\n%s", accountLabel(acc), result))
}
return strings.Join(parts, "\n\n"), nil
}
// SummarizeUnread fasst ungelesene Emails zusammen.
// Wenn email.processed_folder konfiguriert ist, werden die Emails danach dorthin verschoben.
// SummarizeUnread fasst ungelesene Emails für alle konfigurierten Accounts zusammen.
// Wenn email.processed_folder konfiguriert ist, werden Emails danach dorthin verschoben.
func SummarizeUnread() (string, error) {
cl, err := Connect()
accounts := config.AllEmailAccounts()
if len(accounts) == 0 {
return "", fmt.Errorf("Kein Email-Account konfiguriert")
}
if len(accounts) == 1 {
return SummarizeUnreadAccount(accounts[0])
}
var parts []string
allEmpty := true
for _, acc := range accounts {
result, err := SummarizeUnreadAccount(acc)
if err != nil {
slog.Error("Email-Fehler", "account", acc.Name, "fehler", err)
parts = append(parts, fmt.Sprintf("❌ **%s:** %v", accountLabel(acc), err))
continue
}
if result != "📭 Keine ungelesenen Emails." {
allEmpty = false
parts = append(parts, fmt.Sprintf("**%s:**\n%s", accountLabel(acc), result))
}
}
if allEmpty {
return "📭 Keine ungelesenen Emails.", nil
}
return strings.Join(parts, "\n\n"), nil
}
// SummarizeUnreadAccount fasst ungelesene Emails für einen bestimmten Account zusammen.
// Wenn triage_folder konfiguriert ist, werden unwichtige Emails vorher aussortiert.
func SummarizeUnreadAccount(acc config.EmailAccount) (string, error) {
// Phase 1: Triage – Emails sortieren (eigene Verbindung)
if acc.TriageUnimportantFolder != "" || acc.TriageImportantFolder != "" {
if err := triageUnread(acc); err != nil {
slog.Warn("[Triage] fehlgeschlagen, übersprungen", "account", accountLabel(acc), "fehler", err)
}
}
// Phase 2: Zusammenfassung der verbleibenden wichtigen Emails (frische Verbindung)
cl, err := ConnectAccount(acc)
if err != nil {
return "", fmt.Errorf("Email-Verbindung: %w", err)
return "", fmt.Errorf("Email-Verbindung (%s): %w", accountLabel(acc), err)
}
defer cl.Close()
processedFolder := config.Cfg.Email.ProcessedFolder
var msgs []Message
var seqNums []uint32
if processedFolder != "" {
if acc.ProcessedFolder != "" {
msgs, seqNums, err = cl.FetchUnreadSeqNums()
} else {
msgs, err = cl.FetchUnread()
}
if err != nil {
return "", fmt.Errorf("Emails abrufen: %w", err)
return "", fmt.Errorf("Emails abrufen (%s): %w", accountLabel(acc), err)
}
if len(msgs) == 0 {
return "📭 Keine ungelesenen Emails.", nil
}
slog.Info("Email-Zusammenfassung gestartet", "anzahl", len(msgs), "typ", "unread")
result, err := summarizeWithLLM(msgs, "Fasse diese ungelesenen Emails zusammen. Hebe wichtige, dringende oder actionable Emails hervor.")
slog.Info("Email-Zusammenfassung gestartet", "account", accountLabel(acc), "anzahl", len(msgs), "typ", "unread")
result, err := summarizeWithLLMModel(msgs, "Fasse diese ungelesenen Emails zusammen. Hebe wichtige, dringende oder actionable Emails hervor.", accountModel(acc))
if err != nil {
return "", err
}
// Nach dem Zusammenfassen: Emails in Processed-Ordner verschieben
if processedFolder != "" && len(seqNums) > 0 {
if moveErr := cl.MoveMessages(seqNums, processedFolder); moveErr != nil {
slog.Warn("Emails konnten nicht verschoben werden", "fehler", moveErr, "ordner", processedFolder)
if acc.ProcessedFolder != "" && len(seqNums) > 0 {
if moveErr := cl.MoveMessages(seqNums, acc.ProcessedFolder); moveErr != nil {
slog.Warn("Emails konnten nicht verschoben werden", "fehler", moveErr, "ordner", acc.ProcessedFolder)
} else {
slog.Info("Emails verschoben", "anzahl", len(seqNums), "ordner", processedFolder)
slog.Info("Emails verschoben", "anzahl", len(seqNums), "ordner", acc.ProcessedFolder)
}
}
return result, nil
}
// TriageRecentAllAccounts klassifiziert die letzten n Emails aller Accounts manuell
// und verschiebt sie in die konfigurierten Triage-Ordner.
func TriageRecentAllAccounts(n uint32) (string, error) {
accounts := config.AllEmailAccounts()
if len(accounts) == 0 {
return "", fmt.Errorf("kein Email-Account konfiguriert")
}
var lines []string
for _, acc := range accounts {
if acc.TriageImportantFolder == "" && acc.TriageUnimportantFolder == "" {
lines = append(lines, fmt.Sprintf("⚠️ **%s:** kein triage_important_folder / triage_unimportant_folder konfiguriert", accountLabel(acc)))
continue
}
wichtig, unwichtig, err := triageRecentAccount(acc, n)
if err != nil {
lines = append(lines, fmt.Sprintf("❌ **%s:** %v", accountLabel(acc), err))
continue
}
lines = append(lines, fmt.Sprintf("✅ **%s:** %d wichtig → `%s`, %d unwichtig → `%s`",
accountLabel(acc), wichtig, acc.TriageImportantFolder, unwichtig, acc.TriageUnimportantFolder))
}
return strings.Join(lines, "\n"), nil
}
// triageRecentAccount klassifiziert die letzten n Emails eines Accounts.
// Gibt Anzahl wichtiger und unwichtiger Emails zurück.
func triageRecentAccount(acc config.EmailAccount, n uint32) (wichtig, unwichtig int, err error) {
cl, err := ConnectAccount(acc)
if err != nil {
return 0, 0, fmt.Errorf("verbinden: %w", err)
}
defer cl.Close()
// Ordner vorab anlegen, unabhängig davon ob Emails verschoben werden
if acc.TriageImportantFolder != "" {
if err := cl.EnsureFolder(acc.TriageImportantFolder); err != nil {
slog.Warn("[Triage] Ordner anlegen fehlgeschlagen", "ordner", acc.TriageImportantFolder, "fehler", err)
}
}
if acc.TriageUnimportantFolder != "" {
if err := cl.EnsureFolder(acc.TriageUnimportantFolder); err != nil {
slog.Warn("[Triage] Ordner anlegen fehlgeschlagen", "ordner", acc.TriageUnimportantFolder, "fehler", err)
}
}
msgs, err := cl.FetchRecentForSelect(n)
if err != nil {
return 0, 0, fmt.Errorf("fetch: %w", err)
}
if len(msgs) == 0 {
return 0, 0, nil
}
model := accountModel(acc)
var wichtigSeqNums, unwichtigSeqNums []uint32
slog.Info("[Triage] Manuell gestartet", "account", accountLabel(acc), "anzahl", len(msgs))
for _, msg := range msgs {
if ClassifyImportance(msg.Message, model) {
wichtigSeqNums = append(wichtigSeqNums, msg.SeqNum)
} else {
unwichtigSeqNums = append(unwichtigSeqNums, msg.SeqNum)
}
}
if acc.TriageUnimportantFolder != "" && len(unwichtigSeqNums) > 0 {
if ensureErr := cl.EnsureFolder(acc.TriageUnimportantFolder); ensureErr != nil {
slog.Warn("[Triage] Ordner anlegen fehlgeschlagen", "ordner", acc.TriageUnimportantFolder, "fehler", ensureErr)
}
if moveErr := cl.MoveMessages(unwichtigSeqNums, acc.TriageUnimportantFolder); moveErr != nil {
slog.Warn("[Triage] Verschieben unwichtig fehlgeschlagen", "fehler", moveErr)
}
}
if acc.TriageImportantFolder != "" && len(wichtigSeqNums) > 0 {
if ensureErr := cl.EnsureFolder(acc.TriageImportantFolder); ensureErr != nil {
slog.Warn("[Triage] Ordner anlegen fehlgeschlagen", "ordner", acc.TriageImportantFolder, "fehler", ensureErr)
}
if moveErr := cl.MoveMessages(wichtigSeqNums, acc.TriageImportantFolder); moveErr != nil {
slog.Warn("[Triage] Verschieben wichtig fehlgeschlagen", "fehler", moveErr)
}
}
return len(wichtigSeqNums), len(unwichtigSeqNums), nil
}
// triageUnread klassifiziert alle ungelesenen Emails eines Accounts und verschiebt
// wichtige in TriageImportantFolder und unwichtige in TriageUnimportantFolder.
// Läuft sequentiell: eine Email nach der anderen.
func triageUnread(acc config.EmailAccount) error {
cl, err := ConnectAccount(acc)
if err != nil {
return fmt.Errorf("verbinden: %w", err)
}
defer cl.Close()
// Ordner vorab anlegen
if acc.TriageImportantFolder != "" {
cl.EnsureFolder(acc.TriageImportantFolder)
}
if acc.TriageUnimportantFolder != "" {
cl.EnsureFolder(acc.TriageUnimportantFolder)
}
msgs, seqNums, err := cl.FetchUnreadSeqNums()
if err != nil {
return fmt.Errorf("fetch: %w", err)
}
if len(msgs) == 0 {
return nil
}
model := accountModel(acc)
var wichtigSeqNums, unwichtigSeqNums []uint32
slog.Info("[Triage] Klassifizierung gestartet", "account", accountLabel(acc), "anzahl", len(msgs))
for i, msg := range msgs {
if ClassifyImportance(msg, model) {
wichtigSeqNums = append(wichtigSeqNums, seqNums[i])
} else {
unwichtigSeqNums = append(unwichtigSeqNums, seqNums[i])
}
}
if acc.TriageUnimportantFolder != "" && len(unwichtigSeqNums) > 0 {
if err := cl.EnsureFolder(acc.TriageUnimportantFolder); err != nil {
slog.Warn("[Triage] Ordner konnte nicht angelegt werden", "ordner", acc.TriageUnimportantFolder, "fehler", err)
}
if err := cl.MoveMessages(unwichtigSeqNums, acc.TriageUnimportantFolder); err != nil {
slog.Warn("[Triage] Verschieben unwichtig fehlgeschlagen", "ordner", acc.TriageUnimportantFolder, "fehler", err)
} else {
slog.Info("[Triage] Unwichtige Emails verschoben", "anzahl", len(unwichtigSeqNums), "ordner", acc.TriageUnimportantFolder)
}
}
if acc.TriageImportantFolder != "" && len(wichtigSeqNums) > 0 {
if err := cl.EnsureFolder(acc.TriageImportantFolder); err != nil {
slog.Warn("[Triage] Ordner konnte nicht angelegt werden", "ordner", acc.TriageImportantFolder, "fehler", err)
}
if err := cl.MoveMessages(wichtigSeqNums, acc.TriageImportantFolder); err != nil {
slog.Warn("[Triage] Verschieben wichtig fehlgeschlagen", "ordner", acc.TriageImportantFolder, "fehler", err)
} else {
slog.Info("[Triage] Wichtige Emails verschoben", "anzahl", len(wichtigSeqNums), "ordner", acc.TriageImportantFolder)
}
}
return nil
}
// ClassifyImportance klassifiziert eine einzelne Email als wichtig (true) oder unwichtig (false).
// Sucht zuerst ähnliche vergangene Entscheidungen in Qdrant (RAG) und gibt sie als Kontext mit.
// Im Fehlerfall oder bei unklarer Antwort wird true (wichtig) zurückgegeben – sicherer Default.
func ClassifyImportance(msg Message, model string) bool {
// RAG: ähnliche vergangene Triage-Entscheidungen als Few-Shot-Beispiele
ragQuery := fmt.Sprintf("Von: %s Betreff: %s", msg.From, msg.Subject)
examples := triage.SearchSimilar(ragQuery)
var examplesText string
if len(examples) > 0 {
var sb strings.Builder
sb.WriteString("Ähnliche Entscheidungen aus der Vergangenheit:\n")
for _, ex := range examples {
sb.WriteString("- ")
sb.WriteString(ex.Text)
sb.WriteString("\n")
}
sb.WriteString("\n")
examplesText = sb.String()
}
prompt := fmt.Sprintf("%sVon: %s\nBetreff: %s\n\nIst diese Email wichtig? Antworte NUR mit einem einzigen Wort: wichtig oder unwichtig.",
examplesText, msg.From, msg.Subject)
chatClient := config.NewChatClient()
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
resp, err := chatClient.CreateChatCompletion(ctx, openai.ChatCompletionRequest{
Model: model,
Messages: []openai.ChatCompletionMessage{
{Role: openai.ChatMessageRoleSystem, Content: "Du bist ein Email-Filter. Antworte immer nur mit einem einzigen Wort: wichtig oder unwichtig."},
{Role: openai.ChatMessageRoleUser, Content: prompt},
},
Temperature: 0.1,
MaxTokens: 300,
})
if err != nil {
slog.Warn("[Triage] LLM-Fehler, Email als wichtig eingestuft", "betreff", msg.Subject, "fehler", err)
return true
}
if len(resp.Choices) == 0 {
return true
}
raw := resp.Choices[0].Message.Content
// Reasoning-Modelle (z.B. Qwen3) geben Antwort nach </think>-Tag aus
if idx := strings.LastIndex(raw, "</think>"); idx >= 0 {
raw = raw[idx+len("</think>"):]
}
answer := strings.ToLower(strings.TrimSpace(raw))
isImportant := !strings.Contains(answer, "unwichtig")
slog.Info("[Triage] Email klassifiziert",
"betreff", msg.Subject,
"von", msg.From,
"wichtig", isImportant,
"rag_beispiele", len(examples),
"antwort", answer,
)
// Entscheidung für künftiges Lernen in Qdrant speichern
if err := triage.StoreDecision(msg.Subject, msg.From, isImportant); err != nil {
slog.Warn("[Triage] Entscheidung nicht gespeichert", "fehler", err)
}
return isImportant
}
// ExtractReminders sucht in den letzten Emails nach Terminen/Deadlines.
func ExtractReminders() (string, error) {
return fetchAndSummarize(30, "Extrahiere alle Termine, Deadlines, Erinnerungen und wichtigen Daten aus diesen Emails. Liste sie strukturiert auf.")
accounts := config.AllEmailAccounts()
if len(accounts) == 0 {
return "", fmt.Errorf("Kein Email-Account konfiguriert")
}
if len(accounts) == 1 {
return fetchAndSummarizeAccount(accounts[0], 30, "Extrahiere alle Termine, Deadlines, Erinnerungen und wichtigen Daten aus diesen Emails. Liste sie strukturiert auf.")
}
var parts []string
for _, acc := range accounts {
result, err := fetchAndSummarizeAccount(acc, 30, "Extrahiere alle Termine, Deadlines, Erinnerungen und wichtigen Daten aus diesen Emails. Liste sie strukturiert auf.")
if err != nil {
slog.Error("Email-Fehler", "account", acc.Name, "fehler", err)
parts = append(parts, fmt.Sprintf("❌ **%s:** %v", accountLabel(acc), err))
continue
}
parts = append(parts, fmt.Sprintf("**%s:**\n%s", accountLabel(acc), result))
}
return strings.Join(parts, "\n\n"), nil
}
// MoveUnread verschiebt alle ungelesenen Emails eines Accounts in den Zielordner.
// Gibt die Anzahl verschobener Emails zurück.
func MoveUnread(acc config.EmailAccount, destFolder string) (int, error) {
cl, err := ConnectAccount(acc)
if err != nil {
return 0, fmt.Errorf("Email-Verbindung (%s): %w", accountLabel(acc), err)
}
defer cl.Close()
_, seqNums, err := cl.FetchUnreadSeqNums()
if err != nil {
return 0, fmt.Errorf("Emails abrufen: %w", err)
}
if len(seqNums) == 0 {
return 0, nil
}
if err := cl.MoveMessages(seqNums, destFolder); err != nil {
return 0, fmt.Errorf("Verschieben nach %s: %w", destFolder, err)
}
return len(seqNums), nil
}
// AccountSelectMessages enthält ungelesene Emails eines Accounts für die Discord-Auswahl.
type AccountSelectMessages struct {
Account config.EmailAccount
AccIndex int
Messages []SelectMessage
}
// FetchUnreadForSelectAllAccounts holt ungelesene Emails aller Accounts für die Discord-Auswahl.
func FetchUnreadForSelectAllAccounts() ([]AccountSelectMessages, error) {
accounts := config.AllEmailAccounts()
var result []AccountSelectMessages
for i, acc := range accounts {
cl, err := ConnectAccount(acc)
if err != nil {
return nil, fmt.Errorf("Verbindung %s: %w", accountLabel(acc), err)
}
msgs, err := cl.FetchUnreadForSelect()
cl.Close()
if err != nil {
return nil, fmt.Errorf("Emails %s: %w", accountLabel(acc), err)
}
result = append(result, AccountSelectMessages{
Account: acc,
AccIndex: i,
Messages: msgs,
})
}
return result, nil
}
// FetchRecentForSelectAllAccounts holt die letzten n Emails aller Accounts für die Discord-Auswahl.
func FetchRecentForSelectAllAccounts(n uint32) ([]AccountSelectMessages, error) {
accounts := config.AllEmailAccounts()
var result []AccountSelectMessages
for i, acc := range accounts {
cl, err := ConnectAccount(acc)
if err != nil {
return nil, fmt.Errorf("Verbindung %s: %w", accountLabel(acc), err)
}
msgs, err := cl.FetchRecentForSelect(n)
cl.Close()
if err != nil {
return nil, fmt.Errorf("Emails %s: %w", accountLabel(acc), err)
}
result = append(result, AccountSelectMessages{
Account: acc,
AccIndex: i,
Messages: msgs,
})
}
return result, nil
}
// MoveOldEmailsAllAccounts verschiebt alle Emails aller Accounts, die älter als olderThanDays Tage sind, nach destFolder.
// Gibt die Gesamtanzahl verschobener Emails zurück.
func MoveOldEmailsAllAccounts(destFolder string, olderThanDays int) (int, error) {
accounts := config.AllEmailAccounts()
total := 0
for _, acc := range accounts {
cl, err := ConnectAccount(acc)
if err != nil {
return total, fmt.Errorf("Verbindung %s: %w", accountLabel(acc), err)
}
n, err := cl.MoveOldMessages(acc.Folder, destFolder, olderThanDays)
cl.Close()
if err != nil {
return total, fmt.Errorf("Verschieben %s: %w", accountLabel(acc), err)
}
total += n
}
return total, nil
}
// MoveSpecificUnread verschiebt spezifische Emails (per Sequenznummer) eines Accounts in den Zielordner.
func MoveSpecificUnread(accIndex int, seqNums []uint32, destFolder string) (int, error) {
accounts := config.AllEmailAccounts()
if accIndex < 0 || accIndex >= len(accounts) {
return 0, fmt.Errorf("ungültiger Account-Index %d", accIndex)
}
acc := accounts[accIndex]
cl, err := ConnectAccount(acc)
if err != nil {
return 0, fmt.Errorf("Email-Verbindung (%s): %w", accountLabel(acc), err)
}
defer cl.Close()
if err := cl.MoveSpecificMessages(seqNums, destFolder); err != nil {
return 0, err
}
return len(seqNums), nil
}
// CleanupArchiveFolders löscht abgelaufene Emails aus allen konfigurierten Archivordnern.
// Gibt eine menschenlesbare Zusammenfassung zurück.
func CleanupArchiveFolders() (string, error) {
accounts := config.AllEmailAccounts()
var lines []string
var errs []string
total := 0
for _, acc := range accounts {
for _, af := range acc.ArchiveFolders {
if af.RetentionDays <= 0 {
continue
}
cl, err := ConnectAccount(acc)
if err != nil {
errs = append(errs, fmt.Sprintf("%s/%s: %v", accountLabel(acc), af.Name, err))
continue
}
n, err := cl.CleanupOldEmails(af.IMAPFolder, af.RetentionDays)
cl.Close()
if err != nil {
errs = append(errs, fmt.Sprintf("%s/%s: %v", accountLabel(acc), af.Name, err))
continue
}
if n > 0 {
lines = append(lines, fmt.Sprintf("🗑️ %s/%s: %d Email(s) gelöscht (älter als %d Tage)", accountLabel(acc), af.Name, n, af.RetentionDays))
total += n
}
}
}
var result string
if len(lines) == 0 && len(errs) == 0 {
result = "Kein Aufräumen notwendig."
} else {
result = strings.Join(lines, "\n")
}
var combinedErr error
if len(errs) > 0 {
combinedErr = fmt.Errorf("%s", strings.Join(errs, "; "))
}
slog.Info("Archiv-Aufräumen abgeschlossen", "gelöscht", total, "fehler", len(errs))
return result, combinedErr
}
// SummarizeMessages fasst eine übergebene Liste von Nachrichten zusammen (für Tests ohne IMAP).
@@ -72,8 +522,8 @@ func SummarizeMessages(msgs []Message, instruction string) (string, error) {
return summarizeWithLLM(msgs, instruction)
}
func fetchAndSummarize(n uint32, instruction string) (string, error) {
cl, err := Connect()
func fetchAndSummarizeAccount(acc config.EmailAccount, n uint32, instruction string) (string, error) {
cl, err := ConnectAccount(acc)
if err != nil {
return "", fmt.Errorf("Email-Verbindung: %w", err)
}
@@ -87,12 +537,27 @@ func fetchAndSummarize(n uint32, instruction string) (string, error) {
return "📭 Keine Emails gefunden.", nil
}
slog.Info("Email-Zusammenfassung gestartet", "anzahl", len(msgs))
return summarizeWithLLM(msgs, instruction)
slog.Info("Email-Zusammenfassung gestartet", "account", accountLabel(acc), "anzahl", len(msgs))
return summarizeWithLLMModel(msgs, instruction, accountModel(acc))
}
// emailModel gibt das konfigurierte Modell für Email-Zusammenfassungen zurück.
// Fällt auf chat.model zurück wenn email.model nicht gesetzt ist.
// accountLabel gibt einen lesbaren Namen für einen Account zurück.
func accountLabel(acc config.EmailAccount) string {
if acc.Name != "" {
return acc.Name
}
return acc.User
}
// accountModel gibt das konfigurierte LLM-Modell für einen Account zurück.
func accountModel(acc config.EmailAccount) string {
if acc.Model != "" {
return acc.Model
}
return config.Cfg.Chat.Model
}
// emailModel gibt das konfigurierte Modell für den Legacy-Account zurück.
func emailModel() string {
if config.Cfg.Email.Model != "" {
return config.Cfg.Email.Model
@@ -110,8 +575,11 @@ func formatEmailList(msgs []Message) string {
}
func summarizeWithLLM(msgs []Message, instruction string) (string, error) {
return summarizeWithLLMModel(msgs, instruction, emailModel())
}
func summarizeWithLLMModel(msgs []Message, instruction, model string) (string, error) {
emailList := formatEmailList(msgs)
model := emailModel()
chatClient := config.NewChatClient()
ctx := context.Background()

45
internal/agents/tool/email/summary_test.go
View File

@@ -4,6 +4,8 @@ import (
"strings"
"testing"
"time"
"my-brain-importer/internal/config"
)
var testMessages = []Message{
@@ -70,3 +72,46 @@ func TestMessage_DateFormat(t *testing.T) {
t.Errorf("Datumsformat ungültig: %v", err)
}
}
func TestAccountLabel_WithName(t *testing.T) {
acc := config.EmailAccount{Name: "Privat", User: "user@example.de"}
if got := accountLabel(acc); got != "Privat" {
t.Errorf("accountLabel: erwartet %q, got %q", "Privat", got)
}
}
func TestAccountLabel_FallsBackToUser(t *testing.T) {
acc := config.EmailAccount{User: "user@example.de"}
if got := accountLabel(acc); got != "user@example.de" {
t.Errorf("accountLabel: erwartet %q, got %q", "user@example.de", got)
}
}
func TestAccountModel_WithModel(t *testing.T) {
acc := config.EmailAccount{Model: "custom-model"}
if got := accountModel(acc); got != "custom-model" {
t.Errorf("accountModel: erwartet %q, got %q", "custom-model", got)
}
}
func TestAccountModel_FallsBackToChatModel(t *testing.T) {
orig := config.Cfg
defer func() { config.Cfg = orig }()
config.Cfg.Chat.Model = "default-model"
acc := config.EmailAccount{} // kein Model gesetzt
if got := accountModel(acc); got != "default-model" {
t.Errorf("accountModel: erwartet chat.model %q, got %q", "default-model", got)
}
}
func TestSummarizeUnread_NoAccountsConfigured(t *testing.T) {
orig := config.Cfg
defer func() { config.Cfg = orig }()
config.Cfg = config.Config{} // leere Config, kein Email-Account
_, err := SummarizeUnread()
if err == nil {
t.Error("erwartet Fehler wenn kein Account konfiguriert")
}
}

168
internal/agents/tool/rss/watcher.go Normal file
View File

@@ -0,0 +1,168 @@
// rss/watcher.go – Überwacht RSS-Feeds und importiert neue Artikel in Qdrant
package rss
import (
"context"
"fmt"
"log/slog"
"strings"
"time"
"github.com/mmcdole/gofeed"
"my-brain-importer/internal/brain"
"my-brain-importer/internal/config"
)
// FeedResult fasst das Ergebnis eines Feed-Imports zusammen.
type FeedResult struct {
URL string
Title string
Imported int
Err error
}
// IngestFeed fetcht einen RSS-Feed und importiert neue Artikel in Qdrant.
// Gibt Anzahl der importierten Artikel zurück.
func IngestFeed(feedURL string) (int, string, error) {
fp := gofeed.NewParser()
fp.Client = gofeed.NewParser().Client // default HTTP client with timeout
feed, err := fp.ParseURL(feedURL)
if err != nil {
return 0, "", fmt.Errorf("Feed-Parsing fehlgeschlagen: %w", err)
}
feedTitle := feed.Title
if feedTitle == "" {
feedTitle = feedURL
}
imported := 0
for _, item := range feed.Items {
text := buildArticleText(item)
if len(strings.TrimSpace(text)) < 20 {
continue
}
source := fmt.Sprintf("rss/%s", feedURL)
if item.Link != "" {
source = item.Link
}
if err := brain.IngestText(text, source, "rss"); err != nil {
slog.Warn("RSS: Artikel konnte nicht importiert werden", "url", item.Link, "fehler", err)
continue
}
imported++
}
return imported, feedTitle, nil
}
// buildArticleText formatiert einen RSS-Artikel als importierbaren Text.
func buildArticleText(item *gofeed.Item) string {
var sb strings.Builder
if item.Title != "" {
fmt.Fprintf(&sb, "# %s\n\n", item.Title)
}
if item.Published != "" {
fmt.Fprintf(&sb, "Veröffentlicht: %s\n", item.Published)
}
if item.Link != "" {
fmt.Fprintf(&sb, "URL: %s\n\n", item.Link)
}
if item.Description != "" {
sb.WriteString(strings.TrimSpace(item.Description))
}
return sb.String()
}
// IngestAllFeeds importiert alle konfigurierten RSS-Feeds.
// Gibt eine Zusammenfassung der Ergebnisse zurück.
func IngestAllFeeds() []FeedResult {
feeds := config.Cfg.RSSFeeds
if len(feeds) == 0 {
return nil
}
results := make([]FeedResult, 0, len(feeds))
for _, f := range feeds {
n, title, err := IngestFeed(f.URL)
results = append(results, FeedResult{
URL: f.URL,
Title: title,
Imported: n,
Err: err,
})
}
return results
}
// FormatResults gibt eine Discord-formatierte Zusammenfassung zurück.
func FormatResults(results []FeedResult) string {
if len(results) == 0 {
return "📭 Keine RSS-Feeds konfiguriert."
}
var sb strings.Builder
for _, r := range results {
if r.Err != nil {
fmt.Fprintf(&sb, "❌ **%s**: %v\n", r.URL, r.Err)
} else {
name := r.Title
if name == "" {
name = r.URL
}
fmt.Fprintf(&sb, "✅ **%s**: %d Artikel importiert\n", name, r.Imported)
}
}
return strings.TrimSpace(sb.String())
}
// Watcher überwacht alle konfigurierten RSS-Feeds in regelmäßigen Abständen.
type Watcher struct {
OnResults func(summary string)
}
// Run startet die RSS-Überwachungsschleife. Blockiert bis ctx abgebrochen wird.
func (w *Watcher) Run(ctx context.Context) {
feeds := config.Cfg.RSSFeeds
if len(feeds) == 0 {
slog.Info("RSS-Watcher: Keine Feeds konfiguriert, beende")
return
}
// Ersten Durchlauf sofort starten
w.runOnce()
// Dann Timer basierend auf minimalem Intervall
minInterval := 24 * time.Hour
for _, f := range feeds {
h := f.IntervalHours
if h <= 0 {
h = 24
}
d := time.Duration(h) * time.Hour
if d < minInterval {
minInterval = d
}
}
ticker := time.NewTicker(minInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
w.runOnce()
}
}
}
func (w *Watcher) runOnce() {
results := IngestAllFeeds()
if w.OnResults != nil && len(results) > 0 {
summary := FormatResults(results)
if summary != "" {
w.OnResults(summary)
}
}
}

7
internal/brain/ask.go
View File

@@ -41,7 +41,8 @@ func AskQuery(question string, history []agents.HistoryMessage) (string, []Knowl
contextText := buildContext(chunks)
systemPrompt := `Du bist ein hilfreicher persönlicher Assistent.
coreMemory := LoadCoreMemory()
systemPromptBase := `Du bist ein hilfreicher persönlicher Assistent.
Beantworte Fragen primär anhand der bereitgestellten Informationen aus der Wissensdatenbank.
Ergänze fehlende Details mit deinem eigenen Wissen, kennzeichne dies aber klar mit "Aus meinem Wissen:".
@@ -50,6 +51,10 @@ WICHTIGE REGELN:
- Ergänze mit eigenem Wissen wenn sinnvoll, kennzeichne es deutlich
- Antworte auf Deutsch
- Sei präzise und direkt`
systemPrompt := systemPromptBase
if coreMemory != "" {
systemPrompt = systemPromptBase + "\n\n## Fakten über den Nutzer:\n" + coreMemory
}
userPrompt := fmt.Sprintf(`Hier sind die relevanten Informationen aus meiner Wissensdatenbank:

54
internal/brain/core_memory.go Normal file
View File

@@ -0,0 +1,54 @@
// core_memory.go – Persistente Kernfakten über den Nutzer (core_memory.md)
package brain
import (
"fmt"
"os"
"path/filepath"
"strings"
"my-brain-importer/internal/config"
)
// CoreMemoryPath gibt den Pfad zur core_memory.md-Datei zurück.
func CoreMemoryPath() string {
return filepath.Join(config.Cfg.BrainRoot, "core_memory.md")
}
// LoadCoreMemory liest den Inhalt der core_memory.md-Datei.
// Gibt leeren String zurück wenn die Datei nicht existiert.
func LoadCoreMemory() string {
data, err := os.ReadFile(CoreMemoryPath())
if err != nil {
return ""
}
return strings.TrimSpace(string(data))
}
// AppendCoreMemory fügt einen Fakt zur core_memory.md-Datei hinzu.
func AppendCoreMemory(text string) error {
path := CoreMemoryPath()
// Datei erstellen falls nicht vorhanden, sonst anhängen
f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return fmt.Errorf("core_memory.md öffnen: %w", err)
}
defer f.Close()
// Führenden Bindestrich ergänzen wenn nicht vorhanden
line := strings.TrimSpace(text)
if !strings.HasPrefix(line, "-") {
line = "- " + line
}
_, err = fmt.Fprintf(f, "%s\n", line)
return err
}
// ShowCoreMemory gibt den Inhalt der core_memory.md als formatierte Nachricht zurück.
func ShowCoreMemory() string {
content := LoadCoreMemory()
if content == "" {
return "📭 Keine Kernfakten gespeichert. Nutze `/memory profile <text>` um Fakten hinzuzufügen."
}
return fmt.Sprintf("🧠 **Kerngedächtnis:**\n```\n%s\n```", content)
}

27
internal/brain/ingest.go
View File

@@ -256,3 +256,30 @@ func IngestChatMessage(text, author, source string) error {
}
func boolPtr(b bool) *bool { return &b }
// IngestText speichert einen beliebigen Text mit Quelle und Typ in Qdrant.
// Verwendet die gleiche Chunking-Logik wie der Markdown-Ingest.
func IngestText(text, source, docType string) error {
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "api-key", config.Cfg.Qdrant.APIKey)
embClient := config.NewEmbeddingClient()
conn := config.NewQdrantConn()
defer conn.Close()
ensureCollection(ctx, pb.NewCollectionsClient(conn))
pointsClient := pb.NewPointsClient(conn)
var chunks []chunk
for _, part := range splitLongSection(text) {
part = strings.TrimSpace(part)
if len(part) < 20 {
continue
}
chunks = append(chunks, chunk{Text: part, Source: source, Type: docType})
}
if len(chunks) == 0 {
return nil
}
return ingestChunks(ctx, embClient, pointsClient, chunks)
}

98
internal/brain/ingest_email.go Normal file
View File

@@ -0,0 +1,98 @@
// ingest_email.go – Importiert Emails aus einem IMAP-Ordner in Qdrant
package brain
import (
"context"
"fmt"
"log/slog"
"strings"
"time"
pb "github.com/qdrant/go-client/qdrant"
"google.golang.org/grpc/metadata"
"my-brain-importer/internal/agents/tool/email"
"my-brain-importer/internal/config"
)
// IngestEmailFolder importiert alle Emails aus einem IMAP-Ordner in Qdrant.
// Gibt Anzahl der importierten Emails zurück.
// maxEmails = 0 bedeutet: alle (bis max. 500).
func IngestEmailFolder(acc config.EmailAccount, folder string, maxEmails uint32) (int, error) {
if maxEmails == 0 {
maxEmails = 500
}
cl, err := email.ConnectAccount(acc)
if err != nil {
return 0, fmt.Errorf("IMAP-Verbindung: %w", err)
}
defer cl.Close()
slog.Info("Email-Ingest: Lade Emails", "account", acc.Name, "folder", folder, "max", maxEmails)
msgs, err := cl.FetchWithBody(folder, maxEmails)
if err != nil {
return 0, fmt.Errorf("Emails laden: %w", err)
}
if len(msgs) == 0 {
return 0, nil
}
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "api-key", config.Cfg.Qdrant.APIKey)
embClient := config.NewEmbeddingClient()
conn := config.NewQdrantConn()
defer conn.Close()
ensureCollection(ctx, pb.NewCollectionsClient(conn))
pointsClient := pb.NewPointsClient(conn)
var chunks []chunk
for _, m := range msgs {
text := formatEmailForIngest(m)
if len(strings.TrimSpace(text)) < 20 {
continue
}
source := fmt.Sprintf("email/%s/%s", folder, m.Date)
chunks = append(chunks, chunk{Text: text, Source: source, Type: "email"})
}
if len(chunks) == 0 {
return 0, nil
}
slog.Info("Email-Ingest: Starte Embedding", "emails", len(msgs), "chunks", len(chunks))
// In Batches von 20 ingesten (Embeddings können langsam sein)
ingested := 0
for i := 0; i < len(chunks); i += 20 {
end := i + 20
if end > len(chunks) {
end = len(chunks)
}
batch := chunks[i:end]
if err := ingestChunks(ctx, embClient, pointsClient, batch); err != nil {
slog.Warn("Email-Ingest Batch-Fehler", "batch_start", i, "fehler", err)
continue
}
ingested += len(batch)
slog.Info("Email-Ingest Fortschritt", "ingested", ingested, "total", len(chunks))
time.Sleep(50 * time.Millisecond)
}
return ingested, nil
}
// formatEmailForIngest formatiert eine Email als durchsuchbaren Text.
func formatEmailForIngest(m email.MessageWithBody) string {
var sb strings.Builder
fmt.Fprintf(&sb, "Betreff: %s\n", m.Subject)
fmt.Fprintf(&sb, "Von: %s\n", m.From)
fmt.Fprintf(&sb, "Datum: %s\n", m.Date)
if m.Body != "" {
sb.WriteString("\n")
sb.WriteString(m.Body)
}
return sb.String()
}

82
internal/brain/ingest_pdf.go Normal file
View File

@@ -0,0 +1,82 @@
// ingest_pdf.go – Extrahiert Text aus einer PDF-Datei und importiert ihn in Qdrant
package brain
import (
"context"
"fmt"
"strings"
"github.com/ledongthuc/pdf"
pb "github.com/qdrant/go-client/qdrant"
"google.golang.org/grpc/metadata"
"my-brain-importer/internal/config"
)
// IngestPDF extrahiert Text aus einer PDF-Datei und importiert ihn in Qdrant.
// source ist der Anzeigename der Quelle (z.B. Dateiname).
// Gibt Anzahl der importierten Chunks zurück.
func IngestPDF(filePath, source string) (int, error) {
text, err := extractPDFText(filePath)
if err != nil {
return 0, fmt.Errorf("PDF-Parsing fehlgeschlagen: %w", err)
}
text = strings.TrimSpace(text)
if len(text) < 20 {
return 0, fmt.Errorf("kein verwertbarer Text in PDF gefunden")
}
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "api-key", config.Cfg.Qdrant.APIKey)
embClient := config.NewEmbeddingClient()
conn := config.NewQdrantConn()
defer conn.Close()
ensureCollection(ctx, pb.NewCollectionsClient(conn))
pointsClient := pb.NewPointsClient(conn)
var chunks []chunk
for _, part := range splitLongSection(text) {
part = strings.TrimSpace(part)
if len(part) < 20 {
continue
}
chunks = append(chunks, chunk{Text: part, Source: source, Type: "pdf"})
}
if len(chunks) == 0 {
return 0, fmt.Errorf("kein verwertbarer Inhalt nach Aufteilung")
}
if err := ingestChunks(ctx, embClient, pointsClient, chunks); err != nil {
return 0, fmt.Errorf("Ingest fehlgeschlagen: %w", err)
}
return len(chunks), nil
}
// extractPDFText liest alle Seiten einer PDF-Datei und gibt den Text zurück.
func extractPDFText(filePath string) (string, error) {
f, r, err := pdf.Open(filePath)
if err != nil {
return "", err
}
defer f.Close()
var sb strings.Builder
totalPages := r.NumPage()
for pageNum := 1; pageNum <= totalPages; pageNum++ {
page := r.Page(pageNum)
if page.V.IsNull() {
continue
}
text, err := page.GetPlainText(nil)
if err != nil {
continue
}
sb.WriteString(text)
sb.WriteString("\n")
}
return sb.String(), nil
}

124
internal/brain/ingest_url.go Normal file
View File

@@ -0,0 +1,124 @@
// ingest_url.go – Fetcht eine URL und importiert den Textinhalt in Qdrant
package brain
import (
"context"
"fmt"
"io"
"net/http"
"strings"
"time"
pb "github.com/qdrant/go-client/qdrant"
"golang.org/x/net/html"
"google.golang.org/grpc/metadata"
"my-brain-importer/internal/config"
)
// IngestURL fetcht eine URL, extrahiert den Textinhalt und importiert ihn in Qdrant.
// Gibt Anzahl der importierten Chunks zurück.
func IngestURL(rawURL string) (int, error) {
client := &http.Client{Timeout: 30 * time.Second}
resp, err := client.Get(rawURL)
if err != nil {
return 0, fmt.Errorf("HTTP-Fehler: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return 0, fmt.Errorf("HTTP %d: %s", resp.StatusCode, resp.Status)
}
contentType := resp.Header.Get("Content-Type")
var text string
if strings.Contains(contentType, "text/html") {
text, err = extractHTMLText(resp.Body)
if err != nil {
return 0, fmt.Errorf("HTML-Parsing fehlgeschlagen: %w", err)
}
} else {
raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) // max 1MB
if err != nil {
return 0, fmt.Errorf("Lesen fehlgeschlagen: %w", err)
}
text = string(raw)
}
text = strings.TrimSpace(text)
if len(text) < 20 {
return 0, fmt.Errorf("kein verwertbarer Inhalt gefunden")
}
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "api-key", config.Cfg.Qdrant.APIKey)
embClient := config.NewEmbeddingClient()
conn := config.NewQdrantConn()
defer conn.Close()
ensureCollection(ctx, pb.NewCollectionsClient(conn))
pointsClient := pb.NewPointsClient(conn)
var chunks []chunk
for _, part := range splitLongSection(text) {
part = strings.TrimSpace(part)
if len(part) < 20 {
continue
}
chunks = append(chunks, chunk{Text: part, Source: rawURL, Type: "url"})
}
if len(chunks) == 0 {
return 0, fmt.Errorf("kein verwertbarer Inhalt nach Aufteilung")
}
if err := ingestChunks(ctx, embClient, pointsClient, chunks); err != nil {
return 0, fmt.Errorf("Ingest fehlgeschlagen: %w", err)
}
return len(chunks), nil
}
// extractHTMLText extrahiert sichtbaren Text aus einem HTML-Dokument.
func extractHTMLText(r io.Reader) (string, error) {
doc, err := html.Parse(r)
if err != nil {
return "", err
}
var sb strings.Builder
extractTextNode(doc, &sb)
// Mehrfach-Leerzeilen reduzieren
lines := strings.Split(sb.String(), "\n")
var cleaned []string
for _, line := range lines {
line = strings.TrimSpace(line)
if line != "" {
cleaned = append(cleaned, line)
}
}
return strings.Join(cleaned, "\n"), nil
}
// skipTags sind HTML-Elemente deren Inhalt nicht extrahiert wird.
var skipTags = map[string]bool{
"script": true, "style": true, "noscript": true,
"head": true, "meta": true, "link": true,
"nav": true, "footer": true, "header": true,
}
func extractTextNode(n *html.Node, sb *strings.Builder) {
if n.Type == html.TextNode {
text := strings.TrimSpace(n.Data)
if text != "" {
sb.WriteString(text)
sb.WriteString("\n")
}
return
}
if n.Type == html.ElementNode && skipTags[n.Data] {
return
}
for c := n.FirstChild; c != nil; c = c.NextSibling {
extractTextNode(c, sb)
}
}

108
internal/brain/knowledge.go Normal file
View File

@@ -0,0 +1,108 @@
// knowledge.go – Listet und löscht Einträge in der Qdrant-Wissensdatenbank
package brain
import (
"context"
"fmt"
"sort"
pb "github.com/qdrant/go-client/qdrant"
"google.golang.org/grpc/metadata"
"my-brain-importer/internal/config"
)
// ListSources gibt alle eindeutigen Quellen in der Wissensdatenbank zurück.
// Limit begrenzt die Anzahl der zu scrollenden Punkte (0 = Standard 1000).
func ListSources(limit uint32) ([]string, error) {
if limit == 0 {
limit = 1000
}
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "api-key", config.Cfg.Qdrant.APIKey)
conn := config.NewQdrantConn()
defer conn.Close()
pointsClient := pb.NewPointsClient(conn)
seen := map[string]bool{}
var offset *pb.PointId
for {
req := &pb.ScrollPoints{
CollectionName: config.Cfg.Qdrant.Collection,
WithPayload: &pb.WithPayloadSelector{
SelectorOptions: &pb.WithPayloadSelector_Include{
Include: &pb.PayloadIncludeSelector{Fields: []string{"source"}},
},
},
Limit: uint32Ptr(250),
}
if offset != nil {
req.Offset = offset
}
result, err := pointsClient.Scroll(ctx, req)
if err != nil {
return nil, fmt.Errorf("Scroll fehlgeschlagen: %w", err)
}
for _, pt := range result.Result {
if src := pt.Payload["source"].GetStringValue(); src != "" {
seen[src] = true
}
}
if result.NextPageOffset == nil || uint32(len(seen)) >= limit {
break
}
offset = result.NextPageOffset
}
sources := make([]string, 0, len(seen))
for s := range seen {
sources = append(sources, s)
}
sort.Strings(sources)
return sources, nil
}
// DeleteBySource löscht alle Punkte mit dem gegebenen Quellennamen aus Qdrant.
// Gibt Anzahl gelöschter Punkte zurück (Qdrant liefert keine genaue Zahl — gibt 0 zurück wenn erfolgreich).
func DeleteBySource(source string) error {
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "api-key", config.Cfg.Qdrant.APIKey)
conn := config.NewQdrantConn()
defer conn.Close()
pointsClient := pb.NewPointsClient(conn)
_, err := pointsClient.Delete(ctx, &pb.DeletePoints{
CollectionName: config.Cfg.Qdrant.Collection,
Points: &pb.PointsSelector{
PointsSelectorOneOf: &pb.PointsSelector_Filter{
Filter: &pb.Filter{
Must: []*pb.Condition{
{
ConditionOneOf: &pb.Condition_Field{
Field: &pb.FieldCondition{
Key: "source",
Match: &pb.Match{
MatchValue: &pb.Match_Keyword{Keyword: source},
},
},
},
},
},
},
},
},
Wait: boolPtr(true),
})
return err
}
func uint32Ptr(v uint32) *uint32 { return &v }

91
internal/config/config.go
View File

@@ -12,6 +12,30 @@ import (
"gopkg.in/yaml.v3"
)
// ArchiveFolder beschreibt einen IMAP-Archivordner mit optionaler Aufbewahrungsdauer.
type ArchiveFolder struct {
Name string `yaml:"name"` // Anzeigename (z.B. "5Jahre")
IMAPFolder string `yaml:"imap_folder"` // Echter IMAP-Ordnername (z.B. "5Jahre")
RetentionDays int `yaml:"retention_days"` // 0 = dauerhaft behalten
}
// EmailAccount beschreibt einen einzelnen IMAP-Account.
type EmailAccount struct {
Name string `yaml:"name"`
Host string `yaml:"host"`
Port int `yaml:"port"`
User string `yaml:"user"`
Password string `yaml:"password"`
TLS bool `yaml:"tls"`
StartTLS bool `yaml:"starttls"`
Folder string `yaml:"folder"`
ProcessedFolder string `yaml:"processed_folder"`
Model string `yaml:"model"`
ArchiveFolders []ArchiveFolder `yaml:"archive_folders"`
TriageImportantFolder string `yaml:"triage_important_folder"` // Ordner für wichtige Emails (leer = in INBOX lassen)
TriageUnimportantFolder string `yaml:"triage_unimportant_folder"` // Ordner für unwichtige Emails (leer = kein Triage)
}
type Config struct {
Qdrant struct {
Host string `yaml:"host"`
@@ -32,22 +56,30 @@ type Config struct {
} `yaml:"chat"`
Discord struct {
Token string `yaml:"token"`
GuildID string `yaml:"guild_id"`
Token string `yaml:"token"`
GuildID string `yaml:"guild_id"`
AllowedUsers []string `yaml:"allowed_users"` // Wenn gesetzt, dürfen nur diese User-IDs den Bot nutzen
} `yaml:"discord"`
// Email ist der Legacy-Block für einen einzelnen Account.
Email struct {
Host string `yaml:"host"`
Port int `yaml:"port"`
User string `yaml:"user"`
Password string `yaml:"password"`
TLS bool `yaml:"tls"`
StartTLS bool `yaml:"starttls"`
Folder string `yaml:"folder"`
ProcessedFolder string `yaml:"processed_folder"` // Zielordner nach Zusammenfassung (leer = kein Verschieben)
Model string `yaml:"model"` // Optional: überschreibt chat.model für Email-Zusammenfassungen
Host string `yaml:"host"`
Port int `yaml:"port"`
User string `yaml:"user"`
Password string `yaml:"password"`
TLS bool `yaml:"tls"`
StartTLS bool `yaml:"starttls"`
Folder string `yaml:"folder"`
ProcessedFolder string `yaml:"processed_folder"`
Model string `yaml:"model"`
ArchiveFolders []ArchiveFolder `yaml:"archive_folders"`
TriageImportantFolder string `yaml:"triage_important_folder"`
TriageUnimportantFolder string `yaml:"triage_unimportant_folder"`
} `yaml:"email"`
// EmailAccounts ermöglicht mehrere IMAP-Accounts. Hat Vorrang vor email:.
EmailAccounts []EmailAccount `yaml:"email_accounts"`
Tasks struct {
StorePath string `yaml:"store_path"`
} `yaml:"tasks"`
@@ -56,15 +88,52 @@ type Config struct {
ChannelID string `yaml:"channel_id"`
EmailIntervalMin int `yaml:"email_interval_min"`
TaskReminderHour int `yaml:"task_reminder_hour"`
CleanupHour int `yaml:"cleanup_hour"` // Uhrzeit für tägliches Archiv-Aufräumen (Standard: 2)
IngestHour int `yaml:"ingest_hour"` // Uhrzeit für nächtlichen Email-Ingest (Standard: 23, 0 = deaktiviert)
} `yaml:"daemon"`
BrainRoot string `yaml:"brain_root"`
TopK uint64 `yaml:"top_k"`
ScoreThreshold float32 `yaml:"score_threshold"`
// RSSFeeds definiert RSS-Feeds die automatisch überwacht werden.
RSSFeeds []RSSFeed `yaml:"rss_feeds"`
}
// RSSFeed beschreibt einen RSS-Feed mit Polling-Intervall.
type RSSFeed struct {
URL string `yaml:"url"`
IntervalHours int `yaml:"interval_hours"` // 0 = Standard 24h
}
var Cfg Config
// AllEmailAccounts gibt alle konfigurierten Email-Accounts zurück.
// Wenn email_accounts konfiguriert ist, hat das Vorrang vor dem Legacy-email:-Block.
func AllEmailAccounts() []EmailAccount {
if len(Cfg.EmailAccounts) > 0 {
return Cfg.EmailAccounts
}
if Cfg.Email.Host == "" {
return nil
}
return []EmailAccount{{
Name: "Email",
Host: Cfg.Email.Host,
Port: Cfg.Email.Port,
User: Cfg.Email.User,
Password: Cfg.Email.Password,
TLS: Cfg.Email.TLS,
StartTLS: Cfg.Email.StartTLS,
Folder: Cfg.Email.Folder,
ProcessedFolder: Cfg.Email.ProcessedFolder,
Model: Cfg.Email.Model,
ArchiveFolders: Cfg.Email.ArchiveFolders,
TriageImportantFolder: Cfg.Email.TriageImportantFolder,
TriageUnimportantFolder: Cfg.Email.TriageUnimportantFolder,
}}
}
// NewQdrantConn öffnet eine gRPC-Verbindung zur Qdrant-Instanz.
// Der Aufrufer ist verantwortlich für conn.Close().
func NewQdrantConn() *grpc.ClientConn {

87
internal/config/config_test.go Normal file
View File

@@ -0,0 +1,87 @@
package config
import "testing"
func TestAllEmailAccounts_Empty(t *testing.T) {
orig := Cfg
defer func() { Cfg = orig }()
Cfg = Config{}
accounts := AllEmailAccounts()
if len(accounts) != 0 {
t.Errorf("erwartet 0 Accounts, got %d", len(accounts))
}
}
func TestAllEmailAccounts_LegacyFallback(t *testing.T) {
orig := Cfg
defer func() { Cfg = orig }()
Cfg = Config{}
Cfg.Email.Host = "imap.example.de"
Cfg.Email.Port = 143
Cfg.Email.User = "user@example.de"
Cfg.Email.Password = "geheim"
Cfg.Email.Folder = "INBOX"
Cfg.Email.ProcessedFolder = "Processed"
Cfg.Email.Model = "testmodel"
accounts := AllEmailAccounts()
if len(accounts) != 1 {
t.Fatalf("erwartet 1 Account, got %d", len(accounts))
}
a := accounts[0]
if a.Host != "imap.example.de" {
t.Errorf("Host: got %q", a.Host)
}
if a.Port != 143 {
t.Errorf("Port: got %d", a.Port)
}
if a.User != "user@example.de" {
t.Errorf("User: got %q", a.User)
}
if a.ProcessedFolder != "Processed" {
t.Errorf("ProcessedFolder: got %q", a.ProcessedFolder)
}
if a.Model != "testmodel" {
t.Errorf("Model: got %q", a.Model)
}
}
func TestAllEmailAccounts_MultipleAccounts(t *testing.T) {
orig := Cfg
defer func() { Cfg = orig }()
Cfg = Config{}
Cfg.EmailAccounts = []EmailAccount{
{Name: "Privat", Host: "imap1.de", Port: 143},
{Name: "Arbeit", Host: "imap2.de", Port: 993, TLS: true},
}
accounts := AllEmailAccounts()
if len(accounts) != 2 {
t.Fatalf("erwartet 2 Accounts, got %d", len(accounts))
}
if accounts[0].Host != "imap1.de" {
t.Errorf("Account 0 Host: got %q", accounts[0].Host)
}
if accounts[1].Host != "imap2.de" {
t.Errorf("Account 1 Host: got %q", accounts[1].Host)
}
}
func TestAllEmailAccounts_NewTakesPrecedence(t *testing.T) {
orig := Cfg
defer func() { Cfg = orig }()
Cfg = Config{}
Cfg.Email.Host = "legacy.de"
Cfg.EmailAccounts = []EmailAccount{
{Name: "Neu", Host: "new.de"},
}
accounts := AllEmailAccounts()
if len(accounts) != 1 {
t.Fatalf("erwartet 1 Account, got %d", len(accounts))
}
if accounts[0].Host != "new.de" {
t.Errorf("email_accounts sollte Vorrang haben, got host %q", accounts[0].Host)
}
}

15
internal/diag/diag.go
View File

@@ -49,11 +49,18 @@ func RunAll() (results []Result, allOK bool) {
check("LocalAI Embedding ("+cfg.Embedding.URL+")", ok, msg)
}
// IMAP
if cfg.Email.Host != "" {
imapAddr := fmt.Sprintf("%s:%d", cfg.Email.Host, cfg.Email.Port)
// IMAP – alle konfigurierten Accounts prüfen
for _, acc := range config.AllEmailAccounts() {
if acc.Host == "" {
continue
}
imapAddr := fmt.Sprintf("%s:%d", acc.Host, acc.Port)
label := acc.Name
if label == "" {
label = acc.User
}
ok, msg = tcpCheck(imapAddr)
check("IMAP ("+imapAddr+")", ok, msg)
check("IMAP "+label+" ("+imapAddr+")", ok, msg)
}
return results, allOK

138
internal/triage/triage.go Normal file
View File

@@ -0,0 +1,138 @@
// triage/triage.go – Speichert und sucht Email-Triage-Entscheidungen in Qdrant (RAG-Lernen)
// Eigenes Package um Import-Zyklen zwischen brain und email zu vermeiden.
package triage
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"log/slog"
pb "github.com/qdrant/go-client/qdrant"
openai "github.com/sashabaranov/go-openai"
"google.golang.org/grpc/metadata"
"my-brain-importer/internal/config"
)
// TriageResult repräsentiert ein Suchergebnis aus vergangenen Triage-Entscheidungen.
type TriageResult struct {
Text string
Score float32
}
// StoreDecision speichert eine Triage-Entscheidung in Qdrant.
// Bei gleicher Email (deterministischer ID) wird die Entscheidung überschrieben.
func StoreDecision(subject, from string, isImportant bool) error {
label := "wichtig"
if !isImportant {
label = "unwichtig"
}
text := fmt.Sprintf("Email-Triage | Von: %s | Betreff: %s | Entscheidung: %s", from, subject, label)
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "api-key", config.Cfg.Qdrant.APIKey)
embClient := config.NewEmbeddingClient()
embResp, err := embClient.CreateEmbeddings(ctx, openai.EmbeddingRequest{
Input: []string{text},
Model: openai.EmbeddingModel(config.Cfg.Embedding.Model),
})
if err != nil {
return fmt.Errorf("embedding: %w", err)
}
conn := config.NewQdrantConn()
defer conn.Close()
id := triageID(text)
wait := true
_, err = pb.NewPointsClient(conn).Upsert(ctx, &pb.UpsertPoints{
CollectionName: config.Cfg.Qdrant.Collection,
Points: []*pb.PointStruct{{
Id: &pb.PointId{
PointIdOptions: &pb.PointId_Uuid{Uuid: id},
},
Vectors: &pb.Vectors{
VectorsOptions: &pb.Vectors_Vector{
Vector: &pb.Vector{Data: embResp.Data[0].Embedding},
},
},
Payload: map[string]*pb.Value{
"text": {Kind: &pb.Value_StringValue{StringValue: text}},
"source": {Kind: &pb.Value_StringValue{StringValue: "email_triage"}},
"type": {Kind: &pb.Value_StringValue{StringValue: "email_triage"}},
},
}},
Wait: &wait,
})
if err != nil {
return fmt.Errorf("qdrant upsert: %w", err)
}
slog.Debug("[Triage] Entscheidung gespeichert", "betreff", subject, "wichtig", isImportant)
return nil
}
// SearchSimilar sucht ähnliche vergangene Triage-Entscheidungen in Qdrant.
// Gibt bis zu 3 Ergebnisse zurück (nur type=email_triage, Score ≥ 0.7).
func SearchSimilar(query string) []TriageResult {
ctx := context.Background()
ctx = metadata.AppendToOutgoingContext(ctx, "api-key", config.Cfg.Qdrant.APIKey)
embClient := config.NewEmbeddingClient()
embResp, err := embClient.CreateEmbeddings(ctx, openai.EmbeddingRequest{
Input: []string{query},
Model: openai.EmbeddingModel(config.Cfg.Embedding.Model),
})
if err != nil {
slog.Warn("[Triage] Embedding für RAG fehlgeschlagen", "fehler", err)
return nil
}
conn := config.NewQdrantConn()
defer conn.Close()
threshold := float32(0.7)
result, err := pb.NewPointsClient(conn).Search(ctx, &pb.SearchPoints{
CollectionName: config.Cfg.Qdrant.Collection,
Vector: embResp.Data[0].Embedding,
Limit: 3,
WithPayload: &pb.WithPayloadSelector{
SelectorOptions: &pb.WithPayloadSelector_Enable{Enable: true},
},
ScoreThreshold: &threshold,
Filter: &pb.Filter{
Must: []*pb.Condition{{
ConditionOneOf: &pb.Condition_Field{
Field: &pb.FieldCondition{
Key: "type",
Match: &pb.Match{
MatchValue: &pb.Match_Keyword{Keyword: "email_triage"},
},
},
},
}},
},
})
if err != nil {
slog.Warn("[Triage] RAG-Suche fehlgeschlagen", "fehler", err)
return nil
}
var results []TriageResult
for _, hit := range result.Result {
text := hit.Payload["text"].GetStringValue()
if text == "" {
continue
}
results = append(results, TriageResult{Text: text, Score: hit.Score})
}
return results
}
func triageID(text string) string {
hash := sha256.Sum256([]byte("email_triage:" + text))
return hex.EncodeToString(hash[:16])
}