descrybe/apps/api/cmd/tmp-llm-categorize/main.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"strings"
	"syscall"
	"time"

	"github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts"
	"github.com/descrybe/descrybe-v2/apps/api/internal/aiprovider"
	"github.com/descrybe/descrybe-v2/apps/api/internal/config"
	"github.com/descrybe/descrybe-v2/apps/api/internal/db"
	"github.com/descrybe/descrybe-v2/apps/api/internal/eprel"
	"github.com/descrybe/descrybe-v2/apps/api/internal/logredact"
	"github.com/descrybe/descrybe-v2/apps/api/internal/platformsettings"
	"github.com/descrybe/descrybe-v2/apps/api/internal/processing"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

func main() {
	outDir := os.Getenv("PROBE_OUT_DIR")
	if outDir == "" {
		outDir = `F:\laragon\www\_MY\descrybe-v2\.codehelper\_llm_categorize_20260816`
	}
	_ = os.MkdirAll(outDir, 0o755)

	logPath := filepath.Join(outDir, "run.log")
	logFile, err := os.Create(logPath)
	if err != nil {
		log.Fatalf("log file: %v", err)
	}
	defer logFile.Close()
	log.SetOutput(logredact.Writer(io.MultiWriter(os.Stderr, logFile)))

	limit := 90
	companyID := uuid.MustParse("2b3159b0-fc08-415b-b248-35ed02a6baab") // Platform Demo
	userID := uuid.MustParse("0ce3305c-d810-4b56-b1d4-3c1ed510db76")

	cfg, err := config.Load()
	if err != nil {
		log.Fatalf("config: %v", err)
	}
	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer cancel()

	pool, err := db.NewPool(ctx, cfg.DatabaseURL, db.PoolOptions{
		MaxConns:              int32(cfg.DBMaxConns),
		MinConns:              int32(cfg.DBMinConns),
		MaxConnLifetime:       cfg.DBMaxConnLifetime,
		MaxConnLifetimeJitter: cfg.DBMaxConnLifetimeJitter,
		MaxConnIdleTime:       cfg.DBMaxConnIdleTime,
		HealthCheckPeriod:     cfg.DBHealthCheckPeriod,
		StatementTimeout:      cfg.DBStatementTimeout,
	})
	if err != nil {
		log.Fatalf("db: %v", err)
	}
	defer pool.Close()

	platEnv := platformsettings.EnvConfig{
		AppEncryptionKey:         cfg.AppEncryptionKey,
		CredentialsEncryptionKey: cfg.CredentialsEncryptionKey,
		TokenSigningSecret:       cfg.TokenSigningSecret,
		DatabaseURL:              cfg.DatabaseURL,
		OpenAIAPIKey:             cfg.OpenAIAPIKey,
		OpenAIBaseURL:            cfg.OpenAIBaseURL,
		OpenAIModel:              cfg.OpenAIModel,
		OpenAIEmbeddingAPIKey:    cfg.OpenAIEmbeddingAPIKey,
		OpenAIEmbeddingBaseURL:   cfg.OpenAIEmbeddingBaseURL,
		OpenAIEmbeddingModel:     cfg.OpenAIEmbeddingModel,
	}
	platSettings := platformsettings.NewService(pool, platEnv)
	aiSvc := aiprovider.NewService(pool, aiprovider.EnvConfig{
		AppEncryptionKey:         cfg.AppEncryptionKey,
		CredentialsEncryptionKey: cfg.CredentialsEncryptionKey,
		TokenSigningSecret:       cfg.TokenSigningSecret,
		DatabaseURL:              cfg.DatabaseURL,
		OpenAIAPIKey:             cfg.OpenAIAPIKey,
		OpenAIBaseURL:            cfg.OpenAIBaseURL,
		OpenAIModel:              cfg.OpenAIModel,
		ProcessingRPM:            cfg.ProcessingRPM,
		ProcessingMaxRetries:     cfg.ProcessingMaxRetries,
	})
	aiSvc.Platform = platSettings

	roleCfg, rerr := platSettings.ResolveAIConfig(ctx, platformsettings.AIRoleProcessing)
	if rerr != nil {
		failJSON(outDir, "resolve.json", map[string]any{"ok": false, "err": rerr.Error()})
		log.Fatalf("ResolveAIConfig: %v", rerr)
	}
	completer, modeLabel, byok, cerr := aiSvc.ResolveCompleterForRole(ctx, companyID, aiprovider.RoleProcessing)
	if cerr != nil {
		failJSON(outDir, "resolve.json", map[string]any{"ok": false, "err": cerr.Error()})
		log.Fatalf("ResolveCompleterForRole: %v", cerr)
	}
	if completer == nil {
		failJSON(outDir, "resolve.json", map[string]any{"ok": false, "err": "nil completer"})
		log.Fatal("LIVE LLM REQUIRED: nil completer")
	}
	client, ok := completer.(*processing.OpenAIClient)
	if !ok {
		failJSON(outDir, "resolve.json", map[string]any{"ok": false, "err": fmt.Sprintf("type %T", completer)})
		log.Fatalf("completer type %T not *OpenAIClient", completer)
	}
	base := strings.TrimSpace(client.BaseURL)
	model := strings.TrimSpace(client.Model)
	resolveInfo := map[string]any{
		"role_source":    roleCfg.Source,
		"role_base_url":  roleCfg.BaseURL,
		"role_model":     roleCfg.Model,
		"completer_base": base,
		"completer_model": model,
		"mode_label":     modeLabel,
		"byok":           byok,
		"live":           true,
	}
	if processing.IsMockOrLoopbackBaseURL(base) || strings.Contains(strings.ToLower(base), "18767") {
		resolveInfo["live"] = false
		resolveInfo["fail_reason"] = "mock/loopback"
		writeJSON(outDir, "resolve.json", resolveInfo)
		log.Fatalf("LIVE LLM REQUIRED: base=%s — refuse mock", base)
	}
	if strings.TrimSpace(client.APIKey) == "" {
		resolveInfo["live"] = false
		resolveInfo["fail_reason"] = "empty API key"
		writeJSON(outDir, "resolve.json", resolveInfo)
		log.Fatal("LIVE LLM REQUIRED: empty API key")
	}

	// Probe /v1/models
	modelsURL := strings.TrimRight(base, "/") + "/models"
	modelsCtx, modelsCancel := context.WithTimeout(ctx, 30*time.Second)
	defer modelsCancel()
	modelsReq, _ := http.NewRequestWithContext(modelsCtx, http.MethodGet, modelsURL, nil)
	modelsReq.Header.Set("Authorization", "Bearer "+client.APIKey)
	modelsResp, merr := http.DefaultClient.Do(modelsReq)
	if merr != nil {
		resolveInfo["probe_ok"] = false
		resolveInfo["models_error"] = processing.TruncateError(merr)
		writeJSON(outDir, "resolve.json", resolveInfo)
		log.Fatalf("LIVE LLM CONNECTION ERROR: %s", processing.TruncateError(merr))
	}
	modelsBody, _ := io.ReadAll(io.LimitReader(modelsResp.Body, 1024))
	_ = modelsResp.Body.Close()
	resolveInfo["models_http_status"] = modelsResp.StatusCode
	resolveInfo["models_body_snippet"] = truncate(string(modelsBody), 200)
	if modelsResp.StatusCode != http.StatusOK {
		resolveInfo["probe_ok"] = false
		writeJSON(outDir, "resolve.json", resolveInfo)
		log.Fatalf("LIVE LLM CONNECTION ERROR: models HTTP %d", modelsResp.StatusCode)
	}
	resolveInfo["probe_ok"] = true
	writeJSON(outDir, "resolve.json", resolveInfo)
	log.Printf("live LLM ok base=%s model=%s mode=%s", base, model, modeLabel)

	before := coverageStats(ctx, pool, companyID)
	writeJSON(outDir, "coverage_before.json", before)
	log.Printf("coverage before: raw=%d mapped_with=%d mapped_empty=%d processed_with=%d taxonomy=%d",
		before.RawTotal, before.MappedWith, before.MappedEmpty, before.ProcessedWith, before.TaxonomyCount)

	rawIDs, samples, err := pickEmptyCategoryRawIDs(ctx, pool, companyID, limit)
	if err != nil {
		log.Fatalf("pick: %v", err)
	}
	writeJSON(outDir, "batch_sample.json", samples)
	log.Printf("batch size=%d", len(rawIDs))
	if len(rawIDs) == 0 {
		log.Fatal("no empty-category products with name/description found")
	}

	pipeline := processing.NewPipeline(pool)
	pipeline.BatchSize = cfg.ProcessingBatchSize
	pipeline.AI = aiSvc
	pipeline.Prompts = aiprompts.NewService(pool)
	pipeline.Engine = &processing.Engine{
		Completer:    nil,
		Vector:       processing.NoopVectorCategorizer{},
		EPREL:        eprel.NewClient(eprel.Options{Enabled: false, Timeout: 5 * time.Second}),
		ProviderMode: processing.AIProviderInternal,
	}

	_, _ = pool.Exec(ctx, `
		UPDATE processing_jobs
		SET status = 'failed', error = 'yielded to llm categorize batch', updated_at = now()
		WHERE company_id = $1 AND status IN ('pending','running','processing')`, companyID)

	jobs, err := pipeline.StartJob(ctx, companyID, userID, rawIDs, "categorize_only")
	if err != nil {
		log.Fatalf("StartJob: %v", err)
	}
	if len(jobs) == 0 {
		log.Fatal("StartJob returned no jobs")
	}
	jobID := jobs[0].ID
	_ = os.WriteFile(filepath.Join(outDir, "process_id.txt"), []byte(jobID.String()+"\n"), 0o644)
	log.Printf("job_id=%s chunks=%d type=categorize_only", jobID, len(jobs))

	runCtx, runCancel := context.WithTimeout(ctx, 45*time.Minute)
	defer runCancel()
	start := time.Now()
	for i, j := range jobs {
		log.Printf("ProcessJob %d/%d id=%s", i+1, len(jobs), j.ID)
		if err := pipeline.ProcessJob(runCtx, j.ID); err != nil {
			log.Fatalf("ProcessJob %s: %v", j.ID, err)
		}
	}
	elapsed := time.Since(start).Round(time.Millisecond)

	after := coverageStats(ctx, pool, companyID)
	writeJSON(outDir, "coverage_after.json", after)

	results, err := loadBatchResults(ctx, pool, companyID, rawIDs)
	if err != nil {
		log.Fatalf("results: %v", err)
	}
	writeJSON(outDir, "batch_results.json", results)

	okCount, rejected, stillEmpty := 0, 0, 0
	llmCount := 0
	for _, r := range results {
		if r.MappedCategory == "" && r.ProcessedCategory == "" {
			stillEmpty++
			continue
		}
		okCount++
		if r.FieldSource == "llm" {
			llmCount++
		}
		if r.FieldSource == "cleared_invalid" {
			rejected++
		}
	}

	scorecard := map[string]any{
		"date":              time.Now().UTC().Format(time.RFC3339),
		"company_id":        companyID.String(),
		"company":           "Platform Demo",
		"processing_type":   "categorize_only",
		"job_id":            jobID.String(),
		"batch_requested":   limit,
		"batch_actual":      len(rawIDs),
		"elapsed":           elapsed.String(),
		"llm_base":          base,
		"llm_model":         model,
		"coverage_before":   before,
		"coverage_after":    after,
		"batch_ok":          okCount,
		"batch_llm_source":  llmCount,
		"batch_still_empty": stillEmpty,
		"batch_rejected":    rejected,
		"mapped_delta":      after.MappedWith - before.MappedWith,
		"processed_delta":   after.ProcessedWith - before.ProcessedWith,
		"how_to_run_all": []string{
			"Admin UI: Products → select uncategorized → Process with type categorize (or full)",
			"API: POST /v1/products/process with processing_type=categorize_only and product ids lacking mapped_data.category",
			"Chunk 100–500 at a time; avoid full 25k in one job unless worker concurrency + credit budget are confirmed",
			"Seed: npm run seed:a1 restores dump backfill (~17%); remaining empties need categorize_only/full with live LLM",
		},
	}
	writeJSON(outDir, "scorecard.json", scorecard)

	md := fmt.Sprintf(`# LLM categorize scorecard — 2026-08-16

## Verdict
Batch **%d** Platform Demo products with empty `+"`mapped_data.category`"+` via `+"`categorize_only`"+` (vector noop → LLM taxonomy pick).

## LLM
- base: `+"`%s`"+`
- model: `+"`%s`"+`
- job: `+"`%s`"+`
- elapsed: %s

## Coverage (company-wide)
| | before | after | delta |
|--|--:|--:|--:|
| mapped with category | %d | %d | %+d |
| mapped empty | %d | %d | %+d |
| processed with category | %d | %d | %+d |
| taxonomy unique_ids | %d | %d | |

## Batch
- requested/actual: %d / %d
- got category (mapped or processed): %d
- field_sources.category=llm: %d
- still empty: %d

## How it works
1. Full / categorize pipeline: after mapped/prior, if category empty → Pinecone vector (if enabled) → LLM Completer with bounded company taxonomy (unique_id + name, max 200).
2. Response `+"`categoryId`"+` coerced/validated against taxonomy; inventing IDs is rejected (`+"`filterCategoryIfInvalid`"+`).
3. Persists to `+"`processed_products.category`"+` and empty `+"`mapped_data.category`"+`; logs `+"`uid/name source=llm`"+`.

## How to run remaining (~%d empty)
1. Prefer chunks of 100–500 (this probe used %d).
2. Admin Products → filter uncategorized → Process (`+"`categorize`"+` / `+"`categorize_only`"+`).
3. Or API `+"`processing_type=categorize_only`"+` on raw product ids.
4. `+"`full`"+` also categorizes then enhances (more tokens).
5. Seed dump backfill does not invent categories for products missing from MySQL dump — LLM categorize fills those going forward without wiping the 25k catalog.
`,
		len(rawIDs), base, model, jobID.String(), elapsed.String(),
		before.MappedWith, after.MappedWith, after.MappedWith-before.MappedWith,
		before.MappedEmpty, after.MappedEmpty, after.MappedEmpty-before.MappedEmpty,
		before.ProcessedWith, after.ProcessedWith, after.ProcessedWith-before.ProcessedWith,
		before.TaxonomyCount, after.TaxonomyCount,
		limit, len(rawIDs), okCount, llmCount, stillEmpty,
		after.MappedEmpty, len(rawIDs),
	)
	_ = os.WriteFile(filepath.Join(outDir, "SCORECARD.md"), []byte(md), 0o644)
	log.Printf("done ok=%d llm=%d empty=%d mapped_delta=%+d → %s",
		okCount, llmCount, stillEmpty, after.MappedWith-before.MappedWith, outDir)
}

type coverage struct {
	RawTotal       int `json:"raw_total"`
	MappedWith     int `json:"mapped_with"`
	MappedEmpty    int `json:"mapped_empty"`
	ProcessedWith  int `json:"processed_with"`
	TaxonomyCount  int `json:"taxonomy_count"`
}

func coverageStats(ctx context.Context, pool *pgxpool.Pool, companyID uuid.UUID) coverage {
	var c coverage
	_ = pool.QueryRow(ctx, `
		SELECT COUNT(*),
			COUNT(*) FILTER (
				WHERE COALESCE(NULLIF(trim(mapped_data->>'category'), ''), '') <> ''
				  AND lower(trim(mapped_data->>'category')) <> 'none'
			),
			COUNT(*) FILTER (
				WHERE COALESCE(NULLIF(trim(mapped_data->>'category'), ''), '') = ''
				   OR lower(trim(COALESCE(mapped_data->>'category', ''))) = 'none'
			)
		FROM raw_products WHERE company_id = $1`, companyID).
		Scan(&c.RawTotal, &c.MappedWith, &c.MappedEmpty)
	_ = pool.QueryRow(ctx, `
		SELECT COUNT(*) FILTER (
			WHERE COALESCE(NULLIF(trim(category), ''), '') <> ''
			  AND lower(trim(category)) <> 'none'
		)
		FROM processed_products WHERE company_id = $1`, companyID).Scan(&c.ProcessedWith)
	_ = pool.QueryRow(ctx, `
		SELECT COUNT(*) FROM categories
		WHERE company_id = $1 AND COALESCE(NULLIF(trim(unique_id), ''), '') <> ''`, companyID).
		Scan(&c.TaxonomyCount)
	return c
}

type sampleRow struct {
	RawID   string `json:"raw_id"`
	GTIN    string `json:"gtin"`
	Name    string `json:"name"`
}

func pickEmptyCategoryRawIDs(ctx context.Context, pool *pgxpool.Pool, companyID uuid.UUID, limit int) ([]uuid.UUID, []sampleRow, error) {
	rows, err := pool.Query(ctx, `
		SELECT rp.id, COALESCE(rp.gtin, ''), COALESCE(NULLIF(trim(rp.mapped_data->>'name'), ''), NULLIF(trim(rp.mapped_data->>'title'), ''), '')
		FROM raw_products rp
		LEFT JOIN processed_products pp ON pp.company_id = rp.company_id AND pp.raw_product_id = rp.id
		WHERE rp.company_id = $1
		  AND (
			COALESCE(NULLIF(trim(rp.mapped_data->>'category'), ''), '') = ''
			OR lower(trim(COALESCE(rp.mapped_data->>'category', ''))) = 'none'
		  )
		  AND (
			pp.id IS NULL
			OR COALESCE(NULLIF(trim(pp.category), ''), '') = ''
			OR lower(trim(COALESCE(pp.category, ''))) = 'none'
		  )
		  AND (
			COALESCE(NULLIF(trim(rp.mapped_data->>'name'), ''), '') <> ''
			OR COALESCE(NULLIF(trim(rp.mapped_data->>'title'), ''), '') <> ''
			OR COALESCE(NULLIF(trim(rp.mapped_data->>'description'), ''), '') <> ''
		  )
		ORDER BY rp.updated_at DESC NULLS LAST, rp.id
		LIMIT $2`, companyID, limit)
	if err != nil {
		return nil, nil, err
	}
	defer rows.Close()
	ids := make([]uuid.UUID, 0, limit)
	samples := make([]sampleRow, 0, limit)
	for rows.Next() {
		var id uuid.UUID
		var gtin, name string
		if err := rows.Scan(&id, &gtin, &name); err != nil {
			return nil, nil, err
		}
		ids = append(ids, id)
		samples = append(samples, sampleRow{RawID: id.String(), GTIN: gtin, Name: truncate(name, 80)})
	}
	return ids, samples, rows.Err()
}

type resultRow struct {
	RawID             string `json:"raw_id"`
	GTIN              string `json:"gtin"`
	MappedCategory    string `json:"mapped_category"`
	ProcessedCategory string `json:"processed_category"`
	CategoryName      string `json:"category_name"`
	FieldSource       string `json:"field_source"`
	Name              string `json:"name"`
}

func loadBatchResults(ctx context.Context, pool *pgxpool.Pool, companyID uuid.UUID, rawIDs []uuid.UUID) ([]resultRow, error) {
	rows, err := pool.Query(ctx, `
		SELECT rp.id::text, COALESCE(rp.gtin, ''),
			COALESCE(trim(rp.mapped_data->>'category'), ''),
			COALESCE(trim(pp.category), ''),
			COALESCE(c.name, ''),
			COALESCE(pp.field_sources->>'category', ''),
			COALESCE(NULLIF(trim(pp.processed_name), ''), NULLIF(trim(rp.mapped_data->>'name'), ''), '')
		FROM raw_products rp
		LEFT JOIN processed_products pp ON pp.company_id = rp.company_id AND pp.raw_product_id = rp.id
		LEFT JOIN categories c ON c.company_id = rp.company_id AND c.unique_id = NULLIF(trim(pp.category), '')
		WHERE rp.company_id = $1 AND rp.id = ANY($2::uuid[])
		ORDER BY rp.id`, companyID, rawIDs)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]resultRow, 0, len(rawIDs))
	for rows.Next() {
		var r resultRow
		if err := rows.Scan(&r.RawID, &r.GTIN, &r.MappedCategory, &r.ProcessedCategory, &r.CategoryName, &r.FieldSource, &r.Name); err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

func writeJSON(dir, name string, v any) {
	b, _ := json.MarshalIndent(v, "", "  ")
	_ = os.WriteFile(filepath.Join(dir, name), append(b, '\n'), 0o644)
}

func failJSON(dir, name string, v any) {
	writeJSON(dir, name, v)
}

func truncate(s string, n int) string {
	s = strings.TrimSpace(s)
	if len(s) <= n {
		return s
	}
	return s[:n] + "…"
}