connection = $mirror->connection(); } public function available(): bool { $schema = Schema::connection($this->connection->getName()); return $schema->hasTable('business_mirror_table_states') && $schema->hasTable($this->dataset()->targetTable()); } public function ready(): bool { return $this->mirror->ready($this->dataset()); } /** @return array|null */ public function state(): ?array { return $this->mirror->state($this->dataset()); } /** * @param array|null $state * @return array{code:string,lag_minutes:?int} */ public function syncStatus(?array $state = null, ?CarbonInterface $at = null): array { $state ??= $this->state(); if (! is_array($state)) { return ['code' => 'unavailable', 'lag_minutes' => null]; } $lastSucceeded = $this->timestamp($state['last_succeeded_at'] ?? null); $lagMinutes = $lastSucceeded === null ? null : (int) ceil(max(0, ($at ?? now())->getTimestamp() - $lastSucceeded) / 60); $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 = ($state['run_mode'] ?? null) === 'full' ? 'reconciling' : 'syncing'; } elseif ($lagMinutes === null || $lagMinutes > max((int) config('upstream_mirrors.courier_information.stale_after_minutes', 60), 1)) { $code = 'delayed'; } else { $code = 'healthy'; } return ['code' => $code, 'lag_minutes' => $lagMinutes]; } /** * @param array $filters * @return array{data:list,total:int} */ public function paginate(array $filters, int $page, int $perPage): array { if (! $this->ready()) { throw new RuntimeException('business_mirror_dataset_not_ready'); } $page = max($page, 1); $perPage = min(max($perPage, 1), 100); $query = $this->filteredExpressQuery($filters); $total = (clone $query)->count(); $rows = $query->orderBy('Id') ->forPage($page, $perPage) ->get() ->all(); return ['data' => $rows, 'total' => $total]; } /** * 快递资料中的去重快递公司名称，按资料 Id 的首次出现顺序返回。 * * @return list */ public function names(): array { if (! $this->ready()) { throw new RuntimeException('business_mirror_dataset_not_ready'); } return $this->connection ->table($this->dataset()->targetTable()) ->select(['Id', 'expname']) ->whereNotNull('expname') ->orderBy('Id') ->get() ->map(static fn ($row): string => trim((string) ($row->expname ?? ''))) ->filter(static fn (string $name): bool => $name !== '') ->unique() ->values() ->all(); } /** * @param array $filters * @return \Generator */ public function exportRows(array $filters, int $batchSize = 500): \Generator { if (! $this->ready()) { throw new RuntimeException('business_mirror_dataset_not_ready'); } $batchSize = min(max($batchSize, 1), 5000); $lastId = null; while (true) { $query = $this->filteredExpressQuery($filters); if ($lastId !== null) { $query->where('Id', '>', $lastId); } $rows = $query->orderBy('Id')->limit($batchSize)->get()->all(); if ($rows === []) { return; } foreach ($rows as $row) { yield $row; } $lastId = (int) $rows[array_key_last($rows)]->Id; if (count($rows) < $batchSize) { return; } } } /** * 网点标签数据。快递编码为空时返回全部网点；传入编码时只返回对应网点。 * @return array{data:list,total:int} */ public function paginateNetworks(?string $expcode, int $page, int $perPage): array { $dataset = $this->datasets->get(self::NETWORK_DATASET_KEY); if (! $this->mirror->ready($dataset)) { throw new RuntimeException('business_mirror_dataset_not_ready'); } $query = $this->connection->table($dataset->targetTable())->select([ 'id', 'expid', 'expcode', 'netcode', 'netno', 'netname', 'isbill', 'province', 'city', 'area', 'town', 'address', 'linker', 'linkermobile', 'status', 'apitype', 'apipltmname', 'created', 'expnum', ]); if ($expcode !== null && $expcode !== '') { $query->where('expcode', $expcode); } $page = max($page, 1); $perPage = min(max($perPage, 1), 100); $total = (clone $query)->count(); return [ 'data' => $query->orderByDesc('id')->forPage($page, $perPage)->get()->all(), 'total' => $total, ]; } /** * 打款账户标签数据。 * @return array{data:list,total:int} */ public function paginateAccounts(?string $expcode, int $page, int $perPage): array { $dataset = $this->datasets->get(self::ACCOUNT_DATASET_KEY); if (! $this->mirror->ready($dataset)) { throw new RuntimeException('business_mirror_dataset_not_ready'); } $query = $this->connection->table($dataset->targetTable())->select([ 'Id', 'expcode', 'expname', 'accountType', 'account', 'accountName', 'accountkhh', 'createTime', 'invoice', 'erpId', 'erpAccountId', ]); if ($expcode !== null && $expcode !== '') { $query->where('expcode', $expcode); } $page = max($page, 1); $perPage = min(max($perPage, 1), 100); $total = (clone $query)->count(); return [ 'data' => $query->orderByDesc('Id')->forPage($page, $perPage)->get()->all(), 'total' => $total, ]; } /** * 返回指定快递名称对应的电子签章经办人账号。 * * 工作平台将该值保存在 wp_express.dzqzjbr 中，不能用联系人电话 * 代替；两者在部分快递资料中可能不同。镜像未完成首次同步时返回空 * 数组，避免影响协议签署页面的正常打开。 * * @param list $expnames * @return array */ public function signingAccountsByName(array $expnames): array { if (! $this->ready()) { return []; } $expnames = array_values(array_unique(array_filter(array_map( static fn ($name): string => trim((string) $name), $expnames, ), static fn (string $name): bool => $name !== ''))); if ($expnames === []) { return []; } return $this->connection ->table($this->dataset()->targetTable()) ->select(['expname', 'dzqzjbr', 'Id']) ->whereIn('expname', $expnames) ->orderBy('Id') ->get() ->groupBy(static fn ($row): string => (string) ($row->expname ?? '')) ->map(static function ($rows): string { foreach ($rows as $row) { $account = trim((string) ($row->dzqzjbr ?? '')); if ($account !== '') { return $account; } } return ''; }) ->filter(static fn (string $account): bool => $account !== '') ->all(); } /** * 快递资料操作日志。 * @return array{data:list,total:int} */ public function paginateLogs(?string $expcode, int $page, int $perPage): array { $dataset = $this->datasets->get(self::LOG_DATASET_KEY); if (! $this->mirror->ready($dataset)) { throw new RuntimeException('business_mirror_dataset_not_ready'); } $query = $this->connection->table($dataset->targetTable())->select([ 'Id', 'expcode', 'doTime', 'createTime', 'modifyTime', ]); if ($expcode !== null && $expcode !== '') { $query->where('expcode', $expcode); } $page = max($page, 1); $perPage = min(max($perPage, 1), 100); $total = (clone $query)->count(); return [ 'data' => $query->orderByDesc('Id')->forPage($page, $perPage)->get()->all(), 'total' => $total, ]; } /** @return array{main:bool,networks:bool,accounts:bool,logs:bool,network_ready:bool,account_ready:bool,log_ready:bool} */ public function relatedAvailability(): array { try { $schema = Schema::connection($this->connection->getName()); $networkDataset = $this->datasets->get(self::NETWORK_DATASET_KEY); $accountDataset = $this->datasets->get(self::ACCOUNT_DATASET_KEY); $logDataset = $this->datasets->get(self::LOG_DATASET_KEY); return [ 'main' => $schema->hasTable($this->dataset()->targetTable()), 'networks' => $schema->hasTable($networkDataset->targetTable()), 'accounts' => $schema->hasTable($accountDataset->targetTable()), 'logs' => $schema->hasTable($logDataset->targetTable()), 'network_ready' => $this->mirror->ready($networkDataset), 'account_ready' => $this->mirror->ready($accountDataset), 'log_ready' => $this->mirror->ready($logDataset), ]; } catch (Throwable) { return ['main' => false, 'networks' => false, 'accounts' => false, 'logs' => false, 'network_ready' => false, 'account_ready' => false, 'log_ready' => false]; } } /** @param array $filters */ private function filteredExpressQuery(array $filters): Builder { $query = $this->connection->table($this->dataset()->targetTable())->select([ 'Id', 'expcode', 'expname', 'outcode', 'scexpcode', 'status', 'paymethod', 'addvalidsgn', 'kuaiyunsgn', 'autonosgn', 'packnosgn', 'skucountsplitsgn', 'weight0sgn', 'stopdate', 'tradebtime', 'tradeetime', 'signcom', 'signmode', 'linker', 'linkmobile', 'paytype', 'undefined1', 'undefined2', 'undefined3', 'volume0sgn', 'handnosgn', 'decryptsgn', 'pj1sgn', 'undefinedtype', 'undefined4', 'undefined5', 'dzqzjbr', 'guid', 'is_delete', 'exp_properties', 'modified', ]); if (($filters['expcode'] ?? '') !== '') { $query->where('expcode', $filters['expcode']); } if (($filters['expname'] ?? '') !== '') { $query->whereRaw("expname LIKE ? ESCAPE '!'", [$this->contains($filters['expname'])]); } foreach (['undefined1', 'undefined2', 'undefined3'] as $column) { if (($filters[$column] ?? '') !== '') { $query->whereRaw($column . " LIKE ? ESCAPE '!'", [$this->contains($filters[$column])]); } } if (($filters['tradebtime_from'] ?? '') !== '') { $query->whereDate('tradebtime', '>=', $filters['tradebtime_from']); } if (($filters['tradebtime_to'] ?? '') !== '') { $query->whereDate('tradebtime', '<=', $filters['tradebtime_to']); } if (($filters['status'] ?? '') !== '') { $query->where('status', (int) $filters['status']); } return $query; } private function dataset(): \Modules\EcommerceFinance\Contracts\BusinessMirrorDataset { return $this->datasets->get(self::DATASET_KEY); } private function contains(string $value): string { return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], trim($value)) . '%'; } private function timestamp(mixed $value): ?int { if (! is_string($value) || $value === '') { return null; } try { return Carbon::parse($value)->getTimestamp(); } catch (Throwable) { return null; } } }