This commit is contained in:
seb
2026-10-10 18:17:41 +02:00
parent 29057e8d7d
commit 47c61c9f2a
2 changed files with 182 additions and 72 deletions

View File

@@ -568,11 +568,11 @@ export function register(app, basePath = '') {
googleAuthId: req.user.id,
companyId,
importDir,
onProgress: (phase, current, total, item) => {
sseWrite('progress', { phase, current, total, item });
onProgress: (counts) => {
sseWrite('progress', counts);
},
onImage: (imageData) => {
sseWrite('image', imageData);
onImages: (skelehashes) => {
sseWrite('images', skelehashes);
},
});
@@ -589,8 +589,8 @@ export function register(app, basePath = '') {
googleAuthId: req.user.id,
companyId,
importDir,
onProgress: (phase, current, total, item) => {
console.log(`[Import Progress] [${phase}] ${current}/${total} - ${item}`);
onProgress: (counts) => {
console.log('[Import Progress]', counts);
},
});

View File

@@ -195,8 +195,8 @@ export function extractVariantValue(parentName, childName, childSku, parentSku)
* @param {number} opts.googleAuthId
* @param {number} opts.companyId
* @param {string} opts.importDir
* @param {function} [opts.onProgress] - (phase, current, total, item) => void
* @param {function} [opts.onImage] - ({ type, current, total, sku, name, brand, categoryId, brandId, path, filename, skelehash, width, height, hashLength }) => void
* @param {function} [opts.onProgress] - (counts) => void; counts = { categoriestotal, categoriesdone, articlestotal, articlesdone, imagestotal, imagesdone, receiptstotal, receiptsdone }; throttled to 200ms
* @param {function} [opts.onImages] - (skelehashes: string[]) => void; batched every 50ms, a hash is only included after its image was stored in the database
* @returns {Promise<object>} Summary outcome
*/
export async function executeCompanyImport({
@@ -204,7 +204,7 @@ export async function executeCompanyImport({
companyId,
importDir,
onProgress = () => {},
onImage = () => {},
onImages = () => {},
}) {
const store = await getStore();
@@ -223,6 +223,83 @@ export async function executeCompanyImport({
console.log(`[Import] Read ${catRecords.length} category rows and ${artRecords.length} article rows`);
// Global counts tracking for 200ms progress stream — nothing else is sent
const progressCounts = {
categoriestotal: 0,
categoriesdone: 0,
articlestotal: 0,
articlesdone: 0,
imagestotal: 0,
imagesdone: 0,
receiptstotal: 0,
receiptsdone: 0,
};
let lastProgressEmit = 0;
const emitProgress = (force = false) => {
const now = Date.now();
if (force || now - lastProgressEmit >= 200) {
lastProgressEmit = now;
onProgress({ ...progressCounts });
}
};
// Buffer skelehashes and emit array every 50ms (or on flush)
let skeleBuffer = [];
let lastSkeleEmit = Date.now();
const emitSkeleHash = (base62, force = false) => {
if (base62) skeleBuffer.push(base62);
const now = Date.now();
if (skeleBuffer.length > 0 && (force || now - lastSkeleEmit >= 50)) {
const batch = skeleBuffer;
skeleBuffer = [];
lastSkeleEmit = now;
onImages(batch);
}
};
const flushSkeleHashes = () => {
if (skeleBuffer.length > 0) {
const batch = skeleBuffer;
skeleBuffer = [];
lastSkeleEmit = Date.now();
onImages(batch);
}
};
// Done-tracking: an image is done when it is stored in the database, an
// article is done when all its images are done, a category is done when
// all its articles are done.
const imagesBySku = new Map(); // sku -> { total, done, paths }
const articlesByCatId = new Map(); // dashCategoryId -> Set<sku>
const categoriesBySku = new Map(); // sku -> Set<dashCategoryId>
const doneSku = new Set();
const doneCatIds = new Set();
const markArticleDone = (sku) => {
if (doneSku.has(sku)) return;
doneSku.add(sku);
progressCounts.articlesdone++;
const cats = categoriesBySku.get(sku);
if (!cats) return;
for (const catId of cats) {
if (doneCatIds.has(catId)) continue;
const artSet = articlesByCatId.get(catId);
if (artSet && Array.from(artSet).every((s) => doneSku.has(s))) {
doneCatIds.add(catId);
progressCounts.categoriesdone++;
}
}
};
const markImageDone = (sku) => {
progressCounts.imagesdone++;
if (!sku) return;
const rec = imagesBySku.get(sku);
if (!rec) return;
rec.done++;
if (rec.done >= rec.total) markArticleDone(sku);
};
// ─── Phase 1: Category Mapping & Creation ────────────────────────────────
// 1) Find existing category tree & company root category
const treeResult = await store.category.getTreeForCompany(googleAuthId);
@@ -270,6 +347,8 @@ export async function executeCompanyImport({
// Filter included categories
const includedCategories = catRecords.filter((c) => !excludedJtlIds.has(String(c.kKategorie)));
progressCounts.categoriestotal = includedCategories.length;
emitProgress(true);
// Sort categories so parents are created before children
// JTL kOberKategorie === '0' are top-level
@@ -278,8 +357,6 @@ export async function executeCompanyImport({
let createdCatCount = 0;
let catPicCount = 0;
onProgress('categories', 0, includedCategories.length, 'Starting categories import');
let safety = 0;
while (pendingCats.length > 0 && safety < 10) {
safety++;
@@ -334,23 +411,16 @@ export async function executeCompanyImport({
if (picRel) {
const picPath = path.join(importDir, picRel);
if (fs.existsSync(picPath)) {
progressCounts.imagestotal++;
try {
const imgBuf = fs.readFileSync(picPath);
await store.category.addImage(googleAuthId, dashCatId, imgBuf);
catPicCount++;
// Hash is only emitted after the image is stored
const skele = await computeSkeleHash(imgBuf);
onImage({
type: 'category',
categoryId: dashCatId,
name,
path: picRel,
filename: path.basename(picRel),
skelehash: skele?.base62 || null,
width: skele?.width || null,
height: skele?.height || null,
hashLength: skele?.hashLength || 0,
});
emitSkeleHash(skele?.base62 || null);
markImageDone(null);
} catch (err) {
console.warn(`[Import] Category image error (${picRel}):`, err.message);
}
@@ -358,11 +428,13 @@ export async function executeCompanyImport({
}
}
onProgress('categories', jtlCatIdToDashCatId.size, includedCategories.length, name);
emitProgress();
}
pendingCats.length = 0;
pendingCats.push(...remaining);
}
flushSkeleHashes();
emitProgress(true);
console.log(`[Import] Categories resolved: ${jtlCatIdToDashCatId.size}/${includedCategories.length}, created new: ${createdCatCount}, images: ${catPicCount}`);
@@ -382,6 +454,44 @@ export async function executeCompanyImport({
return true;
});
// Precompute per-article image counts and article/category relations for
// done-tracking (article done = all its images stored, category done =
// all its articles done)
let articleCount = 0;
for (const a of includedArticles) {
const sku = String(a.cArtNr || '').trim();
if (!sku) continue;
articleCount++;
const rawPics = (a.pictureFiles || '').trim();
if (rawPics) {
const lines = rawPics.replace(/\r/g, '\n').split('\n').map((l) => l.trim()).filter(Boolean);
if (lines.length > 0) {
imagesBySku.set(sku, { total: lines.length, done: 0, paths: lines });
progressCounts.imagestotal += lines.length;
}
}
let dashCatIds = [];
try {
const rawCats = JSON.parse(a.categories || '[]');
for (const it of rawCats) {
const jtlCId = String(it.kKategorie);
if (jtlCatIdToDashCatId.has(jtlCId)) {
dashCatIds.push(jtlCatIdToDashCatId.get(jtlCId));
}
}
} catch {}
for (const catId of dashCatIds) {
if (!articlesByCatId.has(catId)) articlesByCatId.set(catId, new Set());
articlesByCatId.get(catId).add(sku);
if (!categoriesBySku.has(sku)) categoriesBySku.set(sku, new Set());
categoriesBySku.get(sku).add(catId);
}
}
progressCounts.articlestotal = articleCount;
emitProgress(true);
// ─── Phase 2: Brand (Hersteller) Import ──────────────────────────────────
const brandMap = new Map(); // brandName -> Set of picture relative paths
for (const a of includedArticles) {
@@ -428,23 +538,16 @@ export async function executeCompanyImport({
for (const picRel of picSet) {
const picPath = path.resolve(importDir, picRel);
if (fs.existsSync(picPath)) {
progressCounts.imagestotal++;
try {
const imgBuf = fs.readFileSync(picPath);
await store.brand.addImage(googleAuthId, brand.id, imgBuf);
brandPicCount++;
// Hash is only emitted after the image is stored
const skele = await computeSkeleHash(imgBuf);
onImage({
type: 'brand',
brandId: brand.id,
brand: brandName,
path: picRel,
filename: path.basename(picRel),
skelehash: skele?.base62 || null,
width: skele?.width || null,
height: skele?.height || null,
hashLength: skele?.hashLength || 0,
});
emitSkeleHash(skele?.base62 || null);
markImageDone(null);
break; // 1 primary logo per brand
} catch (err) {
log.warn(`Brand image error for "${brandName}" (${picRel}):`, err.message);
@@ -456,20 +559,21 @@ export async function executeCompanyImport({
log.warn(`Failed to upsert brand "${brandName}":`, err.message);
}
onProgress('brands', i + 1, brandEntries.length, brandName);
emitProgress();
}
flushSkeleHashes();
emitProgress(true);
log.info(`Brands resolved: ${brandNameToId.size}/${brandEntries.length}, brand images: ${brandPicCount}`);
// ─── Phase 3: Article Import Pass 1 (Create or Update) ───────────────────
const totalArticles = includedArticles.length;
log.info(`Articles to process: ${totalArticles}`);
onProgress('articles', 0, totalArticles, 'Starting articles import');
emitProgress(true);
let articlesImported = 0;
let articlesUpdated = 0;
const jtlArtIdToSku = new Map();
const articlesWithPictures = [];
for (let i = 0; i < totalArticles; i++) {
const a = includedArticles[i];
@@ -619,15 +723,13 @@ export async function executeCompanyImport({
console.warn(`[Import] Failed to save article ${sku} ("${name}"):`, err.message);
}
// Keep note of picture references for image phase
const rawPics = (a.pictureFiles || '').trim();
if (rawPics) {
articlesWithPictures.push({ sku, pictureFiles: rawPics });
}
// Articles without images are done as soon as their row is written
if (!imagesBySku.has(sku)) markArticleDone(sku);
onProgress('articles', i + 1, totalArticles, `${sku} - ${name}`);
emitProgress();
}
emitProgress(true);
console.log(`[Import] Article pass 1 complete. New: ${articlesImported}, Updated: ${articlesUpdated}`);
// ─── Phase 3: Article Import Pass 2 (Variant Links) ──────────────────────
@@ -635,7 +737,7 @@ export async function executeCompanyImport({
(a) => a.kVaterArtikel && a.kVaterArtikel !== '0' && a.kVaterArtikel.trim() !== ''
);
console.log(`[Import] Linking ${variantChildren.length} child variants to parents...`);
onProgress('variants', 0, variantChildren.length, 'Linking variants');
emitProgress();
// Fetch existing attribute names for this company
const attributeMap = new Map(); // lowercase name -> attributeId
@@ -780,31 +882,25 @@ export async function executeCompanyImport({
console.warn(`[Import] Failed to attach child ${childSku} to parent ${parentSku}:`, err.message);
}
onProgress('variants', processedCount, variantChildren.length, `${childSku} -> ${parentSku}`);
emitProgress();
}
}
console.log(`[Import] Linked ${variantsLinked}/${variantChildren.length} variants`);
// ─── Phase 4: Article Pictures Import ────────────────────────────────────
const imageTasks = [];
for (const { sku, pictureFiles } of articlesWithPictures) {
const lines = pictureFiles.replace(/\r/g, '\n').split('\n').map((l) => l.trim()).filter(Boolean);
for (const relPath of lines) {
for (const [sku, rec] of imagesBySku.entries()) {
for (const relPath of rec.paths) {
imageTasks.push({ sku, relPath });
}
}
console.log(`[Import] Importing ${imageTasks.length} pictures for ${articlesWithPictures.length} articles...`);
onProgress('images', 0, imageTasks.length, 'Starting article pictures import');
console.log(`[Import] Importing ${imageTasks.length} pictures for ${imagesBySku.size} articles...`);
let totalImagesImported = 0;
for (let i = 0; i < imageTasks.length; i++) {
const { sku, relPath } = imageTasks[i];
const current = i + 1;
const total = imageTasks.length;
const filename = path.basename(relPath);
const imgPath = path.join(importDir, relPath);
let skele = null;
if (fs.existsSync(imgPath)) {
try {
@@ -812,26 +908,32 @@ export async function executeCompanyImport({
await store.article.addImage(googleAuthId, sku, imgBuf);
totalImagesImported++;
skele = await computeSkeleHash(imgBuf);
// Hash is only emitted after the image is stored in the database
const skele = await computeSkeleHash(imgBuf);
emitSkeleHash(skele?.base62 || null);
markImageDone(sku);
} catch (err) {
// If already attached or format issue, log and continue
console.warn(`[Import] Image add failed for SKU ${sku} (${relPath}):`, err.message);
}
}
onImage({
type: 'article',
current,
total,
sku,
path: relPath,
filename,
skelehash: skele?.base62 || null,
width: skele?.width || null,
height: skele?.height || null,
hashLength: skele?.hashLength || 0,
});
emitProgress();
}
flushSkeleHashes();
// Reconcile done-counters so articles/categories reach their totals even
// when individual images failed
for (const sku of imagesBySku.keys()) markArticleDone(sku);
for (const catId of jtlCatIdToDashCatId.values()) {
if (doneCatIds.has(catId)) continue;
const artSet = articlesByCatId.get(catId);
if (!artSet || artSet.size === 0 || Array.from(artSet).every((s) => doneSku.has(s))) {
doneCatIds.add(catId);
progressCounts.categoriesdone++;
}
}
emitProgress(true);
console.log(`[Import] Completed article pictures: ${totalImagesImported} images processed`);
@@ -847,7 +949,8 @@ export async function executeCompanyImport({
const itemRecords = parseCsvRecords(fs.readFileSync(itemsCsvPath, 'utf8'));
totalOrders = manRecords.length;
console.log(`[Import] Manifests found: ${totalOrders} orders, ${itemRecords.length} items`);
onProgress('orders', 0, totalOrders, 'Starting orders import');
progressCounts.receiptstotal = totalOrders;
emitProgress(true);
// Group line items by kAuftrag
const itemsByAuftrag = new Map();
@@ -906,18 +1009,23 @@ export async function executeCompanyImport({
for (let i = 0; i < totalOrders; i++) {
const m = manRecords[i];
const orderNumber = String(m.cAuftragsNr || '').trim();
if (!orderNumber) continue;
if (!orderNumber) {
progressCounts.receiptsdone++;
continue;
}
if (existingOrderNumbers.has(orderNumber)) {
ordersSkipped++;
onProgress('orders', i + 1, totalOrders, `${orderNumber} (skipped)`);
progressCounts.receiptsdone++;
emitProgress();
continue;
}
const rawItems = itemsByAuftrag.get(String(m.kAuftrag || '').trim()) || [];
if (rawItems.length === 0) {
ordersSkipped++;
onProgress('orders', i + 1, totalOrders, `${orderNumber} (skipped)`);
progressCounts.receiptsdone++;
emitProgress();
continue;
}
@@ -964,8 +1072,10 @@ export async function executeCompanyImport({
console.warn(`[Import] Failed to create order ${orderNumber}:`, err.message);
}
onProgress('orders', i + 1, totalOrders, orderNumber);
progressCounts.receiptsdone++;
emitProgress();
}
emitProgress(true);
console.log(`[Import] Orders import complete: ${ordersImported} created, ${ordersSkipped} skipped`);
}