2315 lines
101 KiB
PHP
2315 lines
101 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Repositories;
|
||
|
||
use Carbon\CarbonInterface;
|
||
use Generator;
|
||
use Illuminate\Contracts\Cache\LockTimeoutException;
|
||
use Illuminate\Database\ConnectionInterface;
|
||
use Illuminate\Database\DatabaseManager;
|
||
use Illuminate\Database\Query\Builder;
|
||
use Illuminate\Support\Carbon;
|
||
use Illuminate\Support\Collection;
|
||
use Illuminate\Support\Facades\Cache;
|
||
use Illuminate\Support\Facades\Schema;
|
||
use RuntimeException;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 三方仓进销存的共享业务镜像只读查询。
|
||
*
|
||
* 已导入的历史月份直接读取权威日明细；后续月份从每个 SKU 的固定
|
||
* 月末锚点按已确认流水前推。未导入锚点时保留实时库存兼容逻辑。
|
||
*/
|
||
class ThirdPartyWarehouseStorageMirrorRepository
|
||
{
|
||
private const REPORT_QUERY_TIMEOUT_SECONDS = 30;
|
||
|
||
/** Export queries may aggregate a larger batch, but must stay below the web proxy timeout. */
|
||
private const REPORT_EXPORT_QUERY_TIMEOUT_SECONDS = 90;
|
||
|
||
/** Shared movement aggregate is executed once per full export. */
|
||
private const REPORT_EXPORT_AGGREGATE_TIMEOUT_SECONDS = 180;
|
||
|
||
private const REPORT_EXPORT_BATCH_SIZE = 50000;
|
||
|
||
private const REPORT_CACHE_SECONDS = 93600;
|
||
|
||
private const REPORT_CACHE_LOCK_SECONDS = 45;
|
||
|
||
private const REPORT_CACHE_WAIT_SECONDS = 35;
|
||
|
||
public const DATASET_KEY = 'jiansun.third-party-warehouse-storage';
|
||
|
||
public const WAREHOUSE_ID = 15045364;
|
||
|
||
private const STATE_TABLE = 'third_party_warehouse_sync_states';
|
||
|
||
private const HEADER_TABLE = 'third_party_warehouse_inout_headers';
|
||
|
||
private const ITEM_TABLE = 'third_party_warehouse_inout_items';
|
||
|
||
private const INVENTORY_TABLE = 'third_party_warehouse_inventory_snapshots';
|
||
|
||
private const SUBWAREHOUSE_INVENTORY_TABLE = 'third_party_warehouse_inventory_subwarehouse_snapshots';
|
||
|
||
private const AUTHORITATIVE_STATE_TABLE = 'third_party_warehouse_authoritative_anchor_states';
|
||
|
||
private const AUTHORITATIVE_ANCHOR_TABLE = 'third_party_warehouse_authoritative_anchors';
|
||
|
||
private const AUTHORITATIVE_ROW_TABLE = 'third_party_warehouse_authoritative_rows';
|
||
|
||
private const AUTHORITATIVE_DATASET_KEY = 'jiansun.third-party-warehouse-storage.authoritative-anchor';
|
||
|
||
private const INVENTORY_REPORT_AUTHORITATIVE_STATE_TABLE = 'inventory_report_authoritative_states';
|
||
|
||
private const INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE = 'inventory_report_authoritative_rows';
|
||
|
||
private const INVENTORY_REPORT_AUTHORITATIVE_DATASET_KEY = 'jiansun.inventory-report.authoritative-month';
|
||
|
||
private const PRODUCT_TABLE = 'ops_jst_sku';
|
||
|
||
private const WAREHOUSE_TABLE = 'ops_jst_warehouse_info';
|
||
|
||
private const GENERIC_STATE_TABLE = 'business_mirror_table_states';
|
||
|
||
private const PRODUCT_DATASET_KEY = 'operations_replica.rpt.jst_sku';
|
||
|
||
private const WAREHOUSE_DATASET_KEY = 'operations_replica.rpt.jst_warehouseInfo';
|
||
|
||
public const START_DATE = '2026-08-01';
|
||
|
||
/** Historical monthly report range can start one month before the flow page. */
|
||
public const INVENTORY_REPORT_START_DATE = '2026-07-01';
|
||
|
||
/** @var array<string,string> */
|
||
private const MOVEMENT_COLUMNS = [
|
||
'processing_in' => '加工进仓',
|
||
'processing_out' => '加工出仓',
|
||
'purchase_in' => '采购进仓',
|
||
'purchase_return' => '采购退货',
|
||
'other_in' => '其它进仓',
|
||
'other_return' => '其它退货',
|
||
'sales_out' => '销售出仓',
|
||
'sales_return' => '销售退货',
|
||
'stocktake' => '盘点',
|
||
'transfer_out' => '调拨出',
|
||
'transfer_in' => '调拨入',
|
||
'other_out' => '其它出库',
|
||
];
|
||
|
||
/** @var array<int,string> */
|
||
private const SUBWAREHOUSE_SUFFIXES = [
|
||
1 => '主仓',
|
||
2 => '销退仓',
|
||
3 => '进货仓',
|
||
4 => '次品仓',
|
||
6 => '自定义仓1',
|
||
7 => '自定义仓2',
|
||
8 => '自定义仓3',
|
||
];
|
||
|
||
/**
|
||
* 聚水潭“按商品、按仓库、按月”导出中使用的实际子仓名。
|
||
* 其余仓库按“仓储方 + 子仓后缀”生成，不再只显示通用类型名。
|
||
*
|
||
* @var array<string,array<int,string>>
|
||
*/
|
||
private const SPECIAL_SUBWAREHOUSE_NAMES = [
|
||
'常州简尚家居有限公司' => [
|
||
1 => '配件单仓',
|
||
],
|
||
'简尚-横林二仓' => [
|
||
6 => '客退临时上架仓',
|
||
7 => '数量校验仓',
|
||
],
|
||
'简尚-新苏北仓' => [
|
||
6 => '新苏北客退上架仓',
|
||
],
|
||
'简尚-睢宁仓' => [
|
||
6 => '睢宁客退上架仓',
|
||
],
|
||
'简尚-河南洛宁仓' => [
|
||
6 => '河南洛宁客退上架仓',
|
||
],
|
||
'简尚-山东青岗集仓' => [
|
||
7 => '青岗集客退上架仓',
|
||
],
|
||
];
|
||
|
||
/** @var list<string> */
|
||
private const OUTPUT_COLUMNS = [
|
||
'owner', 'date', 'sku_code', 'image_url', 'product_name', 'short_name',
|
||
'supplier', 'category', 'virtual_category', 'supplier_style_no',
|
||
'color_spec', 'brand', 'unit', 'opening_qty', 'closing_qty',
|
||
'processing_in', 'processing_out', 'purchase_in', 'purchase_return',
|
||
'other_in', 'other_return', 'sales_out', 'sales_return', 'stocktake',
|
||
'transfer_out', 'transfer_in', 'opening_aux', 'other_out',
|
||
];
|
||
|
||
private readonly ConnectionInterface $connection;
|
||
|
||
public function __construct(DatabaseManager $databases)
|
||
{
|
||
$this->connection = $databases->connection(
|
||
(string) config('upstream_mirrors.connection', 'business_mirror'),
|
||
);
|
||
}
|
||
|
||
public function available(): bool
|
||
{
|
||
try {
|
||
$schema = Schema::connection($this->connection->getName());
|
||
|
||
foreach ([
|
||
self::STATE_TABLE,
|
||
self::HEADER_TABLE,
|
||
self::ITEM_TABLE,
|
||
self::INVENTORY_TABLE,
|
||
self::PRODUCT_TABLE,
|
||
self::GENERIC_STATE_TABLE,
|
||
] as $table) {
|
||
if (! $schema->hasTable($table)) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
return $schema->hasColumn(self::ITEM_TABLE, 'qty_type')
|
||
&& $schema->hasColumn(self::HEADER_TABLE, 'source_modified_at')
|
||
&& $schema->hasColumn(self::ITEM_TABLE, 'source_modified_at')
|
||
&& $schema->hasColumn(self::INVENTORY_TABLE, 'source_modified_at')
|
||
&& $schema->hasColumn(self::INVENTORY_TABLE, 'captured_at');
|
||
} catch (Throwable) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
public function ready(): bool
|
||
{
|
||
$state = $this->state();
|
||
|
||
return is_array($state)
|
||
&& ($state['initial_completed_at'] ?? null) !== null
|
||
&& $this->generation($state, 'active_header_generation') !== null
|
||
&& $this->generation($state, 'active_item_generation') !== null
|
||
&& $this->generation($state, 'active_inventory_generation') !== null
|
||
&& $this->dependenciesReady();
|
||
}
|
||
|
||
public function allWarehousesReady(): bool
|
||
{
|
||
$state = $this->state();
|
||
|
||
return $this->ready()
|
||
&& (int) ($state['scope_version'] ?? 1) >= 2
|
||
&& $this->warehouseDependencyReady();
|
||
}
|
||
|
||
/** @return array<string,mixed>|null */
|
||
public function state(): ?array
|
||
{
|
||
if (! $this->available()) {
|
||
return null;
|
||
}
|
||
|
||
try {
|
||
$row = $this->connection->table(self::STATE_TABLE)
|
||
->where('dataset_key', self::DATASET_KEY)
|
||
->first();
|
||
|
||
return $row === null ? null : (array) $row;
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @param array<string,mixed>|null $state
|
||
* @return array{code:string,last_succeeded_at:?string,lag_minutes:?int}
|
||
*/
|
||
public function syncStatus(?array $state = null, ?CarbonInterface $at = null): array
|
||
{
|
||
$state ??= $this->state();
|
||
if (! is_array($state)) {
|
||
return ['code' => 'unavailable', 'last_succeeded_at' => null, 'lag_minutes' => null];
|
||
}
|
||
|
||
$lastSucceededAt = $this->timestamp($state['last_succeeded_at'] ?? null);
|
||
$lastSucceeded = $lastSucceededAt === null
|
||
? null
|
||
: Carbon::createFromTimestamp($lastSucceededAt)->toDateTimeString();
|
||
$lagMinutes = $lastSucceededAt === null
|
||
? null
|
||
: (int) ceil(max(0, ($at ?? now())->getTimestamp() - $lastSucceededAt) / 60);
|
||
$status = (string) ($state['status'] ?? 'pending');
|
||
|
||
if (($state['initial_completed_at'] ?? null) === null) {
|
||
$code = $status === 'failed'
|
||
? 'initial_failed'
|
||
: ($status === 'running' ? 'initializing' : 'pending');
|
||
} elseif (! $this->dependenciesReady()) {
|
||
$code = 'dependency_pending';
|
||
} elseif ($status === 'failed') {
|
||
$code = 'failed';
|
||
} elseif ($status === 'running') {
|
||
$code = ($state['run_mode'] ?? null) === 'full' ? 'reconciling' : 'syncing';
|
||
} elseif ($lagMinutes === null
|
||
|| $lagMinutes > max((int) config(
|
||
'upstream_mirrors.third_party_warehouse_storage.stale_after_minutes',
|
||
1560,
|
||
), 1)) {
|
||
$code = 'delayed';
|
||
} else {
|
||
$code = 'healthy';
|
||
}
|
||
|
||
return [
|
||
'code' => $code,
|
||
'last_succeeded_at' => $lastSucceeded,
|
||
'lag_minutes' => $lagMinutes,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* @param array<string,mixed> $filters
|
||
* @return array{data:list<array<string,mixed>>,total:int,summary:array<string,int|float>,state:array<string,mixed>,sync:array<string,mixed>}
|
||
*/
|
||
public function paginate(array $filters, int $page, int $perPage): array
|
||
{
|
||
$state = $this->requireReadyState();
|
||
$page = max($page, 1);
|
||
$perPage = min(max($perPage, 1), 100);
|
||
$calculated = $this->calculatedFlowRows($filters, $state);
|
||
if ($calculated !== null) {
|
||
$summary = $this->summarizeRows($calculated);
|
||
|
||
return [
|
||
'data' => array_slice($calculated, ($page - 1) * $perPage, $perPage),
|
||
'total' => $summary['count'],
|
||
'summary' => $summary,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
}
|
||
$query = $this->filteredReportQuery($filters, $state);
|
||
$summary = $this->summarizeQuery($this->filteredReportQuery(
|
||
$filters,
|
||
$state,
|
||
$this->hasProductFilters($filters),
|
||
));
|
||
$rows = $query->orderBy('report_date')->orderBy('sku_code')
|
||
->forPage($page, $perPage)
|
||
->get()
|
||
->map(fn (object $row): array => $this->mapRow($row))
|
||
->all();
|
||
|
||
return [
|
||
'data' => $rows,
|
||
'total' => $summary['count'],
|
||
'summary' => $summary,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 商品价格管理的进销存报表与三方仓共用同一套已确认流水、
|
||
* 库存锚点和业务类型口径，只在这里将日明细再汇总为“商品 + 仓库 + 月”。
|
||
*
|
||
* @param array<string,mixed> $filters
|
||
* @return array{data:list<object>,total:int,state:array<string,mixed>,sync:array<string,mixed>}
|
||
*/
|
||
public function paginateMonthlyInventoryReport(
|
||
array $filters,
|
||
int $page,
|
||
int $perPage,
|
||
bool $includeTotal = true,
|
||
?array $state = null,
|
||
?array $sourceCounts = null,
|
||
?int $queryTimeoutSeconds = null,
|
||
?Collection $prefetchedMovementRows = null,
|
||
bool $refreshSourceCounts = false,
|
||
): array
|
||
{
|
||
$state ??= $this->requireReadyState();
|
||
if ((int) ($state['scope_version'] ?? 1) < 2) {
|
||
throw new RuntimeException('inventory_report_all_warehouses_not_ready');
|
||
}
|
||
$page = max($page, 1);
|
||
$perPage = min(max($perPage, 1), $prefetchedMovementRows === null ? 5000 : self::REPORT_EXPORT_BATCH_SIZE);
|
||
$queryTimeoutSeconds ??= self::REPORT_QUERY_TIMEOUT_SECONDS;
|
||
$period = $this->inventoryReportPeriod((string) ($filters['month'] ?? ''));
|
||
// 已导入的当前月份是对账基准，直接返回工作簿中的完整月度明细。
|
||
// 这样不会再把同月镜像流水重复叠加到权威的各类进出数量上。
|
||
$currentAuthority = $this->currentMonthlyInventoryAuthority($period);
|
||
if ($currentAuthority !== null) {
|
||
return $this->paginateCurrentMonthlyInventoryAuthority(
|
||
$period,
|
||
$filters,
|
||
$page,
|
||
$perPage,
|
||
$includeTotal,
|
||
$state,
|
||
$queryTimeoutSeconds,
|
||
);
|
||
}
|
||
|
||
// 没有当前月权威数据时，保留原有的上月期末结转和已确认流水滚算。
|
||
$headerGeneration = $this->generation($state, 'active_header_generation');
|
||
$itemGeneration = $this->generation($state, 'active_item_generation');
|
||
$inventoryGeneration = $this->generation($state, 'active_inventory_generation');
|
||
if ($headerGeneration === null || $itemGeneration === null || $inventoryGeneration === null) {
|
||
throw new RuntimeException('inventory_report_mirror_not_ready');
|
||
}
|
||
|
||
$authority = $this->previousMonthlyInventoryAuthority($period);
|
||
$authorityIsCurrentMonth = false;
|
||
$resolveRollForward = fn (): ?array => $authority === null
|
||
? null
|
||
: $this->monthlyInventoryRollForward(
|
||
$period,
|
||
$filters,
|
||
$authority,
|
||
$headerGeneration,
|
||
$itemGeneration,
|
||
$inventoryGeneration,
|
||
$page,
|
||
$perPage,
|
||
$includeTotal,
|
||
$state,
|
||
$authorityIsCurrentMonth,
|
||
$sourceCounts,
|
||
$queryTimeoutSeconds,
|
||
$prefetchedMovementRows,
|
||
$refreshSourceCounts,
|
||
);
|
||
$rolledForward = $includeTotal && $perPage <= 100 && $authority !== null
|
||
? $this->rememberMonthlyInventoryReport(
|
||
$this->monthlyInventoryReportCacheKey(
|
||
$period,
|
||
$filters,
|
||
$authority,
|
||
$headerGeneration,
|
||
$itemGeneration,
|
||
$inventoryGeneration,
|
||
$page,
|
||
$perPage,
|
||
),
|
||
$resolveRollForward,
|
||
)
|
||
: $resolveRollForward();
|
||
if ($rolledForward !== null) {
|
||
$rolledForward['state'] = $state;
|
||
$rolledForward['sync'] = $this->syncStatus($state);
|
||
|
||
return $rolledForward;
|
||
}
|
||
|
||
$anchorAt = $this->stateAnchor($state);
|
||
$rangeStart = min($period['start'], $anchorAt);
|
||
$rangeEnd = max($period['end'], $anchorAt);
|
||
$latestProducts = $this->connection->table(self::PRODUCT_TABLE)
|
||
->where('sku_type', 'normal')->whereNotNull('sku_id')->where('sku_id', '!=', '')
|
||
->groupBy('sku_id')->select('sku_id')->selectRaw('MAX(autoid) AS autoid');
|
||
$latestWarehouses = $this->connection->table(self::WAREHOUSE_TABLE)
|
||
->whereNotNull('wms_co_id')->groupBy('wms_co_id')
|
||
->select('wms_co_id')->selectRaw('MAX(Id) AS Id');
|
||
$inventoryDimensions = $this->connection->table(self::SUBWAREHOUSE_INVENTORY_TABLE)
|
||
->where('generation', $inventoryGeneration)
|
||
->where('qty', '!=', 0)
|
||
->whereNotNull('sku_code')->where('sku_code', '!=', '')
|
||
->select(['wms_co_id', 'wh_id', 'sku_code as inventory_sku_code', 'captured_at'])
|
||
->selectRaw('qty AS snapshot_qty');
|
||
$candidates = $this->connection->query()->fromSub($inventoryDimensions, 'pairs')
|
||
->joinSub($latestProducts, 'latest_product', 'latest_product.sku_id', '=', 'pairs.inventory_sku_code')
|
||
->join(self::PRODUCT_TABLE . ' as s', function ($join): void {
|
||
$join->on('s.sku_id', '=', 'latest_product.sku_id')
|
||
->on('s.autoid', '=', 'latest_product.autoid');
|
||
})
|
||
->select([
|
||
'pairs.wms_co_id', 'pairs.wh_id', 'pairs.inventory_sku_code',
|
||
's.sku_id as sku_code', 's.category',
|
||
]);
|
||
$productCode = trim((string) ($filters['product_code'] ?? ''));
|
||
if ($productCode !== '') {
|
||
$candidates->where('s.sku_id', $productCode);
|
||
}
|
||
$warehouseFilter = trim((string) ($filters['warehouse_provider'] ?? ''));
|
||
if ($warehouseFilter !== '') {
|
||
$warehouseIds = $this->connection->query()->fromSub(
|
||
clone $latestWarehouses,
|
||
'filter_latest_warehouse',
|
||
)->join(self::WAREHOUSE_TABLE . ' as filter_warehouse', function ($join): void {
|
||
$join->on('filter_warehouse.wms_co_id', '=', 'filter_latest_warehouse.wms_co_id')
|
||
->on('filter_warehouse.Id', '=', 'filter_latest_warehouse.Id');
|
||
});
|
||
$this->containsFilter($warehouseIds, 'filter_warehouse.name', $warehouseFilter);
|
||
$candidates->whereIn('pairs.wms_co_id', $warehouseIds->select('filter_warehouse.wms_co_id'));
|
||
}
|
||
$categoryFilter = trim((string) ($filters['category'] ?? ''));
|
||
if ($categoryFilter !== '') {
|
||
$this->containsFilter($candidates, 's.category', $categoryFilter);
|
||
}
|
||
|
||
$total = $includeTotal ? $this->boundedCount(clone $candidates, $queryTimeoutSeconds) : 0;
|
||
$pagePairs = (clone $candidates)
|
||
->orderBy('pairs.wms_co_id')->orderBy('pairs.wh_id')->orderBy('s.sku_id')
|
||
->forPage($page, $perPage);
|
||
$pairRows = $this->boundedGet(clone $pagePairs, $queryTimeoutSeconds);
|
||
if ($pairRows->isEmpty()) {
|
||
return [
|
||
'data' => [],
|
||
'total' => $total,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
}
|
||
|
||
$dimensionedPagePairs = $this->connection->query()->fromSub($pagePairs, 'pairs')
|
||
->leftJoinSub($latestWarehouses, 'latest_warehouse', 'latest_warehouse.wms_co_id', '=', 'pairs.wms_co_id')
|
||
->leftJoin(self::WAREHOUSE_TABLE . ' as w', function ($join): void {
|
||
$join->on('w.wms_co_id', '=', 'latest_warehouse.wms_co_id')
|
||
->on('w.Id', '=', 'latest_warehouse.Id');
|
||
})
|
||
->select([
|
||
'pairs.wms_co_id', 'pairs.wh_id', 'pairs.inventory_sku_code',
|
||
'pairs.sku_code', 'pairs.category', 'w.name as warehouse_name',
|
||
]);
|
||
$anchors = $this->connection->query()->fromSub($dimensionedPagePairs, 'pairs')
|
||
->leftJoinSub(clone $inventoryDimensions, 'inventory', function ($join): void {
|
||
$join->on('inventory.wms_co_id', '=', 'pairs.wms_co_id')
|
||
->on('inventory.inventory_sku_code', '=', 'pairs.inventory_sku_code')
|
||
->on('inventory.wh_id', '=', 'pairs.wh_id');
|
||
})
|
||
->select([
|
||
'pairs.wms_co_id', 'pairs.wh_id', 'pairs.inventory_sku_code',
|
||
'pairs.sku_code', 'pairs.category', 'pairs.warehouse_name',
|
||
])
|
||
->selectRaw('COALESCE(inventory.snapshot_qty, 0) AS snapshot_qty')
|
||
->selectRaw('COALESCE(inventory.captured_at, ?) AS snapshot_at', [$anchorAt]);
|
||
|
||
$movements = $this->confirmedMovements($headerGeneration, $itemGeneration, null)
|
||
->where('h.occurred_at', '>=', $rangeStart)
|
||
->where('h.occurred_at', '<', $rangeEnd)
|
||
->whereIn('h.wms_co_id', $pairRows->pluck('wms_co_id')->unique()->values()->all())
|
||
->whereIn('i.sku_code', $pairRows->pluck('inventory_sku_code')->unique()->values()->all())
|
||
->select([
|
||
'h.wms_co_id', 'h.io_type', 'h.occurred_at', 'i.sku_code as inventory_sku_code',
|
||
'i.qty as movement_qty', 'i.qty_type as movement_qty_type',
|
||
])
|
||
->selectRaw('COALESCE(NULLIF(h.wh_id, 0), 1) AS wh_id');
|
||
$signed = '(COALESCE(mv.movement_qty, 0) * COALESCE(mv.movement_qty_type, 0))';
|
||
$balanceAt = static fn (string $target): array => [
|
||
'a.snapshot_qty + COALESCE(SUM(CASE '
|
||
. 'WHEN a.snapshot_at < ? AND mv.occurred_at >= a.snapshot_at AND mv.occurred_at < ? THEN ' . $signed . ' '
|
||
. 'WHEN a.snapshot_at >= ? AND mv.occurred_at >= ? AND mv.occurred_at < a.snapshot_at THEN -' . $signed . ' '
|
||
. 'ELSE 0 END), 0)',
|
||
[$target, $target, $target, $target],
|
||
];
|
||
[$openingSql, $openingBindings] = $balanceAt($period['start']);
|
||
[$closingSql, $closingBindings] = $balanceAt($period['end']);
|
||
$monthly = $this->connection->query()->fromSub($anchors, 'a')
|
||
->leftJoinSub($movements, 'mv', function ($join): void {
|
||
$join->on('mv.wms_co_id', '=', 'a.wms_co_id')
|
||
->on('mv.wh_id', '=', 'a.wh_id')
|
||
->on('mv.inventory_sku_code', '=', 'a.inventory_sku_code');
|
||
})
|
||
->groupBy([
|
||
'a.wms_co_id', 'a.wh_id', 'a.inventory_sku_code', 'a.sku_code',
|
||
'a.category', 'a.warehouse_name',
|
||
'a.snapshot_qty', 'a.snapshot_at',
|
||
])
|
||
->select(['a.wms_co_id', 'a.wh_id', 'a.sku_code', 'a.category', 'a.warehouse_name'])
|
||
->selectRaw($openingSql . ' AS opening_qty', $openingBindings)
|
||
->selectRaw($closingSql . ' AS closing_qty', $closingBindings);
|
||
foreach (self::MOVEMENT_COLUMNS as $column => $type) {
|
||
$monthly->selectRaw(
|
||
'COALESCE(SUM(CASE WHEN mv.occurred_at >= ? AND mv.occurred_at < ? '
|
||
. 'AND mv.io_type = ? THEN ' . $signed . ' ELSE 0 END), 0) AS ' . $column,
|
||
[$period['start'], $period['end'], $type],
|
||
);
|
||
}
|
||
$monthly->selectRaw('0 AS opening_aux');
|
||
|
||
$monthlyRows = $this->connection->query()->fromSub($monthly, 'm')
|
||
->select('m.*')->orderBy('m.warehouse_name')->orderBy('m.wh_id')->orderBy('m.sku_code');
|
||
$rows = $this->boundedGet($monthlyRows, $queryTimeoutSeconds)
|
||
->map(function (object $row) use ($period): object {
|
||
$warehouse = trim((string) ($row->warehouse_name ?? ''));
|
||
$subwarehouseId = (int) ($row->wh_id ?? 0);
|
||
$subwarehouse = $this->subwarehouseName($warehouse, $subwarehouseId);
|
||
|
||
return (object) [
|
||
'owner' => (string) config(
|
||
'upstream_mirrors.third_party_warehouse_storage.owner_name',
|
||
'常州简尚家居有限公司',
|
||
),
|
||
'month' => Carbon::parse($period['start'])->toDateString(),
|
||
'warehouse_provider' => $warehouse,
|
||
'warehouse' => $subwarehouse,
|
||
'product_code' => (string) ($row->sku_code ?? ''),
|
||
'category' => (string) ($row->category ?? ''),
|
||
'opening_quantity' => $row->opening_qty,
|
||
'closing_quantity' => $row->closing_qty,
|
||
'processing_in' => $row->processing_in,
|
||
'processing_out' => $row->processing_out,
|
||
'purchase_in' => $row->purchase_in,
|
||
'purchase_return' => $row->purchase_return,
|
||
'other_in' => $row->other_in,
|
||
'other_return' => $row->other_return,
|
||
'sales_out' => $row->sales_out,
|
||
'sales_return' => $row->sales_return,
|
||
'stocktake' => $row->stocktake,
|
||
'transfer_out' => $row->transfer_out,
|
||
'transfer_in' => $row->transfer_in,
|
||
'opening' => $row->opening_aux,
|
||
'other_out' => $row->other_out,
|
||
];
|
||
})->all();
|
||
|
||
return [
|
||
'data' => $rows,
|
||
'total' => $total,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
}
|
||
|
||
/**
|
||
* Read one authoritative month without joining the movement mirror.
|
||
* The imported workbook already contains the opening balance, every movement
|
||
* column and the closing balance at the requested warehouse/SKU grain.
|
||
*
|
||
* @param array{start:string,end:string} $period
|
||
* @param array<string,mixed> $filters
|
||
* @param array<string,mixed> $state
|
||
* @return array{data:list<object>,total:int,state:array<string,mixed>,sync:array<string,mixed>}
|
||
*/
|
||
private function paginateCurrentMonthlyInventoryAuthority(
|
||
array $period,
|
||
array $filters,
|
||
int $page,
|
||
int $perPage,
|
||
bool $includeTotal,
|
||
array $state,
|
||
?int $queryTimeoutSeconds,
|
||
): array {
|
||
$query = $this->monthlyInventoryAuthoritativeQuery($period, $filters);
|
||
if ($query === null) {
|
||
return [
|
||
'data' => [],
|
||
'total' => 0,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
}
|
||
|
||
$total = $includeTotal ? $this->boundedCount(clone $query, $queryTimeoutSeconds) : 0;
|
||
$rows = $this->boundedGet(
|
||
(clone $query)
|
||
->orderBy('warehouse_provider')
|
||
->orderBy('warehouse')
|
||
->orderBy('product_code')
|
||
->forPage($page, $perPage),
|
||
$queryTimeoutSeconds,
|
||
)->map(static function (object $row): object {
|
||
return (object) [
|
||
'owner' => (string) $row->owner,
|
||
'month' => Carbon::parse($row->report_month)->toDateString(),
|
||
'warehouse_provider' => (string) $row->warehouse_provider,
|
||
'warehouse' => (string) $row->warehouse,
|
||
'product_code' => (string) $row->product_code,
|
||
'category' => (string) ($row->category ?? ''),
|
||
'opening_quantity' => $row->opening_quantity,
|
||
'closing_quantity' => $row->closing_quantity,
|
||
'processing_in' => $row->processing_in,
|
||
'processing_out' => $row->processing_out,
|
||
'purchase_in' => $row->purchase_in,
|
||
'purchase_return' => $row->purchase_return,
|
||
'other_in' => $row->other_in,
|
||
'other_return' => $row->other_return,
|
||
'sales_out' => $row->sales_out,
|
||
'sales_return' => $row->sales_return,
|
||
'stocktake' => $row->stocktake,
|
||
'transfer_out' => $row->transfer_out,
|
||
'transfer_in' => $row->transfer_in,
|
||
'opening' => $row->opening,
|
||
'other_out' => $row->other_out,
|
||
];
|
||
})->all();
|
||
|
||
return [
|
||
'data' => $rows,
|
||
'total' => $total,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
}
|
||
|
||
private function subwarehouseName(string $warehouseProvider, int $subwarehouseId): string
|
||
{
|
||
$specialName = self::SPECIAL_SUBWAREHOUSE_NAMES[$warehouseProvider][$subwarehouseId] ?? null;
|
||
if ($specialName !== null) {
|
||
return $specialName;
|
||
}
|
||
|
||
$suffix = self::SUBWAREHOUSE_SUFFIXES[$subwarehouseId] ?? '其他子仓（' . $subwarehouseId . '）';
|
||
|
||
return $warehouseProvider === '' ? $suffix : $warehouseProvider . $suffix;
|
||
}
|
||
|
||
/**
|
||
* @param array<string,mixed> $filters
|
||
* @return Generator<int,object>
|
||
*/
|
||
public function exportMonthlyInventoryReportRows(array $filters, int $batchSize = self::REPORT_EXPORT_BATCH_SIZE): Generator
|
||
{
|
||
$batchSize = min(max($batchSize, 1), self::REPORT_EXPORT_BATCH_SIZE);
|
||
$page = 1;
|
||
$state = $this->requireReadyState();
|
||
$sourceCounts = null;
|
||
$prefetchedMovementRows = $this->prefetchMonthlyMovementRows($filters, $state);
|
||
$refreshSourceCounts = true;
|
||
$exported = 0;
|
||
$expectedTotal = null;
|
||
|
||
while (true) {
|
||
$result = $this->paginateMonthlyInventoryReport(
|
||
$filters,
|
||
$page,
|
||
$batchSize,
|
||
false,
|
||
$state,
|
||
$sourceCounts,
|
||
self::REPORT_EXPORT_QUERY_TIMEOUT_SECONDS,
|
||
$prefetchedMovementRows,
|
||
$refreshSourceCounts,
|
||
);
|
||
$refreshSourceCounts = false;
|
||
|
||
if (isset($result['source_counts']) && is_array($result['source_counts'])) {
|
||
$sourceCounts = $result['source_counts'];
|
||
$expectedTotal ??= array_sum(array_map('intval', $sourceCounts));
|
||
}
|
||
|
||
$rows = $result['data'] ?? [];
|
||
if ($rows === []) {
|
||
return;
|
||
}
|
||
$rowCount = count($rows);
|
||
if ($expectedTotal !== null) {
|
||
$remaining = $expectedTotal - $exported;
|
||
if ($remaining <= 0) {
|
||
return;
|
||
}
|
||
if ($rowCount > $remaining) {
|
||
$rows = array_slice($rows, 0, $remaining);
|
||
$rowCount = count($rows);
|
||
}
|
||
}
|
||
foreach ($rows as $row) {
|
||
yield $row;
|
||
}
|
||
$exported += $rowCount;
|
||
|
||
// Release the previous batch before constructing the next SQL result.
|
||
unset($rows, $result);
|
||
if ($rowCount < $batchSize || ($expectedTotal !== null && $exported >= $expectedTotal)) {
|
||
return;
|
||
}
|
||
$page++;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 用户确认的聚水潭月报保留了“仓储方 + 子仓 + 商品”维度。
|
||
* 命中同月权威数据时整月直接读取，不再用当前库存倒推历史子仓余额。
|
||
*
|
||
* @param array{start:string,end:string} $period
|
||
* @param array<string,mixed> $filters
|
||
*/
|
||
private function monthlyInventoryAuthoritativeQuery(array $period, array $filters): ?Builder
|
||
{
|
||
try {
|
||
$schema = Schema::connection($this->connection->getName());
|
||
if (! $schema->hasTable(self::INVENTORY_REPORT_AUTHORITATIVE_STATE_TABLE)
|
||
|| ! $schema->hasTable(self::INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE)) {
|
||
return null;
|
||
}
|
||
|
||
$month = Carbon::parse($period['start'])->startOfMonth()->toDateString();
|
||
$authority = $this->currentMonthlyInventoryAuthority($period);
|
||
if ($authority === null) {
|
||
return null;
|
||
}
|
||
|
||
$query = $this->connection->table(self::INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE)
|
||
->where('generation', $authority['generation'])
|
||
->where('report_month', $month)
|
||
->select([
|
||
'owner', 'report_month', 'warehouse_provider', 'warehouse',
|
||
'product_code', 'category', 'opening_quantity', 'closing_quantity',
|
||
'processing_in', 'processing_out', 'purchase_in', 'purchase_return',
|
||
'other_in', 'other_return', 'sales_out', 'sales_return', 'stocktake',
|
||
'transfer_out', 'transfer_in', 'opening', 'other_out',
|
||
]);
|
||
|
||
$productCode = trim((string) ($filters['product_code'] ?? ''));
|
||
if ($productCode !== '') {
|
||
$query->where('product_code', $productCode);
|
||
}
|
||
$this->containsFilter($query, 'warehouse_provider', $filters['warehouse_provider'] ?? '');
|
||
$this->containsFilter($query, 'category', $filters['category'] ?? '');
|
||
|
||
return $query;
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Every later month starts from the latest authoritative month closing quantity and
|
||
* rolls confirmed movements forward to the requested month. Current inventory is
|
||
* deliberately ignored because it cannot reproduce historical opening balances.
|
||
*
|
||
* @param array{start:string,end:string} $period
|
||
* @param array<string,mixed> $filters
|
||
* @param array{generation:string,source_month:string} $authority
|
||
* @param array<string,mixed> $state
|
||
* @return array{data:list<object>,total:int,state:array<string,mixed>,sync:array<string,mixed>}|null
|
||
*/
|
||
private function monthlyInventoryRollForward(
|
||
array $period,
|
||
array $filters,
|
||
array $authority,
|
||
string $headerGeneration,
|
||
string $itemGeneration,
|
||
string $inventoryGeneration,
|
||
int $page,
|
||
int $perPage,
|
||
bool $includeTotal,
|
||
array $state,
|
||
bool $authorityIsCurrentMonth = false,
|
||
?array $sourceCounts = null,
|
||
?int $queryTimeoutSeconds = null,
|
||
?Collection $prefetchedMovementRows = null,
|
||
bool $refreshSourceCounts = false,
|
||
): ?array {
|
||
try {
|
||
$queryTimeoutSeconds ??= self::REPORT_QUERY_TIMEOUT_SECONDS;
|
||
$latestProducts = $this->connection->table(self::PRODUCT_TABLE)
|
||
->where('sku_type', 'normal')->whereNotNull('sku_id')->where('sku_id', '!=', '')
|
||
->groupBy('sku_id')->select('sku_id')->selectRaw('MAX(autoid) AS autoid');
|
||
$latestWarehouses = $this->connection->table(self::WAREHOUSE_TABLE)
|
||
->whereNotNull('wms_co_id')->groupBy('wms_co_id')
|
||
->select('wms_co_id')->selectRaw('MAX(Id) AS Id');
|
||
|
||
$authorityRows = $this->connection
|
||
->table(self::INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE . ' as authority')
|
||
->where('authority.generation', $authority['generation'])
|
||
->where('authority.report_month', $authority['source_month']);
|
||
$productCode = trim((string) ($filters['product_code'] ?? ''));
|
||
if ($productCode !== '') {
|
||
$authorityRows->where('authority.product_code', $productCode);
|
||
}
|
||
$this->containsFilter($authorityRows, 'authority.warehouse_provider', $filters['warehouse_provider'] ?? '');
|
||
$this->containsFilter($authorityRows, 'authority.category', $filters['category'] ?? '');
|
||
[$authorityWarehouseIdSql, $authorityWarehouseIdBindings] = $this->subwarehouseIdSql(
|
||
'authority.warehouse_provider',
|
||
'authority.warehouse',
|
||
);
|
||
$authorityOpeningColumn = $authorityIsCurrentMonth
|
||
? 'authority.opening_quantity'
|
||
: 'authority.closing_quantity';
|
||
$authorityCandidates = (clone $authorityRows)
|
||
->leftJoin(self::WAREHOUSE_TABLE . ' as authority_warehouse', function ($join): void {
|
||
$join->on('authority_warehouse.name', '=', 'authority.warehouse_provider');
|
||
})
|
||
->leftJoinSub($latestWarehouses, 'authority_latest_warehouse', function ($join): void {
|
||
$join->on('authority_latest_warehouse.wms_co_id', '=', 'authority_warehouse.wms_co_id')
|
||
->on('authority_latest_warehouse.Id', '=', 'authority_warehouse.Id');
|
||
})->selectRaw(
|
||
'authority.owner AS owner, authority.warehouse_provider AS warehouse_provider, '
|
||
. 'authority.warehouse AS source_warehouse, authority_warehouse.wms_co_id AS wms_co_id, '
|
||
. $authorityWarehouseIdSql . ' AS wh_id, authority.product_code AS sku_code, '
|
||
. 'authority.category AS category, ' . $authorityOpeningColumn . ' AS opening_qty',
|
||
$authorityWarehouseIdBindings,
|
||
);
|
||
$inventoryDimensions = $this->connection->table(self::SUBWAREHOUSE_INVENTORY_TABLE . ' as inventory')
|
||
->where('inventory.generation', $inventoryGeneration)
|
||
->where('inventory.qty', '!=', 0)
|
||
->whereNotNull('inventory.sku_code')->where('inventory.sku_code', '!=', '')
|
||
->joinSub($latestProducts, 'latest_product', 'latest_product.sku_id', '=', 'inventory.sku_code')
|
||
->join(self::PRODUCT_TABLE . ' as s', function ($join): void {
|
||
$join->on('s.sku_id', '=', 'latest_product.sku_id')
|
||
->on('s.autoid', '=', 'latest_product.autoid');
|
||
})
|
||
->joinSub($latestWarehouses, 'latest_warehouse', 'latest_warehouse.wms_co_id', '=', 'inventory.wms_co_id')
|
||
->join(self::WAREHOUSE_TABLE . ' as w', function ($join): void {
|
||
$join->on('w.wms_co_id', '=', 'latest_warehouse.wms_co_id')
|
||
->on('w.Id', '=', 'latest_warehouse.Id');
|
||
});
|
||
if ($productCode !== '') {
|
||
$inventoryDimensions->where('s.sku_id', $productCode);
|
||
}
|
||
$this->containsFilter($inventoryDimensions, 'w.name', $filters['warehouse_provider'] ?? '');
|
||
$this->containsFilter($inventoryDimensions, 's.category', $filters['category'] ?? '');
|
||
$owner = (string) config(
|
||
'upstream_mirrors.third_party_warehouse_storage.owner_name',
|
||
'常州简尚家居有限公司',
|
||
);
|
||
[$inventoryWarehouseSql, $inventoryWarehouseBindings] = $this->subwarehouseNameSql(
|
||
'w.name',
|
||
'inventory.wh_id',
|
||
);
|
||
$inventoryDimensions->whereNotExists(function (Builder $prior) use (
|
||
$authority,
|
||
$inventoryWarehouseSql,
|
||
$inventoryWarehouseBindings,
|
||
): void {
|
||
$prior->selectRaw('1')
|
||
->from(self::INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE . ' as prior')
|
||
->where('prior.generation', $authority['generation'])
|
||
->where('prior.report_month', $authority['source_month'])
|
||
->whereColumn('prior.product_code', 'inventory.sku_code')
|
||
->whereColumn('prior.warehouse_provider', 'w.name')
|
||
->whereRaw('prior.warehouse = ' . $inventoryWarehouseSql, $inventoryWarehouseBindings);
|
||
});
|
||
$inventoryDimensions->distinct()
|
||
->selectRaw(
|
||
'? AS owner, w.name AS warehouse_provider, '
|
||
. $inventoryWarehouseSql . ' AS source_warehouse, '
|
||
. 'inventory.wms_co_id AS wms_co_id, inventory.wh_id AS wh_id, '
|
||
. 'inventory.sku_code AS sku_code, s.category AS category, 0 AS opening_qty',
|
||
array_merge([$owner], $inventoryWarehouseBindings),
|
||
);
|
||
|
||
$authorityEnd = Carbon::parse($authority['source_month'])
|
||
->addMonth()->startOfMonth()->format('Y-m-d 00:00:00');
|
||
$movementStart = $authorityIsCurrentMonth ? $period['start'] : $authorityEnd;
|
||
$movementWarehouseIdSql = 'COALESCE(NULLIF(h.wh_id, 0), 1)';
|
||
$movementPairs = $this->confirmedMovements($headerGeneration, $itemGeneration, null)
|
||
->where('h.occurred_at', '>=', $movementStart)
|
||
->where('h.occurred_at', '<', $period['end'])
|
||
->where('i.qty', '!=', 0)
|
||
->where('i.qty_type', '!=', 0)
|
||
->select('h.wms_co_id', 'i.sku_code')
|
||
->selectRaw($movementWarehouseIdSql . ' AS wh_id')
|
||
->distinct();
|
||
if ($productCode !== '') {
|
||
$movementPairs->where('i.sku_code', $productCode);
|
||
}
|
||
$movementDimensions = $this->connection->query()->fromSub($movementPairs, 'movement_pair')
|
||
->joinSub($latestWarehouses, 'movement_latest_warehouse', 'movement_latest_warehouse.wms_co_id', '=', 'movement_pair.wms_co_id')
|
||
->join(self::WAREHOUSE_TABLE . ' as movement_warehouse', function ($join): void {
|
||
$join->on('movement_warehouse.wms_co_id', '=', 'movement_latest_warehouse.wms_co_id')
|
||
->on('movement_warehouse.Id', '=', 'movement_latest_warehouse.Id');
|
||
});
|
||
$this->containsFilter($movementDimensions, 'movement_warehouse.name', $filters['warehouse_provider'] ?? '');
|
||
[$movementWarehouseSql, $movementWarehouseBindings] = $this->subwarehouseNameSql(
|
||
'movement_warehouse.name',
|
||
'movement_pair.wh_id',
|
||
);
|
||
$movementDimensions->whereNotExists(function (Builder $prior) use (
|
||
$authority,
|
||
$movementWarehouseSql,
|
||
$movementWarehouseBindings,
|
||
): void {
|
||
$prior->selectRaw('1')
|
||
->from(self::INVENTORY_REPORT_AUTHORITATIVE_ROW_TABLE . ' as prior')
|
||
->where('prior.generation', $authority['generation'])
|
||
->where('prior.report_month', $authority['source_month'])
|
||
->whereColumn('prior.product_code', 'movement_pair.sku_code')
|
||
->whereColumn('prior.warehouse_provider', 'movement_warehouse.name')
|
||
->whereRaw('prior.warehouse = ' . $movementWarehouseSql, $movementWarehouseBindings);
|
||
});
|
||
if (! $authorityIsCurrentMonth) {
|
||
$movementDimensions->whereNotExists(function (Builder $snapshot) use ($inventoryGeneration): void {
|
||
$snapshot->selectRaw('1')
|
||
->from(self::SUBWAREHOUSE_INVENTORY_TABLE . ' as matching_inventory')
|
||
->where('matching_inventory.generation', $inventoryGeneration)
|
||
->where('matching_inventory.qty', '!=', 0)
|
||
->whereColumn('matching_inventory.wms_co_id', 'movement_pair.wms_co_id')
|
||
->whereRaw('COALESCE(NULLIF(matching_inventory.wh_id, 0), 1) = movement_pair.wh_id')
|
||
->whereColumn('matching_inventory.sku_code', 'movement_pair.sku_code');
|
||
});
|
||
}
|
||
$movementCount = clone $movementDimensions;
|
||
if (trim((string) ($filters['category'] ?? '')) === '') {
|
||
$movementCount->whereExists(function (Builder $product): void {
|
||
$product->selectRaw('1')->from(self::PRODUCT_TABLE . ' as known_product')
|
||
->where('known_product.sku_type', 'normal')
|
||
->whereColumn('known_product.sku_id', 'movement_pair.sku_code');
|
||
})->selectRaw('1 AS counted');
|
||
}
|
||
$movementDimensions->joinSub($latestProducts, 'movement_latest_product', 'movement_latest_product.sku_id', '=', 'movement_pair.sku_code')
|
||
->join(self::PRODUCT_TABLE . ' as movement_product', function ($join): void {
|
||
$join->on('movement_product.sku_id', '=', 'movement_latest_product.sku_id')
|
||
->on('movement_product.autoid', '=', 'movement_latest_product.autoid');
|
||
});
|
||
$this->containsFilter($movementDimensions, 'movement_product.category', $filters['category'] ?? '');
|
||
if (trim((string) ($filters['category'] ?? '')) !== '') {
|
||
$movementCount = (clone $movementDimensions)->selectRaw('1 AS counted');
|
||
}
|
||
$movementDimensions->selectRaw(
|
||
'? AS owner, movement_warehouse.name AS warehouse_provider, '
|
||
. $movementWarehouseSql . ' AS source_warehouse, '
|
||
. 'movement_pair.wms_co_id AS wms_co_id, movement_pair.wh_id AS wh_id, '
|
||
. 'movement_pair.sku_code AS sku_code, movement_product.category AS category, 0 AS opening_qty',
|
||
array_merge([$owner], $movementWarehouseBindings),
|
||
);
|
||
|
||
$sources = $authorityIsCurrentMonth
|
||
? [
|
||
[$authorityCandidates, $authorityCandidates, 'authority.warehouse_provider', 'authority.product_code'],
|
||
[$movementDimensions, $movementCount, 'movement_warehouse.name', 'movement_pair.sku_code'],
|
||
]
|
||
: [
|
||
[$authorityCandidates, $authorityCandidates, 'authority.warehouse_provider', 'authority.product_code'],
|
||
[$inventoryDimensions, $inventoryDimensions, 'w.name', 'inventory.sku_code'],
|
||
[$movementDimensions, $movementCount, 'movement_warehouse.name', 'movement_pair.sku_code'],
|
||
];
|
||
$counts = is_array($sourceCounts) ? array_values($sourceCounts) : [];
|
||
$total = 0;
|
||
$countCacheKey = $this->monthlyInventoryReportCacheKey(
|
||
$period, $filters, $authority, $headerGeneration, $itemGeneration,
|
||
$inventoryGeneration, 0, 0,
|
||
) . ':source-counts';
|
||
if ($includeTotal || $counts === [] || $refreshSourceCounts) {
|
||
$cachedCounts = $refreshSourceCounts ? null : Cache::get($countCacheKey);
|
||
if (is_array($cachedCounts) && count($cachedCounts) === count($sources)) {
|
||
$counts = array_values($cachedCounts);
|
||
} else {
|
||
foreach ($sources as $index => [, $countQuery]) {
|
||
$counts[$index] = $this->boundedCount(clone $countQuery, $queryTimeoutSeconds);
|
||
}
|
||
Cache::put($countCacheKey, $counts, self::REPORT_CACHE_SECONDS);
|
||
}
|
||
}
|
||
if ($includeTotal) {
|
||
$total = array_sum($counts);
|
||
}
|
||
$offset = ($page - 1) * $perPage;
|
||
$remaining = $perPage;
|
||
$candidatePages = [];
|
||
foreach ($sources as $index => [$candidates, $countQuery, $warehouseOrder, $productOrder]) {
|
||
if ($remaining === 0) {
|
||
break;
|
||
}
|
||
$sourceTotal = $counts[$index] ?? null;
|
||
if ($sourceTotal !== null && $offset >= $sourceTotal) {
|
||
$offset -= $sourceTotal;
|
||
continue;
|
||
}
|
||
$limit = $sourceTotal === null ? $remaining : min($remaining, $sourceTotal - $offset);
|
||
$candidatePages[] = (clone $candidates)
|
||
->orderBy($warehouseOrder)
|
||
->orderBy('wh_id')
|
||
->orderBy($productOrder)
|
||
->offset($offset)
|
||
->limit($limit);
|
||
$remaining -= $limit;
|
||
$offset = 0;
|
||
}
|
||
|
||
if ($candidatePages === []) {
|
||
return [
|
||
'data' => [],
|
||
'total' => $total,
|
||
'source_counts' => $counts,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
}
|
||
|
||
$pageCandidates = array_shift($candidatePages);
|
||
foreach ($candidatePages as $candidatePage) {
|
||
$pageCandidates->unionAll($candidatePage);
|
||
}
|
||
$pairRows = $this->boundedGet(clone $pageCandidates, $queryTimeoutSeconds);
|
||
if ($pairRows->isEmpty()) {
|
||
return [
|
||
'data' => [],
|
||
'total' => $total,
|
||
'source_counts' => $counts,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
}
|
||
|
||
$movementPairs = $pairRows->filter(
|
||
static fn (object $row): bool => $row->wms_co_id !== null && trim((string) $row->sku_code) !== '',
|
||
)->unique(static fn (object $row): string => implode(':', [
|
||
(string) $row->wms_co_id,
|
||
(string) $row->wh_id,
|
||
(string) $row->sku_code,
|
||
]))->values();
|
||
$movementRows = $prefetchedMovementRows ?? collect();
|
||
if ($movementPairs->isNotEmpty()) {
|
||
if ($prefetchedMovementRows === null) {
|
||
$movements = $this->aggregateMonthlyMovementQuery(
|
||
$period,
|
||
$authorityEnd,
|
||
$movementStart,
|
||
$headerGeneration,
|
||
$itemGeneration,
|
||
);
|
||
$movementRows = $this->boundedGet($movements, $queryTimeoutSeconds)->keyBy(
|
||
static fn (object $row): string => implode(':', [
|
||
(string) $row->wms_co_id,
|
||
(string) $row->wh_id,
|
||
(string) $row->sku_code,
|
||
]),
|
||
);
|
||
}
|
||
}
|
||
|
||
$rows = $pairRows->map(function (object $row) use ($movementRows, $period): object {
|
||
$key = implode(':', [(string) $row->wms_co_id, (string) $row->wh_id, (string) $row->sku_code]);
|
||
$movement = $movementRows->get($key);
|
||
$sourceWarehouse = trim((string) ($row->source_warehouse ?? ''));
|
||
$warehouse = $sourceWarehouse !== ''
|
||
? $sourceWarehouse
|
||
: $this->subwarehouseName(
|
||
trim((string) $row->warehouse_provider),
|
||
(int) $row->wh_id,
|
||
);
|
||
$rolledOpening = (float) $row->opening_qty + (float) ($movement->rolled_opening_delta ?? 0);
|
||
$closingQuantity = $rolledOpening;
|
||
foreach (array_keys(self::MOVEMENT_COLUMNS) as $column) {
|
||
$closingQuantity += (float) ($movement->{$column} ?? 0);
|
||
}
|
||
|
||
return (object) [
|
||
'owner' => $row->owner,
|
||
'month' => Carbon::parse($period['start'])->toDateString(),
|
||
'warehouse_provider' => $row->warehouse_provider,
|
||
'warehouse' => $warehouse,
|
||
'product_code' => $row->sku_code,
|
||
'category' => $row->category,
|
||
'opening_quantity' => $rolledOpening,
|
||
'closing_quantity' => $closingQuantity,
|
||
'processing_in' => $movement->processing_in ?? 0,
|
||
'processing_out' => $movement->processing_out ?? 0,
|
||
'purchase_in' => $movement->purchase_in ?? 0,
|
||
'purchase_return' => $movement->purchase_return ?? 0,
|
||
'other_in' => $movement->other_in ?? 0,
|
||
'other_return' => $movement->other_return ?? 0,
|
||
'sales_out' => $movement->sales_out ?? 0,
|
||
'sales_return' => $movement->sales_return ?? 0,
|
||
'stocktake' => $movement->stocktake ?? 0,
|
||
'transfer_out' => $movement->transfer_out ?? 0,
|
||
'transfer_in' => $movement->transfer_in ?? 0,
|
||
'opening' => 0,
|
||
'other_out' => $movement->other_out ?? 0,
|
||
];
|
||
})->all();
|
||
|
||
return [
|
||
'data' => $rows,
|
||
'total' => $total,
|
||
'source_counts' => $counts,
|
||
'state' => $state,
|
||
'sync' => $this->syncStatus($state),
|
||
];
|
||
} catch (Throwable $exception) {
|
||
if ($this->isStatementTimeout($exception)) {
|
||
throw $exception;
|
||
}
|
||
report($exception);
|
||
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Aggregate the complete movement range once for a full export. The normal
|
||
* page query intentionally limits movements to the current page, but doing
|
||
* that for every export batch repeats the same expensive scan thousands of
|
||
* times. The result is keyed by warehouse, subwarehouse and SKU so each
|
||
* page can perform an O(1) lookup.
|
||
*
|
||
* @param array<string,mixed> $filters
|
||
* @param array<string,mixed> $state
|
||
* @return Collection<string,object>|null
|
||
*/
|
||
private function prefetchMonthlyMovementRows(array $filters, array $state): ?Collection
|
||
{
|
||
$period = $this->inventoryReportPeriod((string) ($filters['month'] ?? ''));
|
||
$headerGeneration = $this->generation($state, 'active_header_generation');
|
||
$itemGeneration = $this->generation($state, 'active_item_generation');
|
||
if ($headerGeneration === null || $itemGeneration === null) {
|
||
return null;
|
||
}
|
||
|
||
$currentAuthority = $this->currentMonthlyInventoryAuthority($period);
|
||
if ($currentAuthority !== null) {
|
||
// The current authoritative month already contains all movement
|
||
// columns; exporting it must not scan the mirror movement tables.
|
||
return collect();
|
||
}
|
||
$authority = $currentAuthority ?? $this->previousMonthlyInventoryAuthority($period);
|
||
if ($authority === null) {
|
||
return null;
|
||
}
|
||
|
||
$authorityEnd = Carbon::parse($authority['source_month'])
|
||
->addMonth()->startOfMonth()->format('Y-m-d 00:00:00');
|
||
$movementStart = $currentAuthority !== null ? $period['start'] : $authorityEnd;
|
||
$query = $this->aggregateMonthlyMovementQuery(
|
||
$period,
|
||
$authorityEnd,
|
||
$movementStart,
|
||
$headerGeneration,
|
||
$itemGeneration,
|
||
);
|
||
|
||
return $this->boundedGet($query, self::REPORT_EXPORT_AGGREGATE_TIMEOUT_SECONDS)->keyBy(
|
||
static fn (object $row): string => implode(':', [
|
||
(string) $row->wms_co_id,
|
||
(string) $row->wh_id,
|
||
(string) $row->sku_code,
|
||
]),
|
||
);
|
||
}
|
||
|
||
/**
|
||
* @param array{start:string,end:string} $period
|
||
*/
|
||
private function aggregateMonthlyMovementQuery(
|
||
array $period,
|
||
string $authorityEnd,
|
||
string $movementStart,
|
||
string $headerGeneration,
|
||
string $itemGeneration,
|
||
): Builder {
|
||
$movements = $this->confirmedMovements($headerGeneration, $itemGeneration, null)
|
||
->where('h.occurred_at', '>=', $movementStart)
|
||
->where('h.occurred_at', '<', $period['end'])
|
||
->groupBy(['h.wms_co_id', 'i.sku_code'])
|
||
->groupByRaw('COALESCE(NULLIF(h.wh_id, 0), 1)')
|
||
->select(['h.wms_co_id', 'i.sku_code'])
|
||
->selectRaw('COALESCE(NULLIF(h.wh_id, 0), 1) AS wh_id');
|
||
$signed = '(COALESCE(i.qty, 0) * COALESCE(i.qty_type, 0))';
|
||
$movements->selectRaw(
|
||
'COALESCE(SUM(CASE WHEN h.occurred_at >= ? AND h.occurred_at < ? '
|
||
. 'THEN ' . $signed . ' ELSE 0 END), 0) AS rolled_opening_delta',
|
||
[$authorityEnd, $period['start']],
|
||
);
|
||
foreach (self::MOVEMENT_COLUMNS as $column => $type) {
|
||
$expression = 'COALESCE(SUM(CASE WHEN h.occurred_at >= ? AND h.occurred_at < ? '
|
||
. 'AND h.io_type = ? THEN ' . $signed . ' ELSE 0 END), 0)';
|
||
$movements->selectRaw($expression . ' AS ' . $column, [$period['start'], $period['end'], $type]);
|
||
}
|
||
|
||
return $movements;
|
||
}
|
||
|
||
/**
|
||
* @param array{start:string,end:string} $period
|
||
* @param array<string,mixed> $filters
|
||
* @param array{generation:string,source_month:string} $authority
|
||
*/
|
||
private function monthlyInventoryReportCacheKey(
|
||
array $period,
|
||
array $filters,
|
||
array $authority,
|
||
string $headerGeneration,
|
||
string $itemGeneration,
|
||
string $inventoryGeneration,
|
||
int $page,
|
||
int $perPage,
|
||
): string {
|
||
$normalizedFilters = [
|
||
'product_code' => trim((string) ($filters['product_code'] ?? '')),
|
||
'warehouse_provider' => trim((string) ($filters['warehouse_provider'] ?? '')),
|
||
'category' => trim((string) ($filters['category'] ?? '')),
|
||
];
|
||
|
||
return 'inventory-report:monthly-page:v10:' . hash('sha256', (string) json_encode([
|
||
'period' => $period,
|
||
'filters' => $normalizedFilters,
|
||
'authority' => $authority,
|
||
'header_generation' => $headerGeneration,
|
||
'item_generation' => $itemGeneration,
|
||
'inventory_generation' => $inventoryGeneration,
|
||
'page' => $page,
|
||
'per_page' => $perPage,
|
||
], JSON_UNESCAPED_UNICODE));
|
||
}
|
||
|
||
/** @param callable():?array $resolver */
|
||
private function rememberMonthlyInventoryReport(string $key, callable $resolver, int $waitSeconds = self::REPORT_CACHE_WAIT_SECONDS): ?array
|
||
{
|
||
$cached = Cache::get($key);
|
||
if (is_array($cached)) {
|
||
return $cached;
|
||
}
|
||
|
||
try {
|
||
return Cache::lock($key . ':lock', self::REPORT_CACHE_LOCK_SECONDS)
|
||
->block($waitSeconds, function () use ($key, $resolver): ?array {
|
||
$cached = Cache::get($key);
|
||
if (is_array($cached)) {
|
||
return $cached;
|
||
}
|
||
|
||
$resolved = $resolver();
|
||
if (is_array($resolved)) {
|
||
Cache::put($key, $resolved, self::REPORT_CACHE_SECONDS);
|
||
}
|
||
|
||
return $resolved;
|
||
});
|
||
} catch (LockTimeoutException $exception) {
|
||
$cached = Cache::get($key);
|
||
if (is_array($cached)) {
|
||
return $cached;
|
||
}
|
||
|
||
throw new RuntimeException('inventory_report_generation_busy', 0, $exception);
|
||
}
|
||
}
|
||
|
||
/** @param array{start:string,end:string} $period */
|
||
private function currentMonthlyInventoryAuthority(array $period): ?array
|
||
{
|
||
$month = Carbon::parse($period['start'])->startOfMonth();
|
||
$state = $this->monthlyAuthorityStateQuery()
|
||
?->whereDate('source_month', '=', $month->toDateString())
|
||
->first();
|
||
if ($state === null) {
|
||
$state = $this->connection->table(self::INVENTORY_REPORT_AUTHORITATIVE_STATE_TABLE)
|
||
->where('dataset_key', self::INVENTORY_REPORT_AUTHORITATIVE_DATASET_KEY)
|
||
->whereNotNull('active_generation')
|
||
->whereDate('source_month', '=', $month->toDateString())
|
||
->first();
|
||
}
|
||
if ($state === null || trim((string) $state->active_generation) === '') {
|
||
return null;
|
||
}
|
||
|
||
return [
|
||
'generation' => (string) $state->active_generation,
|
||
'source_month' => $month->toDateString(),
|
||
];
|
||
}
|
||
|
||
/** @param array{start:string,end:string} $period */
|
||
private function previousMonthlyInventoryAuthority(array $period): ?array
|
||
{
|
||
$month = Carbon::parse($period['start'])->startOfMonth();
|
||
$state = $this->monthlyAuthorityStateQuery()
|
||
?->whereDate('source_month', '<', $month)
|
||
->orderByDesc('source_month')
|
||
->first();
|
||
if ($state === null) {
|
||
$state = $this->connection->table(self::INVENTORY_REPORT_AUTHORITATIVE_STATE_TABLE)
|
||
->where('dataset_key', self::INVENTORY_REPORT_AUTHORITATIVE_DATASET_KEY)
|
||
->whereNotNull('active_generation')
|
||
->whereDate('source_month', '<', $month)
|
||
->orderByDesc('source_month')
|
||
->first();
|
||
}
|
||
if ($state === null || trim((string) $state->active_generation) === '') {
|
||
return null;
|
||
}
|
||
|
||
return [
|
||
'generation' => (string) $state->active_generation,
|
||
'source_month' => Carbon::parse($state->source_month)->startOfMonth()->toDateString(),
|
||
];
|
||
}
|
||
|
||
private function monthlyAuthorityStateQuery(): ?Builder
|
||
{
|
||
try {
|
||
$schema = Schema::connection($this->connection->getName());
|
||
if (! $schema->hasTable('inventory_report_authoritative_months')) {
|
||
return null;
|
||
}
|
||
|
||
return $this->connection->table('inventory_report_authoritative_months')
|
||
->where('dataset_key', self::INVENTORY_REPORT_AUTHORITATIVE_DATASET_KEY)
|
||
->where('status', 'succeeded')
|
||
->whereNotNull('active_generation');
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/** @return array{0:string,1:list<mixed>} */
|
||
private function subwarehouseIdSql(string $providerColumn, string $warehouseColumn): array
|
||
{
|
||
$cases = [];
|
||
$bindings = [];
|
||
foreach (self::SPECIAL_SUBWAREHOUSE_NAMES as $provider => $names) {
|
||
foreach ($names as $id => $name) {
|
||
$cases[] = "WHEN {$providerColumn} = ? AND {$warehouseColumn} = ? THEN {$id}";
|
||
array_push($bindings, $provider, $name);
|
||
}
|
||
}
|
||
foreach (self::SUBWAREHOUSE_SUFFIXES as $id => $suffix) {
|
||
$cases[] = 'WHEN ' . $warehouseColumn . ' = '
|
||
. $this->concatSql([$providerColumn, '?']) . " THEN {$id}";
|
||
$bindings[] = $suffix;
|
||
}
|
||
|
||
return ['CASE ' . implode(' ', $cases) . ' ELSE -1 END', $bindings];
|
||
}
|
||
|
||
/** @return array{0:string,1:list<mixed>} */
|
||
private function subwarehouseNameSql(string $providerColumn, string $subwarehouseColumn): array
|
||
{
|
||
$cases = [];
|
||
$bindings = [];
|
||
foreach (self::SPECIAL_SUBWAREHOUSE_NAMES as $provider => $names) {
|
||
foreach ($names as $id => $name) {
|
||
$cases[] = "WHEN {$providerColumn} = ? AND {$subwarehouseColumn} = {$id} THEN ?";
|
||
array_push($bindings, $provider, $name);
|
||
}
|
||
}
|
||
foreach (self::SUBWAREHOUSE_SUFFIXES as $id => $suffix) {
|
||
$cases[] = "WHEN {$subwarehouseColumn} = {$id} THEN " . $this->concatSql([$providerColumn, '?']);
|
||
$bindings[] = $suffix;
|
||
}
|
||
|
||
$fallback = $this->concatSql([$providerColumn, "'其他子仓（'", $subwarehouseColumn, "'）'"]);
|
||
|
||
return ['CASE ' . implode(' ', $cases) . ' ELSE ' . $fallback . ' END', $bindings];
|
||
}
|
||
|
||
/** @param list<string> $parts */
|
||
private function concatSql(array $parts): string
|
||
{
|
||
return $this->connection->getDriverName() === 'sqlite'
|
||
? '(' . implode(' || ', $parts) . ')'
|
||
: 'CONCAT(' . implode(', ', $parts) . ')';
|
||
}
|
||
|
||
/**
|
||
* @param array<string,mixed> $filters
|
||
* @return array{count:int,opening_qty:float,closing_qty:float}
|
||
*/
|
||
public function summary(array $filters): array
|
||
{
|
||
$state = $this->requireReadyState();
|
||
$calculated = $this->calculatedFlowRows($filters, $state, $this->hasProductFilters($filters));
|
||
if ($calculated !== null) {
|
||
return $this->summarizeRows($calculated);
|
||
}
|
||
|
||
return $this->summarizeQuery($this->filteredReportQuery(
|
||
$filters,
|
||
$state,
|
||
$this->hasProductFilters($filters),
|
||
));
|
||
}
|
||
|
||
/**
|
||
* @param array<string,mixed> $filters
|
||
* @return array{suppliers:list<string>,categories:list<string>}
|
||
*/
|
||
public function filterOptions(array $filters = []): array
|
||
{
|
||
$state = $this->requireReadyState();
|
||
$calculated = $this->calculatedFlowRows($filters, $state);
|
||
if ($calculated !== null) {
|
||
$options = [];
|
||
foreach (['suppliers' => 'supplier', 'categories' => 'category'] as $key => $field) {
|
||
$options[$key] = collect($calculated)->pluck($field)->filter(
|
||
static fn (mixed $value): bool => $value !== null && $value !== '',
|
||
)->unique()->sort()->values()->all();
|
||
}
|
||
|
||
return $options;
|
||
}
|
||
$query = $this->filteredReportQuery($filters, $state);
|
||
|
||
$suppliers = (clone $query)->whereNotNull('supplier')->where('supplier', '!=', '')
|
||
->distinct()->orderBy('supplier')->pluck('supplier')
|
||
->map(static fn (mixed $value): string => (string) $value)->all();
|
||
$categories = (clone $query)->whereNotNull('category')->where('category', '!=', '')
|
||
->distinct()->orderBy('category')->pluck('category')
|
||
->map(static fn (mixed $value): string => (string) $value)->all();
|
||
|
||
return ['suppliers' => $suppliers, 'categories' => $categories];
|
||
}
|
||
|
||
/**
|
||
* @param array<string,mixed> $filters
|
||
* @return Generator<int,array<string,mixed>>
|
||
*/
|
||
public function exportRows(
|
||
array $filters,
|
||
int $batchSize = 500,
|
||
bool $includeProductDetails = true,
|
||
): Generator
|
||
{
|
||
$state = $this->requireReadyState();
|
||
$calculated = $this->calculatedFlowRows($filters, $state, $includeProductDetails);
|
||
if ($calculated !== null) {
|
||
yield from $calculated;
|
||
|
||
return;
|
||
}
|
||
$batchSize = min(max($batchSize, 1), 5000);
|
||
$lastDate = null;
|
||
$lastSku = null;
|
||
|
||
while (true) {
|
||
// Settlement only needs quantities and SKU/date fields. Skipping the
|
||
// large product catalogue join keeps the page calculation bounded
|
||
// while the normal CSV export retains product details.
|
||
$query = $this->filteredReportQuery($filters, $state, $includeProductDetails);
|
||
if ($lastDate !== null && $lastSku !== null) {
|
||
$query->where(function (Builder $cursor) use ($lastDate, $lastSku): void {
|
||
$cursor->where('report_date', '>', $lastDate)
|
||
->orWhere(function (Builder $sameDate) use ($lastDate, $lastSku): void {
|
||
$sameDate->where('report_date', $lastDate)
|
||
->where('sku_code', '>', $lastSku);
|
||
});
|
||
});
|
||
}
|
||
|
||
$rows = $query->orderBy('report_date')->orderBy('sku_code')->limit($batchSize)->get();
|
||
if ($rows->isEmpty()) {
|
||
return;
|
||
}
|
||
|
||
foreach ($rows as $row) {
|
||
yield $this->mapRow($row);
|
||
}
|
||
|
||
$lastDate = (string) $rows->last()->report_date;
|
||
$lastSku = (string) $rows->last()->sku_code;
|
||
if ($rows->count() < $batchSize) {
|
||
return;
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Historical imports stay on the indexed SQL path. Later periods aggregate
|
||
* confirmed movements once, then expand only the compact SKU/day balances.
|
||
* Cache quantities by published generations, independently of page/filters;
|
||
* product details remain live and are restricted to these report SKUs.
|
||
*
|
||
* @return list<array<string,mixed>>|null
|
||
*/
|
||
private function calculatedFlowRows(array $filters, array $state, bool $includeProductDetails = true): ?array
|
||
{
|
||
$period = $this->period($filters);
|
||
$authority = $this->authoritativeAnchor();
|
||
if ($this->authoritativeReportQuery($period, $authority, false) !== null) {
|
||
return null;
|
||
}
|
||
|
||
$key = 'third-party-warehouse:daily-quantities:v3:' . hash('sha256', (string) json_encode([
|
||
'database' => $this->connection->getDatabaseName(),
|
||
'period' => $period,
|
||
'authority' => $authority,
|
||
'state' => array_intersect_key($state, array_flip([
|
||
'active_header_generation', 'active_item_generation', 'active_inventory_generation', 'last_succeeded_at',
|
||
])),
|
||
]));
|
||
$rows = $this->rememberMonthlyInventoryReport(
|
||
$key,
|
||
fn (): array => $this->calculateFlowQuantities($period, $state, $authority),
|
||
5,
|
||
) ?? [];
|
||
$owner = (string) config('upstream_mirrors.third_party_warehouse_storage.owner_name', '常州简尚家居有限公司');
|
||
foreach ($rows as &$row) {
|
||
$row['owner'] = $owner;
|
||
}
|
||
unset($row);
|
||
$skuFilter = trim((string) ($filters['sku_code'] ?? ''));
|
||
if ($skuFilter !== '') {
|
||
$rows = array_values(array_filter($rows, static fn (array $row): bool =>
|
||
mb_stripos((string) $row['sku_code'], $skuFilter) !== false));
|
||
}
|
||
if ($rows === [] || (! $includeProductDetails && ! $this->hasProductFilters($filters))) {
|
||
return $rows;
|
||
}
|
||
|
||
$skus = array_values(array_unique(array_column($rows, 'sku_code')));
|
||
$latest = $this->connection->table(self::PRODUCT_TABLE)
|
||
->where('sku_type', 'normal')->whereIn('sku_id', $skus)
|
||
->groupBy('sku_id')->select('sku_id')->selectRaw('MAX(autoid) AS autoid');
|
||
$products = $this->connection->query()->fromSub($latest, 'latest')
|
||
->join(self::PRODUCT_TABLE . ' as s', function ($join): void {
|
||
$join->on('s.sku_id', '=', 'latest.sku_id')->on('s.autoid', '=', 'latest.autoid')
|
||
->where('s.sku_type', '=', 'normal');
|
||
});
|
||
foreach (['product_name' => 's.name', 'supplier' => 's.supplier_name', 'category' => 's.category'] as $filter => $column) {
|
||
$this->containsFilter($products, $column, $filters[$filter] ?? '');
|
||
}
|
||
$products->select([
|
||
's.sku_id', 's.name as product_name', 's.supplier_name as supplier', 's.category',
|
||
's.vc_name as virtual_category', 's.supplier_sku_id as supplier_style_no', 's.brand', 's.unit',
|
||
])->selectRaw('COALESCE(s.pic_big, s.pic) AS image_url')
|
||
->selectRaw('COALESCE(s.short_name, s.name) AS short_name')
|
||
->selectRaw('COALESCE(s.properties_value, s.color) AS color_spec');
|
||
$details = $this->boundedGet($products, 10)->keyBy(fn (object $row): string => $this->flowSkuKey((string) $row->sku_id));
|
||
$requiresProduct = $this->hasProductFilters($filters);
|
||
$result = [];
|
||
foreach ($rows as $row) {
|
||
$product = $details->get($this->flowSkuKey($row['sku_code']));
|
||
if ($requiresProduct && $product === null) {
|
||
continue;
|
||
}
|
||
$result[] = array_replace($row, array_intersect_key((array) $product, array_flip(self::OUTPUT_COLUMNS)));
|
||
}
|
||
|
||
return $result;
|
||
}
|
||
|
||
/** @return list<array<string,mixed>> */
|
||
private function calculateFlowQuantities(array $period, array $state, ?array $authority): array
|
||
{
|
||
$anchorAt = $authority['anchor_at'] ?? $this->stateAnchor($state);
|
||
$inventoryGeneration = $this->generation($state, 'active_inventory_generation');
|
||
$anchors = $authority !== null
|
||
? $this->connection->table(self::AUTHORITATIVE_ANCHOR_TABLE)
|
||
->where('generation', $authority['generation'])->select(['sku_code', 'qty', 'anchor_at as snapshot_at'])
|
||
: $this->connection->table(self::INVENTORY_TABLE)
|
||
->where('generation', $inventoryGeneration)->where('wms_co_id', self::WAREHOUSE_ID)
|
||
->select(['sku_code', 'qty', 'captured_at as snapshot_at']);
|
||
$anchors->whereNotNull('sku_code')->where('sku_code', '!=', '');
|
||
$snapshots = $this->boundedGet($anchors, 10)->keyBy(fn (object $row): string => $this->flowSkuKey((string) $row->sku_code));
|
||
$movements = $this->confirmedMovements(
|
||
$this->generation($state, 'active_header_generation'),
|
||
$this->generation($state, 'active_item_generation'),
|
||
)->where('h.occurred_at', '>=', min($period['start'], $anchorAt))
|
||
->where('h.occurred_at', '<', max($period['end'], $anchorAt));
|
||
if ($authority !== null) {
|
||
$movements->leftJoin(self::AUTHORITATIVE_ANCHOR_TABLE . ' as snapshot', function ($join) use ($authority): void {
|
||
$join->on('snapshot.sku_code', '=', 'i.sku_code')->where('snapshot.generation', $authority['generation']);
|
||
});
|
||
$snapshotTime = 'COALESCE(snapshot.anchor_at, ?)';
|
||
} else {
|
||
$movements->leftJoin(self::INVENTORY_TABLE . ' as snapshot', function ($join) use ($inventoryGeneration): void {
|
||
$join->on('snapshot.sku_code', '=', 'i.sku_code')->where('snapshot.generation', $inventoryGeneration)
|
||
->where('snapshot.wms_co_id', self::WAREHOUSE_ID);
|
||
});
|
||
$snapshotTime = 'COALESCE(snapshot.captured_at, ?)';
|
||
}
|
||
$signed = '(COALESCE(i.qty, 0) * COALESCE(i.qty_type, 0))';
|
||
$movements->groupBy('i.sku_code')->groupByRaw('DATE(h.occurred_at)')->select('i.sku_code')
|
||
->selectRaw('DATE(h.occurred_at) AS report_date')
|
||
->selectRaw('SUM(CASE WHEN h.occurred_at ' . ($authority === null ? '>' : '>=') . ' '
|
||
. $snapshotTime . ' THEN ' . $signed . ' ELSE 0 END) AS forward_qty', [$anchorAt])
|
||
->selectRaw('SUM(CASE WHEN h.occurred_at < ' . $snapshotTime . ' THEN '
|
||
. $signed . ' ELSE 0 END) AS backward_qty', [$anchorAt]);
|
||
foreach (self::MOVEMENT_COLUMNS as $field => $type) {
|
||
$movements->selectRaw('SUM(CASE WHEN h.io_type = ? THEN ' . $signed . ' ELSE 0 END) AS ' . $field, [$type]);
|
||
}
|
||
$daily = $this->boundedGet($movements, 15)->groupBy(fn (object $row): string => $this->flowSkuKey((string) $row->sku_code));
|
||
$skus = array_unique([...$snapshots->keys()->all(), ...$daily->keys()->all()]);
|
||
sort($skus, SORT_STRING);
|
||
$byDate = [];
|
||
foreach ($skus as $sku) {
|
||
$snapshot = $snapshots->get($sku);
|
||
$displaySku = (string) ($snapshot->sku_code ?? $daily->get($sku)?->first()?->sku_code ?? $sku);
|
||
$snapshotAt = (string) ($snapshot->snapshot_at ?? $anchorAt);
|
||
$flows = $daily->get($sku, collect())->sortBy('report_date')->keyBy('report_date');
|
||
$forward = 0.0;
|
||
$backward = (float) $flows->sum('backward_qty');
|
||
foreach ($flows as $date => $flow) {
|
||
if ($date < substr($period['start'], 0, 10)) {
|
||
$forward += (float) $flow->forward_qty;
|
||
$backward -= (float) $flow->backward_qty;
|
||
}
|
||
}
|
||
$cursor = Carbon::parse($period['start']);
|
||
$end = Carbon::parse($period['end']);
|
||
while ($cursor < $end) {
|
||
$date = $cursor->toDateString();
|
||
$flow = $flows->get($date);
|
||
$forward += (float) ($flow->forward_qty ?? 0);
|
||
$backward -= (float) ($flow->backward_qty ?? 0);
|
||
$dayEnd = $cursor->copy()->addDay()->toDateTimeString();
|
||
$closing = round((float) ($snapshot->qty ?? 0) + ($snapshotAt < $dayEnd ? $forward : -$backward), 4);
|
||
$row = ['report_date' => $date, 'sku_code' => $displaySku, 'closing_qty' => $closing, 'opening_aux' => 0];
|
||
$net = 0.0;
|
||
$nonZero = false;
|
||
foreach (self::MOVEMENT_COLUMNS as $field => $type) {
|
||
$row[$field] = round((float) ($flow->{$field} ?? 0), 4);
|
||
$net += $row[$field];
|
||
$nonZero = $nonZero || $row[$field] != 0;
|
||
}
|
||
$row['opening_qty'] = round($closing - $net, 4);
|
||
if ($nonZero || $closing != 0 || $row['opening_qty'] != 0) {
|
||
$byDate[$date][] = $this->mapRow((object) $row);
|
||
}
|
||
$cursor->addDay();
|
||
}
|
||
}
|
||
ksort($byDate);
|
||
|
||
return array_merge([], ...array_values($byDate));
|
||
}
|
||
|
||
private function flowSkuKey(string $sku): string
|
||
{
|
||
// MariaDB's shared mirror uses case-insensitive SKU joins. PHP keys
|
||
// must use the same identity or mixed-case movements duplicate SKUs.
|
||
return $this->connection->getDriverName() === 'mysql' ? mb_strtolower($sku) : $sku;
|
||
}
|
||
|
||
/** @return array{count:int,opening_qty:float,closing_qty:float} */
|
||
private function summarizeRows(array $rows): array
|
||
{
|
||
return [
|
||
'count' => count($rows),
|
||
'opening_qty' => round(array_sum(array_column($rows, 'opening_qty')), 4),
|
||
'closing_qty' => round(array_sum(array_column($rows, 'closing_qty')), 4),
|
||
];
|
||
}
|
||
|
||
/** @param array<string,mixed> $state */
|
||
private function filteredReportQuery(
|
||
array $filters,
|
||
array $state,
|
||
bool $includeProductDetails = true,
|
||
): Builder
|
||
{
|
||
$query = $this->connection->query()->fromSub(
|
||
$this->reportQuery($filters, $state, $includeProductDetails),
|
||
'report',
|
||
);
|
||
|
||
$this->containsFilter($query, 'sku_code', $filters['sku_code'] ?? '');
|
||
$this->containsFilter($query, 'product_name', $filters['product_name'] ?? '');
|
||
$this->containsFilter($query, 'supplier', $filters['supplier'] ?? '');
|
||
$this->containsFilter($query, 'category', $filters['category'] ?? '');
|
||
|
||
$nonZero = ['opening_qty', 'closing_qty', ...array_keys(self::MOVEMENT_COLUMNS)];
|
||
$query->where(function (Builder $values) use ($nonZero): void {
|
||
foreach ($nonZero as $index => $column) {
|
||
$method = $index === 0 ? 'where' : 'orWhere';
|
||
$values->{$method}($column, '<>', 0);
|
||
}
|
||
});
|
||
|
||
return $query;
|
||
}
|
||
|
||
/** @param array<string,mixed> $state */
|
||
private function reportQuery(
|
||
array $filters,
|
||
array $state,
|
||
bool $includeProductDetails,
|
||
): Builder
|
||
{
|
||
$period = $this->period($filters);
|
||
$headerGeneration = $this->generation($state, 'active_header_generation');
|
||
$itemGeneration = $this->generation($state, 'active_item_generation');
|
||
$inventoryGeneration = $this->generation($state, 'active_inventory_generation');
|
||
$authoritativeAnchor = $this->authoritativeAnchor();
|
||
$anchorAt = $authoritativeAnchor['anchor_at'] ?? $this->stateAnchor($state);
|
||
|
||
if ($headerGeneration === null || $itemGeneration === null || $inventoryGeneration === null) {
|
||
throw new RuntimeException('third_party_warehouse_mirror_not_ready');
|
||
}
|
||
|
||
$authoritativeRows = $this->authoritativeReportQuery(
|
||
$period,
|
||
$authoritativeAnchor,
|
||
$includeProductDetails,
|
||
);
|
||
if ($authoritativeRows !== null) {
|
||
return $authoritativeRows;
|
||
}
|
||
|
||
$skuRangeStart = min($period['start'], $anchorAt);
|
||
$skuRangeEnd = max($period['end'], $anchorAt);
|
||
$movementSkus = $this->confirmedMovements($headerGeneration, $itemGeneration)
|
||
->where('h.occurred_at', '>=', $skuRangeStart)
|
||
->where('h.occurred_at', '<', $skuRangeEnd)
|
||
->select('i.sku_code');
|
||
if ($authoritativeAnchor !== null) {
|
||
$skuUnion = $this->connection->table(self::AUTHORITATIVE_ANCHOR_TABLE . ' as inventory_skus')
|
||
->where('inventory_skus.generation', $authoritativeAnchor['generation'])
|
||
->whereNotNull('inventory_skus.sku_code')
|
||
->where('inventory_skus.sku_code', '!=', '')
|
||
->select('inventory_skus.sku_code')
|
||
->union($movementSkus);
|
||
$anchors = $this->connection->query()->fromSub($skuUnion, 'sku_set')
|
||
->leftJoin(self::AUTHORITATIVE_ANCHOR_TABLE . ' as inventory', function ($join) use ($authoritativeAnchor): void {
|
||
$join->on('inventory.sku_code', '=', 'sku_set.sku_code')
|
||
->where('inventory.generation', '=', $authoritativeAnchor['generation']);
|
||
})
|
||
->select('sku_set.sku_code')
|
||
->selectRaw('COALESCE(inventory.qty, 0) AS snapshot_qty')
|
||
->selectRaw('COALESCE(inventory.anchor_at, ?) AS snapshot_at', [$anchorAt]);
|
||
} else {
|
||
$skuUnion = $this->connection->table(self::INVENTORY_TABLE . ' as inventory_skus')
|
||
->where('inventory_skus.generation', $inventoryGeneration)
|
||
->where('inventory_skus.wms_co_id', self::WAREHOUSE_ID)
|
||
->whereNotNull('inventory_skus.sku_code')
|
||
->where('inventory_skus.sku_code', '!=', '')
|
||
->select('inventory_skus.sku_code')
|
||
->union($movementSkus);
|
||
$anchors = $this->connection->query()->fromSub($skuUnion, 'sku_set')
|
||
->leftJoin(self::INVENTORY_TABLE . ' as inventory', function ($join) use ($inventoryGeneration): void {
|
||
$join->on('inventory.sku_code', '=', 'sku_set.sku_code')
|
||
->where('inventory.generation', '=', $inventoryGeneration)
|
||
->where('inventory.wms_co_id', '=', self::WAREHOUSE_ID);
|
||
})
|
||
->select('sku_set.sku_code')
|
||
->selectRaw('COALESCE(inventory.qty, 0) AS snapshot_qty')
|
||
->selectRaw('COALESCE(inventory.captured_at, ?) AS snapshot_at', [$anchorAt]);
|
||
}
|
||
|
||
$dateSkuAnchors = $this->connection->query()
|
||
->fromSub($this->dateSeries($period['start'], $period['end']), 'd')
|
||
->crossJoinSub($anchors, 'a')
|
||
->select([
|
||
'd.report_date',
|
||
'd.day_start',
|
||
'd.day_end',
|
||
'a.sku_code',
|
||
'a.snapshot_qty',
|
||
'a.snapshot_at',
|
||
]);
|
||
$signedQuantity = '(COALESCE(movement_qty, 0) * COALESCE(movement_qty_type, 0))';
|
||
$forwardAnchorOperator = $authoritativeAnchor === null ? '>' : '>=';
|
||
$closingExpression = 'da.snapshot_qty + COALESCE(SUM(CASE '
|
||
. 'WHEN da.snapshot_at < da.day_end AND mv.occurred_at ' . $forwardAnchorOperator . ' da.snapshot_at '
|
||
. 'AND mv.occurred_at < da.day_end THEN ' . $signedQuantity . ' '
|
||
. 'WHEN da.snapshot_at >= da.day_end AND mv.occurred_at >= da.day_end '
|
||
. 'AND mv.occurred_at < da.snapshot_at THEN -' . $signedQuantity . ' '
|
||
. 'ELSE 0 END), 0)';
|
||
$allMovements = $this->confirmedMovements($headerGeneration, $itemGeneration)
|
||
->where('h.occurred_at', '>=', $skuRangeStart)
|
||
->where('h.occurred_at', '<', $skuRangeEnd)
|
||
->select([
|
||
'i.sku_code',
|
||
'i.qty as movement_qty',
|
||
'i.qty_type as movement_qty_type',
|
||
'h.occurred_at',
|
||
]);
|
||
$balances = $this->connection->query()->fromSub($dateSkuAnchors, 'da')
|
||
->leftJoinSub($allMovements, 'mv', 'mv.sku_code', '=', 'da.sku_code')
|
||
->groupBy(
|
||
'da.report_date',
|
||
'da.day_start',
|
||
'da.day_end',
|
||
'da.sku_code',
|
||
'da.snapshot_qty',
|
||
'da.snapshot_at',
|
||
)
|
||
->select(['da.report_date', 'da.sku_code'])
|
||
->selectRaw($closingExpression . ' AS closing_qty');
|
||
|
||
$periodMovements = $this->confirmedMovements($headerGeneration, $itemGeneration)
|
||
->where('h.occurred_at', '>=', $period['start'])
|
||
->where('h.occurred_at', '<', $period['end'])
|
||
->groupBy('i.sku_code')
|
||
->groupByRaw('DATE(h.occurred_at)')
|
||
->select('i.sku_code')
|
||
->selectRaw('DATE(h.occurred_at) AS report_date');
|
||
foreach (self::MOVEMENT_COLUMNS as $column => $type) {
|
||
$periodMovements->selectRaw(
|
||
'COALESCE(SUM(CASE WHEN h.io_type = ? THEN '
|
||
. '(COALESCE(i.qty, 0) * COALESCE(i.qty_type, 0)) ELSE 0 END), 0) AS ' . $column,
|
||
[$type],
|
||
);
|
||
}
|
||
|
||
$report = $this->connection->query()->fromSub($balances, 'b')
|
||
->leftJoinSub($periodMovements, 'm', function ($join): void {
|
||
$join->on('m.sku_code', '=', 'b.sku_code')
|
||
->on('m.report_date', '=', 'b.report_date');
|
||
});
|
||
if ($includeProductDetails) {
|
||
$latestProducts = $this->connection->table(self::PRODUCT_TABLE)
|
||
->where('sku_type', 'normal')
|
||
->whereNotNull('sku_id')->where('sku_id', '!=', '')
|
||
->groupBy('sku_id')->select('sku_id')->selectRaw('MAX(autoid) AS autoid');
|
||
$report->leftJoinSub(
|
||
$latestProducts,
|
||
'latest_product',
|
||
'latest_product.sku_id',
|
||
'=',
|
||
'b.sku_code',
|
||
)->leftJoin(self::PRODUCT_TABLE . ' as s', function ($join): void {
|
||
$join->on('s.sku_id', '=', 'latest_product.sku_id')
|
||
->on('s.autoid', '=', 'latest_product.autoid')
|
||
->where('s.sku_type', '=', 'normal');
|
||
});
|
||
}
|
||
|
||
$movementSelects = [];
|
||
foreach (array_keys(self::MOVEMENT_COLUMNS) as $column) {
|
||
$movementSelects[$column] = 'COALESCE(m.' . $column . ', 0)';
|
||
}
|
||
$netMovement = implode(' + ', array_values($movementSelects));
|
||
$openingQuantity = '(b.closing_qty - (' . $netMovement . '))';
|
||
$owner = (string) config(
|
||
'upstream_mirrors.third_party_warehouse_storage.owner_name',
|
||
'常州简尚家居有限公司',
|
||
);
|
||
|
||
$report->selectRaw('? AS owner', [$owner])
|
||
->addSelect('b.report_date')
|
||
->addSelect('b.sku_code');
|
||
if ($includeProductDetails) {
|
||
$report->addSelect([
|
||
$this->connection->raw('COALESCE(s.pic_big, s.pic) AS image_url'),
|
||
$this->connection->raw('s.name AS product_name'),
|
||
$this->connection->raw('COALESCE(s.short_name, s.name) AS short_name'),
|
||
$this->connection->raw('s.supplier_name AS supplier'),
|
||
$this->connection->raw('s.category AS category'),
|
||
$this->connection->raw('s.vc_name AS virtual_category'),
|
||
$this->connection->raw('s.supplier_sku_id AS supplier_style_no'),
|
||
$this->connection->raw('COALESCE(s.properties_value, s.color) AS color_spec'),
|
||
$this->connection->raw('s.brand AS brand'),
|
||
$this->connection->raw('s.unit AS unit'),
|
||
]);
|
||
}
|
||
$report->selectRaw($openingQuantity . ' AS opening_qty')
|
||
->selectRaw('b.closing_qty AS closing_qty');
|
||
foreach ($movementSelects as $column => $expression) {
|
||
$report->selectRaw($expression . ' AS ' . $column);
|
||
}
|
||
|
||
$report->selectRaw('0 AS opening_aux');
|
||
|
||
return $report;
|
||
}
|
||
|
||
private function dateSeries(string $start, string $end): Builder
|
||
{
|
||
$cursor = Carbon::parse($start)->startOfDay();
|
||
$through = Carbon::parse($end)->startOfDay();
|
||
$dates = null;
|
||
|
||
while ($cursor < $through) {
|
||
$dayStart = $cursor->copy();
|
||
$dayEnd = $cursor->copy()->addDay();
|
||
$row = $this->connection->query()->selectRaw(
|
||
'? AS report_date, ? AS day_start, ? AS day_end',
|
||
[$dayStart->toDateString(), $dayStart->toDateTimeString(), $dayEnd->toDateTimeString()],
|
||
);
|
||
$dates = $dates === null ? $row : $dates->unionAll($row);
|
||
$cursor->addDay();
|
||
}
|
||
|
||
if (! $dates instanceof Builder) {
|
||
throw new RuntimeException('third_party_warehouse_period_invalid');
|
||
}
|
||
|
||
return $dates;
|
||
}
|
||
|
||
private function confirmedMovements(
|
||
string $headerGeneration,
|
||
string $itemGeneration,
|
||
?int $warehouseId = self::WAREHOUSE_ID,
|
||
): Builder
|
||
{
|
||
$query = $this->connection->table(self::HEADER_TABLE . ' as h');
|
||
// MariaDB otherwise scans every historical header for the all-warehouse roll-forward.
|
||
if ($warehouseId === null && $this->connection->getDriverName() === 'mysql') {
|
||
$query->forceIndex('tpw_headers_report_index');
|
||
}
|
||
$query->join(self::ITEM_TABLE . ' as i', function ($join) use ($itemGeneration): void {
|
||
$join->on('i.io_id', '=', 'h.io_id')
|
||
->where('i.generation', '=', $itemGeneration);
|
||
})
|
||
->where('h.generation', $headerGeneration)
|
||
->where('h.status', 'Confirmed')
|
||
->whereIn('h.io_type', array_values(self::MOVEMENT_COLUMNS))
|
||
->whereNotExists(function (Builder $newer) use ($headerGeneration): void {
|
||
$newer->selectRaw('1')->from(self::HEADER_TABLE . ' as newer')
|
||
->where('newer.generation', $headerGeneration)
|
||
->whereColumn('newer.io_id', 'h.io_id')
|
||
->whereColumn('newer.id', '>', 'h.id');
|
||
})
|
||
->whereNotNull('i.sku_code')->where('i.sku_code', '!=', '');
|
||
|
||
if ($warehouseId !== null) {
|
||
$query->where('h.wms_co_id', $warehouseId);
|
||
}
|
||
|
||
return $query;
|
||
}
|
||
|
||
/**
|
||
* @param array<string,mixed> $filters
|
||
* @return array{dimension:string,value:string,start:string,end:string,days:int}
|
||
*/
|
||
private function period(array $filters): array
|
||
{
|
||
$dimension = ($filters['dimension'] ?? 'month') === 'day' ? 'day' : 'month';
|
||
$candidate = trim((string) ($filters['period'] ?? ''));
|
||
|
||
if ($dimension === 'day') {
|
||
if (preg_match('/^\d{4}-\d{2}-\d{2}$/', $candidate) !== 1) {
|
||
$candidate = trim((string) ($filters['date'] ?? ''));
|
||
}
|
||
try {
|
||
$start = preg_match('/^\d{4}-\d{2}-\d{2}$/', $candidate) === 1
|
||
? Carbon::createFromFormat('!Y-m-d', $candidate)
|
||
: now()->startOfDay();
|
||
} catch (Throwable) {
|
||
$start = now()->startOfDay();
|
||
}
|
||
if ($start->lt(Carbon::parse(self::START_DATE)->startOfDay())) {
|
||
$start = Carbon::parse(self::START_DATE)->startOfDay();
|
||
}
|
||
|
||
return [
|
||
'dimension' => 'day',
|
||
'value' => $start->format('Y-m-d'),
|
||
'start' => $start->format('Y-m-d 00:00:00'),
|
||
'end' => $start->copy()->addDay()->format('Y-m-d 00:00:00'),
|
||
'days' => 1,
|
||
];
|
||
}
|
||
|
||
if (preg_match('/^\d{4}-\d{2}$/', $candidate) !== 1) {
|
||
$candidate = trim((string) ($filters['month'] ?? ''));
|
||
}
|
||
try {
|
||
$start = preg_match('/^\d{4}-\d{2}$/', $candidate) === 1
|
||
? Carbon::createFromFormat('!Y-m', $candidate)->startOfMonth()
|
||
: now()->startOfMonth();
|
||
} catch (Throwable) {
|
||
$start = now()->startOfMonth();
|
||
}
|
||
$minimum = Carbon::parse(self::START_DATE)->startOfMonth();
|
||
if ($start->lt($minimum)) {
|
||
$start = $minimum;
|
||
}
|
||
|
||
return [
|
||
'dimension' => 'month',
|
||
'value' => $start->format('Y-m'),
|
||
'start' => $start->format('Y-m-d 00:00:00'),
|
||
'end' => $start->copy()->addMonth()->format('Y-m-d 00:00:00'),
|
||
'days' => $start->daysInMonth,
|
||
];
|
||
}
|
||
|
||
/** @return array{dimension:string,value:string,start:string,end:string,days:int} */
|
||
private function inventoryReportPeriod(string $month): array
|
||
{
|
||
try {
|
||
$start = preg_match('/^\d{4}-\d{2}$/', trim($month)) === 1
|
||
? Carbon::createFromFormat('!Y-m', trim($month))->startOfMonth()
|
||
: now()->startOfMonth();
|
||
} catch (Throwable) {
|
||
$start = now()->startOfMonth();
|
||
}
|
||
|
||
$minimum = Carbon::parse(self::INVENTORY_REPORT_START_DATE)->startOfMonth();
|
||
if ($start->lt($minimum)) {
|
||
$start = $minimum;
|
||
}
|
||
|
||
return [
|
||
'dimension' => 'month',
|
||
'value' => $start->format('Y-m'),
|
||
'start' => $start->format('Y-m-d 00:00:00'),
|
||
'end' => $start->copy()->addMonth()->format('Y-m-d 00:00:00'),
|
||
'days' => $start->daysInMonth,
|
||
];
|
||
}
|
||
|
||
/** @param array<string,mixed> $state */
|
||
private function stateAnchor(array $state): string
|
||
{
|
||
$timestamp = $this->timestamp($state['last_succeeded_at'] ?? null);
|
||
|
||
return $timestamp === null
|
||
? now()->toDateTimeString()
|
||
: Carbon::createFromTimestamp($timestamp)->toDateTimeString();
|
||
}
|
||
|
||
/**
|
||
* @param array{generation:string,anchor_at:string,source_period_start:?string,source_period_end:?string}|null $authoritativeAnchor
|
||
*/
|
||
private function authoritativeReportQuery(
|
||
array $period,
|
||
?array $authoritativeAnchor,
|
||
bool $includeProductDetails,
|
||
): ?Builder {
|
||
if ($authoritativeAnchor === null) {
|
||
return null;
|
||
}
|
||
|
||
try {
|
||
$schema = Schema::connection($this->connection->getName());
|
||
if (! $schema->hasTable(self::AUTHORITATIVE_ROW_TABLE)) {
|
||
return null;
|
||
}
|
||
$sourceStart = $authoritativeAnchor['source_period_start'] ?? null;
|
||
$sourceEnd = $authoritativeAnchor['source_period_end'] ?? null;
|
||
if (! is_string($sourceStart) || ! is_string($sourceEnd)
|
||
|| $sourceStart === '' || $sourceEnd === '') {
|
||
return null;
|
||
}
|
||
$sourceStartAt = Carbon::parse($sourceStart)->startOfDay();
|
||
$sourceEndAt = Carbon::parse($sourceEnd)->addDay()->startOfDay();
|
||
$periodStartAt = Carbon::parse($period['start']);
|
||
$periodEndAt = Carbon::parse($period['end']);
|
||
if ($periodStartAt->lt($sourceStartAt) || $periodEndAt->gt($sourceEndAt)) {
|
||
return null;
|
||
}
|
||
|
||
$rows = $this->connection->table(self::AUTHORITATIVE_ROW_TABLE . ' as b')
|
||
->where('b.generation', $authoritativeAnchor['generation'])
|
||
->where('b.report_date', '>=', $periodStartAt->toDateString())
|
||
->where('b.report_date', '<', $periodEndAt->toDateString());
|
||
$owner = (string) config(
|
||
'upstream_mirrors.third_party_warehouse_storage.owner_name',
|
||
'常州简尚家居有限公司',
|
||
);
|
||
$rows->selectRaw('? AS owner', [$owner])->addSelect(['b.report_date', 'b.sku_code']);
|
||
if ($includeProductDetails) {
|
||
$latestProducts = $this->connection->table(self::PRODUCT_TABLE)
|
||
->where('sku_type', 'normal')
|
||
->whereNotNull('sku_id')->where('sku_id', '!=', '')
|
||
->groupBy('sku_id')->select('sku_id')->selectRaw('MAX(autoid) AS autoid');
|
||
$rows->leftJoinSub(
|
||
$latestProducts,
|
||
'latest_product',
|
||
'latest_product.sku_id',
|
||
'=',
|
||
'b.sku_code',
|
||
)->leftJoin(self::PRODUCT_TABLE . ' as s', function ($join): void {
|
||
$join->on('s.sku_id', '=', 'latest_product.sku_id')
|
||
->on('s.autoid', '=', 'latest_product.autoid')
|
||
->where('s.sku_type', '=', 'normal');
|
||
})->addSelect([
|
||
$this->connection->raw('COALESCE(s.pic_big, s.pic) AS image_url'),
|
||
$this->connection->raw('s.name AS product_name'),
|
||
$this->connection->raw('COALESCE(s.short_name, s.name) AS short_name'),
|
||
$this->connection->raw('s.supplier_name AS supplier'),
|
||
$this->connection->raw('s.category AS category'),
|
||
$this->connection->raw('s.vc_name AS virtual_category'),
|
||
$this->connection->raw('s.supplier_sku_id AS supplier_style_no'),
|
||
$this->connection->raw('COALESCE(s.properties_value, s.color) AS color_spec'),
|
||
$this->connection->raw('s.brand AS brand'),
|
||
$this->connection->raw('s.unit AS unit'),
|
||
]);
|
||
}
|
||
$rows->addSelect([
|
||
'b.opening_qty', 'b.closing_qty', 'b.processing_in', 'b.processing_out',
|
||
'b.purchase_in', 'b.purchase_return', 'b.other_in', 'b.other_return',
|
||
'b.sales_out', 'b.sales_return', 'b.stocktake', 'b.transfer_out',
|
||
'b.transfer_in', 'b.opening_aux', 'b.other_out',
|
||
]);
|
||
|
||
return $rows;
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/** @return array{generation:string,anchor_at:string,source_period_start:?string,source_period_end:?string}|null */
|
||
private function authoritativeAnchor(): ?array
|
||
{
|
||
try {
|
||
$schema = Schema::connection($this->connection->getName());
|
||
if (! $schema->hasTable(self::AUTHORITATIVE_STATE_TABLE)
|
||
|| ! $schema->hasTable(self::AUTHORITATIVE_ANCHOR_TABLE)) {
|
||
return null;
|
||
}
|
||
|
||
$state = $this->connection->table(self::AUTHORITATIVE_STATE_TABLE)
|
||
->where('dataset_key', self::AUTHORITATIVE_DATASET_KEY)
|
||
->whereNotNull('active_generation')
|
||
->whereNotNull('anchor_at')
|
||
->first();
|
||
if ($state === null) {
|
||
return null;
|
||
}
|
||
|
||
$generation = trim((string) $state->active_generation);
|
||
$anchorAt = $this->timestamp($state->anchor_at ?? null);
|
||
if ($generation === '' || $anchorAt === null) {
|
||
return null;
|
||
}
|
||
|
||
return [
|
||
'generation' => $generation,
|
||
'anchor_at' => Carbon::createFromTimestamp($anchorAt)->toDateTimeString(),
|
||
'source_period_start' => isset($state->source_period_start)
|
||
? (string) $state->source_period_start : null,
|
||
'source_period_end' => isset($state->source_period_end)
|
||
? (string) $state->source_period_end : null,
|
||
];
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/** @return array<string,mixed> */
|
||
private function requireReadyState(): array
|
||
{
|
||
$state = $this->state();
|
||
if (! is_array($state)
|
||
|| ($state['initial_completed_at'] ?? null) === null
|
||
|| $this->generation($state, 'active_header_generation') === null
|
||
|| $this->generation($state, 'active_item_generation') === null
|
||
|| $this->generation($state, 'active_inventory_generation') === null
|
||
|| ! $this->dependenciesReady()) {
|
||
throw new RuntimeException('third_party_warehouse_mirror_not_ready');
|
||
}
|
||
|
||
return $state;
|
||
}
|
||
|
||
private function dependenciesReady(): bool
|
||
{
|
||
try {
|
||
$required = [self::PRODUCT_DATASET_KEY];
|
||
$ready = $this->connection->table(self::GENERIC_STATE_TABLE)
|
||
->whereIn('dataset_key', $required)
|
||
->whereNotNull('initial_completed_at')
|
||
->pluck('dataset_key')
|
||
->map(static fn (mixed $dataset): string => (string) $dataset)
|
||
->all();
|
||
|
||
return array_diff($required, $ready) === [];
|
||
} catch (Throwable) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
private function warehouseDependencyReady(): bool
|
||
{
|
||
try {
|
||
if (! $this->connection->getSchemaBuilder()->hasTable(self::WAREHOUSE_TABLE)) {
|
||
return false;
|
||
}
|
||
|
||
return $this->connection->table(self::GENERIC_STATE_TABLE)
|
||
->where('dataset_key', self::WAREHOUSE_DATASET_KEY)
|
||
->whereNotNull('initial_completed_at')
|
||
->whereIn('status', ['succeeded', 'running', 'failed'])
|
||
->exists();
|
||
} catch (Throwable) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
/** @param array<string,mixed> $state */
|
||
private function generation(array $state, string $key): ?string
|
||
{
|
||
$value = trim((string) ($state[$key] ?? ''));
|
||
|
||
return $value === '' ? null : $value;
|
||
}
|
||
|
||
private function timestamp(mixed $value): ?int
|
||
{
|
||
if ($value instanceof CarbonInterface) {
|
||
return $value->getTimestamp();
|
||
}
|
||
if (! is_string($value) || trim($value) === '') {
|
||
return null;
|
||
}
|
||
|
||
try {
|
||
return Carbon::parse($value)->getTimestamp();
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @return array{count:int,opening_qty:float,closing_qty:float}
|
||
*/
|
||
private function summarizeQuery(Builder $query): array
|
||
{
|
||
$row = $query->selectRaw(
|
||
'COUNT(*) AS row_count, '
|
||
. 'COALESCE(SUM(opening_qty), 0) AS opening_qty, '
|
||
. 'COALESCE(SUM(closing_qty), 0) AS closing_qty'
|
||
)->first();
|
||
|
||
return [
|
||
'count' => (int) ($row->row_count ?? 0),
|
||
'opening_qty' => round((float) ($row->opening_qty ?? 0), 4),
|
||
'closing_qty' => round((float) ($row->closing_qty ?? 0), 4),
|
||
];
|
||
}
|
||
|
||
/** @return array<string,mixed> */
|
||
private function mapRow(object $row): array
|
||
{
|
||
$mapped = [];
|
||
foreach (self::OUTPUT_COLUMNS as $column) {
|
||
$mapped[$column] = $column === 'date'
|
||
? ($row->report_date ?? null)
|
||
: ($row->{$column} ?? null);
|
||
}
|
||
foreach (['opening_qty', 'closing_qty', ...array_keys(self::MOVEMENT_COLUMNS), 'opening_aux'] as $column) {
|
||
$mapped[$column] = round((float) ($mapped[$column] ?? 0), 4);
|
||
}
|
||
return $mapped;
|
||
}
|
||
|
||
private function containsFilter(Builder $query, string $column, mixed $value): void
|
||
{
|
||
$value = trim((string) $value);
|
||
if ($value !== '') {
|
||
$query->whereRaw($column . " LIKE ? ESCAPE '!'", [$this->contains($value)]);
|
||
}
|
||
}
|
||
|
||
/** @param array<string,mixed> $filters */
|
||
private function hasProductFilters(array $filters): bool
|
||
{
|
||
foreach (['product_name', 'supplier', 'category'] as $filter) {
|
||
if (trim((string) ($filters[$filter] ?? '')) !== '') {
|
||
return true;
|
||
}
|
||
}
|
||
|
||
return false;
|
||
}
|
||
|
||
private function contains(string $value): string
|
||
{
|
||
return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], $value) . '%';
|
||
}
|
||
|
||
/** @return Collection<int,object> */
|
||
private function boundedGet(Builder $query, ?int $timeoutSeconds = null): Collection
|
||
{
|
||
if ($this->connection->getDriverName() !== 'mysql') {
|
||
return $query->get();
|
||
}
|
||
|
||
$timeoutSeconds = max(1, $timeoutSeconds ?? self::REPORT_QUERY_TIMEOUT_SECONDS);
|
||
|
||
return collect($this->connection->select(
|
||
'SET STATEMENT max_statement_time=' . $timeoutSeconds . ' FOR ' . $query->toSql(),
|
||
$query->getBindings(),
|
||
));
|
||
}
|
||
|
||
private function boundedCount(Builder $query, ?int $timeoutSeconds = null): int
|
||
{
|
||
$count = $this->connection->query()
|
||
->fromSub($query, 'bounded_count')
|
||
->selectRaw('COUNT(*) AS aggregate');
|
||
|
||
return (int) ($this->boundedGet($count, $timeoutSeconds)->first()->aggregate ?? 0);
|
||
}
|
||
|
||
private function isStatementTimeout(Throwable $exception): bool
|
||
{
|
||
$message = strtolower($exception->getMessage());
|
||
|
||
return str_contains($message, 'max_statement_time')
|
||
|| str_contains($message, 'query execution was interrupted');
|
||
}
|
||
}
|