456 lines
16 KiB
Go
456 lines
16 KiB
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, >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] + "…"
|
||
}
|