[['woTypeId', 'woNo', 'Id']], self::PAYMENT_MAP_DATASET => [['woTypeId', 'work_order_create_time', 'woNo']], ]; private readonly ConnectionInterface $connection; public function __construct( private readonly BusinessMirrorDatasetRegistry $datasets, private readonly GenericBusinessMirrorRepository $mirror, ) { $this->connection = $mirror->connection(); } public function available(): bool { $schema = Schema::connection($this->connection->getName()); try { return $schema->hasTable('business_mirror_table_states') && $schema->hasTable($this->dataset(self::WORK_ORDER_DATASET)->targetTable()) && $schema->hasTable($this->dataset(self::PAYMENT_MAP_DATASET)->targetTable()); } catch (Throwable) { return false; } } public function ready(): bool { return $this->mirror->ready($this->dataset(self::WORK_ORDER_DATASET)) && $this->mirror->ready($this->dataset(self::PAYMENT_MAP_DATASET)); } public function queryReady(): bool { $schema = Schema::connection($this->connection->getName()); try { foreach (self::REQUIRED_QUERY_INDEXES as $datasetKey => $requiredIndexes) { $table = $this->dataset($datasetKey)->targetTable(); if (! $schema->hasTable($table)) { return false; } $indexes = collect($schema->getIndexes($table)) ->map(static fn (array $index): array => array_map( 'strtolower', (array) ($index['columns'] ?? []), )); foreach ($requiredIndexes as $required) { if (! $indexes->contains(array_map('strtolower', $required))) { return false; } } } } catch (Throwable) { return false; } return true; } /** @return array|null> */ public function states(): array { return [ self::WORK_ORDER_DATASET => $this->mirror->state($this->dataset(self::WORK_ORDER_DATASET)), self::PAYMENT_MAP_DATASET => $this->mirror->state($this->dataset(self::PAYMENT_MAP_DATASET)), ]; } /** * @param array|null>|null $states * @return array{code:string,last_succeeded_at:?string} */ public function syncStatus(?array $states = null, ?CarbonInterface $at = null): array { $states ??= $this->states(); if (count(array_filter($states, 'is_array')) !== 2) { return ['code' => 'unavailable', 'last_succeeded_at' => null]; } $initialized = collect($states)->every( static fn (?array $state): bool => ($state['initial_completed_at'] ?? null) !== null ); $statuses = collect($states)->pluck('status')->map('strval'); $lastSucceededAt = collect($states) ->pluck('last_succeeded_at') ->filter() ->sort() ->first(); if (! $initialized) { $code = $statuses->contains('failed') ? 'initial_failed' : 'initializing'; } elseif ($statuses->contains('failed')) { $code = 'failed'; } elseif ($statuses->contains('running')) { $code = 'syncing'; } else { $lastSucceeded = $lastSucceededAt ? Carbon::parse($lastSucceededAt) : null; $lagMinutes = $lastSucceeded === null ? null : (int) ceil(max(0, ($at ?? now())->getTimestamp() - $lastSucceeded->getTimestamp()) / 60); $code = $lagMinutes === null || $lagMinutes > 30 ? 'stale' : 'healthy'; } return ['code' => $code, 'last_succeeded_at' => $lastSucceededAt]; } /** * @param array{from:string,to:string,work_order_no:string,shop_name:string,transaction_no:string,online_order_no:string,status:string,payment_reason:string} $filters * @return array{data:list>,total:int,page_payment_amount:string} */ public function paginate(array $filters, int $page, int $perPage): array { $this->ensureReady(); $page = max($page, 1); $perPage = min(max($perPage, 1), 100); $query = $this->baseQuery(); $this->applyFilters($query, $filters); $total = (clone $query)->count(); $rows = $query ->select($this->selectColumns()) ->orderByDesc('p.work_order_create_time') ->orderByDesc('p.woNo') ->forPage($page, $perPage) ->get() ->map(fn (object $row): array => $this->decorate((array) $row)) ->all(); return [ 'data' => $rows, 'total' => $total, 'page_payment_amount' => $this->sumMoney(array_column($rows, 'payment_amount')), ]; } /** * @param array{from:string,to:string,work_order_no:string,shop_name:string,transaction_no:string,online_order_no:string,status:string,payment_reason:string} $filters * @return iterable> */ public function exportRows(array $filters, int $chunkSize = 100): iterable { $this->ensureReady(); $chunkSize = min(max($chunkSize, 1), 100); $query = $this->baseQuery(); $this->applyFilters($query, $filters); $page = 1; do { $rows = (clone $query) ->select($this->selectColumns()) ->orderByDesc('p.work_order_create_time') ->orderByDesc('p.woNo') ->forPage($page, $chunkSize) ->get(); foreach ($rows as $row) { yield $this->decorate((array) $row); } $page++; } while ($rows->count() === $chunkSize); } private function ensureReady(): void { if (! $this->ready() || ! $this->queryReady()) { throw new RuntimeException('business_mirror_dataset_not_ready'); } } private function baseQuery(): Builder { $workOrders = $this->dataset(self::WORK_ORDER_DATASET)->targetTable(); $paymentMap = $this->dataset(self::PAYMENT_MAP_DATASET)->targetTable(); $latestWorkOrders = $this->connection->table($workOrders . ' AS work_order_versions') ->selectRaw('woNo, MAX(Id) AS Id') ->where('woTypeId', self::SMALL_PAYMENT_WORK_TYPE_ID) ->groupBy('woNo'); return $this->connection->table($paymentMap . ' AS p') ->joinSub($latestWorkOrders, 'work_order_versions', static function ($join): void { $join->on('work_order_versions.woNo', '=', 'p.woNo'); }) ->join($workOrders . ' AS w', 'w.Id', '=', 'work_order_versions.Id') // Match the leading columns of (woTypeId, woNo, Id), rather than // rescanning all orders of this type for every payment-map row. ->whereColumn('w.woNo', 'p.woNo') ->where('p.woTypeId', self::SMALL_PAYMENT_WORK_TYPE_ID) ->where('w.woTypeId', self::SMALL_PAYMENT_WORK_TYPE_ID); } /** @return list */ private function selectColumns(): array { return [ 'w.Id AS source_id', 'p.woNo AS work_order_no', 'p.transaction_no AS alipay_batch_no', 'w.status AS work_order_status', 'p.status AS mapped_status', 'w.title AS work_order_title', 'p.title AS mapped_title', 'w.content AS work_order_content', 'w.grade AS priority', 'w.urging AS urging_count', 'w.createby AS creator', 'w.executors', 'w.completer', 'w.completeTime AS completed_at', 'w.created AS work_order_created_at', 'w.updated AS work_order_updated_at', 'p.shop_name', 'p.payment_reason', 'p.parse_status', 'p.parse_error', 'p.work_order_create_time', 'p.work_order_update_time', 'w.businessFieldList AS business_field_list', ]; } /** @param array $filters */ private function applyFilters(Builder $query, array $filters): void { $query->where('p.work_order_create_time', '>=', $filters['from'] . ' 00:00:00') ->where('p.work_order_create_time', '<', Carbon::parse($filters['to'])->addDay()->format('Y-m-d 00:00:00')); if ($filters['work_order_no'] !== '') { $query->whereRaw("p.woNo LIKE ? ESCAPE '!'", [$this->contains($filters['work_order_no'])]); } if ($filters['shop_name'] !== '') { $query->whereRaw("p.shop_name LIKE ? ESCAPE '!'", [$this->contains($filters['shop_name'])]); } if ($filters['transaction_no'] !== '') { $query->whereRaw("p.transaction_no LIKE ? ESCAPE '!'", [$this->contains($filters['transaction_no'])]); } if ($filters['online_order_no'] !== '') { $query->whereRaw("w.businessFieldList LIKE ? ESCAPE '!'", [$this->contains($filters['online_order_no'])]); } if ($filters['status'] !== '') { $query->where(function (Builder $status) use ($filters): void { $status->where('w.status', $filters['status']) ->orWhere('p.status', $filters['status']) ->orWhereRaw("w.businessFieldList LIKE ? ESCAPE '!'", [$this->contains($filters['status'])]); }); } if ($filters['payment_reason'] !== '') { $query->whereRaw("p.payment_reason LIKE ? ESCAPE '!'", [$this->contains($filters['payment_reason'])]); } } /** @param array $row @return array */ private function decorate(array $row): array { $fields = $this->businessFields($row['business_field_list'] ?? null); $row['work_order_status'] = $this->firstFilled($row['work_order_status'] ?? null, $row['mapped_status'] ?? null); $row['work_order_title'] = $this->firstFilled($row['work_order_title'] ?? null, $row['mapped_title'] ?? null); $row['work_order_type'] = $this->field($fields, ['工单类型']) ?: '小额打款'; $row['payment_status'] = $this->firstFilled( $this->field($fields, ['打款状态', '支付打款状态', '付款状态']), $row['mapped_status'] ?? null, ); $row['payment_amount'] = $this->money($this->field($fields, [ '打款金额', '付款金额', '小额打款金额', 'amountField', ])); $row['online_order_no'] = $this->field($fields, [ '线上订单号', '线上单号', '订单号', 'orderField', ]); $row['buyer_account'] = $this->field($fields, [ '买家账号', '买家帐号', '买家', 'buyerIdField', ]); $row['alipay_account'] = $this->field($fields, [ '支付宝账号', '支付宝帐号', 'alipayAccountField', ]); $row['alipay_name'] = $this->field($fields, [ '支付宝姓名', '支付宝实名', '支付宝账户姓名', 'alipayNameField', ]); $row['product_code'] = $this->field($fields, [ '商品编码', '商品ID', '商品编号', 'iid', ]); $row['sku_code'] = $this->field($fields, [ 'SKU编码', 'SKU', '规格编码', 'skuId', ]); $row['platform'] = $this->field($fields, ['平台', '平台名称', 'platform']); $row['actual_paid_amount'] = $this->money($this->field($fields, [ '实际已付金额', '实付金额', '买家实付金额', 'actualPaidAmountField', ])); $row['payment_time'] = $this->field($fields, [ '支付时间', '打款时间', '付款时间', 'payDateField', ]); unset($row['business_field_list'], $row['mapped_status'], $row['mapped_title']); return $row; } /** @return array */ private function businessFields(mixed $json): array { if (! is_string($json) || trim($json) === '') { return []; } try { $decoded = json_decode($json, true, 64, JSON_THROW_ON_ERROR); } catch (Throwable) { return []; } $fields = []; $this->flattenFields($decoded, $fields); return $fields; } /** @param array $fields */ private function flattenFields(mixed $node, array &$fields): void { if (! is_array($node)) { return; } $label = $this->arrayValue($node, [ 'fieldName', 'field_name', 'label', 'name', 'title', 'fieldId', 'field_id', ]); $value = $this->arrayValue($node, [ 'fieldValue', 'field_value', 'valueLabel', 'value_label', 'value', 'text', ]); if ($label !== '' && $value !== '') { $fields[$this->normalizeLabel($label)] = $value; } foreach ($node as $key => $child) { if (is_string($key) && ! is_array($child) && ! is_object($child)) { $text = trim((string) $child); if ($text !== '') { $fields[$this->normalizeLabel($key)] ??= $text; } } if (is_array($child)) { $this->flattenFields($child, $fields); } } } /** @param array $row @param list $keys */ private function arrayValue(array $row, array $keys): string { foreach ($keys as $key) { if (array_key_exists($key, $row) && is_scalar($row[$key])) { $value = trim((string) $row[$key]); if ($value !== '') { return $value; } } } return ''; } /** @param array $fields @param list $labels */ private function field(array $fields, array $labels): string { foreach ($labels as $label) { $value = $fields[$this->normalizeLabel($label)] ?? ''; if ($value !== '') { return $value; } } return ''; } private function normalizeLabel(string $label): string { return mb_strtolower((string) preg_replace('/[\s::()()_\-]+/u', '', trim($label))); } private function money(string $value): ?string { $normalized = str_replace([',', ',', '¥', '¥', '元', ' '], '', trim($value)); if (preg_match('/^(-?)(\d+)(?:\.(\d{1,6}))?$/', $normalized, $matches) !== 1) { return null; } $fraction = str_pad($matches[3] ?? '', 3, '0'); $cents = ((int) $matches[2] * 100) + (int) substr($fraction, 0, 2); if ((int) $fraction[2] >= 5) { $cents++; } $sign = $matches[1] === '-' && $cents !== 0 ? '-' : ''; return $sign . intdiv($cents, 100) . '.' . str_pad((string) ($cents % 100), 2, '0', STR_PAD_LEFT); } /** @param list $amounts */ private function sumMoney(array $amounts): string { $cents = 0; foreach ($amounts as $amount) { if ($amount === null || ! preg_match('/^(-?)(\d+)\.(\d{2})$/', (string) $amount, $matches)) { continue; } $value = ((int) $matches[2] * 100) + (int) $matches[3]; $cents += $matches[1] === '-' ? -$value : $value; } $sign = $cents < 0 ? '-' : ''; $absolute = abs($cents); return $sign . intdiv($absolute, 100) . '.' . str_pad((string) ($absolute % 100), 2, '0', STR_PAD_LEFT); } private function firstFilled(mixed ...$values): string { foreach ($values as $value) { $text = trim((string) $value); if ($text !== '') { return $text; } } return ''; } private function contains(string $value): string { return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], $value) . '%'; } private function dataset(string $key): BusinessMirrorDataset { return $this->datasets->get($key); } }