internal/ingestion/pipeline/builtin_registry.go
Rama main · réplica de lectura del repositorio original.
package pipeline
import (
"bytes"
"embed"
"encoding/json"
"fmt"
"io/fs"
"os"
"path/filepath"
"sort"
"strings"
"sync"
)
const builtinTemplatePrefix = "ingestion_pipeline_"
// builtinParserOrder matches settings.PARSERS order, projected onto
// the canonical builtin templates shipped by Go. Templates not in this list
// are appended deterministically after the shared parser methods.
var builtinParserOrder = map[string]int{
"general": 0,
"qa": 1,
"manual": 2,
"table": 3,
"paper": 4,
"book": 5,
"laws": 6,
"presentation": 7,
"picture": 8,
"one": 9,
"audio": 10,
"email": 11,
}
//go:embed template/ingestion_pipeline_*.json
var builtinTemplateFS embed.FS
// BuiltinPipelineMeta is the API-facing metadata for one built-in ingestion
// pipeline template. The ParserID field is the value stored in the dataset's
// parser_id column for built-in pipelines.
// JSON tag "id" aligns with AgentItem.id for format consistency with
// GET /api/v1/agents?canvas_category=dataflow_canvas.
type BuiltinPipelineMeta struct {
ParserID string `json:"id"`
Title string `json:"title"`
Description string `json:"description,omitempty"`
Filename string `json:"filename"`
}
// BuiltinPipelineListResponse wraps the builtin pipeline list for
// format consistency with ListAgentsResponse.
type BuiltinPipelineListResponse struct {
BuiltinPipelines []*BuiltinPipelineMeta `json:"canvas"`
Total int64 `json:"total"`
}
type BuiltinPipeline struct {
BuiltinPipelineMeta
DSL map[string]any `json:"-"`
}
type Registry struct {
templates map[string]*BuiltinPipeline
// aliases maps legacy parser_id values to their canonical builtin
// pipeline id. Example: "naive" -> "general". Aliases keep existing
// dataset rows runnable after the hardcoded parser list is removed;
// they are hidden from List/Refs so the front end only sees canonical
// templates.
aliases map[string]string
list []*BuiltinPipelineMeta
}
var (
defaultRegistryOnce sync.Once
defaultRegistry *Registry
defaultRegistryErr error
)
func NewRegistryFromDir(dir string) (*Registry, error) {
return NewRegistryFromFS(os.DirFS(dir), ".")
}
func DefaultRegistry() (*Registry, error) {
defaultRegistryOnce.Do(func() {
defaultRegistry, defaultRegistryErr = NewRegistryFromFS(builtinTemplateFS, "template")
})
return defaultRegistry, defaultRegistryErr
}
func NewRegistryFromFS(files fs.FS, dir string) (*Registry, error) {
entries, err := fs.ReadDir(files, dir)
if err != nil {
return nil, err
}
templates := make(map[string]*BuiltinPipeline)
list := make([]*BuiltinPipelineMeta, 0, len(entries))
for _, entry := range entries {
if entry.IsDir() {
continue
}
name := entry.Name()
if !strings.HasPrefix(name, builtinTemplatePrefix) || !strings.HasSuffix(name, ".json") {
continue
}
raw, err := fs.ReadFile(files, filepath.Join(dir, name))
if err != nil {
return nil, err
}
tpl, err := parseTemplate(name, raw)
if err != nil {
return nil, err
}
if _, exists := templates[tpl.ParserID]; exists {
return nil, fmt.Errorf("duplicate parser_id %q", tpl.ParserID)
}
templates[tpl.ParserID] = tpl
meta := tpl.BuiltinPipelineMeta
list = append(list, &meta)
}
sort.SliceStable(list, func(i, j int) bool {
iOrder, iKnown := builtinParserOrder[list[i].ParserID]
jOrder, jKnown := builtinParserOrder[list[j].ParserID]
if iKnown && jKnown {
return iOrder < jOrder
}
if iKnown != jKnown {
return iKnown
}
return list[i].ParserID < list[j].ParserID
})
return &Registry{templates: templates, aliases: builtinAliases(), list: list}, nil
}
// builtinAliases returns the legacy parser_id -> canonical pipeline id
// mappings. "naive" is the historical default parser; its behavior now
// lives in the "general" builtin template, so dataset rows storing
// parser_id="naive" resolve to "general" at lookup time.
func builtinAliases() map[string]string {
return map[string]string{
"naive": "general",
}
}
// canonicalRef resolves a possibly-aliased ref to its canonical pipeline id.
// Unknown refs are returned unchanged so the caller can distinguish "alias
// target missing" from "not an alias".
func (r *Registry) canonicalRef(ref string) string {
if r == nil {
return ref
}
if canonical, ok := r.aliases[ref]; ok {
return canonical
}
return ref
}
func (r *Registry) List() *BuiltinPipelineListResponse {
if r == nil {
return nil
}
out := make([]*BuiltinPipelineMeta, 0, len(r.list))
for _, item := range r.list {
if item == nil {
continue
}
cp := *item
out = append(out, &cp)
}
return &BuiltinPipelineListResponse{
BuiltinPipelines: out,
Total: int64(len(out)),
}
}
func (r *Registry) Refs() []string {
if r == nil {
return nil
}
out := make([]string, 0, len(r.list))
for _, item := range r.list {
if item == nil {
continue
}
out = append(out, item.ParserID)
}
return out
}
func (r *Registry) Get(ref string) (*BuiltinPipeline, bool) {
if r == nil {
return nil, false
}
ref = strings.TrimSpace(ref)
tpl, ok := r.templates[r.canonicalRef(ref)]
if !ok || tpl == nil {
return nil, false
}
cp := *tpl
return &cp, true
}
// IsValid reports whether ref names a builtin template, either directly or
// through an alias. Use this for parser_id validation so legacy values
// (e.g. "naive") stay accepted after the hardcoded allow-list is removed.
func (r *Registry) IsValid(ref string) bool {
if r == nil {
return false
}
ref = strings.TrimSpace(ref)
if ref == "" {
return false
}
_, ok := r.templates[r.canonicalRef(ref)]
return ok
}
func parseTemplate(filename string, raw []byte) (*BuiltinPipeline, error) {
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.UseNumber()
var data map[string]any
if err := decoder.Decode(&data); err != nil {
return nil, err
}
ref := builtinRefFromFilename(filename)
if ref == "" {
return nil, fmt.Errorf("invalid builtin filename %q", filename)
}
dsl, ok := data["dsl"].(map[string]any)
if !ok {
return nil, fmt.Errorf("builtin template %q missing dsl", filename)
}
tpl := &BuiltinPipeline{
BuiltinPipelineMeta: BuiltinPipelineMeta{
ParserID: ref,
Title: englishText(data["title"], ref),
Description: englishText(data["description"], ""),
Filename: filename,
},
DSL: dsl,
}
return tpl, nil
}
func builtinRefFromFilename(filename string) string {
name := strings.TrimSuffix(filename, filepath.Ext(filename))
name = strings.TrimPrefix(name, builtinTemplatePrefix)
if name == filename || name == "" {
return ""
}
return name
}
// LoadBuiltinDSL returns the canonical DSL JSON for a builtin parser_id,
// resolving aliases transparently. "naive" returns the "general" template.
func LoadBuiltinDSL(parserID string) (string, error) {
r, err := DefaultRegistry()
if err != nil {
return "", fmt.Errorf("builtin pipeline registry: %w", err)
}
tpl, ok := r.Get(parserID)
if !ok {
return "", fmt.Errorf("unknown builtin parser_id: %s", parserID)
}
dslBytes, err := json.Marshal(tpl.DSL)
if err != nil {
return "", fmt.Errorf("marshal builtin DSL %s: %w", parserID, err)
}
return string(dslBytes), nil
}
// englishText extracts the English text from a localized template field.
// The API must return English so it aligns with the front-end's hardcoded
// parser labels (web/src/locales/en.ts parserLabel.*); the front end does
// its own i18n and only uses this as the canonical value. When the field
// is a plain string it is returned as-is; when "en" is missing, any
// available translation is used as a last resort.
func englishText(v any, fallback string) string {
switch x := v.(type) {
case string:
return strings.TrimSpace(x)
case map[string]any:
if s, ok := x["en"].(string); ok && strings.TrimSpace(s) != "" {
return strings.TrimSpace(s)
}
for _, raw := range x {
if s, ok := raw.(string); ok && strings.TrimSpace(s) != "" {
return strings.TrimSpace(s)
}
}
}
return fallback
}