302 lines
14 KiB
PHP
302 lines
14 KiB
PHP
<?php
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use Carbon\CarbonImmutable;
|
||
use Illuminate\Console\Command;
|
||
use Illuminate\Support\Facades\DB;
|
||
|
||
/**
|
||
* 按取数逻辑计算某个月的更改价格、定价状态、采购单价、期初金额、入库金额和加权平均单价。
|
||
*/
|
||
class ComputeProductMonthlyWeightedAverage extends Command
|
||
{
|
||
protected $signature = 'product:monthly-weighted-average:compute {month : 报表月份，例如 2026-07}';
|
||
|
||
protected $description = '按取数逻辑计算商品月加权平均单价';
|
||
|
||
public function handle(): int
|
||
{
|
||
$month = CarbonImmutable::parse($this->argument('month'))->startOfMonth();
|
||
$connection = (string) config('upstream_mirrors.connection', 'business_mirror');
|
||
$table = 'product_monthly_weighted_average_rows';
|
||
$deadline = $month->addMonthNoOverflow()->day(2)->startOfDay();
|
||
|
||
$query = DB::connection($connection)->table($table)->where('report_month', $month->toDateString());
|
||
$codes = (clone $query)->distinct()->pluck('product_code')->map(fn ($code) => trim((string) $code))->unique()->values()->all();
|
||
$this->info('待计算 SKU：' . count($codes));
|
||
|
||
$cost = $this->costPrices($connection, $codes);
|
||
$work = $this->workPlatformPrices($connection, $codes);
|
||
$change = $this->priceChanges($connection, $codes);
|
||
$inbound = $this->inboundAggregate($connection, $month);
|
||
$costRecordSkus = $this->costPriceRecordSkus($connection, $codes);
|
||
|
||
$total = (clone $query)->count();
|
||
$computed = 0;
|
||
$now = now();
|
||
|
||
for ($offset = 0; $offset < $total; $offset += 1000) {
|
||
$rows = (clone $query)->orderBy('row_key')->offset($offset)->limit(1000)
|
||
->get([
|
||
'row_key', 'barcode_warehouse', 'warehouse_short_name', 'product_code', 'base_source', 'imported_at',
|
||
'opening_quantity', 'opening_unit_price', 'previous_pricing_status', 'purchase_quantity',
|
||
]);
|
||
|
||
$updates = [];
|
||
foreach ($rows as $row) {
|
||
// 普通商品资料与进销存的商品编码大小写不一致，统一按大写匹配。
|
||
$code = mb_strtoupper(trim((string) $row->product_code));
|
||
$costPrice = is_array($cost) ? ($cost[$code] ?? null) : null;
|
||
$workPrice = is_array($work) ? ($work[$code] ?? null) : null;
|
||
$changePrice = is_array($change) ? ($change[$code] ?? null) : null;
|
||
$inboundKey = $code . "\0" . trim((string) $row->warehouse_short_name);
|
||
$payload = $this->calculate(
|
||
$row,
|
||
$costPrice,
|
||
$workPrice,
|
||
$changePrice,
|
||
$deadline,
|
||
$inbound[$inboundKey] ?? null,
|
||
isset($costRecordSkus[$code]),
|
||
);
|
||
$updates[] = array_merge([
|
||
'report_month' => $month->toDateString(),
|
||
'row_key' => $row->row_key,
|
||
'barcode_warehouse' => $row->barcode_warehouse,
|
||
'warehouse_short_name' => $row->warehouse_short_name,
|
||
'product_code' => $row->product_code,
|
||
'base_source' => $row->base_source,
|
||
'imported_at' => $row->imported_at,
|
||
'computed_at' => $now,
|
||
], $payload);
|
||
$computed++;
|
||
}
|
||
|
||
if ($updates !== []) {
|
||
DB::connection($connection)->table($table)->upsert(
|
||
$updates,
|
||
['report_month', 'row_key'],
|
||
['adjusted_unit_price', 'opening_amount', 'purchase_quantity', 'purchase_unit_price', 'purchase_amount', 'weighted_average_unit_price', 'pricing_status', 'computed_at'],
|
||
);
|
||
}
|
||
}
|
||
|
||
DB::connection($connection)->table('product_monthly_weighted_average_states')->updateOrInsert(
|
||
['report_month' => $month->toDateString()],
|
||
['status' => 'computed', 'computed_rows' => $computed, 'last_computed_at' => $now, 'updated_at' => $now, 'created_at' => $now],
|
||
);
|
||
|
||
$this->info($month->format('Y-m') . " 计算完成：{$computed} 行");
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
|
||
/**
|
||
* @return array<string,float>
|
||
*/
|
||
private function costPrices(string $connection, array $codes): array
|
||
{
|
||
$map = [];
|
||
foreach (array_chunk($codes, 2000) as $chunk) {
|
||
foreach (DB::connection($connection)->table('ops_jst_sku')
|
||
->whereIn('sku_id', $chunk)->whereNotNull('cost_price')->get(['sku_id', 'cost_price']) as $row) {
|
||
$map[mb_strtoupper(trim((string) $row->sku_id))] = (float) $row->cost_price;
|
||
}
|
||
}
|
||
|
||
return $map;
|
||
}
|
||
|
||
/**
|
||
* @return array<string,array{price:float,effect:?string}>
|
||
*/
|
||
private function workPlatformPrices(string $connection, array $codes): array
|
||
{
|
||
$map = [];
|
||
$now = now();
|
||
foreach (array_chunk($codes, 2000) as $chunk) {
|
||
$rows = DB::connection($connection)->table('js_pm_price_adjustment')
|
||
->whereIn('skuCode', $chunk)
|
||
->whereNotNull('purchasePrice')
|
||
->where(function ($query) use ($now): void {
|
||
$query->whereNull('effectTime')->orWhere('effectTime', '<=', $now);
|
||
})
|
||
->orderBy('effectTime')
|
||
->get(['skuCode', 'purchasePrice', 'effectTime']);
|
||
foreach ($rows as $row) {
|
||
$map[mb_strtoupper(trim((string) $row->skuCode))] = [
|
||
'price' => (float) $row->purchasePrice,
|
||
'effect' => $row->effectTime === null ? null : (string) $row->effectTime,
|
||
];
|
||
}
|
||
}
|
||
|
||
return $map;
|
||
}
|
||
|
||
/**
|
||
* @return array<string,array{price:float,time:?string}>
|
||
*/
|
||
/**
|
||
* 入库明细汇总：按「大写条码 + 仓库简称」汇总库存变动量与总计（金额 + 预估金额）。
|
||
*
|
||
* @return array<string,array{quantity:float,total:float}>
|
||
*/
|
||
private function inboundAggregate(string $connection, CarbonImmutable $month): array
|
||
{
|
||
$map = [];
|
||
if (! \Illuminate\Support\Facades\Schema::connection($connection)->hasTable('inventory_inbound_detail_rows')) {
|
||
return $map;
|
||
}
|
||
|
||
foreach (DB::connection($connection)->table('inventory_inbound_detail_rows')
|
||
->where('report_month', $month->toDateString())
|
||
->selectRaw('warehouse_short_name, upper(barcode) as code, sum(quantity) as quantity, sum(coalesce(amount, 0) + coalesce(estimated_amount, 0)) as total')
|
||
->groupBy('warehouse_short_name')
|
||
->groupByRaw('upper(barcode)')
|
||
->get() as $row) {
|
||
$key = mb_strtoupper(trim((string) $row->code)) . "\0" . trim((string) $row->warehouse_short_name);
|
||
$map[$key] = ['quantity' => (float) $row->quantity, 'total' => (float) $row->total];
|
||
}
|
||
|
||
return $map;
|
||
}
|
||
|
||
/**
|
||
* 有"聚水潭成本价记录"（operated_at 非空）的商品编码集合。
|
||
*
|
||
* @return array<string,bool>
|
||
*/
|
||
private function costPriceRecordSkus(string $connection, array $codes): array
|
||
{
|
||
$map = [];
|
||
if (! \Illuminate\Support\Facades\Schema::connection($connection)->hasTable('jushuitan_cost_price_history_rows')) {
|
||
return $map;
|
||
}
|
||
|
||
foreach (array_chunk($codes, 2000) as $chunk) {
|
||
$rows = DB::connection($connection)->table('jushuitan_cost_price_history_rows')
|
||
->whereNotNull('operated_at')
|
||
->where(function ($query) use ($chunk): void {
|
||
$query->whereIn('sku_id', $chunk)->orWhereIn('sku_id_normalized', $chunk);
|
||
})
|
||
->get(['sku_id', 'sku_id_normalized']);
|
||
foreach ($rows as $row) {
|
||
foreach ([$row->sku_id, $row->sku_id_normalized] as $value) {
|
||
$key = mb_strtoupper(trim((string) $value));
|
||
if ($key !== '') {
|
||
$map[$key] = true;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
return $map;
|
||
}
|
||
|
||
private function priceChanges(string $connection, array $codes): array
|
||
{
|
||
$map = [];
|
||
foreach (array_chunk($codes, 2000) as $chunk) {
|
||
$rows = DB::connection($connection)->table('js_pm_price_adjustment_log')
|
||
->whereIn('skucode', $chunk)
|
||
->whereNotNull('afterPrice')
|
||
->orderBy('effectStartTime')
|
||
->get(['skucode', 'afterPrice', 'originalPrice', 'effectStartTime', 'createTime']);
|
||
foreach ($rows as $row) {
|
||
$map[mb_strtoupper(trim((string) $row->skucode))] = [
|
||
'price' => (float) $row->afterPrice,
|
||
'before' => $row->originalPrice === null ? null : (float) $row->originalPrice,
|
||
'time' => $row->effectStartTime === null ? null : (string) $row->effectStartTime,
|
||
'created' => $row->createTime === null ? null : (string) $row->createTime,
|
||
];
|
||
}
|
||
}
|
||
|
||
return $map;
|
||
}
|
||
|
||
/**
|
||
* @param array{price:float,effect:?string}|null $work
|
||
* @param array{price:float,time:?string}|null $change
|
||
* @return array<string,mixed>
|
||
*/
|
||
private function calculate(object $row, ?float $cost, ?array $work, ?array $change, CarbonImmutable $deadline, ?array $inbound = null, bool $hasCostRecord = false): array
|
||
{
|
||
$openingQuantity = $row->opening_quantity === null ? 0.0 : (float) $row->opening_quantity;
|
||
$purchaseQuantity = $row->purchase_quantity === null ? 0.0 : (float) $row->purchase_quantity;
|
||
$openingUnitPrice = $row->opening_unit_price === null ? null : (float) $row->opening_unit_price;
|
||
$previousStatus = trim((string) ($row->previous_pricing_status ?? ''));
|
||
$changePrice = $change['price'] ?? null;
|
||
$workPrice = $work['price'] ?? null;
|
||
// 最终规则：生效开始日期为空 → 未匹配到工作平台价格；
|
||
// 生效开始日期 > 本月，或记录时间 > 次月2号（记录时间为空视为 ≤ 次月2号）→ 新价当月未生效，取变价前价格。
|
||
$changeTime = $change['time'] ?? null;
|
||
$changeCreated = $change['created'] ?? null;
|
||
$monthStart = $deadline->startOfMonth();
|
||
$hasWorkPlatformRecord = $changePrice !== null && $changeTime !== null;
|
||
$newPricePending = false;
|
||
if ($hasWorkPlatformRecord) {
|
||
$newPricePending = CarbonImmutable::parse($changeTime)->greaterThanOrEqualTo($monthStart)
|
||
|| ($changeCreated !== null && CarbonImmutable::parse($changeCreated)->greaterThan($deadline));
|
||
$changePrice = $newPricePending ? ($change['before'] ?? $changePrice) : $changePrice;
|
||
}
|
||
$previousStatus = in_array($previousStatus, ['预估价', '未定价'], true) ? '未定价' : $previousStatus;
|
||
|
||
// 更改价格 + 定价状态（表2 第10条）
|
||
if ($openingUnitPrice !== null) {
|
||
if ($previousStatus !== '未定价') {
|
||
$adjusted = $openingUnitPrice;
|
||
$status = $previousStatus !== '' ? $previousStatus : '已定价';
|
||
} elseif ($hasWorkPlatformRecord) {
|
||
$adjusted = $changePrice;
|
||
$status = $newPricePending ? '未定价' : '已定价';
|
||
} else {
|
||
$adjusted = $openingUnitPrice;
|
||
$status = '未定价';
|
||
}
|
||
} elseif ($hasWorkPlatformRecord) {
|
||
$adjusted = $changePrice;
|
||
$status = $newPricePending ? '未定价' : '已定价';
|
||
} elseif ($cost !== null) {
|
||
// 方案②：只有"聚水潭成本价记录"里有生效记录(operated_at 非空)才算有生效日期 → 已定价。
|
||
$adjusted = $cost;
|
||
$status = $hasCostRecord ? '已定价' : '未定价';
|
||
} else {
|
||
$adjusted = null;
|
||
$status = '待人工核查';
|
||
}
|
||
|
||
// 采购数量/入库金额来自入库明细（大写条码 + 仓库简称汇总），采购单价 = 入库金额 ÷ 采购数量
|
||
$purchaseQuantity = (float) ($inbound['quantity'] ?? 0);
|
||
// 加权平均单价按“先除再舍入”计算，因此这里保留未舍入的金额参与除法。
|
||
$purchaseAmountRaw = $inbound === null ? 0.0 : (float) ($inbound['total'] ?? 0);
|
||
$purchaseAmount = $inbound === null ? null : round($purchaseAmountRaw, 2);
|
||
$purchaseUnitPrice = ($inbound === null || $purchaseQuantity == 0.0)
|
||
? null
|
||
: round((float) ($inbound['total'] ?? 0) / $purchaseQuantity, 2);
|
||
|
||
$openingAmountRaw = $adjusted === null ? 0.0 : $openingQuantity * $adjusted;
|
||
$openingAmount = $adjusted === null ? null : round($openingAmountRaw, 2);
|
||
$denominator = $openingQuantity + $purchaseQuantity;
|
||
if ($denominator === 0.0) {
|
||
// 四项全 0：加权平均单价取"更改价格"（按两位舍入后的值）。
|
||
$weighted = $adjusted === null ? null : round($adjusted, 2);
|
||
} else {
|
||
// 先除再舍入：用未舍入的期初金额与入库金额相除，最后统一保留两位。
|
||
$weighted = round(($openingAmountRaw + $purchaseAmountRaw) / $denominator, 4);
|
||
}
|
||
|
||
return [
|
||
'purchase_quantity' => $purchaseQuantity,
|
||
'adjusted_unit_price' => $adjusted === null ? null : round($adjusted, 2),
|
||
'opening_amount' => $openingAmount,
|
||
'purchase_unit_price' => $purchaseUnitPrice === null ? null : round($purchaseUnitPrice, 2),
|
||
'purchase_amount' => $purchaseAmount,
|
||
'weighted_average_unit_price' => $weighted,
|
||
'pricing_status' => $status,
|
||
];
|
||
}
|
||
}
|