diff --git a/apps/api/cmd/seed-a1/main.go b/apps/api/cmd/seed-a1/main.go index 39db0cd..b83da9d 100644 --- a/apps/api/cmd/seed-a1/main.go +++ b/apps/api/cmd/seed-a1/main.go @@ -27,6 +27,13 @@ // description + feed attributes live on raw_products.mapped_data and are // exported/imported as-is (skip-processed must not strip them). For a // polluted local DB without a full wipe, use -mode backfill-categories. +// +// Remaining empty mapped categories (~products never in the dump): full or +// categorize processing asks the platform LLM (OverloadedBot / OpenAI) to pick +// a company taxonomy unique_id from Available Categories, then persists to +// processed_products.category and empty mapped_data.category (source=llm). +// Do not wipe the catalog — enqueue a process job for the uncategorized subset +// (admin Products → Process, or API processing_type=categorize / full). // Fixture EANs are purged from non-A1 tenants automatically. // Processed attribute junk (poll clean, DB still has zavora/vzmetenje, …): use // -mode backfill-attributes — same SanitizeProductAttributes + category allowlist diff --git a/apps/api/cmd/tmp-cat-coverage/main.go b/apps/api/cmd/tmp-cat-coverage/main.go new file mode 100644 index 0000000..cb21ccd --- /dev/null +++ b/apps/api/cmd/tmp-cat-coverage/main.go @@ -0,0 +1,70 @@ +package main + +import ( + "context" + "fmt" + "os" + "time" + + "github.com/descrybe/descrybe-v2/apps/api/internal/config" + "github.com/jackc/pgx/v5/pgxpool" +) + +func main() { + config.LoadDotEnv() + url := os.Getenv("DATABASE_URL") + if url == "" { + fmt.Println("NO_DATABASE_URL") + os.Exit(1) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + pg, err := pgxpool.New(ctx, url) + if err != nil { + fmt.Println("pg_err", err) + os.Exit(1) + } + defer pg.Close() + company := "2b3159b0-fc08-415b-b248-35ed02a6baab" + var rawTotal, mappedWith, mappedEmpty, processedWith, llmSource, taxonomy int + _ = pg.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::uuid`, company).Scan(&rawTotal, &mappedWith, &mappedEmpty) + _ = pg.QueryRow(ctx, ` + SELECT COUNT(*) FILTER (WHERE COALESCE(NULLIF(trim(category), ''), '') <> '' AND lower(trim(category)) <> 'none'), + COUNT(*) FILTER (WHERE field_sources->>'category' = 'llm') + FROM processed_products WHERE company_id=$1::uuid`, company).Scan(&processedWith, &llmSource) + _ = pg.QueryRow(ctx, `SELECT COUNT(*) FROM categories WHERE company_id=$1::uuid AND COALESCE(NULLIF(trim(unique_id), ''), '') <> ''`, company).Scan(&taxonomy) + fmt.Printf("company=Platform Demo\nraw_total=%d\nmapped_with=%d\nmapped_empty=%d\nprocessed_with=%d\nllm_source=%d\ntaxonomy=%d\n", + rawTotal, mappedWith, mappedEmpty, processedWith, llmSource, taxonomy) + + rows, err := pg.Query(ctx, ` + SELECT COALESCE(rp.gtin,''), COALESCE(trim(pp.category),''), COALESCE(c.name,''), + COALESCE(NULLIF(trim(pp.processed_name),''), NULLIF(trim(rp.mapped_data->>'name'),''), '') + FROM processed_products pp + JOIN raw_products rp ON rp.id = pp.raw_product_id AND rp.company_id = pp.company_id + LEFT JOIN categories c ON c.company_id = pp.company_id AND c.unique_id = NULLIF(trim(pp.category), '') + WHERE pp.company_id=$1::uuid AND pp.field_sources->>'category'='llm' + ORDER BY pp.updated_at DESC + LIMIT 15`, company) + if err != nil { + fmt.Println("sample_err", err) + return + } + defer rows.Close() + fmt.Println("sample_llm:") + for rows.Next() { + var gtin, uid, name, pname string + _ = rows.Scan(>in, &uid, &name, &pname) + fmt.Printf(" gtin=%s uid=%s name=%s product=%s\n", gtin, uid, name, truncate(pname, 60)) + } +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/apps/api/cmd/tmp-llm-categorize/main.go b/apps/api/cmd/tmp-llm-categorize/main.go new file mode 100644 index 0000000..d4bd3d1 --- /dev/null +++ b/apps/api/cmd/tmp-llm-categorize/main.go @@ -0,0 +1,455 @@ +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, >in, &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] + "…" +} diff --git a/apps/api/cmd/tmp-scan-cats/main.go b/apps/api/cmd/tmp-scan-cats/main.go new file mode 100644 index 0000000..6d581b6 --- /dev/null +++ b/apps/api/cmd/tmp-scan-cats/main.go @@ -0,0 +1,140 @@ +package main + +import ( + "context" + "fmt" + "os" + "strings" + "time" + + "github.com/descrybe/descrybe-v2/apps/api/internal/billing" + "github.com/descrybe/descrybe-v2/apps/api/internal/config" + "github.com/descrybe/descrybe-v2/apps/api/internal/processing" + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +func main() { + config.LoadDotEnv() + dumpPath := strings.TrimSpace(os.Getenv("SEED_A1_MYSQL_DUMP")) + if dumpPath == "" && len(os.Args) > 1 { + dumpPath = os.Args[1] + } + if dumpPath == "" { + dumpPath = processing.ResolveMySQLDumpPath("") + } + legacy := billing.A1LegacyCompanyID + companyID := uuid.MustParse("604f23a8-b66e-4b21-8b45-0d72b68f4790") + + fmt.Printf("dump=%s\n", dumpPath) + fmt.Printf("legacy_company=%s\n", legacy) + + if dumpPath != "" { + f, err := os.Open(dumpPath) + if err != nil { + fmt.Printf("dump_open_err=%v\n", err) + } else { + defer f.Close() + byGTIN, err := processing.ScanA1ProcessedCategories(f, legacy) + if err != nil { + fmt.Printf("scan_err=%v\n", err) + } else { + withCat := 0 + for _, c := range byGTIN { + if strings.TrimSpace(c) != "" && !strings.EqualFold(c, "NULL") && !strings.EqualFold(c, "none") { + withCat++ + } + } + fmt.Printf("dump_pairs=%d with_nonempty_cat=%d\n", len(byGTIN), withCat) + } + } + } else { + fmt.Println("dump=NONE") + } + + pgURL := strings.TrimSpace(os.Getenv("DATABASE_URL")) + if pgURL == "" { + fmt.Println("postgres=NO_DATABASE_URL") + return + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + pg, err := pgxpool.New(ctx, pgURL) + if err != nil { + fmt.Printf("postgres_err=%v\n", err) + return + } + defer pg.Close() + + var legacyDB, name string + _ = pg.QueryRow(ctx, `SELECT COALESCE(legacy_company_id::text,''), name FROM companies WHERE id=$1`, companyID).Scan(&legacyDB, &name) + fmt.Printf("pg_company name=%q legacy=%s\n", name, legacyDB) + + var rawTotal, rawWith, rawWithout int + err = pg.QueryRow(ctx, ` + SELECT COUNT(*), + COUNT(*) FILTER (WHERE COALESCE(NULLIF(trim(mapped_data->>'category'), ''), '') <> ''), + COUNT(*) FILTER (WHERE COALESCE(NULLIF(trim(mapped_data->>'category'), ''), '') = '') + FROM raw_products WHERE company_id=$1`, companyID).Scan(&rawTotal, &rawWith, &rawWithout) + if err != nil { + fmt.Printf("raw_count_err=%v\n", err) + } else { + pct := 0.0 + if rawTotal > 0 { + pct = 100.0 * float64(rawWith) / float64(rawTotal) + } + fmt.Printf("pg_raw total=%d with_cat=%d without_cat=%d pct_with=%.1f\n", rawTotal, rawWith, rawWithout, pct) + } + + var ppTotal, ppWith, ppWithout int + err = pg.QueryRow(ctx, ` + SELECT COUNT(*), + COUNT(*) FILTER (WHERE COALESCE(NULLIF(trim(category), ''), '') <> '' AND lower(trim(category)) <> 'none'), + COUNT(*) FILTER (WHERE COALESCE(NULLIF(trim(category), ''), '') = '' OR lower(trim(category)) = 'none') + FROM processed_products WHERE company_id=$1`, companyID).Scan(&ppTotal, &ppWith, &ppWithout) + if err != nil { + fmt.Printf("pp_count_err=%v\n", err) + } else { + fmt.Printf("pg_processed total=%d with_cat=%d without_cat=%d\n", ppTotal, ppWith, ppWithout) + } + + // Dry-run overlap: how many dump GTINs would fill empty mapped categories + if dumpPath != "" { + f2, err := os.Open(dumpPath) + if err == nil { + byGTIN, err := processing.ScanA1ProcessedCategories(f2, legacy) + _ = f2.Close() + if err == nil && len(byGTIN) > 0 { + gtins := make([]string, 0, len(byGTIN)) + cats := make([]string, 0, len(byGTIN)) + for g, c := range byGTIN { + gtins = append(gtins, g) + cats = append(cats, c) + } + var wouldUpdate, alreadyOk, noRawMatch int + err = pg.QueryRow(ctx, ` + WITH dump(gtin, category) AS ( + SELECT * FROM unnest($2::text[], $3::text[]) + ), + joined AS ( + SELECT r.gtin, + COALESCE(NULLIF(trim(r.mapped_data->>'category'), ''), '') AS cur, + COALESCE(NULLIF(trim(d.category), ''), '') AS want + FROM dump d + LEFT JOIN raw_products r ON r.company_id=$1 AND r.gtin=d.gtin + ) + SELECT + COUNT(*) FILTER (WHERE gtin IS NOT NULL AND want <> '' AND (cur = '' OR cur IS DISTINCT FROM want)), + COUNT(*) FILTER (WHERE gtin IS NOT NULL AND want <> '' AND cur <> '' AND cur = want), + COUNT(*) FILTER (WHERE gtin IS NULL) + FROM joined + `, companyID, gtins, cats).Scan(&wouldUpdate, &alreadyOk, &noRawMatch) + if err != nil { + fmt.Printf("dry_run_err=%v\n", err) + } else { + fmt.Printf("dry_run would_update_mapped=%d already_ok=%d dump_gtin_no_raw_row=%d\n", wouldUpdate, alreadyOk, noRawMatch) + } + } + } + } +} diff --git a/apps/api/internal/aiprompts/kinds.go b/apps/api/internal/aiprompts/kinds.go index 8a7ee13..92d2980 100644 --- a/apps/api/internal/aiprompts/kinds.go +++ b/apps/api/internal/aiprompts/kinds.go @@ -78,7 +78,7 @@ var BuiltInDefaults = []DefaultTemplate{ { Key: KeyProductEnhance, Label: "Product title & description", - Description: "Used when processing products (AI enhance step).", + Description: "Used when processing products (AI enhance step). Categorize (taxonomy pick when category is empty) is a separate pipeline step before this.", SystemTemplate: `Retail product copywriter. Rules: - Reply with ONLY JSON (no markdown) diff --git a/apps/api/internal/processing/ai.go b/apps/api/internal/processing/ai.go index fa72b05..789748d 100644 --- a/apps/api/internal/processing/ai.go +++ b/apps/api/internal/processing/ai.go @@ -9,19 +9,23 @@ import ( // Pipeline step names (canonical order for "full"). const ( - StepNormalize = "normalize" - StepParseSpecs = "parse_specs" - StepFillFields = "fill_fields" - StepEPREL = "eprel" - StepAIEnhance = "ai_enhance" + StepNormalize = "normalize" + StepParseSpecs = "parse_specs" + StepFillFields = "fill_fields" + StepEPREL = "eprel" + StepCategorize = "categorize" + StepAIEnhance = "ai_enhance" ) // CanonicalSteps is the default full pipeline order. +// categorize runs before ai_enhance so category formulas / overlays key correctly +// (legacy Descrybe: GPT picks a taxonomy unique_id when mapped category is empty). var CanonicalSteps = []string{ StepNormalize, StepParseSpecs, StepFillFields, StepEPREL, + StepCategorize, StepAIEnhance, } diff --git a/apps/api/internal/processing/categorize_ai.go b/apps/api/internal/processing/categorize_ai.go new file mode 100644 index 0000000..078e475 --- /dev/null +++ b/apps/api/internal/processing/categorize_ai.go @@ -0,0 +1,280 @@ +package processing + +import ( + "context" + "log" + "sort" + "strings" +) + +// MaxCategorizeOptions caps the taxonomy list injected into the categorize LLM +// prompt (token/cost bound). Prefer stable sort by display name then unique_id. +const MaxCategorizeOptions = 200 + +// MaxTokensCategorize budgets a short JSON reply {"categoryId","confidence"}. +// code-fast / reasoning-style models often burn internal tokens before JSON; +// 256 frequently finishes with empty content and forces a length-cap retry. +const MaxTokensCategorize = 4096 + +// categoryOption is one company taxonomy row for the categorize prompt. +type categoryOption struct { + UniqueID string + Name string +} + +// categoryOptionsFromNames builds a stable, capped list from unique_id → name. +// Empty map → no AI categorize (nothing valid to choose). +func categoryOptionsFromNames(namesByUID map[string]string) []categoryOption { + if len(namesByUID) == 0 { + return nil + } + out := make([]categoryOption, 0, len(namesByUID)) + for uid, name := range namesByUID { + uid = strings.TrimSpace(uid) + if uid == "" || strings.EqualFold(uid, "none") { + continue + } + name = strings.TrimSpace(name) + if name == "" { + name = uid + } + out = append(out, categoryOption{UniqueID: uid, Name: name}) + } + sort.Slice(out, func(i, j int) bool { + if out[i].Name != out[j].Name { + return strings.ToLower(out[i].Name) < strings.ToLower(out[j].Name) + } + return out[i].UniqueID < out[j].UniqueID + }) + if len(out) > MaxCategorizeOptions { + out = out[:MaxCategorizeOptions] + } + return out +} + +// categoryUniqueIDsList returns sorted unique_ids for vector SuggestCategory candidates. +func categoryUniqueIDsList(namesByUID map[string]string) []string { + opts := categoryOptionsFromNames(namesByUID) + if len(opts) == 0 { + return nil + } + out := make([]string, len(opts)) + for i, o := range opts { + out[i] = o.UniqueID + } + return out +} + +// formatAvailableCategoriesList mirrors legacy Descrybe "Available Categories:" +// bullets: "Name (ID: unique_id)". +func formatAvailableCategoriesList(opts []categoryOption) string { + if len(opts) == 0 { + return "" + } + var b strings.Builder + for _, o := range opts { + b.WriteString("- ") + b.WriteString(SanitizeText(o.Name)) + b.WriteString(" (ID: ") + b.WriteString(SanitizeText(o.UniqueID)) + b.WriteString(")\n") + } + return strings.TrimSpace(b.String()) +} + +// ProductCategorizeSystem is the built-in system prompt for taxonomy selection. +const ProductCategorizeSystem = `Product categorization expert. +Rules: +- Reply with ONLY a single JSON object (no markdown, no prose, no reasoning) +- Schema: {"categoryId":"string","confidence":0.0} +- categoryId MUST be one of the Available Categories IDs exactly (the value in parentheses after ID:) +- Never invent IDs or names; inventing IDs fails categorization +- Prefer the most specific category that matches the product +Example: +{"categoryId":"50","confidence":0.9}` + +// ProductCategorizeUser builds the user prompt with product context + taxonomy list. +func ProductCategorizeUser(name, description string, opts []categoryOption) string { + var b strings.Builder + b.WriteString("Select the best category for this product from Available Categories only.\n\n") + b.WriteString("Product Information:\n") + b.WriteString("Name: ") + b.WriteString(SanitizeText(truncateRunes(name, 200))) + b.WriteString("\nDesc: ") + b.WriteString(SanitizeText(truncateRunes(description, MaxProductDescRunes))) + b.WriteString("\n\nAvailable Categories:\n") + b.WriteString(formatAvailableCategoriesList(opts)) + b.WriteString("\n\nImportant:\n") + b.WriteString("- Return categoryId as the exact ID from the list\n") + b.WriteString("- Do not invent categories outside the list\n") + return b.String() +} + +// tryAICategorize asks the Completer to pick a company taxonomy unique_id when +// Category is still empty after mapped/prior/vector. Never invents outside taxonomy: +// responses are coerced then filtered to namesByUID / valid unique_ids. +func tryAICategorize(ctx context.Context, e *Engine, out *StepResult, in ProductInput, policy StepPolicy) { + if out == nil || strings.TrimSpace(out.Category) != "" { + return + } + if !policy.AllowAI { + return + } + if e == nil || !e.CompleterEnabled() { + return + } + opts := categoryOptionsFromNames(in.CategoryNamesByUID) + if len(opts) == 0 { + out.Notes = append(out.Notes, "ai_categorize: skipped (no company taxonomy)") + appendStepLog(out.GPTResponse, StepCategorize, map[string]any{ + "status": "skipped", + "reason": "no_taxonomy", + }) + return + } + name := strings.TrimSpace(out.Name) + if name == "" { + name = strings.TrimSpace(out.ProcessedName) + } + desc := strings.TrimSpace(out.Description) + if desc == "" { + desc = strings.TrimSpace(out.ProcessedDescription) + } + if name == "" && desc == "" { + out.Notes = append(out.Notes, "ai_categorize: skipped (empty product text)") + appendStepLog(out.GPTResponse, StepCategorize, map[string]any{ + "status": "skipped", + "reason": "empty_product", + }) + return + } + + system := ProductCategorizeSystem + user := ProductCategorizeUser(name, desc, opts) + comp, obj, err := CompleteJSON(ctx, e.Completer, system, user, CompleteOptions{ + MaxTokens: MaxTokensCategorize, + Temperature: DefaultStructuredTemp, + }) + out.TotalTokens += comp.TotalTokens + if err != nil { + out.Notes = append(out.Notes, "ai_categorize: "+TruncateError(err)) + appendStepLog(out.GPTResponse, StepCategorize, map[string]any{ + "status": "failed", + "error": TruncateError(err), + "tokens": comp.TotalTokens, + }) + return + } + + raw := categoryIDFromCategorizeJSON(obj) + valid := make(map[string]struct{}, len(opts)) + for _, o := range opts { + valid[o.UniqueID] = struct{}{} + } + resolved := resolveCompanyCategoryUniqueID(raw, in.CategoryNamesByUID, valid) + if resolved == "" || isUnusableCategoryValue(resolved, out.ProcessedName, out.Name) { + out.Notes = append(out.Notes, "ai_categorize: ignored (not in company taxonomy)") + appendStepLog(out.GPTResponse, StepCategorize, map[string]any{ + "status": "rejected", + "returned": raw, + "tokens": comp.TotalTokens, + }) + return + } + + out.Category = SanitizeText(resolved) + if out.FieldSources == nil { + out.FieldSources = map[string]any{} + } + out.FieldSources["category"] = "llm" + syncCategoryName(out, in.CategoryNamesByUID) + out.Notes = append(out.Notes, "category: llm") + log.Printf("processing: category choice uid=%s name=%s source=llm", out.Category, out.CategoryName) + appendStepLog(out.GPTResponse, StepCategorize, map[string]any{ + "status": "ok", + "category": out.Category, + "name": out.CategoryName, + "source": "llm", + "tokens": comp.TotalTokens, + "options": len(opts), + }) +} + +func categoryIDFromCategorizeJSON(obj map[string]any) string { + if obj == nil { + return "" + } + for _, k := range []string{"categoryId", "category_id", "categoryUniqueId", "category_unique_id", "unique_id", "id", "category"} { + if s := categoryUniqueIDFromAny(obj[k]); s != "" { + return s + } + } + return "" +} + +// firstAvailableCategoryIDFromPrompt extracts the first "(ID: …)" token from a +// categorize user prompt (HeuristicCompleter / weak-model fallback). +func firstAvailableCategoryIDFromPrompt(user string) string { + const marker = "(id:" + lower := strings.ToLower(user) + at := strings.Index(lower, marker) + if at < 0 { + return "" + } + rest := strings.TrimSpace(user[at+len(marker):]) + end := strings.Index(rest, ")") + if end <= 0 { + return "" + } + return strings.TrimSpace(rest[:end]) +} + +// runCategorizeStep applies vector then LLM taxonomy selection when Category empty. +func runCategorizeStep(ctx context.Context, e *Engine, companyID string, out *StepResult, in ProductInput, categoryNames []string, policy StepPolicy) { + if out == nil { + return + } + if strings.TrimSpace(out.Category) != "" { + appendStepLog(out.GPTResponse, StepCategorize, map[string]any{ + "status": "skipped", + "reason": "already_set", + "category": out.Category, + }) + return + } + if !policy.AllowAI { + out.Notes = append(out.Notes, "categorize: skipped (AI not allowed)") + appendStepLog(out.GPTResponse, StepCategorize, map[string]any{ + "status": "skipped", + "reason": "entitlement_can_use_ai", + }) + return + } + tryVectorCategorize(ctx, e, companyID, out, categoryNames, policy) + if strings.TrimSpace(out.Category) != "" { + return + } + tryAICategorize(ctx, e, out, in, policy) + if strings.TrimSpace(out.Category) == "" { + appendStepLog(out.GPTResponse, StepCategorize, map[string]any{ + "status": "unset", + "reason": "no_vector_or_ai_match", + }) + } +} + +// persistMappedCategorySQL writes a taxonomy unique_id onto mapped_data.category +// when the mapped value is still empty (so Products UI + reprocess stick). +const persistMappedCategorySQL = ` + UPDATE raw_products + SET mapped_data = jsonb_set( + COALESCE(mapped_data, '{}'::jsonb), + '{category}', + to_jsonb($3::text), + true + ), + updated_at = now() + WHERE id = $1 AND company_id = $2 + AND COALESCE(NULLIF(trim(mapped_data->>'category'), ''), '') = '' + AND COALESCE(NULLIF(trim($3), ''), '') <> '' + AND lower(trim($3)) <> 'none'` diff --git a/apps/api/internal/processing/categorize_ai_test.go b/apps/api/internal/processing/categorize_ai_test.go new file mode 100644 index 0000000..1347139 --- /dev/null +++ b/apps/api/internal/processing/categorize_ai_test.go @@ -0,0 +1,174 @@ +package processing + +import ( + "context" + "strings" + "testing" +) + +func TestTryAICategorize_picksTaxonomyUniqueID(t *testing.T) { + e := &Engine{ + Completer: stubCompleter{fn: func(system, user string) (Completion, error) { + if strings.Contains(strings.ToLower(system), "categoryid") { + if !strings.Contains(user, "Available Categories:") { + t.Fatalf("expected Available Categories in user prompt") + } + if !strings.Contains(user, "(ID: 50)") { + t.Fatalf("expected ID 50 in list: %s", user) + } + return Completion{ + Text: `{"categoryId":"50","confidence":0.91}`, + TotalTokens: 12, + PromptTokens: 8, + OutputTokens: 4, + }, nil + } + return Completion{Text: `{"name":"Gorenje Cooker","description":"Freestanding cooker for the kitchen."}`, TotalTokens: 5}, nil + }}, + Vector: NoopVectorCategorizer{}, + } + out, err := e.RunSteps(context.Background(), "co", ProductInput{ + Name: "Gorenje stove", + Description: "Freestanding cooker", + CategoryNamesByUID: map[string]string{ + "28": "TV", + "50": "Štedilniki", + }, + }, "full", nil, StepPolicy{AllowAI: true, AllowEPREL: true}) + if err != nil { + t.Fatal(err) + } + if out.Category != "50" { + t.Fatalf("Category=%q want 50", out.Category) + } + if out.CategoryName != "Štedilniki" { + t.Fatalf("CategoryName=%q want Štedilniki", out.CategoryName) + } + if src, _ := out.FieldSources["category"].(string); src != "llm" { + t.Fatalf("field_sources.category=%v want llm", out.FieldSources["category"]) + } + if out.TotalTokens < 12 { + t.Fatalf("TotalTokens=%d want >=12 from categorize", out.TotalTokens) + } +} + +func TestTryAICategorize_rejectsInventedID(t *testing.T) { + e := &Engine{ + Completer: stubCompleter{fn: func(system, user string) (Completion, error) { + if strings.Contains(strings.ToLower(system), "categoryid") { + return Completion{Text: `{"categoryId":"99999","confidence":0.9}`, TotalTokens: 3}, nil + } + return Completion{Text: `{"name":"X","description":"Y"}`, TotalTokens: 5}, nil + }}, + Vector: NoopVectorCategorizer{}, + } + out, err := e.RunSteps(context.Background(), "co", ProductInput{ + Name: "Widget", + CategoryNamesByUID: map[string]string{ + "50": "Štedilniki", + }, + }, "full", nil, StepPolicy{AllowAI: true}) + if err != nil { + t.Fatal(err) + } + if out.Category != "" { + t.Fatalf("Category=%q want empty (invented id rejected)", out.Category) + } + found := false + for _, n := range out.Notes { + if strings.Contains(n, "ai_categorize: ignored") { + found = true + break + } + } + if !found { + t.Fatalf("expected reject note, got %v", out.Notes) + } +} + +func TestTryAICategorize_resolvesDisplayName(t *testing.T) { + e := &Engine{ + Completer: stubCompleter{fn: func(system, user string) (Completion, error) { + if strings.Contains(strings.ToLower(system), "categoryid") { + return Completion{Text: `{"categoryId":"Štedilniki","confidence":0.8}`, TotalTokens: 3}, nil + } + return Completion{Text: `{"name":"X","description":"Y"}`, TotalTokens: 5}, nil + }}, + Vector: NoopVectorCategorizer{}, + } + out, err := e.RunSteps(context.Background(), "co", ProductInput{ + Name: "Cooker", + CategoryNamesByUID: map[string]string{ + "50": "Štedilniki", + }, + }, "full", nil, StepPolicy{AllowAI: true}) + if err != nil { + t.Fatal(err) + } + if out.Category != "50" { + t.Fatalf("Category=%q want 50 (name coerced)", out.Category) + } +} + +func TestTryAICategorize_skippedWithoutTaxonomy(t *testing.T) { + calls := 0 + e := &Engine{ + Completer: stubCompleter{fn: func(system, user string) (Completion, error) { + calls++ + return Completion{Text: `{"name":"A","description":"B"}`, TotalTokens: 2}, nil + }}, + Vector: NoopVectorCategorizer{}, + } + out, err := e.RunSteps(context.Background(), "co", ProductInput{ + Name: "Lone product", + }, "full", nil, StepPolicy{AllowAI: true}) + if err != nil { + t.Fatal(err) + } + if out.Category != "" { + t.Fatalf("Category=%q want empty", out.Category) + } + // Only enhance should call Completer (categorize skips with no taxonomy). + if calls != 1 { + t.Fatalf("completer calls=%d want 1 (enhance only)", calls) + } +} + +func TestHeuristicCompleter_categorizeReturnsJSON(t *testing.T) { + h := HeuristicCompleter{} + user := ProductCategorizeUser("Stove", "Cooker", []categoryOption{ + {UniqueID: "50", Name: "Štedilniki"}, + {UniqueID: "28", Name: "TV"}, + }) + comp, err := h.Complete(context.Background(), ProductCategorizeSystem, user) + if err != nil { + t.Fatal(err) + } + obj, err := ParseJSONObject(comp.Text) + if err != nil { + t.Fatal(err) + } + got := categoryIDFromCategorizeJSON(obj) + if got != "50" && got != "28" { + t.Fatalf("categoryId=%q want 50 or 28 from Available Categories", got) + } + opts := categoryOptionsFromNames(map[string]string{"50": "Štedilniki", "28": "TV"}) + found := false + for _, o := range opts { + if o.UniqueID == got { + found = true + break + } + } + if !found { + t.Fatalf("categoryId=%q not in options", got) + } +} + +func TestCategoryOptionsFromNames_sortAndCap(t *testing.T) { + names := map[string]string{"2": "Beta", "1": "Alpha", "3": "Gamma"} + opts := categoryOptionsFromNames(names) + if len(opts) != 3 || opts[0].UniqueID != "1" || opts[1].UniqueID != "2" { + t.Fatalf("opts=%v", opts) + } +} diff --git a/apps/api/internal/processing/openai.go b/apps/api/internal/processing/openai.go index 03bfaee..e72648d 100644 --- a/apps/api/internal/processing/openai.go +++ b/apps/api/internal/processing/openai.go @@ -644,6 +644,14 @@ func (h HeuristicCompleter) Complete(_ context.Context, system, user string) (Co user = SanitizeText(user) text := "General" switch { + case strings.Contains(systemL, "categoryid") || (strings.Contains(systemL, "categor") && strings.Contains(systemL, "available categories")): + // Taxonomy pick: return first Available Categories (ID: …) from the user prompt. + id := firstAvailableCategoryIDFromPrompt(user) + if id == "" { + id = "unknown" + } + b, _ := json.Marshal(map[string]any{"categoryId": id, "confidence": 0.5}) + text = string(b) case strings.Contains(systemL, `"name"`) || strings.Contains(systemL, "titles and descriptions"): // Prefer explicit Name:/Desc: (ProductEnhanceUser) or Current name: labels. // Never use firstLine(user) alone — that line is often "Category: …". @@ -659,8 +667,6 @@ func (h HeuristicCompleter) Complete(_ context.Context, system, user string) (Co text = string(b) case strings.Contains(systemL, "attributes") && strings.Contains(systemL, "json"): text = `{"material":"unknown","brand":"unknown"}` - case strings.Contains(systemL, "categor"): - text = "General" default: // Never echo ProductEnhanceUser's first line ("Category: …") as title/output. text = labeledPromptValue(user, "name:", "current name:") diff --git a/apps/api/internal/processing/pipeline.go b/apps/api/internal/processing/pipeline.go index 9e8c3ed..8f4f03b 100644 --- a/apps/api/internal/processing/pipeline.go +++ b/apps/api/internal/processing/pipeline.go @@ -1564,7 +1564,8 @@ func (p *Pipeline) processOne(ctx context.Context, companyID, jobID uuid.UUID, i engine = &Engine{Vector: NoopVectorCategorizer{}} } policy := cache.stepPolicy() - result, err := engine.RunSteps(ctx, companyID.String(), in, processingType, nil, policy) + catNames := categoryUniqueIDsList(categoryNamesByUID) + result, err := engine.RunSteps(ctx, companyID.String(), in, processingType, catNames, policy) if err != nil { return false, 0, StepResult{}, err } @@ -1644,6 +1645,13 @@ func (p *Pipeline) processOne(ctx context.Context, companyID, jobID uuid.UUID, i WHERE id = $1 AND company_id = $2`, it.RawID, companyID); err != nil { return false, 0, result, err } + // Stick taxonomy unique_id onto mapped_data.category when still empty so the + // Products UI (reads mapped) and reprocess keep the AI/vector/mapped pick. + if cat := strings.TrimSpace(result.Category); cat != "" { + if _, err := tx.Exec(ctx, persistMappedCategorySQL, it.RawID, companyID, cat); err != nil { + return false, 0, result, fmt.Errorf("persist mapped category: %w", err) + } + } if err := tx.Commit(ctx); err != nil { return false, 0, result, err } diff --git a/apps/api/internal/processing/steps.go b/apps/api/internal/processing/steps.go index af96225..cdecef3 100644 --- a/apps/api/internal/processing/steps.go +++ b/apps/api/internal/processing/steps.go @@ -207,6 +207,10 @@ func (e *Engine) RunSteps(ctx context.Context, companyID string, in ProductInput out.FieldSources["eprel"] = "eprel_api" appendStepLog(out.GPTResponse, StepEPREL, map[string]any{"status": "ok", "eprel_id": data.ID}) + case StepCategorize: + // Vector (if enabled) then LLM taxonomy pick when category still empty. + runCategorizeStep(ctx, e, companyID, &out, in, categoryNames, policy) + case StepAIEnhance: preservePriorEnhanceHash := func() { if in.PriorEnhanceHash != "" { @@ -501,11 +505,14 @@ func (e *Engine) RunSteps(ctx context.Context, companyID string, in ProductInput } } - // Paths without fill_fields (enhance_only / normalize_only) still get vector - // categorize when AllowAI + embeddings are available. - tryVectorCategorize(ctx, e, companyID, &out, categoryNames, policy) + // Pipelines without StepCategorize (enhance_only / normalize_only) still try + // vector categorize when AllowAI + embeddings are available. Full/categorize + // already ran runCategorizeStep (vector then LLM) inside the loop. + if !stepsContain(steps, StepCategorize) { + tryVectorCategorize(ctx, e, companyID, &out, categoryNames, policy) + noteMissingCategory(&out, policy, e != nil && e.Vector != nil && e.Vector.Enabled()) + } preserveCategoryIfEmpty(&out, in.PriorCategory) - noteMissingCategory(&out, policy, e != nil && e.Vector != nil && e.Vector.Enabled()) syncCategoryName(&out, in.CategoryNamesByUID) out.Name = preferredProductTitle(in.GTIN, out.Name, out.ProcessedName, in.Name, in.PriorProcessedName) out.ProcessedName = preferredProductTitle(in.GTIN, out.ProcessedName, out.Name, in.PriorProcessedName, in.Name) @@ -640,12 +647,14 @@ func tryVectorCategorize(ctx context.Context, e *Engine, companyID string, out * } // noteMissingCategory records why Category stayed empty (mapped absent; vector skipped or failed). +// Used for pipelines that skip StepCategorize (vector-only post-pass). func noteMissingCategory(out *StepResult, policy StepPolicy, vectorEnabled bool) { if out == nil || strings.TrimSpace(out.Category) != "" { return } for _, n := range out.Notes { - if strings.HasPrefix(n, "category: unset") || strings.HasPrefix(n, "category: vector") { + if strings.HasPrefix(n, "category: unset") || strings.HasPrefix(n, "category: vector") || + strings.HasPrefix(n, "category: llm") || strings.HasPrefix(n, "ai_categorize:") { return } } @@ -670,14 +679,25 @@ func resolveSteps(processingType string) []string { return []string{StepNormalize, StepParseSpecs, StepEPREL} case "normalize_only": return []string{StepNormalize} - case "categorize", "categorize_only", "categorize_enhance": - // Legacy aliases → full deterministic + optional AI + case "categorize", "categorize_only": + // Taxonomy assign only (vector then LLM) — no title/description rewrite. + return []string{StepNormalize, StepParseSpecs, StepFillFields, StepEPREL, StepCategorize} + case "categorize_enhance": return append([]string{}, CanonicalSteps...) default: // full return append([]string{}, CanonicalSteps...) } } +func stepsContain(steps []string, want string) bool { + for _, s := range steps { + if s == want { + return true + } + } + return false +} + // InitialStepProgress builds pending step_progress rows for a job. func InitialStepProgress(processingType string) []StepProgress { steps := resolveSteps(processingType) diff --git a/apps/api/internal/processing/steps_test.go b/apps/api/internal/processing/steps_test.go index 4c5587e..b8e18f4 100644 --- a/apps/api/internal/processing/steps_test.go +++ b/apps/api/internal/processing/steps_test.go @@ -19,11 +19,14 @@ func (s stubCompleter) Complete(_ context.Context, system, user string) (Complet func TestResolveSteps(t *testing.T) { cases := map[string][]string{ - "full": {StepNormalize, StepParseSpecs, StepFillFields, StepEPREL, StepAIEnhance}, + "full": {StepNormalize, StepParseSpecs, StepFillFields, StepEPREL, StepCategorize, StepAIEnhance}, "enhance_only": {StepNormalize, StepAIEnhance}, "attributes_only": {StepNormalize, StepParseSpecs, StepFillFields}, "eprel_only": {StepNormalize, StepParseSpecs, StepEPREL}, "normalize_only": {StepNormalize}, + "categorize": {StepNormalize, StepParseSpecs, StepFillFields, StepEPREL, StepCategorize}, + "categorize_only": {StepNormalize, StepParseSpecs, StepFillFields, StepEPREL, StepCategorize}, + "categorize_enhance": {StepNormalize, StepParseSpecs, StepFillFields, StepEPREL, StepCategorize, StepAIEnhance}, } for in, want := range cases { got := resolveSteps(in) diff --git a/docs/features.md b/docs/features.md index e4f4c24..bb484c9 100644 --- a/docs/features.md +++ b/docs/features.md @@ -49,9 +49,9 @@ Legend: `[x]` done · `[~]` partial/stub · `[ ]` missing ## Phase D — AI processing -- [~] Pipeline workers: categorize → attributes → enhance (**stub** marks processed) +- [x] Pipeline workers: categorize (vector + LLM taxonomy) → attributes → enhance - [~] Credits, rate limits, cancel/retry (credits + cancel yes; rate limits/retry incomplete) -- [ ] OpenAI / Pinecone behind interfaces +- [~] OpenAI / Pinecone behind interfaces (OpenAI + optional Pinecone; LLM categorize when category missing) - [~] Job controls UI (list/start/cancel basic; no admin visibility) ## Phase E — WooCommerce + admin / cutover diff --git a/docs/live-pipeline.md b/docs/live-pipeline.md index 94e30bb..9613ce5 100644 --- a/docs/live-pipeline.md +++ b/docs/live-pipeline.md @@ -9,7 +9,7 @@ ## Pipeline (canonical `full`) ``` -normalize -> parse_specs -> fill_fields -> eprel -> ai_enhance +normalize -> parse_specs -> fill_fields -> eprel -> categorize -> ai_enhance ``` | Step | Role | @@ -18,7 +18,8 @@ normalize -> parse_specs -> fill_fields -> eprel -> ai_enhance | `parse_specs` | Parse `specifications` into attributes | | `fill_fields` | Fill missing scalars / standard fields | | `eprel` | Fetch EU energy label when `eprel_id` present + enricher on | -| `ai_enhance` | Green Chat rewrite of name + description | +| `categorize` | When mapped/prior category is empty: Pinecone vector (if configured), else LLM picks a **company taxonomy `unique_id`** from Available Categories (never invents). Persists to `processed_products.category` and empty `mapped_data.category`. | +| `ai_enhance` | Green Chat rewrite of name + description (uses category formulas/overlays when set) | ## Prerequisites