descrybe/apps/api/internal/feeds/extract_schema.go

package feeds

import (
	"bytes"
	"context"
	"encoding/csv"
	"encoding/xml"
	"errors"
	"fmt"
	"io"
	"sort"
	"strings"
	"unicode"

	"github.com/google/uuid"
)

const (
	schemaSampleBytes = 512 << 10 // 512 KiB preview window
	schemaMaxRows     = 25
	schemaMaxSamples  = 5
	schemaMaxFields   = 200
	previewMaxLines   = 80
)

// SchemaField is one discovered source column/xpath with sample values.
type SchemaField struct {
	Path              string   `json:"path"`
	FieldName         string   `json:"field_name"`
	DataType          string   `json:"data_type"`
	SampleValues      []string `json:"sample_values"`
	UniqueValuesCount int      `json:"unique_values_count"`
	SuggestedTarget   string   `json:"suggested_target,omitempty"`
}

// SchemaExtractResult is returned by POST /feeds/{id}/extract-schema.
type SchemaExtractResult struct {
	FeedID           string        `json:"feed_id"`
	Format           string        `json:"format"`
	SuggestedPath    string        `json:"suggested_item_path,omitempty"`
	ItemPath         string        `json:"item_path,omitempty"`
	Fields           []SchemaField `json:"fields"`
	SampleRows       int           `json:"sample_rows"`
	Preview          string        `json:"preview,omitempty"`
	PreviewTruncated bool          `json:"preview_truncated,omitempty"`
}

// ExtractSchema downloads a bounded sample of the feed and returns field paths + samples.
func (s *Service) ExtractSchema(ctx context.Context, companyID, feedID uuid.UUID, itemPathHint string) (*SchemaExtractResult, error) {
	feed, err := s.Get(ctx, companyID, feedID)
	if err != nil {
		if IsNotFound(err) {
			return nil, ErrNotFound
		}
		return nil, err
	}
	urlStr, _ := feed["url"].(string)
	feedType, _ := feed["feed_type"].(string)

	itemPathHint = strings.TrimSpace(itemPathHint)
	if itemPathHint == "" {
		if m, err := s.GetMappings(ctx, companyID, feedID); err == nil {
			itemPathHint = itemPathFromMappings(m["mappings"])
		}
		if itemPathHint == "" {
			if opts, ok := feed["options"].(map[string]any); ok {
				if v, ok := opts["item_path"].(string); ok {
					itemPathHint = strings.TrimSpace(v)
				}
			}
		}
	}

	src, err := s.loadFeedSource(ctx, companyID, feed)
	if err != nil {
		return nil, err
	}
	defer src.Close()

	f, err := src.Open()
	if err != nil {
		return nil, err
	}
	defer f.Close()

	truncated := src.size > schemaSampleBytes
	data, err := io.ReadAll(io.LimitReader(f, schemaSampleBytes))
	if err != nil {
		return nil, err
	}

	format := detectFeedFormat(feedType, src.contentType, urlStr, data)
	out := &SchemaExtractResult{
		FeedID:           feedID.String(),
		Format:           format,
		ItemPath:         itemPathHint,
		PreviewTruncated: truncated,
	}

	switch format {
	case "xml":
		suggested := guessXMLItemPath(data)
		out.SuggestedPath = suggested
		local := itemLocalFromPath(itemPathHint)
		if local == "" {
			local = itemLocalFromPath(suggested)
			out.ItemPath = suggested
		} else {
			out.ItemPath = itemPathHint
		}
		fields, rows, err := extractXMLSchema(data, local)
		if err != nil {
			return nil, err
		}
		out.Fields = fields
		out.SampleRows = rows
		out.Preview = buildXMLPreview(data, local)
	default:
		fields, rows, preview, err := extractCSVSchema(data)
		if err != nil {
			return nil, err
		}
		out.Fields = fields
		out.SampleRows = rows
		out.Preview = preview
		out.SuggestedPath = ""
		out.ItemPath = ""
	}

	return out, nil
}

func itemPathFromMappings(raw any) string {
	switch t := raw.(type) {
	case map[string]any:
		if v, ok := t["item_path"].(string); ok {
			if p := strings.TrimSpace(v); p != "" {
				return p
			}
		}
		if fields, ok := t["fields"]; ok {
			if p := itemPathFromMappings(fields); p != "" {
				return p
			}
		}
		if nested, ok := t["mappings"]; ok {
			return itemPathFromMappings(nested)
		}
		return ""
	case []any, []FieldMapping:
		return deriveItemPathFromMappings(parseMappings(t))
	default:
		if parsed := parseMappings(raw); len(parsed) > 0 {
			return deriveItemPathFromMappings(parsed)
		}
		return ""
	}
}

// deriveItemPathFromMappings picks the common parent path of mapping xpaths
// (e.g. Export/Item/ID + Export/Item/name -> Export/Item).
func deriveItemPathFromMappings(mappings []FieldMapping) string {
	var partsLists [][]string
	for _, m := range mappings {
		src := m.sourceKey()
		if src == "" {
			continue
		}
		src = strings.Trim(strings.ReplaceAll(src, "\\", "/"), "/")
		if !strings.Contains(src, "/") {
			continue
		}
		parts := strings.Split(src, "/")
		if len(parts) < 2 {
			continue
		}
		// Drop the leaf field segment.
		partsLists = append(partsLists, parts[:len(parts)-1])
	}
	if len(partsLists) == 0 {
		return ""
	}
	common := partsLists[0]
	for _, parts := range partsLists[1:] {
		n := len(common)
		if len(parts) < n {
			n = len(parts)
		}
		i := 0
		for i < n && strings.EqualFold(common[i], parts[i]) {
			i++
		}
		common = common[:i]
		if len(common) == 0 {
			return ""
		}
	}
	return strings.Join(common, "/")
}

func itemLocalFromPath(path string) string {
	path = strings.Trim(strings.TrimSpace(path), "/")
	if path == "" {
		return ""
	}
	if i := strings.LastIndex(path, "/"); i >= 0 {
		return path[i+1:]
	}
	return path
}

func guessXMLItemPath(data []byte) string {
	sample := string(data)
	if len(sample) > 64<<10 {
		sample = sample[:64<<10]
	}
	lower := strings.ToLower(sample)

	type cand struct {
		local string
		full  string
	}
	cands := []cand{
		{"item", "rss/channel/item"},
		{"product", "products/product"},
		{"entry", "feed/entry"},
		{"offer", "offers/offer"},
		{"row", "rows/row"},
	}
	for _, c := range cands {
		if strings.Contains(lower, "<"+c.local) || strings.Contains(lower, ":"+c.local) {
			if path := findFirstTagPath(data, c.local); path != "" {
				return path
			}
			return c.full
		}
	}
	return "rss/channel/item"
}

func findFirstTagPath(data []byte, local string) string {
	dec := xml.NewDecoder(bytes.NewReader(data))
	dec.Strict = false
	var stack []string
	for {
		tok, err := dec.Token()
		if err != nil {
			return ""
		}
		switch t := tok.(type) {
		case xml.StartElement:
			stack = append(stack, t.Name.Local)
			if localNameEquals(t.Name, local) {
				return strings.Join(stack, "/")
			}
		case xml.EndElement:
			if len(stack) > 0 {
				stack = stack[:len(stack)-1]
			}
		}
	}
}

type fieldAcc struct {
	path     string
	name     string
	samples  []string
	seen     map[string]struct{}
	dataType string
}

func extractXMLSchema(data []byte, itemLocal string) ([]SchemaField, int, error) {
	if itemLocal == "" {
		itemLocal = guessXMLItemLocal(data)
	}
	acc := map[string]*fieldAcc{}
	rows := 0
	_, err := parseXMLItems(bytes.NewReader(data), itemLocal, func(row feedRow) error {
		rows++
		if rows > schemaMaxRows {
			return errStopSchema
		}
		accumulateRowFields(acc, row)
		return nil
	})
	if err != nil && !errors.Is(err, errStopSchema) {
		return nil, rows, err
	}
	return finalizeSchema(acc), rows, nil
}

func accumulateRowFields(acc map[string]*fieldAcc, row feedRow) {
	for k, v := range row {
		v = strings.TrimSpace(v)
		if v == "" {
			continue
		}
		if strings.HasPrefix(k, "@") && !strings.Contains(k, "/") {
			// Bare attribute dupes are noise; path-qualified @ kept below.
			continue
		}
		fa := acc[k]
		if fa == nil {
			fa = &fieldAcc{
				path: k,
				name: leafName(k),
				seen: map[string]struct{}{},
			}
			acc[k] = fa
		}
		if _, ok := fa.seen[v]; !ok {
			fa.seen[v] = struct{}{}
			if len(fa.samples) < schemaMaxSamples {
				fa.samples = append(fa.samples, truncateSample(v))
			}
		}
		if fa.dataType == "" {
			fa.dataType = inferDataType(v)
		} else if fa.dataType != "string" {
			t := inferDataType(v)
			if t != fa.dataType {
				fa.dataType = "string"
			}
		}
	}
}

var errStopSchema = errors.New("schema sample limit")

func extractCSVSchema(data []byte) ([]SchemaField, int, string, error) {
	r := csv.NewReader(bytes.NewReader(data))
	r.ReuseRecord = true
	r.LazyQuotes = true
	r.TrimLeadingSpace = true
	r.FieldsPerRecord = -1

	header, err := r.Read()
	if err != nil {
		return nil, 0, "", fmt.Errorf("csv header: %w", err)
	}
	cols := make([]string, len(header))
	for i, h := range header {
		cols[i] = strings.TrimSpace(h)
	}

	acc := map[string]*fieldAcc{}
	for _, c := range cols {
		if c == "" {
			continue
		}
		acc[c] = &fieldAcc{path: c, name: c, seen: map[string]struct{}{}, dataType: ""}
	}

	rows := 0
	var previewLines []string
	previewLines = append(previewLines, strings.Join(cols, ","))

	for {
		rec, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, rows, "", fmt.Errorf("csv row %d: %w", rows+1, err)
		}
		rows++
		if rows <= 5 {
			previewLines = append(previewLines, strings.Join(rec, ","))
		}
		if rows > schemaMaxRows {
			break
		}
		row := make(feedRow, len(cols))
		for i, col := range cols {
			if col == "" || i >= len(rec) {
				continue
			}
			row[col] = strings.TrimSpace(rec[i])
		}
		expandSpecificationFields(row)
		accumulateRowFields(acc, row)
	}

	preview := strings.Join(previewLines, "\n")
	return finalizeSchema(acc), rows, preview, nil
}

func finalizeSchema(acc map[string]*fieldAcc) []SchemaField {
	preferNestedFieldPaths(acc)
	keys := make([]string, 0, len(acc))
	for k := range acc {
		keys = append(keys, k)
	}
	sort.SliceStable(keys, func(i, j int) bool {
		di, dj := strings.Count(keys[i], "/"), strings.Count(keys[j], "/")
		if di != dj {
			return di < dj
		}
		return keys[i] < keys[j]
	})
	out := make([]SchemaField, 0, len(keys))
	for _, k := range keys {
		fa := acc[k]
		dt := fa.dataType
		if dt == "" {
			dt = "string"
		}
		// Nested CDATA/HTML parent blobs stay as string; children are preferred for mapping.
		if isSpecFieldKey(k) && hasPrefixedChildrenAcc(acc, k) {
			dt = "object"
		}
		out = append(out, SchemaField{
			Path:              fa.path,
			FieldName:         fa.name,
			DataType:          dt,
			SampleValues:      fa.samples,
			UniqueValuesCount: len(fa.seen),
			SuggestedTarget:   SuggestTarget(fa.name),
		})
		if out[len(out)-1].SuggestedTarget == "" {
			out[len(out)-1].SuggestedTarget = SuggestTarget(fa.path)
		}
		if len(out) >= schemaMaxFields {
			break
		}
	}
	return out
}

// preferNestedFieldPaths drops bare leaf keys when a nested path ending with
// the same leaf exists (e.g. keep specifications/Color, drop Color).
func preferNestedFieldPaths(acc map[string]*fieldAcc) {
	nestedLeaves := map[string]struct{}{}
	for k := range acc {
		if strings.Contains(k, "/") {
			nestedLeaves[strings.ToLower(leafName(k))] = struct{}{}
		}
	}
	for k := range acc {
		if strings.Contains(k, "/") {
			continue
		}
		if _, ok := nestedLeaves[strings.ToLower(k)]; ok {
			delete(acc, k)
		}
	}
}

func hasPrefixedChildrenAcc(acc map[string]*fieldAcc, prefix string) bool {
	p := strings.TrimSuffix(prefix, "/") + "/"
	for k := range acc {
		if strings.HasPrefix(k, p) {
			return true
		}
	}
	return false
}

func leafName(path string) string {
	path = strings.TrimSpace(path)
	if i := strings.LastIndex(path, "/"); i >= 0 {
		return path[i+1:]
	}
	return path
}

func truncateSample(s string) string {
	if len(s) > 120 {
		return s[:117] + "..."
	}
	return s
}

func inferDataType(v string) string {
	v = strings.TrimSpace(v)
	if v == "" {
		return "string"
	}
	lower := strings.ToLower(v)
	if lower == "true" || lower == "false" {
		return "boolean"
	}
	dot := 0
	digits := 0
	for i, r := range v {
		if r == '-' && i == 0 {
			continue
		}
		if r == '.' {
			dot++
			if dot > 1 {
				return "string"
			}
			continue
		}
		if !unicode.IsDigit(r) {
			return "string"
		}
		digits++
	}
	if digits == 0 {
		return "string"
	}
	if dot == 1 {
		return "number"
	}
	return "integer"
}

func buildXMLPreview(data []byte, itemLocal string) string {
	sample := string(data)
	if len(sample) > schemaSampleBytes {
		sample = sample[:schemaSampleBytes]
	}
	lines := strings.Split(sample, "\n")
	out := make([]string, 0, previewMaxLines)
	for _, line := range lines {
		trimmed := strings.TrimSpace(line)
		if trimmed == "" {
			continue
		}
		out = append(out, line)
		if len(out) >= previewMaxLines {
			break
		}
	}
	_ = itemLocal
	return strings.Join(out, "\n")
}