feat(product-archive): use marketing daily sales with daily sync and arrival monitor
Aggregate rpt.op_marketingDailySales by complete SKU and calculate weighted purchase sales. Persist daily source fingerprints, skip unchanged archive writes and prevent old enrichment from overwriting sales. Run daily at 09:30 Beijing time and observe source arrival every 10 minutes on 2026-10-11 08:00-09:30.
This commit is contained in:
WuYongJie
parent
f665a6d271
commit
de7518bb16
@ -0,0 +1,25 @@
|
||||
ALTER TABLE "SupplyChainProductArchive"
|
||||
ADD COLUMN "sales30d" TEXT NOT NULL DEFAULT '0.00',
|
||||
ADD COLUMN "sales30dWithBig" TEXT NOT NULL DEFAULT '0.00';
|
||||
|
||||
CREATE TABLE "ProductArchiveMarketingSalesState" (
|
||||
"id" TEXT NOT NULL,
|
||||
"state" JSONB NOT NULL,
|
||||
"snapshot" JSONB NOT NULL,
|
||||
"updatedAt" TIMESTAMP(3) NOT NULL,
|
||||
CONSTRAINT "ProductArchiveMarketingSalesState_pkey" PRIMARY KEY ("id")
|
||||
);
|
||||
|
||||
CREATE TABLE "ProductArchiveMarketingSalesCheck" (
|
||||
"id" TEXT NOT NULL,
|
||||
"checkedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"status" TEXT NOT NULL,
|
||||
"fingerprint" TEXT NOT NULL DEFAULT '',
|
||||
"sourceDate" TEXT NOT NULL DEFAULT '',
|
||||
"rowCount" INTEGER NOT NULL DEFAULT 0,
|
||||
"skuCount" INTEGER NOT NULL DEFAULT 0,
|
||||
"changedRows" INTEGER NOT NULL DEFAULT 0,
|
||||
"errorCode" TEXT NOT NULL DEFAULT '',
|
||||
CONSTRAINT "ProductArchiveMarketingSalesCheck_pkey" PRIMARY KEY ("id")
|
||||
);
|
||||
CREATE INDEX "ProductArchiveMarketingSalesCheck_checkedAt_idx" ON "ProductArchiveMarketingSalesCheck"("checkedAt");
|
||||
@ -728,6 +728,8 @@ model SupplyChainProductArchive {
|
||||
sales3d String @default("")
|
||||
sales7d String @default("")
|
||||
sales15d String @default("")
|
||||
sales30d String @default("0.00")
|
||||
sales30dWithBig String @default("0.00")
|
||||
stockJianshang String @default("")
|
||||
stockClearance String @default("")
|
||||
stockFactory String @default("")
|
||||
@ -756,6 +758,28 @@ model SupplyChainProductArchive {
|
||||
@@index([firstSourceCode, createdAt])
|
||||
}
|
||||
|
||||
// 产品全量档案日销同步状态：保存来源指纹和固定每日同步计划。
|
||||
model ProductArchiveMarketingSalesState {
|
||||
id String @id
|
||||
state Json
|
||||
snapshot Json
|
||||
updatedAt DateTime @updatedAt
|
||||
}
|
||||
|
||||
model ProductArchiveMarketingSalesCheck {
|
||||
id String @id @default(cuid())
|
||||
checkedAt DateTime @default(now())
|
||||
status String
|
||||
fingerprint String @default("")
|
||||
sourceDate String @default("")
|
||||
rowCount Int @default(0)
|
||||
skuCount Int @default(0)
|
||||
changedRows Int @default(0)
|
||||
errorCode String @default("")
|
||||
|
||||
@@index([checkedAt])
|
||||
}
|
||||
|
||||
// 月度 SABC 核算快照。主表只保留当前标签，历史月份和计算过程在这里追溯。
|
||||
model SupplyChainProductArchiveSabcResult {
|
||||
id String @id @default(cuid())
|
||||
|
||||
@ -0,0 +1,13 @@
|
||||
param([string]$TaskName = 'Furniture-ProductArchive-MarketingDailySales-3001')
|
||||
$ErrorActionPreference = 'Stop'
|
||||
$appRoot = Split-Path -Parent $PSScriptRoot
|
||||
$nodePath = (Get-Command node -ErrorAction Stop).Source
|
||||
$runner = Join-Path $PSScriptRoot 'run-product-archive-marketing-sales-local.cjs'
|
||||
if (-not (Test-Path -LiteralPath $runner)) { throw 'Marketing daily sales runner missing' }
|
||||
$action = New-ScheduledTaskAction -Execute $nodePath -Argument ("`"" + $runner + "`"") -WorkingDirectory $appRoot
|
||||
$trigger = New-ScheduledTaskTrigger -Daily -At (Get-Date -Hour 9 -Minute 30 -Second 0)
|
||||
$settings = New-ScheduledTaskSettingsSet -StartWhenAvailable -MultipleInstances IgnoreNew -ExecutionTimeLimit (New-TimeSpan -Hours 2) -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries
|
||||
$principal = New-ScheduledTaskPrincipal -UserId ([System.Security.Principal.WindowsIdentity]::GetCurrent().Name) -LogonType Interactive -RunLevel Limited
|
||||
Register-ScheduledTask -TaskName $TaskName -Action $action -Trigger $trigger -Settings $settings -Principal $principal -Description 'Daily sync of rpt.op_marketingDailySales at 09:30 Asia/Shanghai. The source exposes only a business-date createTime at midnight, so 09:30 is a conservative observed-after-arrival time.' -Force | Out-Null
|
||||
Get-ScheduledTask -TaskName $TaskName | Select-Object TaskName, State
|
||||
Get-ScheduledTaskInfo -TaskName $TaskName | Select-Object NextRunTime, LastRunTime, LastTaskResult
|
||||
@ -0,0 +1,27 @@
|
||||
param(
|
||||
[string]$TaskName = 'Furniture-ProductArchive-MarketingDailySales-Monitor-20261011-3001'
|
||||
)
|
||||
$ErrorActionPreference = 'Stop'
|
||||
$appRoot = Split-Path -Parent $PSScriptRoot
|
||||
$nodePath = (Get-Command node -ErrorAction Stop).Source
|
||||
$runner = Join-Path $PSScriptRoot 'run-product-archive-marketing-sales-local.cjs'
|
||||
if (-not (Test-Path -LiteralPath $runner)) { throw 'Marketing daily sales runner missing' }
|
||||
|
||||
$beijingZone = [TimeZoneInfo]::FindSystemTimeZoneById('China Standard Time')
|
||||
$beijingNow = [TimeZoneInfo]::ConvertTimeFromUtc([DateTime]::UtcNow, $beijingZone)
|
||||
$monitorDate = $beijingNow.Date.AddDays(1)
|
||||
$startAt = $monitorDate.AddHours(8)
|
||||
$dateOption = '--date=' + $monitorDate.ToString('yyyy-MM-dd')
|
||||
$action = New-ScheduledTaskAction -Execute $nodePath -Argument ("`"" + $runner + "`" --monitor --once " + $dateOption) -WorkingDirectory $appRoot
|
||||
$triggers = @(0..9 | ForEach-Object {
|
||||
$slot = $startAt.AddMinutes($_ * 10)
|
||||
$trigger = New-ScheduledTaskTrigger -Once -At $slot
|
||||
$trigger.StartBoundary = $slot.ToString('yyyy-MM-ddTHH:mm:ss') + '+08:00'
|
||||
$trigger
|
||||
})
|
||||
$settings = New-ScheduledTaskSettingsSet -StartWhenAvailable -MultipleInstances IgnoreNew -ExecutionTimeLimit (New-TimeSpan -Hours 2) -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries
|
||||
$principal = New-ScheduledTaskPrincipal -UserId ([System.Security.Principal.WindowsIdentity]::GetCurrent().Name) -LogonType Interactive -RunLevel Limited
|
||||
$description = 'Read-only monitor of rpt.op_marketingDailySales from 08:00 to 09:30 Beijing time on the next day; samples every 10 minutes and records the first target-date snapshot.'
|
||||
Register-ScheduledTask -TaskName $TaskName -Action $action -Trigger $triggers -Settings $settings -Principal $principal -Description $description -Force | Out-Null
|
||||
Get-ScheduledTask -TaskName $TaskName | Select-Object TaskName, State
|
||||
Get-ScheduledTaskInfo -TaskName $TaskName | Select-Object NextRunTime, LastRunTime, LastTaskResult
|
||||
@ -0,0 +1,79 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
BASE="${FURNITURE_PROD_BASE:-/opt/furniture-3d-system-prod}"
|
||||
APP_ROOT="$BASE/current/furniture-3d-system"
|
||||
SERVICE="furniture-product-archive-marketing-sales"
|
||||
PERSONAL_ENV="$BASE/shared/marketing-sales-personal.env"
|
||||
MONITOR_DATE="${MARKETING_SALES_MONITOR_DATE:-}"
|
||||
if [ "$(id -u)" -ne 0 ]; then echo "Root required" >&2; exit 1; fi
|
||||
if [ ! -f "$APP_ROOT/scripts/run-product-archive-marketing-sales-production.cjs" ] || [ ! -f "$PERSONAL_ENV" ]; then
|
||||
echo "Missing deployed runner or personal report credentials" >&2; exit 1
|
||||
fi
|
||||
chmod 600 "$PERSONAL_ENV"
|
||||
mkdir -p "$BASE/shared/logs/marketing-sales-monitor"
|
||||
cat >"/etc/systemd/system/$SERVICE.service" <<EOF
|
||||
[Unit]
|
||||
Description=Synchronize product archive marketing daily sales
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
[Service]
|
||||
Type=oneshot
|
||||
WorkingDirectory=$APP_ROOT
|
||||
Environment=NODE_ENV=production
|
||||
EnvironmentFile=$BASE/shared/.env
|
||||
EnvironmentFile=$PERSONAL_ENV
|
||||
ExecStart=/usr/bin/node scripts/run-product-archive-marketing-sales-production.cjs --sync
|
||||
TimeoutStartSec=10min
|
||||
EOF
|
||||
cat >"/etc/systemd/system/$SERVICE.timer" <<EOF
|
||||
[Unit]
|
||||
Description=Marketing daily sales at Beijing 09:30
|
||||
[Timer]
|
||||
OnCalendar=*-*-* 09:30:00 Asia/Shanghai
|
||||
AccuracySec=1s
|
||||
Persistent=true
|
||||
Unit=$SERVICE.service
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
EOF
|
||||
|
||||
if [ -n "$MONITOR_DATE" ]; then
|
||||
if [[ ! "$MONITOR_DATE" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ ]]; then echo "Invalid monitor date" >&2; exit 1; fi
|
||||
cat >"/etc/systemd/system/$SERVICE-monitor.service" <<EOF
|
||||
[Unit]
|
||||
Description=Read-only observation of marketing sales arrival
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
[Service]
|
||||
Type=oneshot
|
||||
WorkingDirectory=$APP_ROOT
|
||||
Environment=NODE_ENV=production
|
||||
Environment=MARKETING_SALES_MONITOR_DATE=$MONITOR_DATE
|
||||
Environment=MARKETING_SALES_MONITOR_LOG_DIR=$BASE/shared/logs/marketing-sales-monitor
|
||||
EnvironmentFile=$BASE/shared/.env
|
||||
EnvironmentFile=$PERSONAL_ENV
|
||||
ExecStart=/usr/bin/node scripts/run-product-archive-marketing-sales-production.cjs --monitor --once --date=$MONITOR_DATE
|
||||
TimeoutStartSec=5min
|
||||
EOF
|
||||
{
|
||||
cat <<EOF
|
||||
[Unit]
|
||||
Description=One-day marketing sales observation 08:00-09:30 Beijing
|
||||
[Timer]
|
||||
EOF
|
||||
for minute in 00 10 20 30 40 50; do echo "OnCalendar=$MONITOR_DATE 08:$minute:00 Asia/Shanghai"; done
|
||||
for minute in 00 10 20 30; do echo "OnCalendar=$MONITOR_DATE 09:$minute:00 Asia/Shanghai"; done
|
||||
cat <<EOF
|
||||
AccuracySec=1s
|
||||
Persistent=false
|
||||
Unit=$SERVICE-monitor.service
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
EOF
|
||||
} >"/etc/systemd/system/$SERVICE-monitor.timer"
|
||||
fi
|
||||
systemctl daemon-reload
|
||||
systemctl enable --now "$SERVICE.timer"
|
||||
if [ -n "$MONITOR_DATE" ]; then systemctl enable --now "$SERVICE-monitor.timer"; fi
|
||||
systemctl list-timers --all "$SERVICE*" --no-pager
|
||||
@ -0,0 +1,102 @@
|
||||
import fs from 'node:fs/promises'
|
||||
import path from 'node:path'
|
||||
import { JstMarketingDailySalesRepository } from '../src/server/repositories/jst-marketing-daily-sales.repository'
|
||||
|
||||
const INTERVAL_MINUTES = 10
|
||||
|
||||
function beijingDate(value = new Date()) {
|
||||
return new Intl.DateTimeFormat('en-CA', { timeZone: 'Asia/Shanghai' }).format(value)
|
||||
}
|
||||
|
||||
function errorCode(error: unknown) {
|
||||
const candidate = error instanceof Error ? error.message : ''
|
||||
return /^[A-Z][A-Z0-9_]{1,79}$/.test(candidate) ? candidate : 'MARKETING_SALES_MONITOR_READ_FAILED'
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const dateOption = process.argv.find(arg => arg.startsWith('--date='))?.slice('--date='.length)
|
||||
const targetDate = dateOption || process.env.MARKETING_SALES_MONITOR_DATE || beijingDate()
|
||||
if (!/^\d{4}-\d{2}-\d{2}$/.test(targetDate)) throw new Error('INVALID_MONITOR_DATE')
|
||||
if (beijingDate() !== targetDate) throw new Error('MONITOR_DATE_NOT_TODAY')
|
||||
const logDir = process.env.MARKETING_SALES_MONITOR_LOG_DIR || 'D:/ChanPing/.local-services/logs/product-archive-marketing-sales-monitor'
|
||||
await fs.mkdir(logDir, { recursive: true })
|
||||
const outputPath = path.join(logDir, `${targetDate}.json`)
|
||||
let previous: { records?: Array<Record<string, unknown>>; firstAvailableAt?: string | null } = {}
|
||||
try {
|
||||
previous = JSON.parse(await fs.readFile(outputPath, 'utf8'))
|
||||
} catch (error) {
|
||||
if (!(error && typeof error === 'object' && 'code' in error && error.code === 'ENOENT')) throw error
|
||||
}
|
||||
const records = previous.records || []
|
||||
const previousDate = new Date(new Date(`${targetDate}T00:00:00+08:00`).getTime() - 86_400_000)
|
||||
let baseline: { rowCount?: number; skuCount?: number } | undefined
|
||||
try {
|
||||
const prior = JSON.parse(await fs.readFile(path.join(logDir, `${beijingDate(previousDate)}.json`), 'utf8'))
|
||||
baseline = prior.records?.filter((record: Record<string, unknown>) => typeof record.fingerprint === 'string').at(-1)
|
||||
} catch (error) {
|
||||
if (!(error && typeof error === 'object' && 'code' in error && error.code === 'ENOENT')) throw error
|
||||
}
|
||||
const source = new JstMarketingDailySalesRepository()
|
||||
let firstAvailableAt = previous.firstAvailableAt || null
|
||||
let successfulReads = 0
|
||||
|
||||
try {
|
||||
const queryStartedAt = new Date()
|
||||
try {
|
||||
const snapshot = await source.readSnapshot()
|
||||
const observedAt = new Date()
|
||||
const previousRead = records.filter(record => typeof record.fingerprint === 'string').at(-1)
|
||||
const countsComplete = snapshot.rowCount >= Number(baseline?.rowCount || 0) * 0.8
|
||||
&& snapshot.rows.length >= Number(baseline?.skuCount || 0) * 0.8
|
||||
const available = snapshot.sourceDate === targetDate && countsComplete
|
||||
if (available && !firstAvailableAt) firstAvailableAt = observedAt.toISOString()
|
||||
records.push({
|
||||
queryStartedAt: queryStartedAt.toISOString(),
|
||||
observedAt: observedAt.toISOString(),
|
||||
beijingObservedAt: observedAt.toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai', hour12: false }),
|
||||
sourceDate: snapshot.sourceDate,
|
||||
rowCount: snapshot.rowCount,
|
||||
skuCount: snapshot.rows.length,
|
||||
fingerprint: snapshot.fingerprint,
|
||||
changedFromPrevious: previousRead ? previousRead.fingerprint !== snapshot.fingerprint : null,
|
||||
targetDate,
|
||||
available,
|
||||
countsComplete,
|
||||
})
|
||||
successfulReads += 1
|
||||
} catch (error) {
|
||||
records.push({ observedAt: new Date().toISOString(), queryStartedAt: queryStartedAt.toISOString(), targetDate, error: errorCode(error) })
|
||||
}
|
||||
await fs.writeFile(outputPath, JSON.stringify({
|
||||
targetDate,
|
||||
window: { start: '08:00', end: '09:30', intervalMinutes: INTERVAL_MINUTES },
|
||||
firstAvailableAt,
|
||||
firstAvailableBeijingAt: firstAvailableAt
|
||||
? new Date(firstAvailableAt).toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai', hour12: false })
|
||||
: null,
|
||||
lastNotAvailableAt: records.filter(record => record.available === false
|
||||
&& (!firstAvailableAt || String(record.observedAt) < firstAvailableAt)).at(-1)?.observedAt || null,
|
||||
records,
|
||||
updatedAt: new Date().toISOString(),
|
||||
}, null, 2) + '\n', 'utf8')
|
||||
} finally {
|
||||
await source.close().catch(() => undefined)
|
||||
}
|
||||
|
||||
console.log(JSON.stringify({
|
||||
targetDate,
|
||||
firstAvailableAt,
|
||||
firstAvailableBeijingAt: firstAvailableAt
|
||||
? new Date(firstAvailableAt).toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai', hour12: false })
|
||||
: null,
|
||||
successfulReads,
|
||||
recordCount: records.length,
|
||||
outputPath,
|
||||
}, null, 2))
|
||||
if (successfulReads === 0) process.exitCode = 1
|
||||
}
|
||||
|
||||
void main().catch(error => {
|
||||
console.error(errorCode(error))
|
||||
process.exitCode = 1
|
||||
})
|
||||
@ -0,0 +1,59 @@
|
||||
const fs = require('node:fs')
|
||||
const path = require('node:path')
|
||||
const { spawn } = require('node:child_process')
|
||||
const { loadEnvConfig } = require('@next/env')
|
||||
|
||||
const appRoot = path.resolve(__dirname, '..')
|
||||
loadEnvConfig(appRoot)
|
||||
|
||||
if (!process.env.DATABASE_URL) throw new Error('LOCAL_DATABASE_URL_UNAVAILABLE')
|
||||
const database = new URL(process.env.DATABASE_URL)
|
||||
if (!['127.0.0.1', 'localhost'].includes(database.hostname)
|
||||
|| database.port !== '55432'
|
||||
|| database.pathname !== '/furniture3d_local_clean_20260807') {
|
||||
throw new Error('UNEXPECTED_LOCAL_DATABASE_TARGET')
|
||||
}
|
||||
|
||||
function readCredentials() {
|
||||
const secretsPath = 'D:/ChanPing/LOCAL_SECRETS.md'
|
||||
const text = fs.readFileSync(secretsPath, 'utf8')
|
||||
const section = text.split(/^##\s+/m).find(part => part.startsWith('公司数据库')) || ''
|
||||
const account = section.match(/账号[：:]\s*`([^`]+)`/)
|
||||
const password = section.match(/密码[：:]\s*`([^`]+)`/)
|
||||
if (!account || !password) throw new Error('COMPANY_DB_CREDENTIALS_UNAVAILABLE')
|
||||
return { user: account[1], password: password[1] }
|
||||
}
|
||||
|
||||
if (process.env.JST_REPORT_DB_HOST !== '218.93.9.21'
|
||||
|| process.env.JST_REPORT_DB_NAME !== 'JSReportReplica'
|
||||
|| (process.env.JST_REPORT_DB_SCHEMA || 'rpt') !== 'rpt') {
|
||||
throw new Error('UNEXPECTED_COMPANY_DB_TARGET')
|
||||
}
|
||||
const credentials = readCredentials()
|
||||
process.env.JST_REPORT_DB_USER = credentials.user
|
||||
process.env.JST_REPORT_DB_PASSWORD = credentials.password
|
||||
|
||||
const logDir = 'D:/ChanPing/.local-services/logs/product-archive-marketing-sales'
|
||||
fs.mkdirSync(logDir, { recursive: true })
|
||||
const file = path.join(logDir, new Date().toISOString().slice(0, 10) + '.log')
|
||||
const log = fs.openSync(file, 'a')
|
||||
const mode = process.argv[2] || '--sync'
|
||||
if (!['--sync', '--monitor'].includes(mode)) throw new Error('INVALID_MARKETING_SALES_MODE')
|
||||
const entry = mode === '--monitor'
|
||||
? 'scripts/monitor-product-archive-marketing-sales.ts'
|
||||
: 'scripts/sync-product-archive-marketing-sales.ts'
|
||||
const childArgs = [require.resolve('tsx/cli'), entry, ...process.argv.slice(3)]
|
||||
fs.writeSync(log, '\nSTART ' + new Date().toISOString() + ' mode=' + mode + '\n')
|
||||
const child = spawn(process.execPath, childArgs, {
|
||||
cwd: appRoot, env: process.env, windowsHide: true, stdio: ['ignore', log, log],
|
||||
})
|
||||
child.on('error', () => {
|
||||
fs.writeSync(log, 'Runner failed to launch\n')
|
||||
fs.closeSync(log)
|
||||
process.exitCode = 1
|
||||
})
|
||||
child.on('exit', code => {
|
||||
fs.writeSync(log, 'END ' + new Date().toISOString() + ' exit=' + code + '\n')
|
||||
fs.closeSync(log)
|
||||
process.exitCode = code === 0 ? 0 : 1
|
||||
})
|
||||
@ -0,0 +1,24 @@
|
||||
const path = require('node:path')
|
||||
const { spawn } = require('node:child_process')
|
||||
const { loadEnvConfig } = require('@next/env')
|
||||
|
||||
const appRoot = path.resolve(__dirname, '..')
|
||||
loadEnvConfig(appRoot)
|
||||
if (process.env.JST_REPORT_DB_HOST !== '218.93.9.21'
|
||||
|| process.env.JST_REPORT_DB_NAME !== 'JSReportReplica'
|
||||
|| (process.env.JST_REPORT_DB_SCHEMA || 'rpt') !== 'rpt') {
|
||||
throw new Error('UNEXPECTED_COMPANY_DB_TARGET')
|
||||
}
|
||||
if (!process.env.JST_REPORT_DB_USER || !process.env.JST_REPORT_DB_PASSWORD) {
|
||||
throw new Error('PERSONAL_REPORT_DB_CREDENTIALS_REQUIRED')
|
||||
}
|
||||
const mode = process.argv[2] || '--sync'
|
||||
if (!['--sync', '--monitor'].includes(mode)) throw new Error('INVALID_MARKETING_SALES_MODE')
|
||||
const entry = mode === '--monitor'
|
||||
? 'scripts/monitor-product-archive-marketing-sales.ts'
|
||||
: 'scripts/sync-product-archive-marketing-sales.ts'
|
||||
const child = spawn(process.execPath, [require.resolve('tsx/cli'), entry, ...process.argv.slice(3)], {
|
||||
cwd: appRoot, env: process.env, stdio: 'inherit',
|
||||
})
|
||||
child.on('error', () => { console.error('MARKETING_SALES_RUNNER_LAUNCH_FAILED'); process.exitCode = 1 })
|
||||
child.on('exit', code => { process.exitCode = code === 0 ? 0 : 1 })
|
||||
@ -0,0 +1,15 @@
|
||||
import { loadEnvConfig } from '@next/env'
|
||||
import path from 'node:path'
|
||||
import { productArchiveMarketingSalesService } from '../src/server/services/product-archive-marketing-sales.service'
|
||||
|
||||
loadEnvConfig(path.resolve(__dirname, '..'))
|
||||
|
||||
async function main() {
|
||||
const result = await productArchiveMarketingSalesService.run()
|
||||
console.log(JSON.stringify({ status: result.status, sourceDate: result.snapshot?.sourceDate, fingerprint: result.snapshot?.fingerprint, matched: result.result?.matched }, null, 2))
|
||||
}
|
||||
|
||||
main().catch(error => {
|
||||
console.error(error instanceof Error ? error.stack ?? error.message : 'MARKETING_SALES_SYNC_FAILED')
|
||||
process.exitCode = 1
|
||||
})
|
||||
@ -108,6 +108,7 @@ export const ARCHIVE_COLUMNS: ArchiveColumnGroup[] = [
|
||||
{ prop: 'sales3d', label: '3日日销', width: 90 },
|
||||
{ prop: 'sales7d', label: '7日日销', width: 90 },
|
||||
{ prop: 'sales15d', label: '15日日销', width: 90 },
|
||||
{ prop: 'sales30d', label: '30日日销', width: 90 },
|
||||
{ prop: 'sales30dWithBig', label: '含大单30日日销', width: 140 },
|
||||
{ prop: 'jianshangStockDays', label: '简尚库存可使用天数', width: 160 },
|
||||
{ prop: 'stockDays', label: '总库存可使用天数', width: 150 },
|
||||
|
||||
@ -0,0 +1,92 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
|
||||
export const MARKETING_SALES_SOURCE_CODE = 'JST_MARKETING_DAILY_SALES'
|
||||
export const MARKETING_SALES_COLUMNS = {
|
||||
dailySales: 'SUM(fifteenAvgSale)*0.2+SUM(sevenAvgSale)*0.3+SUM(threeAvgSale)*0.5',
|
||||
sales3d: 'SUM(threeAvgSale)',
|
||||
sales7d: 'SUM(sevenAvgSale)',
|
||||
sales15d: 'SUM(fifteenAvgSale)',
|
||||
sales30d: 'SUM(thirtyAvgSale)',
|
||||
sales30dWithBig: 'SUM(thirtyBigSale)',
|
||||
} as const
|
||||
|
||||
export type MarketingSalesValues = Record<keyof typeof MARKETING_SALES_COLUMNS, string>
|
||||
export type MarketingSalesRow = MarketingSalesValues & { sku: string; rowCount: number }
|
||||
export type MarketingSalesSnapshot = {
|
||||
sourceDate: string
|
||||
fingerprint: string
|
||||
rowCount: number
|
||||
rows: MarketingSalesRow[]
|
||||
}
|
||||
export type MarketingSalesSnapshotSummary = {
|
||||
sourceDate: string
|
||||
fingerprint: string
|
||||
rowCount: number
|
||||
skuCount: number
|
||||
}
|
||||
export type MarketingSalesAggregate = {
|
||||
sku: string; sourceDate: string; rowCount: number; invalidCount: number
|
||||
threeAvgSale: string; sevenAvgSale: string; fifteenAvgSale: string
|
||||
thirtyAvgSale: string; thirtyBigSale: string
|
||||
}
|
||||
|
||||
function cents(value: string) {
|
||||
if (!/^-?\d+\.\d{2}$/.test(value)) throw new Error('INVALID_SALES_DECIMAL')
|
||||
return BigInt(value.replace('.', ''))
|
||||
}
|
||||
|
||||
function text(value: bigint) {
|
||||
const zero = BigInt(0)
|
||||
const hundred = BigInt(100)
|
||||
const absolute = value < zero ? -value : value
|
||||
return `${value < zero ? '-' : ''}${absolute / hundred}.${String(absolute % hundred).padStart(2, '0')}`
|
||||
}
|
||||
|
||||
export function marketingSalesValues(row?: MarketingSalesAggregate): MarketingSalesValues {
|
||||
if (!row) return { dailySales: '0.00', sales3d: '0.00', sales7d: '0.00',
|
||||
sales15d: '0.00', sales30d: '0.00', sales30dWithBig: '0.00' }
|
||||
const three = cents(row.threeAvgSale), seven = cents(row.sevenAvgSale), fifteen = cents(row.fifteenAvgSale)
|
||||
const weighted = fifteen * BigInt(2) + seven * BigInt(3) + three * BigInt(5)
|
||||
// Integer arithmetic avoids binary floating point; round half away from zero.
|
||||
const rounded = weighted < BigInt(0)
|
||||
? -((-weighted + BigInt(5)) / BigInt(10))
|
||||
: (weighted + BigInt(5)) / BigInt(10)
|
||||
return { dailySales: text(rounded), sales3d: text(three), sales7d: text(seven),
|
||||
sales15d: text(fifteen), sales30d: text(cents(row.thirtyAvgSale)),
|
||||
sales30dWithBig: text(cents(row.thirtyBigSale)) }
|
||||
}
|
||||
|
||||
export function buildMarketingSalesSnapshot(aggregates: MarketingSalesAggregate[]): MarketingSalesSnapshot {
|
||||
if (!aggregates.length) throw new Error('EMPTY_SOURCE_SNAPSHOT')
|
||||
const batches = new Set(aggregates.map(row => row.sourceDate))
|
||||
if (batches.size !== 1 || !/^\d{4}-\d{2}-\d{2}$/.test(aggregates[0].sourceDate)) {
|
||||
throw new Error('MIXED_OR_MISSING_SOURCE_BATCH')
|
||||
}
|
||||
const seen = new Set<string>()
|
||||
const rows = aggregates.map(row => {
|
||||
const sku = row.sku.trim()
|
||||
if (!sku || seen.has(sku) || row.invalidCount || row.rowCount < 1) {
|
||||
throw new Error('INCOMPLETE_SOURCE_SNAPSHOT')
|
||||
}
|
||||
seen.add(sku)
|
||||
return { sku, rowCount: row.rowCount, ...marketingSalesValues(row) }
|
||||
}).sort((a, b) => a.sku < b.sku ? -1 : a.sku > b.sku ? 1 : 0)
|
||||
const sourceDate = aggregates[0].sourceDate
|
||||
const fingerprint = createHash('sha256').update(JSON.stringify([sourceDate, rows])).digest('hex')
|
||||
return { sourceDate, fingerprint, rows, rowCount: rows.reduce((sum, row) => sum + row.rowCount, 0) }
|
||||
}
|
||||
|
||||
export function validateMarketingSalesSnapshot(
|
||||
next: MarketingSalesSnapshot,
|
||||
previous?: MarketingSalesSnapshot | MarketingSalesSnapshotSummary | null,
|
||||
) {
|
||||
if (next.sourceDate !== new Intl.DateTimeFormat('en-CA', { timeZone: 'Asia/Shanghai' }).format(new Date())) {
|
||||
throw new Error('STALE_OR_FUTURE_SOURCE_BATCH')
|
||||
}
|
||||
if (!previous) return
|
||||
if (next.sourceDate < previous.sourceDate) throw new Error('SOURCE_BATCH_REGRESSED')
|
||||
const previousSkuCount = 'rows' in previous ? previous.rows.length : previous.skuCount
|
||||
if (next.rowCount < previous.rowCount * 0.8 || next.rows.length < previousSkuCount * 0.8) {
|
||||
throw new Error('SOURCE_COUNT_DROPPED_REQUIRES_REVIEW')
|
||||
}
|
||||
}
|
||||
@ -52,6 +52,8 @@ export const PRODUCT_ARCHIVE_CHANGE_FIELDS = [
|
||||
{ key: 'sales3d', label: '3日日销' },
|
||||
{ key: 'sales7d', label: '7日日销' },
|
||||
{ key: 'sales15d', label: '15日日销' },
|
||||
{ key: 'sales30d', label: '30日日销' },
|
||||
{ key: 'sales30dWithBig', label: '含大单30日日销' },
|
||||
{ key: 'stockJianshang', label: '简尚库存' },
|
||||
{ key: 'stockClearance', label: '清货库存' },
|
||||
{ key: 'stockFactory', label: '工厂库存' },
|
||||
|
||||
@ -7,6 +7,7 @@ export const YUNSHAN_PRODUCT_EXPORT_ARCHIVE_SOURCE_CODE = 'YUNSHAN_PRODUCT_EXPOR
|
||||
export const DRAWING_MEDIA_ARCHIVE_SOURCE_CODE = 'DRAWING_MEDIA_EXPORT'
|
||||
export const JST_SUPPLIER_ARCHIVE_SOURCE_CODE = 'JST_SUPPLIER'
|
||||
export const JST_STOCK_MONITOR_ARCHIVE_SOURCE_CODE = 'JST_STOCK_MONITOR'
|
||||
export const JST_MARKETING_DAILY_SALES_ARCHIVE_SOURCE_CODE = 'JST_MARKETING_DAILY_SALES'
|
||||
export const JST_CLEARANCE_STOCK_ARCHIVE_SOURCE_CODE = 'JST_CLEARANCE_STOCK'
|
||||
export const JST_FACTORY_STOCK_ARCHIVE_SOURCE_CODE = 'JST_FACTORY_STOCK'
|
||||
export const SUPPLY_CHAIN_OWNER_CONFIG_ARCHIVE_SOURCE_CODE = 'SUPPLY_CHAIN_OWNER_CONFIG'
|
||||
|
||||
@ -56,6 +56,7 @@ export type SupplyChainProductArchiveRow = {
|
||||
sales3d: string
|
||||
sales7d: string
|
||||
sales15d: string
|
||||
sales30d: string
|
||||
stockClearance: string
|
||||
purchasePrice: string
|
||||
costPrice: string
|
||||
@ -231,6 +232,7 @@ export type ProductArchiveFilters = {
|
||||
priceType?: string[]
|
||||
dataTag?: string[]
|
||||
sales30dWithBig?: string[]
|
||||
sales30d?: string[]
|
||||
jianshangStockDays?: string[]
|
||||
stockDays?: string[]
|
||||
stockJianshang?: string[]
|
||||
@ -279,7 +281,7 @@ export const PRODUCT_ARCHIVE_FILTER_OPTION_KEYS = [
|
||||
'palletSpec', 'backupSupplier', 'opsDirector', 'opsDeptHead', 'opsHead',
|
||||
'inOutWarehouse', 'warehouseHead', 'warehouseOwnership', 'warehouseType', 'priceType',
|
||||
'purchasePrice', 'costPrice', 'costPriceSecondary', 'dataTag', 'dailySales', 'sales3d',
|
||||
'sales7d', 'sales15d', 'sales30dWithBig', 'jianshangStockDays', 'stockDays',
|
||||
'sales7d', 'sales15d', 'sales30d', 'sales30dWithBig', 'jianshangStockDays', 'stockDays',
|
||||
'stockJianshang', 'stockClearance', 'stockFactory', 'stockCloud', 'inTransitQty',
|
||||
'stockTurnoverDays', 'stockValue', 'slowStock', 'slowStockValue', 'afterSaleAmount', 'afterSalesCount',
|
||||
'afterSalesRate', 'afterSalesCost', 'afterSalesCostRate', 'returnQty', 'returnRate',
|
||||
|
||||
@ -0,0 +1,57 @@
|
||||
import sql from 'mssql'
|
||||
import { buildMarketingSalesSnapshot, type MarketingSalesAggregate } from '@/lib/product-archive-marketing-sales'
|
||||
|
||||
function required(name: string) {
|
||||
const value = process.env[name]?.trim()
|
||||
if (!value) throw new Error(`MISSING_ENV_${name}`)
|
||||
return value
|
||||
}
|
||||
|
||||
export class JstMarketingDailySalesRepository {
|
||||
private pool: sql.ConnectionPool | null = null
|
||||
|
||||
async readSnapshot() {
|
||||
if (!this.pool?.connected) {
|
||||
this.pool = await new sql.ConnectionPool({
|
||||
server: required('JST_REPORT_DB_HOST'),
|
||||
port: Number(process.env.JST_REPORT_DB_PORT || 1433),
|
||||
database: required('JST_REPORT_DB_NAME'),
|
||||
user: required('JST_REPORT_DB_USER'), password: required('JST_REPORT_DB_PASSWORD'),
|
||||
options: { encrypt: process.env.JST_REPORT_DB_ENCRYPT !== 'false',
|
||||
trustServerCertificate: process.env.JST_REPORT_DB_TRUST_SERVER_CERTIFICATE !== 'false' },
|
||||
pool: { min: 0, max: 1, idleTimeoutMillis: 30_000 },
|
||||
requestTimeout: 60_000, connectionTimeout: 15_000,
|
||||
}).connect()
|
||||
}
|
||||
// One read transaction prevents a concurrent replacement from producing a torn scan.
|
||||
// This connection executes SELECT only; no source writes, DDL or NOLOCK.
|
||||
const transaction = new sql.Transaction(this.pool)
|
||||
await transaction.begin(sql.ISOLATION_LEVEL.SERIALIZABLE)
|
||||
try {
|
||||
const result = await new sql.Request(transaction).query<MarketingSalesAggregate>(`
|
||||
SELECT LTRIM(RTRIM(CAST(sku AS varchar(100)))) AS sku,
|
||||
CONVERT(varchar(10), createTime, 23) AS sourceDate,
|
||||
COUNT(*) AS [rowCount],
|
||||
SUM(CASE WHEN createTime IS NULL OR threeAvgSale IS NULL OR sevenAvgSale IS NULL
|
||||
OR fifteenAvgSale IS NULL OR thirtyAvgSale IS NULL OR thirtyBigSale IS NULL THEN 1 ELSE 0 END) AS invalidCount,
|
||||
CONVERT(varchar(60), SUM(CAST(threeAvgSale AS decimal(28,2)))) AS threeAvgSale,
|
||||
CONVERT(varchar(60), SUM(CAST(sevenAvgSale AS decimal(28,2)))) AS sevenAvgSale,
|
||||
CONVERT(varchar(60), SUM(CAST(fifteenAvgSale AS decimal(28,2)))) AS fifteenAvgSale,
|
||||
CONVERT(varchar(60), SUM(CAST(thirtyAvgSale AS decimal(28,2)))) AS thirtyAvgSale,
|
||||
CONVERT(varchar(60), SUM(CAST(thirtyBigSale AS decimal(28,2)))) AS thirtyBigSale
|
||||
FROM [rpt].[op_marketingDailySales]
|
||||
GROUP BY LTRIM(RTRIM(CAST(sku AS varchar(100)))), CONVERT(varchar(10), createTime, 23)
|
||||
`)
|
||||
await transaction.commit()
|
||||
return buildMarketingSalesSnapshot(result.recordset)
|
||||
} catch (error) {
|
||||
await transaction.rollback().catch(() => undefined)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
async close() {
|
||||
await this.pool?.close()
|
||||
this.pool = null
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,128 @@
|
||||
import { Prisma } from '@prisma/client'
|
||||
import { prisma } from '@/lib/prisma'
|
||||
import type { MarketingSalesSnapshot } from '@/lib/product-archive-marketing-sales'
|
||||
import { MARKETING_SALES_SOURCE_CODE } from '@/lib/product-archive-marketing-sales'
|
||||
import type { MarketingSalesState } from '@/server/services/product-archive-marketing-sales-schedule'
|
||||
|
||||
const STATE_ID = 'default'
|
||||
const SALES_FIELDS = ['dailySales', 'sales3d', 'sales7d', 'sales15d', 'sales30d', 'sales30dWithBig'] as const
|
||||
|
||||
function sourceRowsCte(payload: string, sourceUpdatedAt: string) {
|
||||
return Prisma.sql`
|
||||
WITH source_rows AS (
|
||||
SELECT r.*,
|
||||
jsonb_build_object(
|
||||
'dailySales', jsonb_build_object('sourceCode', CAST(${MARKETING_SALES_SOURCE_CODE} AS text), 'sourceSystem', 'JSReportReplica', 'sourceSchema', 'rpt', 'sourceObject', 'op_marketingDailySales', 'sourceColumn', 'weighted', 'sourceRecordId', r.sku, 'sourceUpdatedAt', CAST(${sourceUpdatedAt} AS text)),
|
||||
'sales3d', jsonb_build_object('sourceCode', CAST(${MARKETING_SALES_SOURCE_CODE} AS text), 'sourceSystem', 'JSReportReplica', 'sourceSchema', 'rpt', 'sourceObject', 'op_marketingDailySales', 'sourceColumn', 'threeAvgSale', 'sourceRecordId', r.sku, 'sourceUpdatedAt', CAST(${sourceUpdatedAt} AS text)),
|
||||
'sales7d', jsonb_build_object('sourceCode', CAST(${MARKETING_SALES_SOURCE_CODE} AS text), 'sourceSystem', 'JSReportReplica', 'sourceSchema', 'rpt', 'sourceObject', 'op_marketingDailySales', 'sourceColumn', 'sevenAvgSale', 'sourceRecordId', r.sku, 'sourceUpdatedAt', CAST(${sourceUpdatedAt} AS text)),
|
||||
'sales15d', jsonb_build_object('sourceCode', CAST(${MARKETING_SALES_SOURCE_CODE} AS text), 'sourceSystem', 'JSReportReplica', 'sourceSchema', 'rpt', 'sourceObject', 'op_marketingDailySales', 'sourceColumn', 'fifteenAvgSale', 'sourceRecordId', r.sku, 'sourceUpdatedAt', CAST(${sourceUpdatedAt} AS text)),
|
||||
'sales30d', jsonb_build_object('sourceCode', CAST(${MARKETING_SALES_SOURCE_CODE} AS text), 'sourceSystem', 'JSReportReplica', 'sourceSchema', 'rpt', 'sourceObject', 'op_marketingDailySales', 'sourceColumn', 'thirtyAvgSale', 'sourceRecordId', r.sku, 'sourceUpdatedAt', CAST(${sourceUpdatedAt} AS text)),
|
||||
'sales30dWithBig', jsonb_build_object('sourceCode', CAST(${MARKETING_SALES_SOURCE_CODE} AS text), 'sourceSystem', 'JSReportReplica', 'sourceSchema', 'rpt', 'sourceObject', 'op_marketingDailySales', 'sourceColumn', 'thirtyBigSale', 'sourceRecordId', r.sku, 'sourceUpdatedAt', CAST(${sourceUpdatedAt} AS text))
|
||||
) AS source_map
|
||||
FROM jsonb_to_recordset(${payload}::jsonb) AS r(sku text, "dailySales" text, "sales3d" text, "sales7d" text, "sales15d" text, "sales30d" text, "sales30dWithBig" text)
|
||||
)
|
||||
`
|
||||
}
|
||||
|
||||
function sourceMapDrift(alias: 'a', rowAlias: 'r') {
|
||||
const checks = SALES_FIELDS.map(field => Prisma.raw(`${alias}."fieldSourceMap"->'${field}' IS DISTINCT FROM ${rowAlias}.source_map->'${field}'`))
|
||||
return Prisma.sql`(${Prisma.join(checks, ' OR ')})`
|
||||
}
|
||||
|
||||
export class ProductArchiveMarketingSalesRepository {
|
||||
async getState() {
|
||||
return prisma.productArchiveMarketingSalesState.findUnique({ where: { id: STATE_ID } })
|
||||
}
|
||||
|
||||
async applySnapshot(snapshot: MarketingSalesSnapshot, state: MarketingSalesState, options: { status?: string } = {}) {
|
||||
if (snapshot.rows.length === 0) throw new Error('EMPTY_APPLY_SNAPSHOT')
|
||||
const rows = snapshot.rows.map(row => ({
|
||||
sku: row.sku,
|
||||
dailySales: row.dailySales,
|
||||
sales3d: row.sales3d,
|
||||
sales7d: row.sales7d,
|
||||
sales15d: row.sales15d,
|
||||
sales30d: row.sales30d,
|
||||
sales30dWithBig: row.sales30dWithBig,
|
||||
}))
|
||||
const payload = JSON.stringify(rows)
|
||||
const sourceDate = new Date(`${snapshot.sourceDate}T00:00:00.000Z`)
|
||||
const sourceUpdatedAt = sourceDate.toISOString()
|
||||
const snapshotSummary = { sourceDate: snapshot.sourceDate, fingerprint: snapshot.fingerprint, rowCount: snapshot.rowCount, skuCount: snapshot.rows.length }
|
||||
return prisma.$transaction(async tx => {
|
||||
const reset = await tx.$executeRaw(Prisma.sql`
|
||||
${sourceRowsCte(payload, sourceUpdatedAt)}
|
||||
UPDATE "SupplyChainProductArchive" AS a
|
||||
SET "dailySales" = '0.00', "sales3d" = '0.00', "sales7d" = '0.00', "sales15d" = '0.00', "sales30d" = '0.00', "sales30dWithBig" = '0.00',
|
||||
"fieldSourceMap" = COALESCE(a."fieldSourceMap", '{}'::jsonb) - ARRAY['dailySales', 'sales3d', 'sales7d', 'sales15d', 'sales30d', 'sales30dWithBig']::text[]
|
||||
WHERE NOT EXISTS (SELECT 1 FROM source_rows r WHERE r.sku = a."barcode")
|
||||
AND (a."dailySales" IS DISTINCT FROM '0.00' OR a."sales3d" IS DISTINCT FROM '0.00' OR a."sales7d" IS DISTINCT FROM '0.00'
|
||||
OR a."sales15d" IS DISTINCT FROM '0.00' OR a."sales30d" IS DISTINCT FROM '0.00' OR a."sales30dWithBig" IS DISTINCT FROM '0.00'
|
||||
OR COALESCE(a."fieldSourceMap", '{}'::jsonb) ?| ARRAY['dailySales', 'sales3d', 'sales7d', 'sales15d', 'sales30d', 'sales30dWithBig']::text[])
|
||||
`)
|
||||
const matched = await tx.$executeRaw(Prisma.sql`
|
||||
${sourceRowsCte(payload, sourceUpdatedAt)}
|
||||
UPDATE "SupplyChainProductArchive" AS a
|
||||
SET "dailySales" = r."dailySales", "sales3d" = r."sales3d", "sales7d" = r."sales7d",
|
||||
"sales15d" = r."sales15d", "sales30d" = r."sales30d", "sales30dWithBig" = r."sales30dWithBig",
|
||||
"fieldSourceMap" = COALESCE(a."fieldSourceMap", '{}'::jsonb) || r.source_map
|
||||
FROM source_rows r
|
||||
WHERE a."barcode" = r.sku
|
||||
AND ((a."dailySales", a."sales3d", a."sales7d", a."sales15d", a."sales30d", a."sales30dWithBig")
|
||||
IS DISTINCT FROM (r."dailySales", r."sales3d", r."sales7d", r."sales15d", r."sales30d", r."sales30dWithBig")
|
||||
OR ${sourceMapDrift('a', 'r')})
|
||||
`)
|
||||
await tx.$executeRaw(Prisma.sql`
|
||||
INSERT INTO "SupplyChainProductArchiveSource"
|
||||
("id", "archiveId", "sourceCode", "entryMode", "sourceRecordId", "sourceProductId", "sourceSkuItemId", "sourceSystem", "sourceSchema", "sourceObject", "sourceCreatedAt", "lastSyncedAt", "createdAt", "updatedAt")
|
||||
SELECT 'marketing-' || md5(a."id" || r.sku), a."id", ${MARKETING_SALES_SOURCE_CODE}, 'AUTO', a."barcode", a."spu", a."barcode", 'JSReportReplica', 'rpt', 'op_marketingDailySales', ${sourceDate}, NOW(), NOW(), NOW()
|
||||
FROM "SupplyChainProductArchive" a JOIN jsonb_to_recordset(${payload}::jsonb) AS r(sku text) ON a."barcode" = r.sku
|
||||
ON CONFLICT ("sourceCode", "sourceRecordId") DO UPDATE SET "archiveId" = EXCLUDED."archiveId", "lastSyncedAt" = NOW(), "updatedAt" = NOW()
|
||||
`)
|
||||
await tx.productArchiveMarketingSalesState.upsert({ where: { id: STATE_ID }, create: { id: STATE_ID, state, snapshot: snapshotSummary }, update: { state, snapshot: snapshotSummary } })
|
||||
await tx.productArchiveMarketingSalesCheck.create({ data: { status: options.status ?? 'APPLIED', fingerprint: snapshot.fingerprint, sourceDate: snapshot.sourceDate, rowCount: snapshot.rowCount, skuCount: snapshot.rows.length, changedRows: matched } })
|
||||
return { reset, matched, sourceRows: snapshot.rows.length }
|
||||
}, { maxWait: 10_000, timeout: 120_000 })
|
||||
}
|
||||
|
||||
async countSnapshotDrift(snapshot: MarketingSalesSnapshot) {
|
||||
const payload = JSON.stringify(snapshot.rows)
|
||||
const sourceUpdatedAt = new Date(`${snapshot.sourceDate}T00:00:00.000Z`).toISOString()
|
||||
const result = await prisma.$queryRaw<Array<{ count: bigint }>>(Prisma.sql`
|
||||
${sourceRowsCte(payload, sourceUpdatedAt)}
|
||||
SELECT COUNT(*)::bigint AS count
|
||||
FROM "SupplyChainProductArchive" AS a
|
||||
LEFT JOIN source_rows r ON r.sku = a."barcode"
|
||||
WHERE (r.sku IS NOT NULL
|
||||
AND ((a."dailySales", a."sales3d", a."sales7d", a."sales15d", a."sales30d", a."sales30dWithBig")
|
||||
IS DISTINCT FROM (r."dailySales", r."sales3d", r."sales7d", r."sales15d", r."sales30d", r."sales30dWithBig")
|
||||
OR ${sourceMapDrift('a', 'r')}))
|
||||
OR (r.sku IS NULL
|
||||
AND (a."dailySales" IS DISTINCT FROM '0.00' OR a."sales3d" IS DISTINCT FROM '0.00' OR a."sales7d" IS DISTINCT FROM '0.00'
|
||||
OR a."sales15d" IS DISTINCT FROM '0.00' OR a."sales30d" IS DISTINCT FROM '0.00' OR a."sales30dWithBig" IS DISTINCT FROM '0.00'
|
||||
OR COALESCE(a."fieldSourceMap", '{}'::jsonb) ?| ARRAY['dailySales', 'sales3d', 'sales7d', 'sales15d', 'sales30d', 'sales30dWithBig']::text[]))
|
||||
`)
|
||||
return Number(result[0]?.count ?? BigInt(0))
|
||||
}
|
||||
|
||||
async recordUnchanged(snapshot: MarketingSalesSnapshot, state: MarketingSalesState) {
|
||||
const snapshotSummary = { sourceDate: snapshot.sourceDate, fingerprint: snapshot.fingerprint, rowCount: snapshot.rowCount, skuCount: snapshot.rows.length }
|
||||
await prisma.$transaction(async tx => {
|
||||
await tx.productArchiveMarketingSalesState.upsert({ where: { id: STATE_ID }, create: { id: STATE_ID, state, snapshot: snapshotSummary }, update: { state, snapshot: snapshotSummary } })
|
||||
await tx.productArchiveMarketingSalesCheck.create({ data: { status: 'UNCHANGED', fingerprint: snapshot.fingerprint, sourceDate: snapshot.sourceDate, rowCount: snapshot.rowCount, skuCount: snapshot.rows.length } })
|
||||
})
|
||||
}
|
||||
|
||||
async recordFailure(state: MarketingSalesState, snapshot: unknown, errorCode: string) {
|
||||
await prisma.$transaction(async tx => {
|
||||
await tx.productArchiveMarketingSalesState.upsert({
|
||||
where: { id: STATE_ID },
|
||||
create: { id: STATE_ID, state, snapshot: (snapshot ?? {}) as Prisma.InputJsonValue },
|
||||
update: { state, ...(snapshot ? { snapshot: snapshot as Prisma.InputJsonValue } : {}) },
|
||||
})
|
||||
await tx.productArchiveMarketingSalesCheck.create({ data: { status: 'FAILED', errorCode } })
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
export const productArchiveMarketingSalesRepository = new ProductArchiveMarketingSalesRepository()
|
||||
@ -31,10 +31,6 @@ export const PRODUCT_ARCHIVE_ENRICHMENT_FIELDS = [
|
||||
'minimumOrderQuantity',
|
||||
'firstLeadTime',
|
||||
'palletSpec',
|
||||
'dailySales',
|
||||
'sales3d',
|
||||
'sales7d',
|
||||
'sales15d',
|
||||
'stockJianshang',
|
||||
'stockClearance',
|
||||
'stockFactory',
|
||||
@ -102,10 +98,6 @@ const currentArchiveSelect = {
|
||||
minimumOrderQuantity: true,
|
||||
firstLeadTime: true,
|
||||
palletSpec: true,
|
||||
dailySales: true,
|
||||
sales3d: true,
|
||||
sales7d: true,
|
||||
sales15d: true,
|
||||
stockJianshang: true,
|
||||
stockClearance: true,
|
||||
stockFactory: true,
|
||||
|
||||
@ -235,6 +235,8 @@ export const FILTER_OPTION_DATABASE_FIELDS = {
|
||||
sales3d: 'sales3d',
|
||||
sales7d: 'sales7d',
|
||||
sales15d: 'sales15d',
|
||||
sales30d: 'sales30d',
|
||||
sales30dWithBig: 'sales30dWithBig',
|
||||
stockClearance: 'stockClearance',
|
||||
stockFactory: 'stockFactory',
|
||||
stockJianshang: 'stockJianshang',
|
||||
|
||||
@ -64,6 +64,8 @@ export const productArchiveSelect = {
|
||||
sales3d: true,
|
||||
sales7d: true,
|
||||
sales15d: true,
|
||||
sales30d: true,
|
||||
sales30dWithBig: true,
|
||||
stockJianshang: true,
|
||||
stockClearance: true,
|
||||
stockFactory: true,
|
||||
|
||||
@ -0,0 +1,133 @@
|
||||
jest.mock('@/server/repositories/jst-marketing-daily-sales.repository', () => ({
|
||||
JstMarketingDailySalesRepository: jest.fn(),
|
||||
}))
|
||||
|
||||
import {
|
||||
buildMarketingSalesSnapshot,
|
||||
marketingSalesValues,
|
||||
validateMarketingSalesSnapshot,
|
||||
type MarketingSalesAggregate,
|
||||
} from '@/lib/product-archive-marketing-sales'
|
||||
import {
|
||||
INITIAL_MARKETING_SALES_STATE,
|
||||
MARKETING_SALES_DAILY_MINUTE,
|
||||
marketingSalesCheckDue,
|
||||
observeMarketingSales,
|
||||
} from '@/server/services/product-archive-marketing-sales-schedule'
|
||||
import { ProductArchiveMarketingSalesService } from '@/server/services/product-archive-marketing-sales.service'
|
||||
|
||||
const today = new Intl.DateTimeFormat('en-CA', { timeZone: 'Asia/Shanghai' }).format(new Date())
|
||||
const sample = (overrides: Partial<MarketingSalesAggregate> = {}): MarketingSalesAggregate => ({
|
||||
sku: 'SJ000569N1028V01', sourceDate: today, rowCount: 81, invalidCount: 0,
|
||||
threeAvgSale: '4.29', sevenAvgSale: '3.37', fifteenAvgSale: '3.54',
|
||||
thirtyAvgSale: '4.00', thirtyBigSale: '4.00', ...overrides,
|
||||
})
|
||||
|
||||
describe('marketing daily sales', () => {
|
||||
it('sums existing SQL aggregates and rounds weighted purchase sales to two decimals', () => {
|
||||
expect(marketingSalesValues(sample())).toEqual({ dailySales: '3.86', sales3d: '4.29', sales7d: '3.37', sales15d: '3.54', sales30d: '4.00', sales30dWithBig: '4.00' })
|
||||
})
|
||||
|
||||
it('uses exact integer arithmetic at half-cent boundaries and zero for unmatched codes', () => {
|
||||
expect(marketingSalesValues(sample({ threeAvgSale: '0.01', sevenAvgSale: '0.01', fifteenAvgSale: '0.01' })).dailySales).toBe('0.01')
|
||||
expect(marketingSalesValues()).toEqual({ dailySales: '0.00', sales3d: '0.00', sales7d: '0.00', sales15d: '0.00', sales30d: '0.00', sales30dWithBig: '0.00' })
|
||||
})
|
||||
|
||||
it('rejects empty, mixed, duplicate, invalid and sharply regressed source snapshots', () => {
|
||||
expect(() => buildMarketingSalesSnapshot([])).toThrow('EMPTY_SOURCE_SNAPSHOT')
|
||||
expect(() => buildMarketingSalesSnapshot([sample(), sample({ sku: 'other', sourceDate: '2026-10-08' })])).toThrow('MIXED_OR_MISSING_SOURCE_BATCH')
|
||||
expect(() => buildMarketingSalesSnapshot([sample(), sample()])).toThrow('INCOMPLETE_SOURCE_SNAPSHOT')
|
||||
const current = buildMarketingSalesSnapshot([sample()])
|
||||
const smaller = buildMarketingSalesSnapshot([sample({ rowCount: 50 })])
|
||||
expect(() => validateMarketingSalesSnapshot(smaller, current)).toThrow('SOURCE_COUNT_DROPPED_REQUIRES_REVIEW')
|
||||
const summary = { sourceDate: current.sourceDate, fingerprint: current.fingerprint, rowCount: current.rowCount, skuCount: current.rows.length }
|
||||
expect(() => validateMarketingSalesSnapshot(current, summary)).not.toThrow()
|
||||
expect(() => validateMarketingSalesSnapshot(smaller, summary)).toThrow('SOURCE_COUNT_DROPPED_REQUIRES_REVIEW')
|
||||
})
|
||||
|
||||
it('uses one fixed daily check at 09:30 Beijing time', () => {
|
||||
const now = new Date('2026-10-10T09:05:00+08:00')
|
||||
const oldState = { ...INITIAL_MARKETING_SALES_STATE, mode: 'hourly' as const }
|
||||
const state = observeMarketingSales(oldState, '2026-10-10', 'same-source', now)
|
||||
expect(state.mode).toBe('daily')
|
||||
expect(state.dailyMinute).toBe(MARKETING_SALES_DAILY_MINUTE)
|
||||
expect(state.nextCheckAt).toBe('2026-10-11T01:30:00.000Z')
|
||||
expect(marketingSalesCheckDue(state, new Date('2026-10-10T10:30:00+08:00'))).toBe(false)
|
||||
expect(marketingSalesCheckDue(state, new Date('2026-10-11T09:29:59+08:00'))).toBe(false)
|
||||
expect(marketingSalesCheckDue(state, new Date('2026-10-11T09:30:00+08:00'))).toBe(true)
|
||||
expect(marketingSalesCheckDue(state, new Date('2026-10-11T11:30:00+08:00'))).toBe(true)
|
||||
})
|
||||
|
||||
it('does not contact the source again before the next daily check', async () => {
|
||||
const now = new Date('2026-10-10T10:30:00+08:00')
|
||||
const state = observeMarketingSales(INITIAL_MARKETING_SALES_STATE, '2026-10-10', 'same-source', now)
|
||||
const source = { readSnapshot: jest.fn(), close: jest.fn() }
|
||||
const repository = { getState: jest.fn().mockResolvedValue({ state }) }
|
||||
const result = await new ProductArchiveMarketingSalesService(source as never, repository as never).run(now)
|
||||
expect(result.status).toBe('SKIPPED_SCHEDULE')
|
||||
expect(source.readSnapshot).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('skips archive writes when the fingerprint and full row metadata are unchanged', async () => {
|
||||
const now = new Date('2026-10-09T10:00:00Z')
|
||||
const snapshot = buildMarketingSalesSnapshot([sample()])
|
||||
const source = { readSnapshot: jest.fn().mockResolvedValue(snapshot), close: jest.fn().mockResolvedValue(undefined) }
|
||||
const repository = {
|
||||
getState: jest.fn().mockResolvedValue({ state: INITIAL_MARKETING_SALES_STATE, snapshot }),
|
||||
countSnapshotDrift: jest.fn().mockResolvedValue(0),
|
||||
recordUnchanged: jest.fn().mockResolvedValue(undefined),
|
||||
applySnapshot: jest.fn(),
|
||||
recordFailure: jest.fn(),
|
||||
}
|
||||
|
||||
const result = await new ProductArchiveMarketingSalesService(source as never, repository as never).run(now)
|
||||
|
||||
expect(result.status).toBe('UNCHANGED')
|
||||
expect(repository.recordUnchanged).toHaveBeenCalledTimes(1)
|
||||
expect(repository.applySnapshot).not.toHaveBeenCalled()
|
||||
expect(source.close).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('repairs archive drift for an unchanged source snapshot', async () => {
|
||||
const now = new Date('2026-10-09T10:00:00Z')
|
||||
const snapshot = buildMarketingSalesSnapshot([sample()])
|
||||
const source = { readSnapshot: jest.fn().mockResolvedValue(snapshot), close: jest.fn().mockResolvedValue(undefined) }
|
||||
const repository = {
|
||||
getState: jest.fn().mockResolvedValue({ state: INITIAL_MARKETING_SALES_STATE, snapshot }),
|
||||
countSnapshotDrift: jest.fn().mockResolvedValue(2),
|
||||
recordUnchanged: jest.fn(),
|
||||
applySnapshot: jest.fn().mockResolvedValue({ reset: 1, matched: 1, sourceRows: 1 }),
|
||||
recordFailure: jest.fn(),
|
||||
}
|
||||
|
||||
const result = await new ProductArchiveMarketingSalesService(source as never, repository as never).run(now)
|
||||
|
||||
expect(result.status).toBe('REPAIRED')
|
||||
expect(repository.applySnapshot).toHaveBeenCalledWith(expect.anything(), expect.anything(), { status: 'REPAIRED' })
|
||||
expect(repository.recordUnchanged).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('records source and close failures while retaining the fixed daily schedule', async () => {
|
||||
const now = new Date('2026-10-09T10:00:00Z')
|
||||
const source = {
|
||||
readSnapshot: jest.fn().mockRejectedValue(new Error('SOURCE_COUNT_DROPPED_REQUIRES_REVIEW')),
|
||||
close: jest.fn().mockRejectedValue(new Error('close failure')),
|
||||
}
|
||||
const repository = {
|
||||
getState: jest.fn().mockResolvedValue({ state: { mode: 'daily', dailyMinute: 600, observations: [] }, snapshot: { fingerprint: 'old' } }),
|
||||
countSnapshotDrift: jest.fn(),
|
||||
recordUnchanged: jest.fn(),
|
||||
applySnapshot: jest.fn(),
|
||||
recordFailure: jest.fn().mockResolvedValue(undefined),
|
||||
}
|
||||
|
||||
await expect(new ProductArchiveMarketingSalesService(source as never, repository as never).run(now))
|
||||
.rejects.toThrow('SOURCE_COUNT_DROPPED_REQUIRES_REVIEW')
|
||||
|
||||
expect(repository.recordFailure).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ mode: 'daily', dailyMinute: MARKETING_SALES_DAILY_MINUTE, observations: [], lastCheckedAt: now.toISOString() }),
|
||||
{ fingerprint: 'old' },
|
||||
'SOURCE_COUNT_DROPPED_REQUIRES_REVIEW',
|
||||
)
|
||||
})
|
||||
})
|
||||
@ -0,0 +1,50 @@
|
||||
export type MarketingSalesObservation = {
|
||||
sourceDate: string; lower: string; upper: string; corrected: boolean
|
||||
}
|
||||
export type MarketingSalesState = {
|
||||
mode: 'hourly' | 'daily'
|
||||
lastCheckedAt?: string
|
||||
fingerprint?: string
|
||||
sourceDate?: string
|
||||
dailyMinute?: number
|
||||
nextCheckAt?: string
|
||||
observations: MarketingSalesObservation[]
|
||||
}
|
||||
|
||||
const HOUR = 3_600_000
|
||||
export const MARKETING_SALES_DAILY_MINUTE = 9 * 60 + 30
|
||||
export const INITIAL_MARKETING_SALES_STATE: MarketingSalesState = {
|
||||
mode: 'daily', dailyMinute: MARKETING_SALES_DAILY_MINUTE, observations: [],
|
||||
}
|
||||
|
||||
export function beijingDate(now: Date) {
|
||||
return new Date(now.getTime() + 8 * HOUR).toISOString().slice(0, 10)
|
||||
}
|
||||
|
||||
export function marketingSalesCheckDue(state: MarketingSalesState, now: Date) {
|
||||
// Accept the previous hourly state once, then persist the fixed daily schedule.
|
||||
return state.mode === 'hourly' || !state.nextCheckAt || now >= new Date(state.nextCheckAt)
|
||||
}
|
||||
|
||||
export function resetMarketingSalesSchedule(state: MarketingSalesState, now: Date): MarketingSalesState {
|
||||
return {
|
||||
mode: 'daily',
|
||||
dailyMinute: MARKETING_SALES_DAILY_MINUTE,
|
||||
nextCheckAt: nextDailyCheck(now),
|
||||
observations: [],
|
||||
lastCheckedAt: now.toISOString(),
|
||||
...(state.sourceDate ? { sourceDate: state.sourceDate } : {}),
|
||||
...(state.fingerprint ? { fingerprint: state.fingerprint } : {}),
|
||||
}
|
||||
}
|
||||
|
||||
function nextDailyCheck(now: Date) {
|
||||
// Each invocation consumes today's check; the next check is tomorrow at 09:30.
|
||||
const time = new Date(`${beijingDate(now)}T00:00:00+08:00`).getTime()
|
||||
+ 24 * HOUR + MARKETING_SALES_DAILY_MINUTE * 60_000
|
||||
return new Date(time).toISOString()
|
||||
}
|
||||
|
||||
export function observeMarketingSales(state: MarketingSalesState, sourceDate: string, fingerprint: string, now: Date) {
|
||||
return { ...resetMarketingSalesSchedule(state, now), sourceDate, fingerprint }
|
||||
}
|
||||
@ -0,0 +1,66 @@
|
||||
import { validateMarketingSalesSnapshot, type MarketingSalesSnapshot, type MarketingSalesSnapshotSummary } from '@/lib/product-archive-marketing-sales'
|
||||
import { JstMarketingDailySalesRepository } from '@/server/repositories/jst-marketing-daily-sales.repository'
|
||||
import { productArchiveMarketingSalesRepository, type ProductArchiveMarketingSalesRepository } from '@/server/repositories/product-archive-marketing-sales.repository'
|
||||
import { INITIAL_MARKETING_SALES_STATE, marketingSalesCheckDue, observeMarketingSales, resetMarketingSalesSchedule, type MarketingSalesState } from './product-archive-marketing-sales-schedule'
|
||||
|
||||
function safeErrorCode(error: unknown) {
|
||||
const candidate = error && typeof error === 'object' && 'code' in error
|
||||
? String(error.code)
|
||||
: error instanceof Error ? error.message : ''
|
||||
return /^[A-Z][A-Z0-9_]{1,79}$/.test(candidate) ? candidate : 'MARKETING_SALES_SYNC_FAILED'
|
||||
}
|
||||
|
||||
export class ProductArchiveMarketingSalesService {
|
||||
constructor(
|
||||
private readonly source = new JstMarketingDailySalesRepository(),
|
||||
private readonly repository: Pick<ProductArchiveMarketingSalesRepository,
|
||||
'getState' | 'applySnapshot' | 'countSnapshotDrift' | 'recordUnchanged' | 'recordFailure'> = productArchiveMarketingSalesRepository,
|
||||
) {}
|
||||
|
||||
async run(now = new Date()) {
|
||||
const persisted = await this.repository.getState()
|
||||
const state = (persisted?.state as unknown as MarketingSalesState | undefined) ?? INITIAL_MARKETING_SALES_STATE
|
||||
const persistedSnapshot = persisted?.snapshot as unknown as MarketingSalesSnapshotSummary | undefined
|
||||
if (!marketingSalesCheckDue(state, now)) return { status: 'SKIPPED_SCHEDULE', state }
|
||||
let snapshot: MarketingSalesSnapshot | undefined
|
||||
let readError: unknown
|
||||
try {
|
||||
snapshot = await this.source.readSnapshot()
|
||||
validateMarketingSalesSnapshot(snapshot, persistedSnapshot)
|
||||
} catch (error) {
|
||||
readError = error
|
||||
}
|
||||
try {
|
||||
await this.source.close()
|
||||
} catch (error) {
|
||||
readError ??= error
|
||||
}
|
||||
if (readError || !snapshot) {
|
||||
const retryState = resetMarketingSalesSchedule(state, now)
|
||||
const failure = readError ?? new Error('EMPTY_SOURCE_SNAPSHOT')
|
||||
await this.repository.recordFailure(retryState, persistedSnapshot ?? null, safeErrorCode(failure))
|
||||
throw failure
|
||||
}
|
||||
|
||||
const nextState = observeMarketingSales(state, snapshot.sourceDate, snapshot.fingerprint, now)
|
||||
try {
|
||||
if (persistedSnapshot && persistedSnapshot.fingerprint === snapshot.fingerprint) {
|
||||
const drift = await this.repository.countSnapshotDrift(snapshot)
|
||||
if (drift === 0) {
|
||||
await this.repository.recordUnchanged(snapshot, nextState)
|
||||
return { status: 'UNCHANGED', state: nextState, snapshot }
|
||||
}
|
||||
const result = await this.repository.applySnapshot(snapshot, nextState, { status: 'REPAIRED' })
|
||||
return { status: 'REPAIRED', state: nextState, snapshot, result }
|
||||
}
|
||||
const result = await this.repository.applySnapshot(snapshot, nextState)
|
||||
return { status: 'APPLIED', state: nextState, snapshot, result }
|
||||
} catch (error) {
|
||||
const retryState = resetMarketingSalesSchedule(state, now)
|
||||
await this.repository.recordFailure(retryState, persistedSnapshot ?? null, safeErrorCode(error))
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export const productArchiveMarketingSalesService = new ProductArchiveMarketingSalesService()
|
||||
@ -351,10 +351,6 @@ export function buildProductArchiveEnrichmentUpdates(
|
||||
: {}),
|
||||
firstLeadTime: stockProfile?.produceCycle ?? '',
|
||||
palletSpec: stockProfile?.trayParameter ?? '',
|
||||
dailySales: stockProfile?.purchaseSale ?? '',
|
||||
sales3d: stockProfile?.threeAvgSale ?? '',
|
||||
sales7d: stockProfile?.sevenAvgSale ?? '',
|
||||
sales15d: stockProfile?.fifteenAvgSale ?? '',
|
||||
},
|
||||
source: source(
|
||||
JST_STOCK_MONITOR_ARCHIVE_SOURCE_CODE,
|
||||
@ -366,10 +362,6 @@ export function buildProductArchiveEnrichmentUpdates(
|
||||
minimumOrderQuantity: 'minimum',
|
||||
firstLeadTime: 'produceCycle',
|
||||
palletSpec: 'trayParameter',
|
||||
dailySales: 'purchaseSale',
|
||||
sales3d: 'threeAvgSale',
|
||||
sales7d: 'sevenAvgSale',
|
||||
sales15d: 'fifteenAvgSale',
|
||||
},
|
||||
),
|
||||
},
|
||||
|
||||
@ -84,6 +84,7 @@ const productArchiveFiltersSchema = z.object({
|
||||
sales7d: archiveValueArraySchema.optional(),
|
||||
sales15d: archiveValueArraySchema.optional(),
|
||||
sales30dWithBig: archiveValueArraySchema.optional(),
|
||||
sales30d: archiveValueArraySchema.optional(),
|
||||
jianshangStockDays: archiveValueArraySchema.optional(),
|
||||
stockDays: archiveValueArraySchema.optional(),
|
||||
stockJianshang: archiveValueArraySchema.optional(),
|
||||
|
||||
@ -114,6 +114,7 @@ export function mapProductArchiveRow({
|
||||
sales3d: record.sales3d,
|
||||
sales7d: record.sales7d,
|
||||
sales15d: record.sales15d,
|
||||
sales30d: record.sales30d,
|
||||
stockJianshang: record.stockJianshang,
|
||||
stockClearance: record.stockClearance,
|
||||
stockFactory: record.stockFactory,
|
||||
@ -138,7 +139,7 @@ export function mapProductArchiveRow({
|
||||
productStatus: record.productStatus,
|
||||
realPerformanceText: decimalText(record.sabcPeriodRealPerformance ?? null),
|
||||
}),
|
||||
sales30dWithBig: '',
|
||||
sales30dWithBig: record.sales30dWithBig,
|
||||
jianshangStockDays: '',
|
||||
stockDays: '',
|
||||
stockCloud: '',
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user