1242 lines
37 KiB
Go
1242 lines
37 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/descrybe/descrybe-v2/apps/api/internal/processing"
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
func migrateCatalogAndFeeds(
|
|
ctx context.Context,
|
|
mysqlDB *sql.DB,
|
|
pg *pgxpool.Pool,
|
|
companyMap, userMap map[string]string,
|
|
allow map[string]bool,
|
|
domains domainSet,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) (categoryMap, attributeMap, feedMap, rawMap, fileMap map[string]string) {
|
|
categoryMap = map[string]string{}
|
|
attributeMap = map[string]string{}
|
|
feedMap = map[string]string{}
|
|
rawMap = map[string]string{}
|
|
fileMap = map[string]string{}
|
|
|
|
// Feeds before products so feed_id can be remapped when possible.
|
|
if domains.has("feeds") {
|
|
migrateInputFeeds(ctx, mysqlDB, pg, companyMap, feedMap, allow, report, dryRun)
|
|
}
|
|
if domains.has("files") {
|
|
migrateFiles(ctx, mysqlDB, pg, companyMap, userMap, fileMap, allow, report, dryRun)
|
|
}
|
|
if domains.has("catalog") {
|
|
migrateCategories(ctx, mysqlDB, pg, companyMap, categoryMap, allow, report, dryRun)
|
|
migrateAttributes(ctx, mysqlDB, pg, companyMap, attributeMap, allow, report, dryRun)
|
|
migrateCategoryAttributes(ctx, mysqlDB, pg, companyMap, attributeMap, allow, report, dryRun)
|
|
migrateCustomVariables(ctx, mysqlDB, pg, companyMap, allow, report, dryRun)
|
|
}
|
|
if domains.has("products") {
|
|
migrateRawProducts(ctx, mysqlDB, pg, companyMap, feedMap, rawMap, allow, report, dryRun)
|
|
migrateProcessedProducts(ctx, mysqlDB, pg, companyMap, feedMap, rawMap, allow, report, dryRun)
|
|
backfillProcessedDescriptionsFromMapped(ctx, pg, companyMap, allow, report, dryRun)
|
|
backfillProcessedNamesFromMapped(ctx, pg, companyMap, allow, report, dryRun)
|
|
backfillProcessedCategoriesFromMapped(ctx, pg, companyMap, allow, report, dryRun)
|
|
}
|
|
if domains.has("feeds") {
|
|
migrateExportFeeds(ctx, mysqlDB, pg, companyMap, feedMap, allow, report, dryRun)
|
|
}
|
|
|
|
return categoryMap, attributeMap, feedMap, rawMap, fileMap
|
|
}
|
|
|
|
func migrateCategories(
|
|
ctx context.Context,
|
|
mysqlDB *sql.DB,
|
|
pg *pgxpool.Pool,
|
|
companyMap, categoryMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if !mysqlTableExists(ctx, mysqlDB, "categories") {
|
|
log.Printf("categories skipped: table missing")
|
|
return
|
|
}
|
|
q := mysqlSelectList(
|
|
"id",
|
|
"company_id",
|
|
"name",
|
|
"unique_id",
|
|
mysqlCol(ctx, mysqlDB, "categories", "parent_id", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "categories", "path", "NULL"),
|
|
mysqlCoalesce(ctx, mysqlDB, "categories", "level", "0"),
|
|
mysqlCoalesce(ctx, mysqlDB, "categories", "position", "0"),
|
|
mysqlCoalesce(ctx, mysqlDB, "categories", "is_active", "1"),
|
|
mysqlCol(ctx, mysqlDB, "categories", "description", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "categories", "prompt", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "categories", "metadata", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "categories", "config", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "categories", "title_template", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "categories", "description_template", "NULL"),
|
|
) + " FROM categories WHERE 1=1"
|
|
clause, cargs := mysqlCompanyFilter("company_id", allow)
|
|
q += clause
|
|
rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
|
|
if err != nil {
|
|
rows, err = mysqlDB.QueryContext(ctx, `
|
|
SELECT id, company_id, name, unique_id, NULL, NULL, 0, 0, 1,
|
|
NULL, NULL, NULL, NULL, NULL, NULL
|
|
FROM categories WHERE 1=1`+clause, cargs...)
|
|
}
|
|
if err != nil {
|
|
log.Printf("categories skipped: %v", err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var legacyID int64
|
|
var companyLegacy, name, uniqueID string
|
|
var parentID, path, desc, prompt sql.NullString
|
|
var level, position, isActive int
|
|
var metadata, config, titleTpl, descTpl []byte
|
|
if err := rows.Scan(&legacyID, &companyLegacy, &name, &uniqueID, &parentID, &path,
|
|
&level, &position, &isActive, &desc, &prompt, &metadata, &config, &titleTpl, &descTpl); err != nil {
|
|
log.Printf("category scan: %v", err)
|
|
report["categories_skipped"]++
|
|
continue
|
|
}
|
|
cid, ok := companyMap[companyLegacy]
|
|
if !ok {
|
|
report["categories_skipped"]++
|
|
continue
|
|
}
|
|
newID := uuid.New()
|
|
categoryMap[strconv.FormatInt(legacyID, 10)] = newID.String()
|
|
if dryRun {
|
|
report["categories"]++
|
|
continue
|
|
}
|
|
_, err = pg.Exec(ctx, `
|
|
INSERT INTO categories (
|
|
id, company_id, name, unique_id, parent_unique_id, path, level, position,
|
|
is_active, description, prompt, metadata, config, title_template, description_template
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11,
|
|
COALESCE($12::jsonb, '{}'::jsonb), COALESCE($13::jsonb, '{}'::jsonb), $14::jsonb, $15::jsonb
|
|
)
|
|
ON CONFLICT (company_id, unique_id) DO UPDATE SET
|
|
name = EXCLUDED.name,
|
|
parent_unique_id = EXCLUDED.parent_unique_id,
|
|
title_template = EXCLUDED.title_template,
|
|
description_template = EXCLUDED.description_template,
|
|
updated_at = now()`,
|
|
newID, cid, name, uniqueID, nullString(parentID), nullString(path), level, position,
|
|
isActive == 1, nullString(desc), nullString(prompt),
|
|
jsonOrNull(metadata), jsonOrNull(config), jsonOrNull(titleTpl), jsonOrNull(descTpl))
|
|
if err != nil {
|
|
log.Printf("category insert %d: %v", legacyID, err)
|
|
var existing uuid.UUID
|
|
if err2 := pg.QueryRow(ctx, `SELECT id FROM categories WHERE company_id = $1 AND unique_id = $2`, cid, uniqueID).Scan(&existing); err2 == nil {
|
|
categoryMap[strconv.FormatInt(legacyID, 10)] = existing.String()
|
|
} else {
|
|
report["categories_skipped"]++
|
|
continue
|
|
}
|
|
} else {
|
|
var existing uuid.UUID
|
|
_ = pg.QueryRow(ctx, `SELECT id FROM categories WHERE company_id = $1 AND unique_id = $2`, cid, uniqueID).Scan(&existing)
|
|
if existing != uuid.Nil {
|
|
categoryMap[strconv.FormatInt(legacyID, 10)] = existing.String()
|
|
}
|
|
}
|
|
report["categories"]++
|
|
}
|
|
}
|
|
|
|
func migrateAttributes(
|
|
ctx context.Context,
|
|
mysqlDB *sql.DB,
|
|
pg *pgxpool.Pool,
|
|
companyMap, attributeMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if !mysqlTableExists(ctx, mysqlDB, "attributes") {
|
|
log.Printf("attributes skipped: table missing")
|
|
return
|
|
}
|
|
q := mysqlSelectList(
|
|
"id",
|
|
"company_id",
|
|
"attribute_key",
|
|
"name",
|
|
mysqlCoalesce(ctx, mysqlDB, "attributes", "value_type", "'string'"),
|
|
mysqlCol(ctx, mysqlDB, "attributes", "unit", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "attributes", "example", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "attributes", "parent_key", "NULL"),
|
|
) + " FROM attributes WHERE 1=1"
|
|
clause, cargs := mysqlCompanyFilter("company_id", allow)
|
|
q += clause
|
|
rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
|
|
if err != nil {
|
|
rows, err = mysqlDB.QueryContext(ctx, `
|
|
SELECT id, company_id, attribute_key, name, 'string', NULL, NULL, NULL FROM attributes WHERE 1=1`+clause, cargs...)
|
|
}
|
|
if err != nil {
|
|
log.Printf("attributes skipped: %v", err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
|
|
const batchSize = 400
|
|
type pending struct {
|
|
legacyID int64
|
|
cid, key, name, valueType string
|
|
unit, example, parentKey *string
|
|
newID uuid.UUID
|
|
}
|
|
flush := func(batch []pending) {
|
|
if len(batch) == 0 {
|
|
return
|
|
}
|
|
if dryRun {
|
|
report["attributes"] += len(batch)
|
|
return
|
|
}
|
|
b := &pgx.Batch{}
|
|
for _, p := range batch {
|
|
b.Queue(`
|
|
INSERT INTO attributes (id, company_id, attribute_key, name, value_type, unit, example, parent_key)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
|
|
ON CONFLICT (company_id, attribute_key) DO UPDATE SET
|
|
name = EXCLUDED.name, value_type = EXCLUDED.value_type, updated_at = now()`,
|
|
p.newID, p.cid, p.key, p.name, p.valueType, p.unit, p.example, p.parentKey)
|
|
}
|
|
br := pg.SendBatch(ctx, b)
|
|
if err := br.Close(); err != nil {
|
|
log.Printf("attributes batch: %v — resolving ids individually", err)
|
|
for _, p := range batch {
|
|
var existing uuid.UUID
|
|
if err2 := pg.QueryRow(ctx, `SELECT id FROM attributes WHERE company_id = $1 AND attribute_key = $2`, p.cid, p.key).Scan(&existing); err2 == nil {
|
|
attributeMap[strconv.FormatInt(p.legacyID, 10)] = existing.String()
|
|
report["attributes"]++
|
|
} else {
|
|
report["attributes_skipped"]++
|
|
}
|
|
}
|
|
return
|
|
}
|
|
for _, p := range batch {
|
|
var existing uuid.UUID
|
|
if err := pg.QueryRow(ctx, `SELECT id FROM attributes WHERE company_id = $1 AND attribute_key = $2`, p.cid, p.key).Scan(&existing); err == nil {
|
|
attributeMap[strconv.FormatInt(p.legacyID, 10)] = existing.String()
|
|
}
|
|
report["attributes"]++
|
|
}
|
|
}
|
|
|
|
batch := make([]pending, 0, batchSize)
|
|
for rows.Next() {
|
|
var legacyID int64
|
|
var companyLegacy, key, name, valueType string
|
|
var unit, example, parentKey sql.NullString
|
|
if err := rows.Scan(&legacyID, &companyLegacy, &key, &name, &valueType, &unit, &example, &parentKey); err != nil {
|
|
log.Printf("attribute scan: %v", err)
|
|
report["attributes_skipped"]++
|
|
continue
|
|
}
|
|
cid, ok := companyMap[companyLegacy]
|
|
if !ok {
|
|
report["attributes_skipped"]++
|
|
continue
|
|
}
|
|
if valueType == "" {
|
|
valueType = "string"
|
|
}
|
|
newID := uuid.New()
|
|
attributeMap[strconv.FormatInt(legacyID, 10)] = newID.String()
|
|
batch = append(batch, pending{
|
|
legacyID: legacyID, cid: cid, key: key, name: name, valueType: valueType,
|
|
unit: nullString(unit), example: nullString(example), parentKey: nullString(parentKey),
|
|
newID: newID,
|
|
})
|
|
if len(batch) >= batchSize {
|
|
flush(batch)
|
|
batch = batch[:0]
|
|
}
|
|
}
|
|
flush(batch)
|
|
}
|
|
|
|
func migrateCategoryAttributes(
|
|
ctx context.Context,
|
|
mysqlDB *sql.DB,
|
|
pg *pgxpool.Pool,
|
|
companyMap, attributeMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if !mysqlTableExists(ctx, mysqlDB, "category_attributes") {
|
|
log.Printf("category_attributes skipped: table missing")
|
|
return
|
|
}
|
|
clause, cargs := mysqlCompanyFilter("company_id", allow)
|
|
rows, err := mysqlDB.QueryContext(ctx, `
|
|
SELECT company_id, category_id, attribute_id, COALESCE(required, 0)
|
|
FROM category_attributes WHERE 1=1`+clause, cargs...)
|
|
if err != nil {
|
|
log.Printf("category_attributes skipped: %v", err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
const batchSize = 500
|
|
batch := &pgx.Batch{}
|
|
pending := 0
|
|
flush := func() {
|
|
if dryRun || pending == 0 {
|
|
return
|
|
}
|
|
br := pg.SendBatch(ctx, batch)
|
|
if err := br.Close(); err != nil {
|
|
log.Printf("category_attributes batch: %v", err)
|
|
}
|
|
batch = &pgx.Batch{}
|
|
pending = 0
|
|
}
|
|
for rows.Next() {
|
|
var companyLegacy, categoryUnique string
|
|
var attrLegacy int64
|
|
var required int
|
|
if err := rows.Scan(&companyLegacy, &categoryUnique, &attrLegacy, &required); err != nil {
|
|
log.Printf("category_attribute scan: %v", err)
|
|
report["category_attributes_skipped"]++
|
|
continue
|
|
}
|
|
cid, okC := companyMap[companyLegacy]
|
|
attrID, okA := attributeMap[strconv.FormatInt(attrLegacy, 10)]
|
|
if !okC || !okA {
|
|
report["category_attributes_skipped"]++
|
|
continue
|
|
}
|
|
if dryRun {
|
|
report["category_attributes"]++
|
|
continue
|
|
}
|
|
batch.Queue(`
|
|
INSERT INTO category_attributes (company_id, category_unique_id, attribute_id, required)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (company_id, category_unique_id, attribute_id) DO UPDATE SET required = EXCLUDED.required`,
|
|
cid, categoryUnique, attrID, required == 1)
|
|
pending++
|
|
report["category_attributes"]++
|
|
if pending >= batchSize {
|
|
flush()
|
|
}
|
|
}
|
|
flush()
|
|
}
|
|
|
|
func migrateCustomVariables(
|
|
ctx context.Context,
|
|
mysqlDB *sql.DB,
|
|
pg *pgxpool.Pool,
|
|
companyMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if !mysqlTableExists(ctx, mysqlDB, "custom_variables") {
|
|
log.Printf("custom_variables skipped: table missing")
|
|
return
|
|
}
|
|
// Legacy has name/label/description/example; v2 has name/value/description.
|
|
clause, cargs := mysqlCompanyFilter("company_id", allow)
|
|
rows, err := mysqlDB.QueryContext(ctx, `
|
|
SELECT company_id, name,
|
|
COALESCE(NULLIF(label, ''), NULLIF(example, ''), ''),
|
|
description
|
|
FROM custom_variables WHERE 1=1`+clause, cargs...)
|
|
if err != nil {
|
|
rows, err = mysqlDB.QueryContext(ctx, `
|
|
SELECT company_id, name, COALESCE(value, ''), description
|
|
FROM custom_variables WHERE 1=1`+clause, cargs...)
|
|
}
|
|
if err != nil {
|
|
log.Printf("custom_variables skipped: %v", err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var companyLegacy, name, value string
|
|
var desc sql.NullString
|
|
if err := rows.Scan(&companyLegacy, &name, &value, &desc); err != nil {
|
|
log.Printf("custom_variable scan: %v", err)
|
|
report["custom_variables_skipped"]++
|
|
continue
|
|
}
|
|
cid, ok := companyMap[companyLegacy]
|
|
if !ok {
|
|
report["custom_variables_skipped"]++
|
|
continue
|
|
}
|
|
if dryRun {
|
|
report["custom_variables"]++
|
|
continue
|
|
}
|
|
_, err = pg.Exec(ctx, `
|
|
INSERT INTO custom_variables (company_id, name, value, description)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (company_id, name) DO UPDATE SET
|
|
value = EXCLUDED.value, description = EXCLUDED.description, updated_at = now()`,
|
|
cid, name, value, nullString(desc))
|
|
if err != nil {
|
|
log.Printf("custom_variable: %v", err)
|
|
report["custom_variables_skipped"]++
|
|
continue
|
|
}
|
|
report["custom_variables"]++
|
|
}
|
|
}
|
|
|
|
func migrateInputFeeds(
|
|
ctx context.Context,
|
|
mysqlDB *sql.DB,
|
|
pg *pgxpool.Pool,
|
|
companyMap, feedMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if mysqlTableExists(ctx, mysqlDB, "xml_feeds") {
|
|
hasMappings := mysqlColumnExists(ctx, mysqlDB, "xml_feeds", "field_mappings")
|
|
hasVersion := mysqlColumnExists(ctx, mysqlDB, "xml_feeds", "mapping_version")
|
|
hasSourceType := mysqlColumnExists(ctx, mysqlDB, "xml_feeds", "source_type")
|
|
q := `SELECT id, company_id, name, url, COALESCE(status, 'active'), COALESCE(sync_frequency, 24)`
|
|
if hasMappings {
|
|
q += `, field_mappings`
|
|
} else {
|
|
q += `, NULL`
|
|
}
|
|
if hasVersion {
|
|
q += `, COALESCE(mapping_version, 1)`
|
|
} else {
|
|
q += `, 1`
|
|
}
|
|
if hasSourceType {
|
|
q += `, COALESCE(source_type, 'url')`
|
|
} else {
|
|
q += `, 'url'`
|
|
}
|
|
clause, cargs := mysqlCompanyFilter("company_id", allow)
|
|
q += ` FROM xml_feeds WHERE 1=1` + clause
|
|
rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
|
|
if err != nil {
|
|
log.Printf("xml_feeds skipped: %v", err)
|
|
} else {
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var legacyID int64
|
|
var companyLegacy, name string
|
|
var url, status sql.NullString
|
|
var syncFreq, mapVersion int
|
|
var fieldMappings []byte
|
|
var sourceType string
|
|
if err := rows.Scan(&legacyID, &companyLegacy, &name, &url, &status, &syncFreq, &fieldMappings, &mapVersion, &sourceType); err != nil {
|
|
log.Printf("xml_feed scan: %v", err)
|
|
report["input_feeds_skipped"]++
|
|
continue
|
|
}
|
|
cid, ok := companyMap[companyLegacy]
|
|
if !ok {
|
|
report["input_feeds_skipped"]++
|
|
continue
|
|
}
|
|
st := "active"
|
|
if status.Valid && status.String != "" {
|
|
st = status.String
|
|
}
|
|
interval := syncFreq * 60
|
|
if interval <= 0 {
|
|
interval = 60
|
|
}
|
|
feedType := "xml"
|
|
if sourceType == "csv" || sourceType == "file" {
|
|
feedType = sourceType
|
|
}
|
|
newID := uuid.New()
|
|
feedMap[strconv.FormatInt(legacyID, 10)] = newID.String()
|
|
if dryRun {
|
|
report["input_feeds"]++
|
|
if len(fieldMappings) > 0 {
|
|
report["feed_mappings"]++
|
|
}
|
|
continue
|
|
}
|
|
_, err = pg.Exec(ctx, `
|
|
INSERT INTO input_feeds (id, company_id, name, url, feed_type, status, sync_interval_minutes)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
|
|
newID, cid, name, nullString(url), feedType, st, interval)
|
|
if err != nil {
|
|
log.Printf("input_feed insert %d: %v", legacyID, err)
|
|
report["input_feeds_skipped"]++
|
|
delete(feedMap, strconv.FormatInt(legacyID, 10))
|
|
continue
|
|
}
|
|
report["input_feeds"]++
|
|
if err := insertFeedMappings(ctx, pg, newID, cid, fieldMappings, mapVersion, report); err != nil {
|
|
log.Printf("feed_mappings insert %d: %v", legacyID, err)
|
|
}
|
|
}
|
|
return
|
|
}
|
|
} else {
|
|
log.Printf("xml_feeds skipped: table missing")
|
|
}
|
|
|
|
if !mysqlTableExists(ctx, mysqlDB, "product_feeds") {
|
|
log.Printf("product_feeds skipped: table missing")
|
|
return
|
|
}
|
|
rows, err := mysqlDB.QueryContext(ctx, `
|
|
SELECT id, feed_name FROM product_feeds`)
|
|
if err != nil {
|
|
log.Printf("product_feeds skipped: %v", err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var legacyID int64
|
|
var feedName sql.NullString
|
|
if err := rows.Scan(&legacyID, &feedName); err != nil {
|
|
report["input_feeds_skipped"]++
|
|
continue
|
|
}
|
|
name := feedName.String
|
|
if name == "" {
|
|
name = fmt.Sprintf("product_feed_%d", legacyID)
|
|
}
|
|
// product_feeds has no company_id — attach to first mapped company when only one, else skip.
|
|
if len(companyMap) != 1 {
|
|
report["input_feeds_skipped"]++
|
|
continue
|
|
}
|
|
var cid string
|
|
for _, v := range companyMap {
|
|
cid = v
|
|
break
|
|
}
|
|
newID := uuid.New()
|
|
feedMap[strconv.FormatInt(legacyID, 10)] = newID.String()
|
|
if dryRun {
|
|
report["input_feeds"]++
|
|
continue
|
|
}
|
|
_, err = pg.Exec(ctx, `
|
|
INSERT INTO input_feeds (id, company_id, name, feed_type, status)
|
|
VALUES ($1, $2, $3, 'xml', 'active')`, newID, cid, name)
|
|
if err != nil {
|
|
log.Printf("product_feed insert %d: %v", legacyID, err)
|
|
delete(feedMap, strconv.FormatInt(legacyID, 10))
|
|
report["input_feeds_skipped"]++
|
|
continue
|
|
}
|
|
report["input_feeds"]++
|
|
}
|
|
}
|
|
|
|
func migrateRawProducts(
|
|
ctx context.Context,
|
|
mysqlDB *sql.DB,
|
|
pg *pgxpool.Pool,
|
|
companyMap, feedMap, rawMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if !mysqlTableExists(ctx, mysqlDB, "raw_products") {
|
|
log.Printf("raw_products skipped: table missing")
|
|
return
|
|
}
|
|
q := mysqlSelectList(
|
|
"id",
|
|
"company_id",
|
|
"gtin",
|
|
mysqlCol(ctx, mysqlDB, "raw_products", "feed_id", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "raw_products", "feed_ids", "NULL"),
|
|
mysqlCoalesce(ctx, mysqlDB, "raw_products", "raw_data", "'{}'"),
|
|
mysqlCol(ctx, mysqlDB, "raw_products", "mapped_data", "NULL"),
|
|
mysqlCoalesce(ctx, mysqlDB, "raw_products", "is_processed", "0"),
|
|
mysqlCoalesce(ctx, mysqlDB, "raw_products", "processing_status", "'unprocessed'"),
|
|
) + " FROM raw_products WHERE 1=1"
|
|
clause, cargs := mysqlCompanyFilter("company_id", allow)
|
|
q += clause
|
|
rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
|
|
if err != nil {
|
|
rows, err = mysqlDB.QueryContext(ctx, `
|
|
SELECT id, company_id, gtin, feed_id, NULL, raw_data, NULL, 0, 'unprocessed'
|
|
FROM raw_products WHERE 1=1`+clause, cargs...)
|
|
}
|
|
if err != nil {
|
|
log.Printf("raw_products skipped: %v", err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
|
|
const batchSize = 500
|
|
type pending struct {
|
|
legacyID int64
|
|
cid, gtin, status string
|
|
feedID *uuid.UUID
|
|
feedIDs, rawData, mappedData []byte
|
|
isProcessed bool
|
|
newID uuid.UUID
|
|
}
|
|
flush := func(batch []pending) {
|
|
if len(batch) == 0 {
|
|
return
|
|
}
|
|
if dryRun {
|
|
report["raw_products"] += len(batch)
|
|
return
|
|
}
|
|
tx, err := pg.Begin(ctx)
|
|
if err != nil {
|
|
log.Printf("raw_products tx: %v", err)
|
|
report["raw_products_skipped"] += len(batch)
|
|
return
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
copyRows := make([][]any, 0, len(batch))
|
|
for _, p := range batch {
|
|
copyRows = append(copyRows, []any{
|
|
p.newID, p.cid, p.gtin, p.feedID,
|
|
jsonOrNull(p.feedIDs), string(ensureJSON(p.rawData)), string(ensureJSON(p.mappedData)),
|
|
p.isProcessed, p.status,
|
|
})
|
|
}
|
|
_, err = tx.CopyFrom(ctx,
|
|
pgx.Identifier{"raw_products"},
|
|
[]string{"id", "company_id", "gtin", "feed_id", "feed_ids", "raw_data", "mapped_data", "is_processed", "processing_status"},
|
|
pgx.CopyFromRows(copyRows),
|
|
)
|
|
if err != nil {
|
|
log.Printf("raw_products copy: %v — falling back to per-row insert", err)
|
|
for _, p := range batch {
|
|
var inserted uuid.UUID
|
|
err = tx.QueryRow(ctx, `
|
|
INSERT INTO raw_products (
|
|
id, company_id, gtin, feed_id, feed_ids, raw_data, mapped_data,
|
|
is_processed, processing_status
|
|
) VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb, $7::jsonb, $8, $9)
|
|
ON CONFLICT (company_id, gtin) DO UPDATE SET
|
|
feed_id = COALESCE(EXCLUDED.feed_id, raw_products.feed_id),
|
|
raw_data = EXCLUDED.raw_data,
|
|
mapped_data = EXCLUDED.mapped_data,
|
|
is_processed = EXCLUDED.is_processed,
|
|
processing_status = EXCLUDED.processing_status,
|
|
updated_at = now()
|
|
RETURNING id`,
|
|
p.newID, p.cid, p.gtin, p.feedID,
|
|
jsonOrNull(p.feedIDs), string(ensureJSON(p.rawData)), string(ensureJSON(p.mappedData)),
|
|
p.isProcessed, p.status).Scan(&inserted)
|
|
if err != nil {
|
|
log.Printf("raw_product insert %d: %v", p.legacyID, err)
|
|
report["raw_products_skipped"]++
|
|
continue
|
|
}
|
|
rawMap[strconv.FormatInt(p.legacyID, 10)] = inserted.String()
|
|
report["raw_products"]++
|
|
}
|
|
} else {
|
|
report["raw_products"] += len(batch)
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
log.Printf("raw_products commit: %v", err)
|
|
}
|
|
}
|
|
|
|
batch := make([]pending, 0, batchSize)
|
|
gtinOwner := map[string]string{} // companyUUID|gtin → new UUID string (dedupe for PG unique index)
|
|
for rows.Next() {
|
|
var legacyID int64
|
|
var companyLegacy string
|
|
var gtin, status sql.NullString
|
|
var feedID sql.NullInt64
|
|
var feedIDs, rawData, mappedData []byte
|
|
var isProcessed int
|
|
if err := rows.Scan(&legacyID, &companyLegacy, >in, &feedID, &feedIDs, &rawData, &mappedData, &isProcessed, &status); err != nil {
|
|
log.Printf("raw_product scan: %v", err)
|
|
report["raw_products_skipped"]++
|
|
continue
|
|
}
|
|
cid, ok := companyMap[companyLegacy]
|
|
if !ok {
|
|
report["raw_products_skipped"]++
|
|
continue
|
|
}
|
|
gtinVal := ""
|
|
if gtin.Valid {
|
|
gtinVal = strings.TrimSpace(gtin.String)
|
|
}
|
|
if gtinVal == "" {
|
|
// PG requires NOT NULL gtin; synthesize a stable placeholder from legacy id.
|
|
gtinVal = fmt.Sprintf("legacy-missing-%d", legacyID)
|
|
report["raw_products_gtin_synthesized"]++
|
|
}
|
|
statusVal := "unprocessed"
|
|
if status.Valid && strings.TrimSpace(status.String) != "" {
|
|
statusVal = strings.TrimSpace(status.String)
|
|
}
|
|
switch statusVal {
|
|
case "unprocessed", "processing", "processed", "failed":
|
|
default:
|
|
statusVal = "unprocessed"
|
|
report["raw_products_status_normalized"]++
|
|
}
|
|
rawObj := map[string]any{}
|
|
if len(rawData) > 0 {
|
|
_ = json.Unmarshal(rawData, &rawObj)
|
|
}
|
|
var mappedFeed *uuid.UUID
|
|
if feedID.Valid {
|
|
rawObj["_legacy_feed_id"] = feedID.Int64
|
|
if fid, ok := feedMap[strconv.FormatInt(feedID.Int64, 10)]; ok {
|
|
u, err := uuid.Parse(fid)
|
|
if err == nil {
|
|
mappedFeed = &u
|
|
}
|
|
}
|
|
}
|
|
rawBytes, _ := json.Marshal(rawObj)
|
|
dedupeKey := cid + "|" + gtinVal
|
|
if existing, ok := gtinOwner[dedupeKey]; ok {
|
|
rawMap[strconv.FormatInt(legacyID, 10)] = existing
|
|
report["raw_products_gtin_deduped"]++
|
|
continue
|
|
}
|
|
newID := uuid.New()
|
|
gtinOwner[dedupeKey] = newID.String()
|
|
rawMap[strconv.FormatInt(legacyID, 10)] = newID.String()
|
|
batch = append(batch, pending{
|
|
legacyID: legacyID, cid: cid, gtin: gtinVal, status: statusVal,
|
|
feedID: mappedFeed, feedIDs: feedIDs, rawData: rawBytes, mappedData: mappedData,
|
|
isProcessed: isProcessed == 1,
|
|
newID: newID,
|
|
})
|
|
if len(batch) >= batchSize {
|
|
flush(batch)
|
|
batch = batch[:0]
|
|
}
|
|
}
|
|
flush(batch)
|
|
}
|
|
|
|
func migrateProcessedProducts(
|
|
ctx context.Context,
|
|
mysqlDB *sql.DB,
|
|
pg *pgxpool.Pool,
|
|
companyMap, feedMap, rawMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if !mysqlTableExists(ctx, mysqlDB, "processed_products") {
|
|
log.Printf("processed_products skipped: table missing")
|
|
return
|
|
}
|
|
q := mysqlSelectList(
|
|
"id",
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "company_id", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "product_id", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "name", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "category", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "description", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "processed_description", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "attributes", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "processed_attributes", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "status", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "gpt_response", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "total_tokens", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "feed_id", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "raw_product_id", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "processed_name", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "meta_title", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "meta_description", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "structured_description", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "processed_products", "field_sources", "NULL"),
|
|
) + " FROM processed_products WHERE 1=1"
|
|
clause, cargs := mysqlCompanyFilter("company_id", allow)
|
|
q += clause
|
|
rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
|
|
if err != nil {
|
|
rows, err = mysqlDB.QueryContext(ctx, `
|
|
SELECT id, company_id, product_id, name, category, description, processed_description,
|
|
attributes, processed_attributes, status, gpt_response, total_tokens,
|
|
feed_id, raw_product_id, NULL, NULL, NULL, NULL, NULL
|
|
FROM processed_products WHERE 1=1`+clause, cargs...)
|
|
}
|
|
if err != nil {
|
|
log.Printf("processed_products skipped: %v", err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var legacyID int64
|
|
var companyLegacy sql.NullString
|
|
var productID, name, category, desc, procDesc, status, procName, metaTitle, metaDesc sql.NullString
|
|
var attrs, procAttrs, gptResp, structured, fieldSources []byte
|
|
var totalTokens sql.NullInt64
|
|
var feedID, rawProductID sql.NullInt64
|
|
if err := rows.Scan(&legacyID, &companyLegacy, &productID, &name, &category, &desc, &procDesc,
|
|
&attrs, &procAttrs, &status, &gptResp, &totalTokens, &feedID, &rawProductID,
|
|
&procName, &metaTitle, &metaDesc, &structured, &fieldSources); err != nil {
|
|
log.Printf("processed_product scan: %v", err)
|
|
report["processed_products_skipped"]++
|
|
continue
|
|
}
|
|
if !companyLegacy.Valid || companyLegacy.String == "" {
|
|
report["processed_products_skipped"]++
|
|
continue
|
|
}
|
|
cid, ok := companyMap[companyLegacy.String]
|
|
if !ok {
|
|
report["processed_products_skipped"]++
|
|
continue
|
|
}
|
|
var mappedFeed, mappedRaw *uuid.UUID
|
|
if feedID.Valid {
|
|
if fid, ok := feedMap[strconv.FormatInt(feedID.Int64, 10)]; ok {
|
|
u, err := uuid.Parse(fid)
|
|
if err == nil {
|
|
mappedFeed = &u
|
|
}
|
|
}
|
|
}
|
|
if rawProductID.Valid {
|
|
if rid, ok := rawMap[strconv.FormatInt(rawProductID.Int64, 10)]; ok {
|
|
u, err := uuid.Parse(rid)
|
|
if err == nil {
|
|
mappedRaw = &u
|
|
}
|
|
}
|
|
}
|
|
if dryRun {
|
|
report["processed_products"]++
|
|
continue
|
|
}
|
|
var tokens *int
|
|
if totalTokens.Valid {
|
|
t := int(totalTokens.Int64)
|
|
tokens = &t
|
|
}
|
|
_, err = pg.Exec(ctx, `
|
|
INSERT INTO processed_products (
|
|
company_id, product_id, name, category, description, processed_description,
|
|
attributes, processed_attributes, status, gpt_response, total_tokens,
|
|
feed_id, raw_product_id, processed_name, meta_title, meta_description,
|
|
structured_description, field_sources
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5, $6, $7::jsonb, $8::jsonb, $9, $10::jsonb, $11,
|
|
$12, $13, $14, $15, $16, $17::jsonb, $18::jsonb
|
|
)`,
|
|
cid, nullString(productID), nullString(name), nullString(category), nullString(desc), nullString(procDesc),
|
|
jsonOrNull(attrs), jsonOrNull(procAttrs), nullString(status), jsonOrNull(gptResp), tokens,
|
|
mappedFeed, mappedRaw, nullString(procName), nullString(metaTitle), nullString(metaDesc),
|
|
jsonOrNull(structured), jsonOrNull(fieldSources))
|
|
if err != nil {
|
|
log.Printf("processed_product insert %d: %v", legacyID, err)
|
|
report["processed_products_skipped"]++
|
|
continue
|
|
}
|
|
report["processed_products"]++
|
|
}
|
|
}
|
|
|
|
// backfillProcessedDescriptionsFromMapped fills empty processed_products.description from the
|
|
// linked raw mapped_data.description. Legacy MySQL often left description blank while the feed
|
|
// original lived only on raw_products.mapped_data.
|
|
func backfillProcessedDescriptionsFromMapped(
|
|
ctx context.Context,
|
|
pg *pgxpool.Pool,
|
|
companyMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if dryRun || pg == nil {
|
|
return
|
|
}
|
|
cids := make([]uuid.UUID, 0, len(companyMap))
|
|
for legacy, cid := range companyMap {
|
|
if len(allow) > 0 && !allow[legacy] {
|
|
continue
|
|
}
|
|
u, err := uuid.Parse(cid)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
cids = append(cids, u)
|
|
}
|
|
if len(cids) == 0 {
|
|
return
|
|
}
|
|
ct, err := pg.Exec(ctx, `
|
|
UPDATE processed_products p
|
|
SET description = NULLIF(r.mapped_data->>'description', ''),
|
|
updated_at = now()
|
|
FROM raw_products r
|
|
WHERE p.raw_product_id = r.id
|
|
AND p.company_id = ANY($1::uuid[])
|
|
AND COALESCE(p.description, '') = ''
|
|
AND COALESCE(r.mapped_data->>'description', '') <> ''`, cids)
|
|
if err != nil {
|
|
log.Printf("processed description backfill: %v", err)
|
|
return
|
|
}
|
|
n := int(ct.RowsAffected())
|
|
if n > 0 {
|
|
report["processed_descriptions_backfilled"] = n
|
|
log.Printf("backfilled %d processed_products.description from mapped_data", n)
|
|
}
|
|
}
|
|
|
|
// backfillProcessedNamesFromMapped fills empty processed_products.name from mapped_data name/title.
|
|
func backfillProcessedNamesFromMapped(
|
|
ctx context.Context,
|
|
pg *pgxpool.Pool,
|
|
companyMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if dryRun || pg == nil {
|
|
return
|
|
}
|
|
cids := make([]uuid.UUID, 0, len(companyMap))
|
|
for legacy, cid := range companyMap {
|
|
if len(allow) > 0 && !allow[legacy] {
|
|
continue
|
|
}
|
|
u, err := uuid.Parse(cid)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
cids = append(cids, u)
|
|
}
|
|
if len(cids) == 0 {
|
|
return
|
|
}
|
|
ct, err := pg.Exec(ctx, `
|
|
UPDATE processed_products p
|
|
SET name = COALESCE(NULLIF(r.mapped_data->>'name', ''), NULLIF(r.mapped_data->>'title', '')),
|
|
updated_at = now()
|
|
FROM raw_products r
|
|
WHERE p.raw_product_id = r.id
|
|
AND p.company_id = ANY($1::uuid[])
|
|
AND COALESCE(p.name, '') = ''
|
|
AND COALESCE(NULLIF(r.mapped_data->>'name', ''), NULLIF(r.mapped_data->>'title', ''), '') <> ''`, cids)
|
|
if err != nil {
|
|
log.Printf("processed name backfill: %v", err)
|
|
return
|
|
}
|
|
n := int(ct.RowsAffected())
|
|
if n > 0 {
|
|
report["processed_names_backfilled"] = n
|
|
log.Printf("backfilled %d processed_products.name from mapped_data", n)
|
|
}
|
|
}
|
|
|
|
// backfillProcessedCategoriesFromMapped fills empty processed.category from mapped unique_ids.
|
|
func backfillProcessedCategoriesFromMapped(
|
|
ctx context.Context,
|
|
pg *pgxpool.Pool,
|
|
companyMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if dryRun || pg == nil {
|
|
return
|
|
}
|
|
total := 0
|
|
for legacy, cid := range companyMap {
|
|
if len(allow) > 0 && !allow[legacy] {
|
|
continue
|
|
}
|
|
u, err := uuid.Parse(cid)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
res, err := processing.BackfillProcessedCategoriesFromMapped(ctx, pg, u)
|
|
if err != nil {
|
|
log.Printf("processed category backfill company=%s: %v", cid, err)
|
|
continue
|
|
}
|
|
total += int(res.Updated)
|
|
}
|
|
if total > 0 {
|
|
report["processed_categories_backfilled"] = total
|
|
log.Printf("backfilled %d processed_products.category from mapped_data", total)
|
|
}
|
|
}
|
|
|
|
func migrateExportFeeds(
|
|
ctx context.Context,
|
|
mysqlDB *sql.DB,
|
|
pg *pgxpool.Pool,
|
|
companyMap, feedMap map[string]string,
|
|
allow map[string]bool,
|
|
report map[string]int,
|
|
dryRun bool,
|
|
) {
|
|
if !mysqlTableExists(ctx, mysqlDB, "export_feeds") {
|
|
log.Printf("export_feeds skipped: table missing")
|
|
return
|
|
}
|
|
q := mysqlSelectList(
|
|
"id",
|
|
"company_id",
|
|
"name",
|
|
mysqlCoalesce(ctx, mysqlDB, "export_feeds", "format", "'xml'"),
|
|
mysqlCol(ctx, mysqlDB, "export_feeds", "source_feed_id", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "export_feeds", "mappings", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "export_feeds", "structure", "NULL"),
|
|
mysqlCol(ctx, mysqlDB, "export_feeds", "last_generated_at", "NULL"),
|
|
) + " FROM export_feeds WHERE 1=1"
|
|
clause, cargs := mysqlCompanyFilter("company_id", allow)
|
|
q += clause
|
|
rows, err := mysqlDB.QueryContext(ctx, q, cargs...)
|
|
if err != nil {
|
|
rows, err = mysqlDB.QueryContext(ctx, `
|
|
SELECT id, company_id, name, format, source_feed_id, mappings, NULL, NULL
|
|
FROM export_feeds WHERE 1=1`+clause, cargs...)
|
|
}
|
|
if err != nil {
|
|
log.Printf("export_feeds skipped: %v", err)
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var legacyID, companyLegacy, name, format string
|
|
var sourceFeed sql.NullInt64
|
|
var mappings, structure []byte
|
|
var lastGen sql.NullTime
|
|
if err := rows.Scan(&legacyID, &companyLegacy, &name, &format, &sourceFeed, &mappings, &structure, &lastGen); err != nil {
|
|
log.Printf("export_feed scan: %v", err)
|
|
report["export_feeds_skipped"]++
|
|
continue
|
|
}
|
|
cid, ok := companyMap[companyLegacy]
|
|
if !ok {
|
|
report["export_feeds_skipped"]++
|
|
continue
|
|
}
|
|
var src *uuid.UUID
|
|
if sourceFeed.Valid {
|
|
if fid, ok := feedMap[strconv.FormatInt(sourceFeed.Int64, 10)]; ok {
|
|
u, err := uuid.Parse(fid)
|
|
if err == nil {
|
|
src = &u
|
|
}
|
|
}
|
|
}
|
|
tpl := map[string]any{}
|
|
if len(mappings) > 0 {
|
|
var m any
|
|
if json.Unmarshal(mappings, &m) == nil {
|
|
tpl["mappings"] = m
|
|
}
|
|
}
|
|
if len(structure) > 0 {
|
|
var s any
|
|
if json.Unmarshal(structure, &s) == nil {
|
|
tpl["structure"] = s
|
|
}
|
|
}
|
|
tplBytes, _ := json.Marshal(tpl)
|
|
if format == "" {
|
|
format = "xml"
|
|
}
|
|
if dryRun {
|
|
report["export_feeds"]++
|
|
continue
|
|
}
|
|
newID := uuid.New()
|
|
_, err = pg.Exec(ctx, `
|
|
INSERT INTO export_feeds (id, company_id, name, source_feed_id, format, template, last_generated_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7)`,
|
|
newID, cid, name, src, format, string(tplBytes), nullTime(lastGen))
|
|
if err != nil {
|
|
log.Printf("export_feed insert %s: %v", legacyID, err)
|
|
report["export_feeds_skipped"]++
|
|
continue
|
|
}
|
|
report["export_feeds"]++
|
|
}
|
|
}
|
|
|
|
func insertFeedMappings(
|
|
ctx context.Context,
|
|
pg *pgxpool.Pool,
|
|
feedID uuid.UUID,
|
|
companyID string,
|
|
fieldMappings []byte,
|
|
version int,
|
|
report map[string]int,
|
|
) error {
|
|
if len(fieldMappings) == 0 {
|
|
return nil
|
|
}
|
|
payload := ensureJSON(fieldMappings)
|
|
// Normalize object maps → JSON array of entries for feed_mappings.mappings default shape.
|
|
var asObj map[string]any
|
|
if json.Unmarshal(payload, &asObj) == nil && len(asObj) > 0 {
|
|
arr := make([]any, 0, len(asObj))
|
|
for k, v := range asObj {
|
|
entry := map[string]any{"key": k, "mapping": normalizeLegacyMappingValue(v)}
|
|
arr = append(arr, entry)
|
|
}
|
|
if b, err := json.Marshal(arr); err == nil {
|
|
payload = b
|
|
}
|
|
} else {
|
|
var asArr []any
|
|
if json.Unmarshal(payload, &asArr) == nil && len(asArr) > 0 {
|
|
for i, item := range asArr {
|
|
m, ok := item.(map[string]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
if nested, ok := m["mapping"]; ok {
|
|
m["mapping"] = normalizeLegacyMappingValue(nested)
|
|
asArr[i] = m
|
|
}
|
|
}
|
|
if b, err := json.Marshal(asArr); err == nil {
|
|
payload = b
|
|
}
|
|
}
|
|
}
|
|
if version <= 0 {
|
|
version = 1
|
|
}
|
|
_, err := pg.Exec(ctx, `
|
|
INSERT INTO feed_mappings (feed_id, company_id, version, mappings, is_active)
|
|
VALUES ($1, $2, $3, $4::jsonb, true)`,
|
|
feedID, companyID, version, string(payload))
|
|
if err != nil {
|
|
report["feed_mappings_skipped"]++
|
|
return err
|
|
}
|
|
report["feed_mappings"]++
|
|
return nil
|
|
}
|
|
|
|
// normalizeLegacyMappingValue rewrites nested mapping objects so fieldName uses
|
|
// snake_case aliases the v2 mapping UI / ecommerce catalog understand.
|
|
func normalizeLegacyMappingValue(v any) any {
|
|
m, ok := v.(map[string]any)
|
|
if !ok {
|
|
return v
|
|
}
|
|
out := make(map[string]any, len(m))
|
|
for k, val := range m {
|
|
out[k] = val
|
|
}
|
|
for _, key := range []string{"fieldName", "field", "target"} {
|
|
raw, _ := out[key].(string)
|
|
if strings.TrimSpace(raw) == "" {
|
|
continue
|
|
}
|
|
out[key] = canonicalizeLegacyFieldKey(raw)
|
|
break
|
|
}
|
|
return out
|
|
}
|
|
|
|
func canonicalizeLegacyFieldKey(raw string) string {
|
|
compact := strings.ToLower(strings.TrimSpace(raw))
|
|
compact = strings.ReplaceAll(compact, "_", "")
|
|
compact = strings.ReplaceAll(compact, "-", "")
|
|
compact = strings.ReplaceAll(compact, " ", "")
|
|
aliases := map[string]string{
|
|
"name": "title", "productname": "title", "title": "title",
|
|
"purchaseprice": "purchase_price", "productmodel": "product_model",
|
|
"moreimages": "additional_image_urls", "imageurl": "image_url",
|
|
"producturl": "product_url", "officiallink": "official_link",
|
|
"mainimage": "main_image", "eprelid": "eprel_id",
|
|
"stockstatus": "availability", "videourl": "video_url",
|
|
"netdepth": "net_depth", "netheight": "net_height",
|
|
"netwidth": "net_width", "netmass": "net_mass",
|
|
}
|
|
if v, ok := aliases[compact]; ok {
|
|
return v
|
|
}
|
|
return strings.TrimSpace(raw)
|
|
}
|
|
|
|
func mysqlTableExists(ctx context.Context, db *sql.DB, name string) bool {
|
|
var n int
|
|
err := db.QueryRowContext(ctx, `
|
|
SELECT COUNT(*) FROM information_schema.tables
|
|
WHERE table_schema = DATABASE() AND table_name = ?`, name).Scan(&n)
|
|
return err == nil && n > 0
|
|
}
|
|
|
|
func mysqlColumnExists(ctx context.Context, db *sql.DB, table, column string) bool {
|
|
var n int
|
|
err := db.QueryRowContext(ctx, `
|
|
SELECT COUNT(*) FROM information_schema.columns
|
|
WHERE table_schema = DATABASE() AND table_name = ? AND column_name = ?`, table, column).Scan(&n)
|
|
return err == nil && n > 0
|
|
}
|
|
|
|
func nullString(ns sql.NullString) *string {
|
|
if !ns.Valid {
|
|
return nil
|
|
}
|
|
s := ns.String
|
|
return &s
|
|
}
|
|
|
|
func nullTime(nt sql.NullTime) any {
|
|
if !nt.Valid {
|
|
return nil
|
|
}
|
|
return nt.Time
|
|
}
|
|
|
|
func jsonOrNull(b []byte) *string {
|
|
if len(b) == 0 || string(b) == "null" {
|
|
return nil
|
|
}
|
|
s := string(b)
|
|
if !json.Valid(b) {
|
|
enc, err := json.Marshal(s)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
s = string(enc)
|
|
}
|
|
return &s
|
|
}
|
|
|
|
func ensureJSON(b []byte) []byte {
|
|
if len(b) == 0 || !json.Valid(b) {
|
|
return []byte("{}")
|
|
}
|
|
return b
|
|
}
|