*/ private const REQUIRED_DATASETS = [ self::DETAILS_DATASET, self::EXTENSION_DATASET, self::OUTBOUND_DATASET, ]; /** 仅暴露附件中有演示值、且已经核对来源的字段。 */ private const COLUMNS = [ 'detail_id', 'waybill', 'product_id', 'send_time', 'transaction_order_no', 'province', 'city', 'district', 'barcode', 'express_name', 'express_name_2', 'quantity', 'product_name', 'platform', 'shop_name', 'warehouse', 'volume', 'supplier', 'length', 'width', 'height', 'freight', 'reconciliation_weight', 'reconciliation_volume', 'month', 'status', ]; private readonly ConnectionInterface $connection; public function __construct( private readonly BusinessMirrorDatasetRegistry $datasets, private readonly GenericBusinessMirrorRepository $mirror, ) { $this->connection = $mirror->connection(); } public function available(): bool { try { $schema = Schema::connection($this->connection->getName()); if (! $schema->hasTable('business_mirror_table_states')) { return false; } foreach (self::REQUIRED_DATASETS as $key) { if (! $schema->hasTable($this->datasets->get($key)->targetTable())) { return false; } } return true; } catch (Throwable) { return false; } } public function ready(): bool { try { foreach (self::REQUIRED_DATASETS as $key) { if (! $this->mirror->ready($this->datasets->get($key))) { return false; } } return true; } catch (Throwable) { return false; } } /** @return array|null */ public function state(): ?array { try { $states = []; foreach (self::REQUIRED_DATASETS as $key) { $state = $this->mirror->state($this->datasets->get($key)); if (is_array($state)) { $states[$key] = $state; } } if ($states === []) { return null; } $first = $states[self::DETAILS_DATASET] ?? reset($states); $initial = array_values(array_filter(array_map( static fn (array $state): ?string => $state['initial_completed_at'] ?? null, $states, ))); $succeeded = array_values(array_filter(array_map( static fn (array $state): ?string => $state['last_succeeded_at'] ?? null, $states, ))); $statuses = array_map(static fn (array $state): string => (string) ($state['status'] ?? 'pending'), $states); $first['initial_completed_at'] = count($initial) === count(self::REQUIRED_DATASETS) ? min($initial) : null; $first['last_succeeded_at'] = count($succeeded) === count(self::REQUIRED_DATASETS) ? min($succeeded) : null; $first['status'] = in_array('failed', $statuses, true) ? 'failed' : (in_array('running', $statuses, true) ? 'running' : (count(array_filter($statuses, static fn (string $value): bool => $value === 'succeeded')) === count(self::REQUIRED_DATASETS) ? 'succeeded' : 'pending')); $first['datasets'] = $states; return $first; } catch (Throwable) { return null; } } /** @return array{code:string,last_succeeded_at:?string,lag_minutes:?int} */ public function syncStatus(?array $state = null): array { $state ??= $this->state(); if (! is_array($state)) { return ['code' => 'unavailable', 'last_succeeded_at' => null, 'lag_minutes' => null]; } $lastSucceeded = $state['last_succeeded_at'] ?? null; $lagMinutes = null; if (is_string($lastSucceeded) && $lastSucceeded !== '') { try { $lagMinutes = (int) ceil(max(0, now()->getTimestamp() - Carbon::parse($lastSucceeded)->getTimestamp()) / 60); } catch (Throwable) { $lagMinutes = null; } } $status = (string) ($state['status'] ?? 'pending'); if (($state['initial_completed_at'] ?? null) === null) { $code = $status === 'failed' ? 'initial_failed' : ($status === 'running' ? 'initializing' : 'pending'); } elseif ($status === 'failed') { $code = 'failed'; } elseif ($status === 'running') { $code = 'syncing'; } elseif ($lagMinutes === null || $lagMinutes > max((int) config('upstream_mirrors.courier_information.stale_after_minutes', 60), 1)) { $code = 'stale'; } else { $code = 'fresh'; } return ['code' => $code, 'last_succeeded_at' => $lastSucceeded, 'lag_minutes' => $lagMinutes]; } /** * @param array $filters * @return array{data:list>,total:int,summary:array,state:array|null,sync:array} */ public function paginate(array $filters, int $page, int $perPage): array { if (! $this->ready()) { throw new RuntimeException('courier_freight_reconciliation_mirror_not_ready'); } $page = max($page, 1); $perPage = min(max($perPage, 1), 100); $query = $this->filteredQuery($filters); $total = (clone $query)->count('d.id'); $summary = (clone $query)->selectRaw( 'COUNT(DISTINCT d.id) AS aggregate_count, COALESCE(SUM(o.freight), 0) AS freight', )->first(); $rows = (clone $query)->select($this->selectColumns()) ->orderByDesc('d.id') ->forPage($page, $perPage) ->get() ->map(fn (object $row): array => $this->mapRow($row)) ->all(); $state = $this->state(); $freight = (float) ($summary->freight ?? 0); return [ 'data' => $rows, 'total' => $total, 'summary' => [ 'count' => (int) ($summary->aggregate_count ?? 0), 'freight' => $freight, 'payable_amount' => $freight, 'actual_payment' => 0.0, 'confirmed_amount' => 0.0, 'reconciliation_difference' => 0.0, ], 'state' => $state, 'sync' => $this->syncStatus($state), ]; } private function filteredQuery(array $filters): Builder { $query = $this->detailQuery()->whereNotNull('d.logistic_id')->where('d.logistic_id', '!=', ''); $this->containsFilter($query, 'd.logistic_id', $filters['express_no'] ?? ''); if (($filters['express_name'] ?? '') !== '') { $needle = $this->contains((string) $filters['express_name']); $query->where(function (Builder $name) use ($needle): void { $name->whereRaw("d.logistic_company_name LIKE ? ESCAPE '!'", [$needle]) ->orWhereRaw("x.expressName LIKE ? ESCAPE '!'", [$needle]); }); } if (($filters['month'] ?? '') !== '' && preg_match('/^\d{4}-\d{2}$/', (string) $filters['month']) === 1) { $from = (string) $filters['month'] . '-01'; $query->where('d.send_time', '>=', $from . ' 00:00:00') ->where('d.send_time', '<', date('Y-m-d 00:00:00', strtotime($from . ' +1 month'))); } if (($filters['status'] ?? '') !== '') { $query->where('d.logistic_status_name', '=', $filters['status']); } if (($filters['modified_from'] ?? '') !== '') { $query->where('d.updatedAt', '>=', $filters['modified_from'] . ' 00:00:00'); } if (($filters['modified_to'] ?? '') !== '') { $query->where('d.updatedAt', '<', date('Y-m-d 00:00:00', strtotime($filters['modified_to'] . ' +1 day'))); } return $query; } private function detailQuery(): Builder { $details = $this->datasets->get(self::DETAILS_DATASET)->targetTable(); $extension = $this->datasets->get(self::EXTENSION_DATASET)->targetTable(); $outbound = $this->datasets->get(self::OUTBOUND_DATASET)->targetTable(); return $this->connection->table($details . ' as d') ->leftJoinSub($this->latestBy($extension, 'so_id', 'Id'), 'x', 'x.so_id', '=', 'd.so_id') ->leftJoinSub($this->latestBy($outbound, 'io_id', 'Id'), 'o', 'o.io_id', '=', 'x.io_id'); } /** 固定取关联键下 Id 最大的一行，避免一对多重复页面行和运费汇总。 */ private function latestBy(string $table, string $relationColumn, string $idColumn): Builder { return $this->connection->table($table . ' as candidate') ->whereNotExists(function (Builder $query) use ($table, $relationColumn, $idColumn): void { $query->selectRaw('1') ->from($table . ' as newer') ->whereColumn('newer.' . $relationColumn, '=', 'candidate.' . $relationColumn) ->whereColumn('newer.' . $idColumn, '>', 'candidate.' . $idColumn); }); } /** @return list */ private function selectColumns(): array { return [ 'd.id as detail_id', 'd.logistic_id as waybill', 'x.shop_i_id as product_id', 'd.send_time', $this->connection->raw('COALESCE(x.so_id, d.so_id) as transaction_order_no'), 'x.receiver_state as province', 'x.receiver_city as city', 'x.receiver_district as district', 'x.sku_ids as barcode', 'd.logistic_company_name as express_name', 'x.expressName as express_name_2', 'x.qty as quantity', 'x.sku_name as product_name', 'd.platform_name as platform', $this->connection->raw('COALESCE(x.shop_name, d.shop_name) as shop_name'), $this->connection->raw('COALESCE(x.wareHoursName, d.warehouse_name) as warehouse'), 'x.volume as volume', 'x.supplier_name as supplier', 'x.l as length', 'x.w as width', 'x.h as height', 'o.freight as freight', 'd.f_weight as reconciliation_weight', 'x.volume as reconciliation_volume', 'd.logistic_status_name as status', ]; } /** @return array */ private function mapRow(object $row): array { $mapped = []; foreach (self::COLUMNS as $column) { $mapped[$column] = $row->{$column} ?? null; } $sendTime = $mapped['send_time']; $mapped['month'] = is_string($sendTime) ? substr($sendTime, 0, 7) : null; foreach (['length', 'width', 'height', 'volume', 'freight', 'reconciliation_weight', 'reconciliation_volume'] as $number) { $mapped[$number] = $mapped[$number] === null || $mapped[$number] === '' ? null : (float) $mapped[$number]; } if ($mapped['quantity'] !== null && $mapped['quantity'] !== '') { $mapped['quantity'] = (int) $mapped['quantity']; } 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)]); } } private function contains(string $value): string { return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], $value) . '%'; } }