//
//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
//

package chunker

import (
	"fmt"
	"regexp"
	"sort"
	"strings"

	"ragflow/internal/agent/runtime"
	"ragflow/internal/ingestion/component/schema"
	"ragflow/internal/tokenizer"
)

// newChunkerByName dispatches the DSL name to a typed constructor.
// Centralised here so each chunker file only needs an init() that
// declares its registered name (see register.go). The returned
// runtime.Component interface is satisfied directly by each
// constructor (NewTokenChunker etc.) — no intermediate assertion
// is needed.
func newChunkerByName(name string, params map[string]any) (runtime.Component, error) {
	switch name {
	case ComponentNameTokenChunker:
		return NewTokenChunker(params)
	case ComponentNameTitleChunker:
		return NewTitleChunker(params)
	case ComponentNameGroupTitleChunker:
		return NewGroupTitleChunker(params)
	case ComponentNameHierarchyTitleChunker:
		return NewHierarchyTitleChunker(params)
	case ComponentNameQAChunker:
		return NewQAChunker(params)
	case ComponentNameOneChunker:
		return NewOneChunker(params)
	case ComponentNameTagChunker:
		return NewTagChunker(params)
	case ComponentNameTableChunker:
		return NewTableChunker(params)
	case ComponentNamePresentationChunker:
		return NewPresentationChunker(params)
	default:
		return nil, fmt.Errorf("chunker: unknown component %q", name)
	}
}

// ---------------------------------------------------------------------------
// numeric / list conversion helpers (shared across chunker variants)
// ---------------------------------------------------------------------------

func stringListFromAny(in []any) []string {
	out := make([]string, 0, len(in))
	for _, x := range in {
		if s, ok := x.(string); ok && s != "" {
			out = append(out, s)
		}
	}
	return out
}

// ---------------------------------------------------------------------------
// regex / split helpers
// ---------------------------------------------------------------------------

// backtickDelimRE extracts backtick-wrapped delimiter tokens.
// Case-sensitive on purpose (mirrors Python after #17384 / PR #17386):
// never add (?i) here — delimiter matching must preserve letter casing.
var backtickDelimRE = regexp.MustCompile("`([^`]+)`")

// escapeAndSort QuoteMeta-escapes tokens and sorts longest-first.
// Empty tokens are dropped. When dedup is true, exact duplicate tokens
// are collapsed (first occurrence wins). Matching stays case-sensitive.
func escapeAndSort(tokens []string, dedup bool) []string {
	out := make([]string, 0, len(tokens))
	var seen map[string]struct{}
	if dedup {
		seen = make(map[string]struct{}, len(tokens))
	}
	for _, tok := range tokens {
		if tok == "" {
			continue
		}
		if dedup {
			if _, ok := seen[tok]; ok {
				continue
			}
			seen[tok] = struct{}{}
		}
		out = append(out, regexp.QuoteMeta(tok))
	}
	sort.SliceStable(out, func(i, j int) bool { return len(out[i]) > len(out[j]) })
	return out
}

// getDelimiters ports rag.nlp.get_delimiters (rag/nlp/__init__.py).
//
// It walks a single delimiter string, pulling out backtick-wrapped tokens
// and every bare character outside those spans, then returns a
// length-sorted, QuoteMeta-escaped alternation suitable for regexp.Split.
// Matching is case-sensitive: "a" does not match "A", and "`end`" does not
// match "End" / "END".
func getDelimiters(delimiters string) string {
	var dels []string
	s := 0
	for _, m := range backtickDelimRE.FindAllStringSubmatchIndex(delimiters, -1) {
		// m = [fullStart, fullEnd, g1Start, g1End]
		f, t := m[0], m[1]
		dels = append(dels, delimiters[m[2]:m[3]])
		for _, r := range delimiters[s:f] {
			dels = append(dels, string(r))
		}
		s = t
	}
	if s < len(delimiters) {
		for _, r := range delimiters[s:] {
			dels = append(dels, string(r))
		}
	}
	// Python get_delimiters does not dedup; preserve that behavior.
	return strings.Join(escapeAndSort(dels, false), "|")
}

// compileDelimPattern builds an alternation from backtick-wrapped tokens
// across delimiter entries (mirrors Python _compile_delimiter_pattern).
// Each entry is scanned independently so token boundaries cannot span
// adjacent slice elements. Extraction is case-sensitive — see backtickDelimRE.
//
// Plain (non-backtick) delimiters are not compiled here; callers that
// need bare-char splitting use getDelimiters (naive_merge path).
func compileDelimPattern(delims []string) *regexp.Regexp {
	var tokens []string
	for _, d := range delims {
		if d == "" {
			continue
		}
		for _, m := range backtickDelimRE.FindAllStringSubmatch(d, -1) {
			tokens = append(tokens, m[1])
		}
	}
	// Python _compile_delimiter_pattern dedups via set(...).
	custom := escapeAndSort(tokens, true)
	if len(custom) == 0 {
		return nil
	}
	return regexp.MustCompile(strings.Join(custom, "|"))
}

// splitKeepingDelim mirrors Python token_chunker._split_text_by_pattern
// (token_chunker.py:79-94): re.split with a captured delimiter group yields
// [text, delim, text, delim, ...]; each delimiter is glued to the END of the
// preceding text segment, so it never surfaces as a standalone chunk. A
// delimiter with no preceding text (a leading delimiter or one adjacent to
// another) is dropped together with the empty segment, matching Python's
// `if not chunk: continue`.
func splitKeepingDelim(text string, pattern *regexp.Regexp) []string {
	if pattern == nil {
		return []string{text}
	}
	idxs := pattern.FindAllStringIndex(text, -1)
	if len(idxs) == 0 {
		return []string{text}
	}
	var out []string
	cursor := 0
	for _, idx := range idxs {
		start, end := idx[0], idx[1]
		if start == cursor {
			cursor = end
			continue
		}
		out = append(out, text[cursor:end])
		cursor = end
	}
	if cursor < len(text) {
		out = append(out, text[cursor:])
	}
	return out
}

// ---------------------------------------------------------------------------
// chunk-doc helpers
// ---------------------------------------------------------------------------

// itemText returns the text payload from a JSON-style chunk item,
// preferring "text", then "content_with_weight".
func itemText(it schema.ChunkDoc) (string, bool) {
	if it.Text != "" {
		return it.Text, true
	}
	if it.ContentWithWeight != "" {
		return it.ContentWithWeight, true
	}
	return "", false
}

// itemDocType mirrors _build_json_chunks's type derivation.
func itemDocType(it schema.ChunkDoc) string {
	switch strings.ToLower(strings.TrimSpace(it.DocType)) {
	case "table":
		return "table"
	case "image":
		return "image"
	}
	return "text"
}

// itemTextOrFallback returns the item's preferred text, or "".
func itemTextOrFallback(it schema.ChunkDoc) string {
	if t, ok := itemText(it); ok {
		return t
	}
	return ""
}

// tokenizeStr is the shared NumTokensFromString wrapper used by
// Table/Image context attachment. Lives here so we can centrally
// swizzle the count strategy in one place if needed.
func tokenizeStr(s string) int { return tokenizer.NumTokensFromString(s) }

// toString normalises a chunk-map field to a string. Empty strings
// for missing fields.
func toString(v any) string {
	if v == nil {
		return ""
	}
	if s, ok := v.(string); ok {
		return s
	}
	return ""
}

// emptyOutputs returns the canonical no-chunks payload.
func emptyOutputs() map[string]any {
	return map[string]any{
		"output_format": "chunks",
		"chunks":        []map[string]any{},
	}
}

func emptyChunkDocs() []schema.ChunkDoc { return []schema.ChunkDoc{} }

// chunkOutputs builds the canonical chunker output (output_format="chunks" +
// chunks). The Go runtime passes only this explicit output to the next node,
// so the run-level metadata that downstream components still need (e.g.
// `name` for Tokenizer title embedding, or tenant_id/kb_id for embedding
// model resolution) is NOT re-emitted here — it lives in the workflow-wide
// CanvasState.Globals bag (seeded at pipeline start, published by the File
// component) and read directly from ctx. See runtime.CanvasState.Globals.
func chunkOutputs(chunks []schema.ChunkDoc) map[string]any {
	return map[string]any{
		"output_format": "chunks",
		"chunks":        schema.ChunkDocsToMaps(chunks),
	}
}

// withName returns a shallow copy of inputs with name set, so a component can
// guarantee `name` is present on the map it forwards to a decode step without
// mutating the caller's snapshot.
func withName(inputs map[string]any, name string) map[string]any {
	cp := make(map[string]any, len(inputs)+1)
	for k, v := range inputs {
		cp[k] = v
	}
	cp["name"] = name
	return cp
}

// cloneInputs returns a shallow copy of m with room for one extra key. Used to
// inject the Globals-resolved `name` into the decode input without mutating
// the caller's input snapshot.
func cloneInputs(m map[string]any) map[string]any {
	if m == nil {
		return map[string]any{}
	}
	cp := make(map[string]any, len(m)+1)
	for k, v := range m {
		cp[k] = v
	}
	return cp
}
