385 lines
12 KiB
Go
385 lines
12 KiB
Go
package ingest
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
htmlstd "html"
|
|
"os"
|
|
"path/filepath"
|
|
"regexp"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/local/glpi-neural-brain/internal/activity"
|
|
"github.com/local/glpi-neural-brain/internal/glpi"
|
|
"github.com/local/glpi-neural-brain/internal/graph"
|
|
"github.com/local/glpi-neural-brain/internal/model"
|
|
"github.com/local/glpi-neural-brain/internal/persist"
|
|
)
|
|
|
|
type GLPIKBConfig struct {
|
|
Enabled bool
|
|
Path string
|
|
Filter string
|
|
Limit int
|
|
SyncInterval time.Duration
|
|
Source string
|
|
CachePath string
|
|
ShouldSync func() bool
|
|
}
|
|
|
|
type GLPIKBStatus struct {
|
|
Enabled bool `json:"enabled"`
|
|
Path string `json:"path,omitempty"`
|
|
LastSync time.Time `json:"last_sync,omitempty"`
|
|
LastError string `json:"last_error,omitempty"`
|
|
Documents int `json:"documents"`
|
|
CachePath string `json:"cache_path,omitempty"`
|
|
NextSync time.Time `json:"next_sync,omitempty"`
|
|
}
|
|
|
|
type GLPIKBSource interface {
|
|
DiscoverKnowledgeBasePath(context.Context, string) (string, error)
|
|
ListKnowledgeBaseItems(context.Context, string, int, string) ([]glpi.KnowledgeItem, error)
|
|
GetCategories(context.Context) ([]glpi.ITILCategory, error)
|
|
}
|
|
|
|
type GLPIKBSyncer struct {
|
|
cfg GLPIKBConfig
|
|
client GLPIKBSource
|
|
graph *graph.Store
|
|
broker *activity.Broker
|
|
persistence *persist.Coordinator
|
|
|
|
syncMu sync.Mutex
|
|
mu sync.RWMutex
|
|
path string
|
|
lastSync time.Time
|
|
lastError string
|
|
documents int
|
|
nextSync time.Time
|
|
lastFingerprint string
|
|
contents map[string]string
|
|
}
|
|
|
|
type glpiKBCache struct {
|
|
SyncedAt time.Time `json:"synced_at"`
|
|
Path string `json:"path"`
|
|
Nodes []model.Node `json:"nodes"`
|
|
Edges []model.Edge `json:"edges"`
|
|
Contents map[string]string `json:"contents,omitempty"`
|
|
}
|
|
|
|
func NewGLPIKBSyncer(cfg GLPIKBConfig, client GLPIKBSource, g *graph.Store, b *activity.Broker, p *persist.Coordinator) *GLPIKBSyncer {
|
|
return &GLPIKBSyncer{cfg: cfg, client: client, graph: g, broker: b, persistence: p, contents: map[string]string{}}
|
|
}
|
|
|
|
func (s *GLPIKBSyncer) Start(ctx context.Context) {
|
|
if !s.cfg.Enabled {
|
|
return
|
|
}
|
|
_ = s.LoadCache()
|
|
go func() {
|
|
if s.syncAllowed() {
|
|
s.runSync(ctx, "startup")
|
|
}
|
|
ticker := time.NewTicker(s.cfg.SyncInterval)
|
|
defer ticker.Stop()
|
|
for {
|
|
s.setNextSync(time.Now().Add(s.cfg.SyncInterval))
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
if s.syncAllowed() {
|
|
s.runSync(ctx, "interval")
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (s *GLPIKBSyncer) syncAllowed() bool {
|
|
return s.cfg.ShouldSync == nil || s.cfg.ShouldSync()
|
|
}
|
|
|
|
func (s *GLPIKBSyncer) runSync(ctx context.Context, trigger string) {
|
|
timeout := s.cfg.SyncInterval / 2
|
|
if timeout < 30*time.Second {
|
|
timeout = 30 * time.Second
|
|
}
|
|
if timeout > 5*time.Minute {
|
|
timeout = 5 * time.Minute
|
|
}
|
|
syncCtx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
if err := s.Sync(syncCtx, trigger); err != nil {
|
|
s.mu.Lock()
|
|
s.lastError = err.Error()
|
|
s.mu.Unlock()
|
|
if s.broker != nil {
|
|
s.broker.Publish(model.Activity{Type: "glpi.kb.failed", Source: "glpi", Phase: "knowledge-sync", Message: "GLPI-Wissensdatenbank konnte nicht synchronisiert werden", Strength: .35, Metadata: map[string]any{"error": err.Error(), "trigger": trigger}})
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *GLPIKBSyncer) LoadCache() error {
|
|
if strings.TrimSpace(s.cfg.CachePath) == "" {
|
|
return nil
|
|
}
|
|
data, err := os.ReadFile(s.cfg.CachePath)
|
|
if os.IsNotExist(err) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var cache glpiKBCache
|
|
if err := json.Unmarshal(data, &cache); err != nil {
|
|
return err
|
|
}
|
|
s.graph.ReplaceOrigins([]string{"glpi-kb", "glpi-kb-taxonomy"}, cache.Nodes, cache.Edges)
|
|
fingerprint, _ := glpiKnowledgeFingerprint(cache.Nodes, cache.Edges, cache.Contents)
|
|
s.mu.Lock()
|
|
s.path = cache.Path
|
|
s.lastSync = cache.SyncedAt
|
|
s.documents = countKnowledge(cache.Nodes)
|
|
s.lastError = ""
|
|
s.lastFingerprint = fingerprint
|
|
s.contents = cloneStringMap(cache.Contents)
|
|
s.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
func (s *GLPIKBSyncer) Sync(ctx context.Context, trigger string) error {
|
|
s.syncMu.Lock()
|
|
defer s.syncMu.Unlock()
|
|
path := s.currentPath()
|
|
if path == "" {
|
|
var err error
|
|
path, err = s.client.DiscoverKnowledgeBasePath(ctx, s.cfg.Path)
|
|
if err != nil {
|
|
return fmt.Errorf("discover GLPI KB path: %w", err)
|
|
}
|
|
}
|
|
items, err := s.client.ListKnowledgeBaseItems(ctx, path, s.cfg.Limit, s.cfg.Filter)
|
|
if err != nil {
|
|
return fmt.Errorf("list GLPI KB items: %w", err)
|
|
}
|
|
categories, err := s.client.GetCategories(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("load GLPI categories: %w", err)
|
|
}
|
|
nodes, edges, contents := s.buildGraph(items, categories)
|
|
fingerprint, err := glpiKnowledgeFingerprint(nodes, edges, contents)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
now := time.Now().UTC()
|
|
s.mu.RLock()
|
|
unchanged := fingerprint != "" && fingerprint == s.lastFingerprint
|
|
s.mu.RUnlock()
|
|
if unchanged {
|
|
s.mu.Lock()
|
|
s.path = path
|
|
s.lastSync = now
|
|
s.lastError = ""
|
|
s.documents = len(items)
|
|
s.mu.Unlock()
|
|
if s.broker != nil {
|
|
s.broker.Publish(model.Activity{Type: "glpi.kb.unchanged", Source: "glpi", Phase: "knowledge-sync", Message: "GLPI-Wissensdatenbank ist unverändert", Strength: .18, Metadata: map[string]any{"documents": len(items), "path": path, "trigger": trigger}})
|
|
}
|
|
return nil
|
|
}
|
|
s.graph.ReplaceOrigins([]string{"glpi-kb", "glpi-kb-taxonomy"}, nodes, edges)
|
|
cache := glpiKBCache{SyncedAt: now, Path: path, Nodes: nodes, Edges: edges, Contents: contents}
|
|
if s.persistence != nil && s.cfg.CachePath != "" {
|
|
data, err := json.Marshal(cache)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := s.persistence.QueueFile(s.cfg.CachePath, data, 0o640); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
s.mu.Lock()
|
|
s.path = path
|
|
s.lastSync = now
|
|
s.lastError = ""
|
|
s.documents = len(items)
|
|
s.lastFingerprint = fingerprint
|
|
s.contents = cloneStringMap(contents)
|
|
s.mu.Unlock()
|
|
if s.broker != nil {
|
|
s.broker.Publish(model.Activity{Type: "glpi.kb.synced", Source: "glpi", Phase: "knowledge-sync", Message: fmt.Sprintf("%d GLPI-KB-Beiträge wurden in den Wissensgraphen übernommen", len(items)), Strength: .64, Metadata: map[string]any{"documents": len(items), "nodes": len(nodes), "edges": len(edges), "path": path, "trigger": trigger}})
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *GLPIKBSyncer) Status() GLPIKBStatus {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
return GLPIKBStatus{Enabled: s.cfg.Enabled, Path: s.path, LastSync: s.lastSync, LastError: s.lastError, Documents: s.documents, CachePath: s.cfg.CachePath, NextSync: s.nextSync}
|
|
}
|
|
|
|
func (s *GLPIKBSyncer) setNextSync(t time.Time) { s.mu.Lock(); s.nextSync = t; s.mu.Unlock() }
|
|
func (s *GLPIKBSyncer) currentPath() string { s.mu.RLock(); defer s.mu.RUnlock(); return s.path }
|
|
|
|
func (s *GLPIKBSyncer) buildGraph(items []glpi.KnowledgeItem, categories []glpi.ITILCategory) ([]model.Node, []model.Edge, map[string]string) {
|
|
kbCategoryNames := map[int64][]string{}
|
|
for _, category := range categories {
|
|
if category.KnowbaseCategoryID <= 0 {
|
|
continue
|
|
}
|
|
name := strings.TrimSpace(category.CompleteName)
|
|
if name == "" {
|
|
name = strings.TrimSpace(category.Name)
|
|
}
|
|
if name != "" {
|
|
kbCategoryNames[category.KnowbaseCategoryID] = append(kbCategoryNames[category.KnowbaseCategoryID], name)
|
|
}
|
|
}
|
|
now := time.Now().UTC()
|
|
nodesByID := map[string]model.Node{}
|
|
contents := map[string]string{}
|
|
var edges []model.Edge
|
|
sourceLabel := s.cfg.Source
|
|
if sourceLabel == "" {
|
|
sourceLabel = "GLPI Knowledge Base"
|
|
}
|
|
sourceID := graph.ID("source", strings.ToLower(sourceLabel))
|
|
nodesByID[sourceID] = model.Node{ID: sourceID, Kind: "source", Label: sourceLabel, Origin: "glpi-kb-taxonomy", ExternalID: sourceLabel, Weight: .85, UpdatedAt: now}
|
|
for _, item := range items {
|
|
text := cleanGLPIHTML(item.Content)
|
|
if text == "" {
|
|
continue
|
|
}
|
|
externalID := "GLPI-KB-" + strconv.FormatInt(item.ID, 10)
|
|
nodeID := graph.ID("knowledge", externalID)
|
|
categoriesForNode := []string{"GLPI KB"}
|
|
for _, categoryID := range item.CategoryIDs {
|
|
categoriesForNode = append(categoriesForNode, kbCategoryNames[categoryID]...)
|
|
}
|
|
categoriesForNode = uniqueStrings(categoriesForNode)
|
|
keywords := extractKeywords(item.Title, categoriesForNode)
|
|
nodesByID[nodeID] = model.Node{ID: nodeID, Kind: "knowledge", Label: strings.TrimSpace(item.Title), Summary: clamp(text, 1100), Status: "production", Origin: "glpi-kb", ExternalID: externalID, URI: "glpi://KnowbaseItem/" + strconv.FormatInt(item.ID, 10), Categories: categoriesForNode, Keywords: keywords, Weight: 1.35, Metadata: map[string]any{"source": sourceLabel, "glpi_id": item.ID, "glpi_category_ids": item.CategoryIDs, "language": item.Language, "modified_at": item.ModifiedAt}, UpdatedAt: now}
|
|
contents[nodeID] = text
|
|
edges = append(edges, model.Edge{Source: nodeID, Target: sourceID, Type: "derived_from", Origin: "glpi-kb", Status: "verified", Confidence: 1, Weight: .45})
|
|
for _, category := range categoriesForNode {
|
|
categoryID := graph.ID("category", strings.ToLower(category))
|
|
if _, ok := nodesByID[categoryID]; !ok {
|
|
nodesByID[categoryID] = model.Node{ID: categoryID, Kind: "category", Label: category, Origin: "glpi-kb-taxonomy", ExternalID: category, Weight: .8, UpdatedAt: now}
|
|
}
|
|
edges = append(edges, model.Edge{Source: nodeID, Target: categoryID, Type: "categorized_as", Origin: "glpi-kb", Status: "verified", Confidence: 1, Weight: .58})
|
|
}
|
|
}
|
|
nodes := make([]model.Node, 0, len(nodesByID))
|
|
for _, node := range nodesByID {
|
|
nodes = append(nodes, node)
|
|
}
|
|
sort.Slice(nodes, func(i, j int) bool { return nodes[i].ID < nodes[j].ID })
|
|
return nodes, edges, contents
|
|
}
|
|
|
|
func (s *GLPIKBSyncer) Content(nodeID string) (string, bool) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
value, ok := s.contents[nodeID]
|
|
return value, ok
|
|
}
|
|
|
|
func glpiKnowledgeFingerprint(nodes []model.Node, edges []model.Edge, contents map[string]string) (string, error) {
|
|
base, err := knowledgeFingerprint(nodes, edges)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
keys := make([]string, 0, len(contents))
|
|
for key := range contents {
|
|
keys = append(keys, key)
|
|
}
|
|
sort.Strings(keys)
|
|
var b strings.Builder
|
|
b.WriteString(base)
|
|
for _, key := range keys {
|
|
b.WriteString("\x00")
|
|
b.WriteString(key)
|
|
b.WriteString("\x00")
|
|
b.WriteString(contents[key])
|
|
}
|
|
sum := sha256.Sum256([]byte(b.String()))
|
|
return hex.EncodeToString(sum[:]), nil
|
|
}
|
|
|
|
func cloneStringMap(in map[string]string) map[string]string {
|
|
out := make(map[string]string, len(in))
|
|
for key, value := range in {
|
|
out[key] = value
|
|
}
|
|
return out
|
|
}
|
|
|
|
var glpiTagRE = regexp.MustCompile(`(?s)<[^>]*>`)
|
|
var glpiSpaceRE = regexp.MustCompile(`[\t\r\n ]+`)
|
|
var keywordSplitRE = regexp.MustCompile(`[^\pL\pN_-]+`)
|
|
|
|
func cleanGLPIHTML(value string) string {
|
|
replacer := strings.NewReplacer("<br>", "\n", "<br/>", "\n", "<br />", "\n", "</p>", "\n", "</li>", "\n")
|
|
value = replacer.Replace(value)
|
|
value = glpiTagRE.ReplaceAllString(value, " ")
|
|
value = htmlstd.UnescapeString(value)
|
|
return strings.TrimSpace(glpiSpaceRE.ReplaceAllString(value, " "))
|
|
}
|
|
|
|
func extractKeywords(title string, categories []string) []string {
|
|
var out []string
|
|
for _, part := range keywordSplitRE.Split(title, -1) {
|
|
part = strings.TrimSpace(part)
|
|
if len([]rune(part)) >= 4 {
|
|
out = append(out, part)
|
|
}
|
|
}
|
|
out = append(out, categories...)
|
|
return uniqueStrings(out)
|
|
}
|
|
|
|
func uniqueStrings(in []string) []string {
|
|
seen := map[string]struct{}{}
|
|
var out []string
|
|
for _, value := range in {
|
|
value = strings.TrimSpace(value)
|
|
key := strings.ToLower(value)
|
|
if value == "" {
|
|
continue
|
|
}
|
|
if _, ok := seen[key]; ok {
|
|
continue
|
|
}
|
|
seen[key] = struct{}{}
|
|
out = append(out, value)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func countKnowledge(nodes []model.Node) int {
|
|
count := 0
|
|
for _, node := range nodes {
|
|
if node.Kind == "knowledge" {
|
|
count++
|
|
}
|
|
}
|
|
return count
|
|
}
|
|
|
|
func ensureCacheDir(path string) error {
|
|
if path == "" {
|
|
return nil
|
|
}
|
|
return os.MkdirAll(filepath.Dir(path), 0o750)
|
|
}
|