diff --git a/.dockerignore b/.dockerignore index 63855c1d..776b9a12 100644 --- a/.dockerignore +++ b/.dockerignore @@ -11,6 +11,7 @@ runtime JetBrainsMono-字体 node_modules vendor +bootstrap/cache/* .caiwu-build public/js public/css/app.css diff --git a/app/Business/TableRegistry.php b/app/Business/TableRegistry.php index 6b7e0a75..2e2d2d84 100644 --- a/app/Business/TableRegistry.php +++ b/app/Business/TableRegistry.php @@ -154,6 +154,7 @@ class TableRegistry 'trademark-management/renewal' => Trademark::class, 'trademark-management/information' => Trademark::class, 'store-management/basic-information' => StoreInformation::class, + 'store-management/store-report' => StoreInformation::class, 'store-management/opening-process' => StoreOpening::class, 'store-management/changes' => StoreChange::class, 'store-management/closing-process' => StoreClosing::class, @@ -241,7 +242,9 @@ class TableRegistry return in_array("{$section}/{$page}", [ 'company-management/basic-information', 'store-management/basic-information', + 'store-management/store-report', 'trademark-management/information', + 'financial-operations/installation-details', ], true); } diff --git a/app/Business/UpstreamBusinessInformation.php b/app/Business/UpstreamBusinessInformation.php index 121c7ec9..5a3e61da 100644 --- a/app/Business/UpstreamBusinessInformation.php +++ b/app/Business/UpstreamBusinessInformation.php @@ -7,6 +7,9 @@ use InvalidArgumentException; use Modules\EcommerceFinance\Repositories\BusinessInformationMirrorRepository; use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory; use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery; +use Modules\EcommerceFinance\Repositories\InstallationDetailMirrorRepository; +use Modules\EcommerceFinance\Repositories\StoreReportRepository; +use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException; class UpstreamBusinessInformation { @@ -15,6 +18,9 @@ class UpstreamBusinessInformation public function __construct( private readonly UpstreamQueryCacheFactory $caches, private readonly BusinessInformationMirrorRepository $mirror, + private readonly InstallationDetailMirrorRepository $installationMirror, + // 兼容仍按旧构造方式调用的代码;容器解析时会正常注入仓储。 + private readonly ?StoreReportRepository $storeReports = null, ) {} public function paginate(string $section, string $page, int $perPage = 50): LengthAwarePaginator @@ -26,7 +32,9 @@ class UpstreamBusinessInformation $payload = match ($key) { 'company-management/basic-information' => $this->companyPage($currentPage, $perPage), 'store-management/basic-information' => $this->storePage($currentPage, $perPage), + 'store-management/store-report' => $this->storeReportPage($currentPage, $perPage), 'trademark-management/information' => $this->trademarkPage($currentPage, $perPage), + 'financial-operations/installation-details' => $this->installationPage($currentPage, $perPage), default => ['data' => [], 'total' => 0], }; @@ -42,6 +50,15 @@ class UpstreamBusinessInformation ); } + private function installationPage(int $page, int $perPage): array + { + $payload = $this->installationMirror->paginate($page, $perPage, InstallationDetailMirrorRepository::requestFilters()); + if (! $payload['ready']) { + throw new ServiceUnavailableHttpException(null, '安装明细镜像尚未完成首次同步'); + } + return ['data' => $payload['data'], 'total' => $payload['total']]; + } + /** * 将已登记的经营分析副本库查询还原成不经过页面缓存的固定只读查询。 * @@ -122,6 +139,32 @@ SQL); ); } + private function storeReportPage(int $page, int $perPage): array + { + $stores = $this->allStores(); + try { + $repository = $this->storeReports ?? app(StoreReportRepository::class); + $rows = $repository->rows($stores['data']); + } catch (\RuntimeException $exception) { + throw new ServiceUnavailableHttpException(null, $exception->getMessage(), $exception); + } + $offset = ($page - 1) * $perPage; + + return ['data' => array_slice($rows, $offset, $perPage), 'total' => count($rows)]; + } + + /** @return array{data:array>,total:int} */ + private function allStores(): array + { + if (! $this->mirror->ready('store-information.read')) { + throw new ServiceUnavailableHttpException(null, '店铺资料共享镜像尚未完成首次同步。'); + } + + $rows = $this->mirror->allStores(); + $rows = array_map(fn (array $row): array => $this->mapStore($row), $rows); + return ['data' => $rows, 'total' => count($rows)]; + } + private function storeSnapshot(UpstreamReadOnlyQuery $query, int $page, int $perPage): array { $offset = ($page - 1) * $perPage; @@ -134,19 +177,20 @@ SELECT s.erpShopName AS erp_store_name, s.unifiedSocialCreditCode AS unified_social_credit_code, c.businessLicenseName AS company_name, + c.businessLicenseType AS company_type, s.trademarkNumber AS trademark_number, NULL AS trademark_name, s.platform AS platform, s.responsibleDepartment AS profit_center, - NULL AS store_manager, - NULL AS department_manager, + s.shopHead AS store_manager, + s.customerDeptManager AS department_manager, s.shopState AS store_status, s.shopBankAccount AS withdrawal_account, s.shopType AS store_type, s.channelType AS channel_mode FROM rpt.fys_shopInfo AS s OUTER APPLY ( - SELECT TOP 1 company.businessLicenseName + SELECT TOP 1 company.businessLicenseName, company.businessLicenseType FROM rpt.fys_companyInfo AS company WHERE company.unifiedSocialCreditCode = NULLIF(s.unifiedSocialCreditCode, '') ORDER BY company.Id DESC @@ -209,6 +253,11 @@ SQL); return $this->normalize($scope, $this->mirror->paginate($scope, $page, $perPage)); } + if ((app(\App\Utilities\SystemDatabaseConfiguration::class)->current()['mode'] ?? '') === 'bootstrap' + && ! config('upstreams.local_configuration.enabled', false)) { + throw new ServiceUnavailableHttpException(null, '当前仍使用本机临时引导库。请在“系统设置 → 数据库管理 → 系统数据库”连接团队共享数据库,然后重新登录;无需另配上游 SQL Server。'); + } + $cache = $this->caches->forSource(self::SOURCE); return $cache->remember($scope, compact('page', 'perPage'), function () use ($resolver) { @@ -266,7 +315,6 @@ SQL); $row = $this->toUtf8($row); $row['registration_date'] = $this->date($row['registration_date'] ?? null); $row['expiry_date'] = $this->date($row['expiry_date'] ?? null); - $row['trademark_type'] = $this->numericClassifications($row['trademark_type'] ?? null); $row['trademark_status'] = match ((string) ($row['trademark_status'] ?? '')) { '0' => '未启用', '1' => '有效', @@ -276,22 +324,6 @@ SQL); return $row; } - /** - * 国际分类在上游偶尔带有“第”“类”等中文标记,Web 表只展示分类数字。 - * 多个分类保留为逗号分隔的数字,空值继续保持为空。 - */ - private function numericClassifications(mixed $value): ?string - { - if ($value === null || trim((string) $value) === '') { - return null; - } - - preg_match_all('/\d+/', (string) $value, $matches); - $values = array_values(array_unique($matches[0] ?? [])); - - return $values === [] ? null : implode(',', $values); - } - private function toUtf8(array $row): array { foreach ($row as &$value) { diff --git a/app/Console/Commands/ApplicationReady.php b/app/Console/Commands/ApplicationReady.php index eac78802..34ca2d43 100644 --- a/app/Console/Commands/ApplicationReady.php +++ b/app/Console/Commands/ApplicationReady.php @@ -2,6 +2,7 @@ namespace App\Console\Commands; +use App\Utilities\SystemDatabaseConfiguration; use Illuminate\Console\Command; use Illuminate\Support\Facades\DB; use Throwable; @@ -18,7 +19,7 @@ class ApplicationReady extends Command protected $description = '检查应用安装、运行目录和数据库是否已经可用'; - public function handle(): int + public function handle(SystemDatabaseConfiguration $configuration): int { if (! config('app.installed')) { return $this->failed('APP_INSTALLED 未被 Laravel 加载。', self::EXIT_NOT_INSTALLED); @@ -42,7 +43,11 @@ class ApplicationReady extends Command return $this->failed('Laravel 数据库连接失败。', self::EXIT_DATABASE_CONNECTION_FAILED); } - $this->components->info('应用已就绪。'); + if ($configuration->current()['mode'] === 'bootstrap') { + $this->components->info('本地网页和临时引导库已就绪;团队共享数据尚未接入。'); + } else { + $this->components->info('应用运行正常,当前数据库可查询。团队数据接入状态请运行 system-database:verify 检查。'); + } return self::SUCCESS; } diff --git a/app/Console/Commands/BusinessMirrorStatus.php b/app/Console/Commands/BusinessMirrorStatus.php new file mode 100644 index 00000000..298e02d0 --- /dev/null +++ b/app/Console/Commands/BusinessMirrorStatus.php @@ -0,0 +1,40 @@ +argument('dataset'); + $selected = $requested ? [$requested => $datasets->get($requested)] : $datasets->all(); + $rows = []; + foreach ($selected as $key => $dataset) { + $state = $mirror->state($dataset); + $rows[] = [ + 'dataset' => $key, 'enabled' => $datasets->scheduled($key), + 'initial_completed' => ($state['initial_completed_at'] ?? null) !== null, + 'status' => $state['status'] ?? ($datasets->automaticInitialSyncAllowed($key) ? 'queued' : 'not_initialized'), + 'last_attempted_at' => $state['last_attempted_at'] ?? null, + 'last_succeeded_at' => $state['last_succeeded_at'] ?? null, + 'error_code' => $state['last_error_code'] ?? null, + ]; + } + if ($this->option('json')) $this->line(json_encode($rows, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR)); + else $this->table(['数据集', '启用', '首次完成', '状态', '最近尝试', '最近完成', '错误类型'], array_map(fn ($r) => array_values($r), $rows)); + return self::SUCCESS; + } catch (Throwable) { + $this->error('无法读取镜像状态,请确认共享系统数据库和业务镜像库已接入。'); + return self::FAILURE; + } + } +} diff --git a/app/Console/Commands/ConnectTeamSystemDatabase.php b/app/Console/Commands/ConnectTeamSystemDatabase.php new file mode 100644 index 00000000..576183d0 --- /dev/null +++ b/app/Console/Commands/ConnectTeamSystemDatabase.php @@ -0,0 +1,75 @@ +environment(['local', 'development', 'testing'])) { + $this->error('此命令仅用于本地开发环境,不修改生产服务器连接。'); + return self::FAILURE; + } + + $current = $configuration->current(); + $password = $this->option('password-stdin') + ? rtrim((string) fgets(STDIN, 1002), "\r\n") + : (string) ($current['password'] ?? ''); + + try { + $settings = $profile->settings($password); + } catch (Throwable) { + $this->warn('本机尚未保存有效的团队密码。请在系统数据库页面输入一次密码;已有配置未修改,无需另配上游账号。'); + return VerifySystemDatabase::EXIT_SHARED_NOT_CONFIGURED; + } + + try { + // 只探测最新端点,不先连接可能已停用的旧服务器。 + $result = $configuration->probe($settings); + $settings['prefix'] = $result['prefix']; + if ($this->matches($current, $settings)) { + $this->info('本机已连接最新团队系统数据库,业务镜像共用此连接。'); + return self::SUCCESS; + } + } catch (Throwable) { + $this->error('最新团队数据库接入未通过验证,原配置已保留。请检查网络、TLS和共享库访问权限;不要改填上游账号。'); + return self::FAILURE; + } + + try { + $configuration->persist($settings); + } catch (Throwable) { + $this->error('本机配置保存或复核失败。请核对私有配置备份和系统连接状态;不要仅修改运行模式或绕过验证。'); + return self::FAILURE; + } + $this->info('已备份并更新本机系统库、业务镜像库连接。请重新登录;未迁移或同步业务数据。'); + return self::SUCCESS; + } + + private function matches(array $current, array $settings): bool + { + foreach (['host', 'port', 'database', 'username', 'password', 'prefix', 'tls_required', 'tls_verify_server_cert', 'tls_ca'] as $key) { + if (($current[$key] ?? null) != ($settings[$key] ?? null)) return false; + } + $mirror = (array) config('database.connections.business_mirror'); + return ($current['mode'] ?? '') === 'shared' + && config('upstreams.stored_configuration.enabled', false) + && ! config('upstreams.local_configuration.enabled', false) + && data_get($mirror, 'read.host.0') === $settings['host'] + && data_get($mirror, 'write.host.0') === $settings['host'] + && (int) ($mirror['port'] ?? 0) === (int) $settings['port'] + && ($mirror['database'] ?? '') === $settings['mirror_database'] + && ($mirror['username'] ?? '') === $settings['username'] + && ($mirror['password'] ?? '') === $settings['password'] + && empty($mirror['unix_socket']); + } +} diff --git a/app/Console/Commands/ImportStoreReportData.php b/app/Console/Commands/ImportStoreReportData.php new file mode 100644 index 00000000..0ad33885 --- /dev/null +++ b/app/Console/Commands/ImportStoreReportData.php @@ -0,0 +1,44 @@ +option('bills')); + $lifecycles = trim((string) $this->option('lifecycles')); + if ($bills === '' && $lifecycles === '') { + $this->error('请至少指定 --bills 或 --lifecycles。'); + return self::INVALID; + } + + try { + if ($bills !== '') { + $result = $importer->importBills($bills, (int) $this->option('batch')); + $this->info('平台账单共享镜像导入完成:' . number_format($result['row_count']) . ' 行。'); + } + if ($lifecycles !== '') { + $result = $importer->importLifecycles($lifecycles, (int) $this->option('batch')); + $this->info('店铺生命周期共享镜像导入完成:' . number_format($result['row_count']) . ' 行。'); + } + } catch (Throwable $exception) { + report($exception); + $this->error('店铺维度报表数据导入失败,已保留上一版共享数据。'); + return self::FAILURE; + } + + return self::SUCCESS; + } +} diff --git a/app/Console/Commands/InspectUpstream.php b/app/Console/Commands/InspectUpstream.php new file mode 100644 index 00000000..6613f338 --- /dev/null +++ b/app/Console/Commands/InspectUpstream.php @@ -0,0 +1,66 @@ +argument('source'); + $schema = (string) $this->argument('schema'); + $object = (string) $this->argument('object'); + try { + if ($source === '') { + $rows = []; + foreach ($sources->all() as $key => $definition) $rows[] = [$key, $definition['label'], $sources->driver($key), $sources->enabled($key) ? '启用' : '停用']; + $this->table(['内部标识', '数据库名称', '类型', '状态'], $rows); + return self::SUCCESS; + } + $sources->definition($source); + if ($schema === '' || $object === '') { + $objects = DB::table('upstream_schema_columns')->where('source_key', $source) + ->when($schema !== '', fn ($q) => $q->where('schema_name', $schema)) + ->select('schema_name', 'object_name')->distinct()->orderBy('schema_name')->orderBy('object_name')->get(); + $this->table(['Schema', '表 / 视图'], $objects->map(fn ($row) => [$row->schema_name, $row->object_name])->all()); + return self::SUCCESS; + } + $catalog = DB::table('upstream_schema_columns')->where('source_key', $source) + ->where('schema_name', $schema)->where('object_name', $object)->orderBy('ordinal_position') + ->get(['column_name', 'data_type', 'is_nullable']); + if ($catalog->isEmpty()) { + $this->error('尚无该对象的字段目录,请先核实来源并同步数据字典。'); + return self::FAILURE; + } + if (! $this->option('columns')) { + $this->table(['字段', '类型', '允许空值'], $catalog->map(fn ($c) => [$c->column_name, $c->data_type, $c->is_nullable ? '是' : '否'])->all()); + return self::SUCCESS; + } + $columns = array_values(array_unique(array_filter(array_map('trim', explode(',', (string) $this->option('columns')))))); + if ($columns === [] || array_diff($columns, $catalog->pluck('column_name')->all())) { + $this->error('样本字段必须来自该对象的数据字典。'); + return self::FAILURE; + } + $dialect = new UpstreamSqlDialect($sources->driver($source)); + $sql = $dialect->select(implode(', ', array_map($dialect->quote(...), $columns)), + 'FROM ' . $dialect->quote($schema) . '.' . $dialect->quote($object), + min(max((int) $this->option('limit'), 1), 20)); + $rows = $queries->forSource($source)->select($sql); + $this->table($columns, array_map(fn ($row) => array_values($row), $rows)); + $this->info('样本仅用于字段核实,不能作为完整业务数据。'); + return self::SUCCESS; + } catch (Throwable) { + $this->error('只读检查未完成,请检查共享连接、来源启用状态和访问权限。'); + return self::FAILURE; + } + } +} diff --git a/app/Console/Commands/RegisterBusinessMirror.php b/app/Console/Commands/RegisterBusinessMirror.php new file mode 100644 index 00000000..a6c96d17 --- /dev/null +++ b/app/Console/Commands/RegisterBusinessMirror.php @@ -0,0 +1,94 @@ +error('尚未执行镜像数据集登记表迁移。'); + return self::FAILURE; + } + $source = (string) $this->argument('source'); + $schema = (string) $this->argument('schema'); + $object = (string) $this->argument('object'); + $key = $source . '.' . $schema . '.' . $object; + try { + $sources->definition($source); + if (strlen($key) > 191 || isset(config('upstream_mirrors.datasets', [])[$key])) { + throw new \RuntimeException('数据集键过长或由现有代码维护,不能覆盖。'); + } + if ($this->option('disable-sync')) { + if ($this->option('enable-sync')) throw new \RuntimeException('不能同时启用和停用。'); + if (DB::table('business_mirror_datasets')->where('dataset_key', $key)->update(['sync_enabled' => false, 'updated_at' => now()]) !== 1) { + throw new \RuntimeException('未找到该登记数据集。'); + } + $this->info('已停止自动同步,现有镜像保留。'); + return self::SUCCESS; + } + $keys = $this->csv((string) $this->option('keys')); + $columns = $this->csv((string) $this->option('columns')); + $watermark = trim((string) $this->option('watermark')); + if ($keys === [] || $columns === [] || array_diff([...$keys, ...($watermark === '' ? [] : [$watermark])], $columns)) { + throw new \RuntimeException('请明确字段和稳定键;同步字段必须包含稳定键及增量字段。'); + } + $catalog = DB::table('upstream_schema_columns')->where('source_key', $source) + ->where('schema_name', $schema)->where('object_name', $object)->pluck('column_name')->all(); + if (array_diff($columns, $catalog)) { + throw new \RuntimeException('字段不在数据字典中,请先同步该来源字典并核实字段。'); + } + $definition = ['source_key' => $source, 'schema' => $schema, 'object' => $object, + 'target' => 'mirror_' . substr(hash('sha256', $key), 0, 24), + 'keys' => $keys, 'columns' => $columns, 'watermark' => $watermark]; + $datasets->validate($key, $definition); + $enable = (bool) $this->option('enable-sync'); + DB::transaction(function () use ($key, $source, $definition, $enable): void { + $existing = DB::table('business_mirror_datasets')->where('dataset_key', $key)->lockForUpdate()->first(); + if ($existing !== null && json_decode($existing->definition, true) != $definition) { + throw new \RuntimeException('已有数据集定义不同,不能覆盖稳定键、字段或同步游标。请先完成变更评估。'); + } + if ($existing !== null) { + if ($enable) DB::table('business_mirror_datasets')->where('dataset_key', $key)->update([ + 'sync_enabled' => true, 'initial_sync_approved' => true, 'updated_at' => now(), + ]); + return; + } + DB::table('business_mirror_datasets')->insert([ + 'dataset_key' => $key, 'source_key' => $source, 'definition' => json_encode($definition, JSON_THROW_ON_ERROR), + 'sync_enabled' => $enable, 'initial_sync_approved' => $enable, 'created_at' => now(), 'updated_at' => now(), + ]); + }); + $this->info($key . ($enable ? ':已启用,生产调度将自动执行首次同步并持续增量。' : ':已登记,自动同步未启用。')); + return self::SUCCESS; + } catch (\Illuminate\Database\QueryException) { + $this->error('登记失败,请检查系统数据库状态;原数据集保留。'); + } catch (\RuntimeException|\InvalidArgumentException $e) { + $this->error($e->getMessage()); + } catch (Throwable) { + $this->error('登记失败,未启用新同步。请检查系统数据库状态。'); + } + return self::FAILURE; + } + + private function csv(string $value): array + { + return array_values(array_unique(array_filter(array_map('trim', explode(',', $value))))); + } +} diff --git a/app/Console/Commands/SyncBusinessMirrors.php b/app/Console/Commands/SyncBusinessMirrors.php index 46772a7c..7c3bb48c 100644 --- a/app/Console/Commands/SyncBusinessMirrors.php +++ b/app/Console/Commands/SyncBusinessMirrors.php @@ -35,23 +35,27 @@ class SyncBusinessMirrors extends Command $requested = $this->argument('dataset'); $keys = $requested === null ? $this->scheduledKeys($datasets, $mirror) - : [(string) $requested]; + : [['key' => (string) $requested, 'full' => (bool) $this->option('full'), 'catchup' => false]]; $maxDatasets = max((int) ($this->option('max-datasets') - ?? config('upstream_mirrors.table_sync.max_datasets_per_run', 1)), 1); + ?? config('upstream_pipeline.sync.max_datasets', 8)), 1); $processed = 0; $failed = false; + $deadline = microtime(true) + max((int) config('upstream_pipeline.sync.cycle_seconds', 30), 1); - foreach ($keys as $key) { - if ($processed >= $maxDatasets) { + foreach ($keys as $task) { + if ($processed >= $maxDatasets || ($processed > 0 && microtime(true) >= $deadline)) { break; } + $key = $task['key']; try { $result = $synchronizer->sync( $key, - (bool) $this->option('full'), + (bool) $task['full'], $this->option('batch') === null ? null : (int) $this->option('batch'), - $this->option('max-batches') === null ? null : (int) $this->option('max-batches'), + $this->option('max-batches') === null + ? ($task['catchup'] ? max((int) config('upstream_pipeline.sync.catchup_batches', 6), 1) : null) + : (int) $this->option('max-batches'), ); } catch (Throwable $exception) { $failed = true; @@ -105,7 +109,7 @@ class SyncBusinessMirrors extends Command return $failed ? self::FAILURE : self::SUCCESS; } - /** @return list */ + /** @return list */ private function scheduledKeys( BusinessMirrorDatasetRegistry $datasets, GenericBusinessMirrorRepository $mirror, @@ -117,17 +121,20 @@ class SyncBusinessMirrors extends Command ), true); foreach ($datasets->all() as $key => $dataset) { - if (isset($dedicated[$key])) { + if (isset($dedicated[$key]) || ! $datasets->scheduled($key)) { continue; } $state = $mirror->state($dataset); - if (! is_array($state) || ($state['initial_completed_at'] ?? null) === null) { - continue; - } + $initial = ! is_array($state) || ($state['initial_completed_at'] ?? null) === null; + if ($initial && ! $datasets->automaticInitialSyncAllowed($key)) continue; + if (($state['status'] ?? '') === 'failed' && ! empty($state['last_attempted_at']) + && now()->diffInSeconds(\Illuminate\Support\Carbon::parse($state['last_attempted_at'])) < 300) continue; $scheduled[] = [ 'key' => $key, + 'full' => $initial || (bool) $this->option('full'), + 'catchup' => $initial || ($state['status'] ?? '') === 'running', 'last_attempted_at' => (string) ($state['last_attempted_at'] ?? ''), ]; } @@ -136,6 +143,15 @@ class SyncBusinessMirrors extends Command [$left['last_attempted_at'], $left['key']] <=> [$right['last_attempted_at'], $right['key']] ); - return array_column($scheduled, 'key'); + // Alternate unfinished windows and routine refreshes: catch-up progresses + // every cycle without starving small tables or monopolizing a source. + $catchup = array_values(array_filter($scheduled, fn ($task) => $task['catchup'])); + $routine = array_values(array_filter($scheduled, fn ($task) => ! $task['catchup'])); + $ordered = []; + while ($catchup !== [] || $routine !== []) { + if ($routine !== []) $ordered[] = array_shift($routine); + if ($catchup !== []) $ordered[] = array_shift($catchup); + } + return $ordered; } } diff --git a/app/Console/Commands/UpstreamSyncSchemaCatalog.php b/app/Console/Commands/UpstreamSyncSchemaCatalog.php index 25318160..cc78fbc1 100644 --- a/app/Console/Commands/UpstreamSyncSchemaCatalog.php +++ b/app/Console/Commands/UpstreamSyncSchemaCatalog.php @@ -32,7 +32,7 @@ class UpstreamSyncSchemaCatalog extends Command $requested = $this->argument('source'); $sourceKeys = $requested ? [(string) $requested] - : array_keys((array) config('upstreams.sources', [])); + : array_keys($sources->all()); $cache = $cacheManager->store((string) config('upstreams.snapshots.cache_store', 'upstream_shared')); $failed = false; diff --git a/app/Console/Commands/VerifySystemDatabase.php b/app/Console/Commands/VerifySystemDatabase.php index 658965b0..52d8ddb0 100644 --- a/app/Console/Commands/VerifySystemDatabase.php +++ b/app/Console/Commands/VerifySystemDatabase.php @@ -15,6 +15,8 @@ class VerifySystemDatabase extends Command public const EXIT_SYSTEM_TABLE_VALIDATION_FAILED = 12; + public const EXIT_SHARED_NOT_CONFIGURED = 13; + protected $signature = 'system-database:verify'; protected $description = '在新的 Laravel 进程中验证当前共享系统数据库'; @@ -30,12 +32,16 @@ class VerifySystemDatabase extends Command ); } - if ( - $settings['mode'] !== 'shared' - || ! config('upstreams.stored_configuration.enabled', false) - ) { + if ($settings['mode'] === 'bootstrap') { + $this->components->warn('当前仍使用本机临时引导库,尚未接入团队共享系统数据库;本次没有测试远程连接。'); + $this->line('下一步:在“系统设置 → 数据库管理 → 系统数据库”中输入一次负责人提供的系统数据库密码。'); + + return self::EXIT_SHARED_NOT_CONFIGURED; + } + + if ($settings['mode'] !== 'shared' || ! config('upstreams.stored_configuration.enabled', false)) { return $this->failed( - '新配置未被 Laravel 加载。', + '共享配置未完整启用,请通过系统数据库连接页面重新保存;不要仅手工修改运行模式。', self::EXIT_CONFIG_NOT_LOADED ); } diff --git a/app/Http/Controllers/Business/Pages.php b/app/Http/Controllers/Business/Pages.php index 289212a7..c0917408 100644 --- a/app/Http/Controllers/Business/Pages.php +++ b/app/Http/Controllers/Business/Pages.php @@ -21,6 +21,7 @@ use Modules\EcommerceFinance\Repositories\AlibabaBillRepository; use Modules\EcommerceFinance\Repositories\CourierInformationMirrorRepository; use Modules\EcommerceFinance\Repositories\EdgeBandingRepository; use Modules\EcommerceFinance\Repositories\InventoryRepository; +use Modules\EcommerceFinance\Repositories\InstallationDetailMirrorRepository; use Modules\EcommerceFinance\Repositories\JdWalletBillRepository; use Modules\EcommerceFinance\Repositories\NonJingxiRefundMirrorRepository; use Modules\EcommerceFinance\Repositories\PinduoduoSpreadBillRepository; @@ -58,6 +59,7 @@ class Pages extends Controller public function __construct( private readonly EdgeBandingRepository $edgeBanding, private readonly InventoryRepository $inventory, + private readonly InstallationDetailMirrorRepository $installationDetails, private readonly NonJingxiRefundMirrorRepository $nonJingxiRefunds, private readonly ProductCatalogMirrorRepository $productCatalog, private readonly ProductPriceMirrorRepository $productPrices, @@ -161,6 +163,18 @@ class Pages extends Controller )); } + public function installationDetails(string $guid) + { + $section = collect(SidebarPermissions::navigationFor(user()))->firstWhere('slug', 'financial-operations'); + abort_unless($section && collect($section['pages'] ?? [])->contains('slug', 'installation-details'), 403, '当前账号没有该页面权限。'); + $payload = $this->installationDetails->details($guid); + if (! $payload['ready']) { + return response()->json(['message' => '安装明细镜像尚未完成首次同步'], 503); + } + + return response()->json(['data' => $payload['data'], 'has_more' => $payload['has_more'] ?? false]); + } + protected function viewData( array $section, ?array $page = null, diff --git a/app/Http/Controllers/SystemSettings/Databases.php b/app/Http/Controllers/SystemSettings/Databases.php index 0eb9b009..abaa3df5 100644 --- a/app/Http/Controllers/SystemSettings/Databases.php +++ b/app/Http/Controllers/SystemSettings/Databases.php @@ -15,6 +15,7 @@ use Illuminate\Routing\Controller; use Illuminate\Validation\Rule; use Illuminate\Validation\ValidationException; use Modules\EcommerceFinance\Services\UpstreamOdbcConnection; +use Modules\EcommerceFinance\Services\UpstreamDatabaseProbe; use Modules\EcommerceFinance\Services\UpstreamSourceRegistry; use Modules\EcommerceFinance\Services\UpstreamStoredConfiguration; use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException; @@ -210,7 +211,7 @@ class Databases extends Controller $data = $this->validateConnection($request); $connection = DatabaseConnection::create($this->attributes($data, null, $request)); - $this->resetSource($connection->source_key, $storedConfiguration); + $this->resetSource($connection->sourceIdentifier(), $storedConfiguration); flash('数据库配置已新增。')->success(); @@ -233,12 +234,12 @@ class Databases extends Controller { abort_unless($storedConfiguration->editable() && $storedConfiguration->enabled(), 403); $connection = $this->connection($connection); - $oldSource = $connection->source_key; + $oldSource = $connection->sourceIdentifier(); $data = $this->validateConnection($request, $connection); $connection->update($this->attributes($data, $connection, $request)); $this->resetSource($oldSource, $storedConfiguration); - $this->resetSource($connection->source_key, $storedConfiguration); + $this->resetSource($connection->sourceIdentifier(), $storedConfiguration); flash('数据库配置已修改。')->success(); @@ -249,7 +250,7 @@ class Databases extends Controller { abort_unless($storedConfiguration->editable() && $storedConfiguration->enabled(), 403); $connection = $this->connection($connection); - $source = $connection->source_key; + $source = $connection->sourceIdentifier(); $connection->delete(); $this->resetSource($source, $storedConfiguration); @@ -258,24 +259,27 @@ class Databases extends Controller return redirect()->route('system-settings.databases.index'); } - public function test(int $connection, UpstreamStoredConfiguration $storedConfiguration) + public function test(int $connection, UpstreamStoredConfiguration $storedConfiguration, UpstreamDatabaseProbe $probe) { abort_unless($storedConfiguration->editable() && $storedConfiguration->enabled(), 403); $connection = $this->connection($connection); try { - UpstreamOdbcConnection::probeConfiguration($this->odbcConfiguration($connection), $connection->name); + $probe->probe($connection->driver, $this->odbcConfiguration($connection), $connection->name); } catch (DecryptException $exception) { throw ValidationException::withMessages([ 'connection' => $exception->getMessage(), ]); - } catch (ServiceUnavailableHttpException) { + } catch (ServiceUnavailableHttpException $exception) { throw ValidationException::withMessages([ - 'connection' => '连接失败。请检查地址、端口、数据库、账号、网络白名单和本机 ODBC 驱动。', + 'connection' => $connection->driver !== 'sqlsrv' + ? $exception->getMessage() + : 'SQL Server 连接失败。请检查地址、端口、数据库、账号、网络白名单和 ODBC 驱动。', ]); } - flash($connection->name . '连接成功,仅执行了 SELECT 1 只读测试。')->success(); + $protocol = match ($connection->driver) { 'pgsql' => ' PostgreSQL SSL ', 'mysql' => ' MySQL/MariaDB TLS ', default => '' }; + flash($connection->name . $protocol . '连接成功,仅执行了 SELECT 1 只读测试。')->success(); return redirect()->route('system-settings.databases.index'); } @@ -283,8 +287,17 @@ class Databases extends Controller private function validateConnection(Request $request, ?DatabaseConnection $connection = null): array { $sourceKeys = array_keys($this->sourceLabels()); + $request->merge(['driver' => $request->input('driver', $connection?->driver ?? 'sqlsrv')]); + $postgres = in_array($request->input('driver'), ['pgsql', 'mysql'], true); + + if ($postgres && $request->input('source_key', $connection?->source_key)) { + throw ValidationException::withMessages([ + 'source_key' => '现有业务来源使用 SQL Server,不能更换其数据库类型。请单独添加新数据库。', + ]); + } return $request->validate([ + 'driver' => ['required', Rule::in(['sqlsrv', 'pgsql', 'mysql'])], 'name' => [ 'required', 'string', @@ -300,15 +313,20 @@ class Databases extends Controller ->where(fn ($query) => $query->where('company_id', UpstreamStoredConfiguration::SYSTEM_COMPANY_ID)) ->ignore($connection?->id), ], - 'host' => ['required', 'string', 'max:191', 'not_regex:/[;\x00-\x1F]/'], + 'host' => ['required', 'string', 'max:191', function ($attribute, $value, $fail) { + if (! filter_var($value, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6) + && ! preg_match('/^[a-z0-9_.-]+$/i', $value)) { + $fail('主机地址只填写域名或 IP,端口请单独填写。'); + } + }], 'port' => ['required', 'integer', 'between:1,65535'], 'database_name' => ['required', 'string', 'max:191', 'not_regex:/[;\x00-\x1F]/'], 'username' => ['required', 'string', 'max:191', 'not_regex:/[;\x00-\x1F]/'], 'password' => [$connection ? 'nullable' : 'required', 'string', 'max:1000'], - 'odbc_driver' => ['required', 'string', 'max:128', 'not_regex:/[;{}\x00-\x1F]/'], - 'tds_version' => ['required', 'regex:/^\d+(?:\.\d+)?$/', 'max:16'], + 'odbc_driver' => [$postgres ? 'nullable' : 'required', 'string', 'max:128', 'not_regex:/[;{}\x00-\x1F]/'], + 'tds_version' => [$postgres ? 'nullable' : 'required', 'regex:/^\d+(?:\.\d+)?$/', 'max:16'], 'connect_timeout' => ['required', 'integer', 'between:1,30'], - 'query_timeout' => ['required', 'integer', 'between:1,120'], + 'query_timeout' => ['required', 'integer', $postgres ? 'between:1,15' : 'between:1,120'], 'allowed_scopes' => ['nullable', 'string', 'max:4000'], 'enabled' => ['nullable', 'boolean'], ]); @@ -318,18 +336,19 @@ class Databases extends Controller { $attributes = [ 'company_id' => UpstreamStoredConfiguration::SYSTEM_COMPANY_ID, - 'source_key' => $data['source_key'] ?? null, + 'source_key' => array_key_exists('source_key', $data) ? $data['source_key'] : $connection?->source_key, 'name' => trim($data['name']), - 'driver' => 'sqlsrv', + 'driver' => $data['driver'], 'host' => trim($data['host']), 'port' => (int) $data['port'], 'database_name' => trim($data['database_name']), 'username' => trim($data['username']), - 'odbc_driver' => trim($data['odbc_driver']), - 'tds_version' => trim($data['tds_version']), + 'odbc_driver' => $data['driver'] !== 'sqlsrv' ? null : trim($data['odbc_driver']), + 'tds_version' => $data['driver'] !== 'sqlsrv' ? null : trim($data['tds_version']), 'connect_timeout' => (int) $data['connect_timeout'], 'query_timeout' => (int) $data['query_timeout'], - 'allowed_scopes' => $this->parseScopes((string) ($data['allowed_scopes'] ?? '')), + 'allowed_scopes' => array_key_exists('allowed_scopes', $data) + ? $this->parseScopes((string) $data['allowed_scopes']) : ($connection?->allowed_scopes ?? []), 'charset' => 'UTF-8', 'enabled' => $request->boolean('enabled'), ]; @@ -365,7 +384,7 @@ class Databases extends Controller private function sourceLabels(): array { - return collect((array) config('upstreams.sources', [])) + return collect(app(UpstreamSourceRegistry::class)->all()) ->mapWithKeys(fn (array $source, string $key) => [$key => $source['label'] ?? $key]) ->all(); } @@ -374,8 +393,9 @@ class Databases extends Controller { return [ 'name' => '', + 'driver' => 'sqlsrv', 'source_key' => '', - 'enabled' => false, + 'enabled' => true, 'host' => '', 'port' => 1433, 'database_name' => '', @@ -392,6 +412,7 @@ class Databases extends Controller { return [ 'name' => $connection->name, + 'driver' => $connection->driver, 'source_key' => $connection->source_key, 'enabled' => $connection->enabled, 'host' => $connection->host, @@ -441,6 +462,6 @@ class Databases extends Controller } $storedConfiguration->forget($source); - UpstreamOdbcConnection::reset($source); + \Modules\EcommerceFinance\Services\UpstreamConnection::reset($source); } } diff --git a/app/Listeners/Menu/ShowInAdmin.php b/app/Listeners/Menu/ShowInAdmin.php index 7f78f6a1..38f9c8d9 100644 --- a/app/Listeners/Menu/ShowInAdmin.php +++ b/app/Listeners/Menu/ShowInAdmin.php @@ -42,7 +42,9 @@ class ShowInAdmin continue; } - $pages = $this->sortByTitleLength($section['pages'] ?? []); + $pages = $section['slug'] === 'order-data' + ? array_values($section['pages'] ?? []) + : $this->sortByTitleLength($section['pages'] ?? []); $order = ($sectionIndex + 1) * 10; if (empty($pages)) { diff --git a/app/Models/Business/InstallationDetail.php b/app/Models/Business/InstallationDetail.php new file mode 100644 index 00000000..cc8edd0d --- /dev/null +++ b/app/Models/Business/InstallationDetail.php @@ -0,0 +1,78 @@ + 'integer', + 'item_quantity' => 'integer', + 'actual_install_quantity' => 'integer', + 'detail_requested_quantity' => 'integer', + 'requested_amount' => 'decimal:2', + 'settled_amount' => 'decimal:2', + 'actual_install_fee' => 'decimal:2', + 'payment_amount' => 'decimal:2', + 'detail_requested_amount' => 'decimal:2', + 'detail_settled_amount' => 'decimal:2', + 'detail_received_amount' => 'decimal:2', + ]; + + public static function columnLabels(): array + { + return [ + 'installation_no' => '安装单号', 'delete_status' => '删除状态', 'quote_status' => '报价状态', + 'quote_deadline' => '报价截至时间', 'installation_company' => '安装公司', 'submitted_at' => '提交时间', + 'shipped_at' => '发货时间', 'buyer_nickname' => '买家昵称', 'store_name' => '店铺名称', + 'installation_type' => '类型', 'pre_push_verification' => '推送前验证', 'push_status' => '推送状态', + 'push_result' => '推送结果', 'arranged_status' => '是否安排', 'order_no' => '订单编号', + 'item_quantity' => '商品数量', 'buyer_name' => '买家姓名', 'buyer_phone' => '买家号码', 'notes' => '备注', + 'province' => '省', 'city' => '市', 'district' => '区', 'address' => '地址', + 'requested_amount' => '申请总金额', 'settled_amount' => '结算总金额', 'image_urls' => '图片地址', + 'completed_status' => '是否完工', 'customer_service_review' => '客服审核', + 'customer_service_reviewed_at' => '客服审核时间', 'finance_review' => '财务审核', + 'finance_reviewed_at' => '财务审核时间', 'actual_install_quantity' => '实际安装数量', + 'actual_install_fee' => '实际安装费用', 'actual_install_product_name' => '实际安装商品名称', + 'installer_phone' => '安装师傅电话', 'installation_company_notes' => '安装公司备注', + 'installation_fee_payer' => '安装费承担方', 'payment_channel' => '收款渠道', + 'payment_transaction_no' => '收款交易单号', 'payment_voucher' => '收款凭证', 'payment_amount' => '收款金额', + 'creator' => '创建人', 'related_guid' => '关联guid', 'product_id' => '商品id', 'id_owner' => 'ID负责人', + 'department_owner' => '部门负责人', 'auxiliary_spu' => '辅助码Spu', 'product_store_code' => '商品店铺编码', + 'item_no' => '货号', 'barcode' => '条形码', 'product_name' => '商品名称', 'supplier_name' => '供应商名称', + 'detail_requested_amount' => '申请明细金额', 'detail_requested_quantity' => '申请明细数量', + 'detail_settled_amount' => '结算明细金额', 'detail_received_amount' => '收款明细金额', + 'change_reason' => '变更原因', 'source_created_at' => '创建时间', 'source_modified_at' => '修改时间', + 'product_name_1' => '商品名称(1)', + ]; + } + + public static function fieldTypes(): array + { + return [ + 'item_quantity' => 'number', 'actual_install_quantity' => 'number', 'detail_requested_quantity' => 'number', + 'requested_amount' => 'currency', 'settled_amount' => 'currency', 'actual_install_fee' => 'currency', + 'payment_amount' => 'currency', 'detail_requested_amount' => 'currency', + 'detail_settled_amount' => 'currency', 'detail_received_amount' => 'currency', + ]; + } +} diff --git a/app/Models/Settings/DatabaseConnection.php b/app/Models/Settings/DatabaseConnection.php index de430301..62df065b 100644 --- a/app/Models/Settings/DatabaseConnection.php +++ b/app/Models/Settings/DatabaseConnection.php @@ -65,4 +65,10 @@ class DatabaseConnection extends Model { return $this->password; } + + public function sourceIdentifier(): string + { + // Legacy names stay stable; new connections need no manually assigned binding. + return $this->source_key ?: 'database_' . $this->id; + } } diff --git a/app/Utilities/SystemDatabaseConfiguration.php b/app/Utilities/SystemDatabaseConfiguration.php index 312100f4..e2b437f9 100644 --- a/app/Utilities/SystemDatabaseConfiguration.php +++ b/app/Utilities/SystemDatabaseConfiguration.php @@ -175,9 +175,15 @@ class SystemDatabaseConfiguration : $this->current()['prefix']; $this->assertValidPrefix($prefix); - $updated = $this->replaceEnvironmentValues($current, [ + $values = [ 'DB_CONNECTION' => 'mysql', 'DB_HOST' => (string) $settings['host'], + // Laravel prioritizes split read/write hosts, URL and socket over + // DB_HOST. They must point to the same newly tested connection. + 'DB_HOST_READ' => (string) $settings['host'], + 'DB_HOST_WRITE' => (string) $settings['host'], + 'DATABASE_URL' => '', + 'DB_SOCKET' => '', 'DB_PORT' => (string) $settings['port'], 'DB_DATABASE' => (string) $settings['database'], 'DB_USERNAME' => (string) $settings['username'], @@ -196,7 +202,22 @@ class SystemDatabaseConfiguration // 另外维护完整的 DB__* 连接参数。 'UPSTREAM_STORED_CONFIG_ENABLED' => 'true', 'UPSTREAM_LOCAL_CONFIG_ENABLED' => 'false', - ]); + ]; + if (isset($settings['mirror_database'])) { + // 团队接入统一两个库的端点,避免旧 BUSINESS_MIRROR_DB_* 覆盖新系统连接。 + $values += [ + 'APP_INSTALLED' => 'true', + 'BUSINESS_MIRROR_DB_HOST' => (string) $settings['host'], + 'BUSINESS_MIRROR_DB_HOST_READ' => (string) $settings['host'], + 'BUSINESS_MIRROR_DB_HOST_WRITE' => (string) $settings['host'], + 'BUSINESS_MIRROR_DB_PORT' => (string) $settings['port'], + 'BUSINESS_MIRROR_DB_DATABASE' => (string) $settings['mirror_database'], + 'BUSINESS_MIRROR_DB_USERNAME' => (string) $settings['username'], + 'BUSINESS_MIRROR_DB_PASSWORD' => (string) $settings['password'], + 'BUSINESS_MIRROR_DB_SOCKET' => '', + ]; + } + $updated = $this->replaceEnvironmentValues($current, $values); try { if (file_put_contents($this->environmentPath, $updated, LOCK_EX) === false) { @@ -462,6 +483,7 @@ class SystemDatabaseConfiguration VerifySystemDatabase::EXIT_CONFIG_NOT_LOADED => '新配置未被加载', VerifySystemDatabase::EXIT_DATABASE_CONNECTION_FAILED => 'Laravel 数据库连接失败', VerifySystemDatabase::EXIT_SYSTEM_TABLE_VALIDATION_FAILED => '系统表验证失败', + VerifySystemDatabase::EXIT_SHARED_NOT_CONFIGURED => '新配置仍指向本机临时引导库', default => '应用级验证失败', }; } @@ -504,6 +526,14 @@ class SystemDatabaseConfiguration 'DB_PASSWORD', 'DB_PREFIX', 'DB_SOCKET', + 'BUSINESS_MIRROR_DB_HOST', + 'BUSINESS_MIRROR_DB_HOST_READ', + 'BUSINESS_MIRROR_DB_HOST_WRITE', + 'BUSINESS_MIRROR_DB_PORT', + 'BUSINESS_MIRROR_DB_DATABASE', + 'BUSINESS_MIRROR_DB_USERNAME', + 'BUSINESS_MIRROR_DB_PASSWORD', + 'BUSINESS_MIRROR_DB_SOCKET', 'MYSQL_ATTR_SSL_CIPHER', 'MYSQL_ATTR_SSL_CA', 'MYSQL_ATTR_SSL_VERIFY_SERVER_CERT', @@ -554,10 +584,9 @@ class SystemDatabaseConfiguration $found = false; foreach ($lines as $index => $line) { - if (preg_match('/^' . preg_quote($key, '/') . '=/', $line) === 1) { + if (preg_match('/^\s*(?:export\s+)?' . preg_quote($key, '/') . '\s*=/', $line) === 1) { $lines[$index] = $replacement; $found = true; - break; } } diff --git a/app/Utilities/TeamSystemDatabaseProfile.php b/app/Utilities/TeamSystemDatabaseProfile.php index a4616ccd..1443a3f0 100644 --- a/app/Utilities/TeamSystemDatabaseProfile.php +++ b/app/Utilities/TeamSystemDatabaseProfile.php @@ -48,6 +48,7 @@ class TeamSystemDatabaseProfile 'host' => $this->requiredString($profile, 'host'), 'port' => $this->requiredPort($profile), 'database' => $this->requiredString($profile, 'database'), + 'mirror_database' => 'caiwu_business_mirror', 'username' => $this->requiredString($profile, 'username'), 'password' => $password, 'tls_required' => true, diff --git a/config/business_navigation.php b/config/business_navigation.php index 94b2b3ff..83d9ea59 100644 --- a/config/business_navigation.php +++ b/config/business_navigation.php @@ -11,9 +11,9 @@ return [ 'title' => '主体维度经营报表', 'directory' => [ 'icon' => 'query_stats', - 'status' => '已接入店铺/公司资料', - 'description' => '已接入店铺和公司主体资料;销量、变更负责人等上游尚未提供的字段保留为空。', - 'empty' => '尚未读取到店铺或公司资料。请检查共享镜像或上游同步状态。', + 'status' => '尚未同步', + 'description' => '按店铺唯一 ID 和主体生效、失效日期归属销量;自动汇总前,先固定 Excel 中的报表字段。', + 'empty' => '经营报表尚未同步。完成店铺主体和销量镜像接入后,这里会显示可追溯的主体数据。', 'fields' => [ ['name' => 'city', 'label' => '市'], ['name' => 'entity', 'label' => '主体'], @@ -39,7 +39,7 @@ return [ 'directory' => [ 'icon' => 'business', 'status' => '既有只读上游', - 'description' => '沿用现有公司基础资料只读链路;Excel 中暂未匹配到的城市、认缴时间等字段保留为空。', + 'description' => '优先读取共享业务镜像中的公司资料;未匹配的城市等字段保留为空。', 'empty' => '尚未读取到公司基础资料。请检查共享镜像或上游同步状态。', 'fields' => [ ['name' => 'unified_social_credit_code', 'label' => '统一社会信用代码'], @@ -112,7 +112,7 @@ return [ 'directory' => [ 'icon' => 'verified', 'status' => '既有只读上游', - 'description' => '沿用现有商标资料只读链路;当前已匹配名称、状态、注册时间和有效期,其余 Excel 字段待接入。', + 'description' => '优先读取共享业务镜像中的商标权属、有效期和保管资料。', 'empty' => '尚未读取到商标基础资料。请检查共享镜像或上游同步状态。', 'fields' => [ ['name' => 'trademark_name', 'label' => '商标名称'], @@ -159,9 +159,9 @@ return [ 'title' => '店铺维度报表', 'directory' => [ 'icon' => 'query_stats', - 'status' => '已接入店铺/公司资料', - 'description' => '已接入店铺、公司主体和店铺状态资料;期间销量及 2026 年月度销量待对应销量镜像接入。', - 'empty' => '尚未读取到店铺资料。请检查共享镜像或上游同步状态。', + 'status' => '共享镜像', + 'description' => '店铺资料、主体生效区间和平台账单统一读取团队共享业务镜像;销量按账单收入汇总。', + 'empty' => '店铺维度报表尚未完成共享数据导入。', 'fields' => [ ['name' => 'erp_store_name', 'label' => 'ERP系统店铺名称'], ['name' => 'store_id', 'label' => '店铺ID(唯一值)'], @@ -190,7 +190,7 @@ return [ 'directory' => [ 'icon' => 'storefront', 'status' => '既有只读上游', - 'description' => '沿用现有店铺资料只读链路;Excel 中的店铺 ID、续签时间和提现账户待字段映射后补充。', + 'description' => '优先读取共享业务镜像中的店铺资料、店铺 ID 和续签时间;提现账户仅显示配置状态。', 'empty' => '尚未读取到店铺基础资料。请检查共享镜像或上游同步状态。', 'fields' => [ ['name' => 'platform', 'label' => '平台'], @@ -209,30 +209,6 @@ return [ ], ], ], - [ - 'slug' => 'store-accounts', - 'title' => '店铺账号密码', - 'directory' => [ - 'icon' => 'key', - 'status' => '尚未同步', - 'description' => '按账号系统字段展示店铺账号、绑定信息和维护时间;账号密码等敏感字段仅按安全规则展示。', - 'empty' => '店铺账号数据尚未同步。接入账号系统只读数据后,这里会显示账号记录。', - 'fields' => [ - ['name' => 'store_name', 'label' => '店铺名称'], - ['name' => 'store_operator', 'label' => '店铺运营负责人'], - ['name' => 'operation_department_manager', 'label' => '店铺运营部门负责人'], - ['name' => 'operation_director', 'label' => '店铺运营总监'], - ['name' => 'profit_center', 'label' => '店铺所属利润中心'], - ['name' => 'account_type', 'label' => '账号类型'], - ['name' => 'account', 'label' => '账号'], - ['name' => 'bound_mobile', 'label' => '绑定手机号', 'kind' => 'secret'], - ['name' => 'mobile_code_recipient', 'label' => '手机号验证码接收人'], - ['name' => 'account_group_name', 'label' => '账号分组名称'], - ['name' => 'created_at', 'label' => '店铺创建时间', 'kind' => 'date'], - ['name' => 'updated_at', 'label' => '店铺修改时间', 'kind' => 'date'], - ], - ], - ], [ 'slug' => 'change-progress', 'title' => '店铺变更进度表', diff --git a/config/upstream_mirrors.php b/config/upstream_mirrors.php index 80b376ec..133cd081 100644 --- a/config/upstream_mirrors.php +++ b/config/upstream_mirrors.php @@ -157,6 +157,24 @@ return [ 'jiansun.dbo.alibaba_bill' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'alibaba_bill', 'target' => 'js_alibaba_bill', 'keys' => ['id']], 'jiansun.dbo.jd_spreadBills' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'jd_spreadBills', 'target' => 'js_jd_spread_bills', 'keys' => ['id'], 'watermark' => 'updateTime'], 'jiansun.dbo.jd_costBills' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'jd_costBills', 'target' => 'js_jd_cost_bills', 'keys' => ['id'], 'watermark' => 'creatTime'], + // 安装明细页面使用已整理的只读视图,避免在 HTTP 请求中临时 JOIN + // 安装主表和明细表;视图中的安装字段与 Excel 模板字段保持稳定别名。 + 'jiansun.dbo.fys_Ticketing_install_view' => [ + 'source_key' => 'jiansun', + 'schema' => 'dbo', + 'object' => 'fys_Ticketing_install_view', + 'target' => 'js_fys_ticketing_install_view', + 'keys' => ['Id'], + 'watermark' => null, + 'indexes' => [ + ['SubmitTime', 'Id'], + ], + ], + 'jiansun.dbo.fys_Ticketing_install_item' => [ + 'source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'fys_Ticketing_install_item', + 'target' => 'js_fys_ticketing_install_item', 'keys' => ['Id'], 'watermark' => 'modifyTime', + 'indexes' => [['guid', 'Id']], + ], 'jiansun.dbo.finance_purchaseReturn' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'finance_purchaseReturn', 'target' => 'js_finance_purchase_return', 'keys' => ['id']], 'jiansun.dbo.finance_invoiceDeduction' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'finance_invoiceDeduction', 'target' => 'js_finance_invoice_deduction', 'keys' => ['id'], 'watermark' => 'createTime'], 'jiansun.dbo.fys_supplierPaymentSummary' => ['source_key' => 'jiansun', 'schema' => 'dbo', 'object' => 'fys_supplierPaymentSummary', 'target' => 'js_supplier_payment_summary', 'keys' => ['id'], 'watermark' => 'modifiedTime'], diff --git a/config/upstream_pipeline.php b/config/upstream_pipeline.php new file mode 100644 index 00000000..b1c515bd --- /dev/null +++ b/config/upstream_pipeline.php @@ -0,0 +1,12 @@ + ['enabled' => false, 'connection' => ['connect_timeout' => 5, 'query_timeout' => 15]], + 'sync' => [ + 'max_datasets' => (int) env('BUSINESS_MIRROR_MAX_DATASETS', 8), + 'cycle_seconds' => (int) env('BUSINESS_MIRROR_CYCLE_SECONDS', 30), + 'dataset_seconds' => (int) env('BUSINESS_MIRROR_DATASET_SECONDS', 8), + 'catchup_batches' => (int) env('BUSINESS_MIRROR_CATCHUP_BATCHES', 6), + ], +]; diff --git a/database/migrations/2026_08_31_000000_create_business_installation_details_table.php b/database/migrations/2026_08_31_000000_create_business_installation_details_table.php new file mode 100644 index 00000000..a0e6327d --- /dev/null +++ b/database/migrations/2026_08_31_000000_create_business_installation_details_table.php @@ -0,0 +1,87 @@ +id(); + $table->unsignedInteger('company_id')->index(); + $table->text('installation_no')->nullable(); + $table->text('delete_status')->nullable(); + $table->text('quote_status')->nullable(); + $table->text('quote_deadline')->nullable(); + $table->text('installation_company')->nullable(); + $table->text('submitted_at')->nullable(); + $table->text('shipped_at')->nullable(); + $table->text('buyer_nickname')->nullable(); + $table->text('store_name')->nullable(); + $table->text('installation_type')->nullable(); + $table->text('pre_push_verification')->nullable(); + $table->text('push_status')->nullable(); + $table->text('push_result')->nullable(); + $table->text('arranged_status')->nullable(); + $table->text('order_no')->nullable(); + $table->integer('item_quantity')->nullable(); + $table->text('buyer_name')->nullable(); + $table->text('buyer_phone')->nullable(); + $table->text('notes')->nullable(); + $table->text('province')->nullable(); + $table->text('city')->nullable(); + $table->text('district')->nullable(); + $table->text('address')->nullable(); + $table->decimal('requested_amount', 18, 2)->nullable(); + $table->decimal('settled_amount', 18, 2)->nullable(); + $table->text('image_urls')->nullable(); + $table->text('completed_status')->nullable(); + $table->text('customer_service_review')->nullable(); + $table->text('customer_service_reviewed_at')->nullable(); + $table->text('finance_review')->nullable(); + $table->text('finance_reviewed_at')->nullable(); + $table->integer('actual_install_quantity')->nullable(); + $table->decimal('actual_install_fee', 18, 2)->nullable(); + $table->text('actual_install_product_name')->nullable(); + $table->text('installer_phone')->nullable(); + $table->text('installation_company_notes')->nullable(); + $table->text('installation_fee_payer')->nullable(); + $table->text('payment_channel')->nullable(); + $table->text('payment_transaction_no')->nullable(); + $table->text('payment_voucher')->nullable(); + $table->decimal('payment_amount', 18, 2)->nullable(); + $table->text('creator')->nullable(); + $table->text('related_guid')->nullable(); + $table->text('product_id')->nullable(); + $table->text('id_owner')->nullable(); + $table->text('department_owner')->nullable(); + $table->text('auxiliary_spu')->nullable(); + $table->text('product_store_code')->nullable(); + $table->text('item_no')->nullable(); + $table->text('barcode')->nullable(); + $table->text('product_name')->nullable(); + $table->text('supplier_name')->nullable(); + $table->decimal('detail_requested_amount', 18, 2)->nullable(); + $table->integer('detail_requested_quantity')->nullable(); + $table->decimal('detail_settled_amount', 18, 2)->nullable(); + $table->decimal('detail_received_amount', 18, 2)->nullable(); + $table->text('change_reason')->nullable(); + $table->text('source_created_at')->nullable(); + $table->text('source_modified_at')->nullable(); + $table->text('product_name_1')->nullable(); + $table->text('installation_company_1')->nullable(); + $table->timestamps(); + $table->softDeletes(); + // installation_no is a TEXT field and cannot be included in a full + // composite index under MySQL's 3072-byte key limit; company_id + // already provides the tenant filter index above. + }); + } + + public function down(): void + { + Schema::dropIfExists('business_installation_details'); + } +}; diff --git a/database/migrations/2026_08_31_180000_create_business_mirror_datasets_table.php b/database/migrations/2026_08_31_180000_create_business_mirror_datasets_table.php new file mode 100644 index 00000000..7bef0672 --- /dev/null +++ b/database/migrations/2026_08_31_180000_create_business_mirror_datasets_table.php @@ -0,0 +1,24 @@ +string('dataset_key', 191)->primary(); + $table->string('source_key', 64)->index(); + $table->json('definition'); + $table->boolean('sync_enabled')->default(false); + $table->boolean('initial_sync_approved')->default(false); + $table->timestamps(); + }); + } + + public function down(): void + { + Schema::dropIfExists('business_mirror_datasets'); + } +}; diff --git a/database/migrations/2026_09_01_000000_create_store_report_mirror_tables.php b/database/migrations/2026_09_01_000000_create_store_report_mirror_tables.php new file mode 100644 index 00000000..fe88a291 --- /dev/null +++ b/database/migrations/2026_09_01_000000_create_store_report_mirror_tables.php @@ -0,0 +1,79 @@ +connection()); + + if (! $schema->hasTable('store_report_import_states')) { + $schema->create('store_report_import_states', function (Blueprint $table): void { + $table->string('dataset_key', 100)->primary(); + $table->string('status', 32)->default('pending'); + $table->uuid('active_generation')->nullable(); + $table->uuid('pending_generation')->nullable(); + $table->unsignedBigInteger('row_count')->default(0); + $table->string('source_sha256', 64)->nullable(); + $table->longText('source_files_json')->nullable(); + $table->dateTime('last_attempted_at')->nullable(); + $table->dateTime('last_succeeded_at')->nullable(); + $table->string('last_error_code', 191)->nullable(); + $table->timestamps(); + }); + } + + if (! $schema->hasTable('store_report_bill_rows')) { + $schema->create('store_report_bill_rows', function (Blueprint $table): void { + $table->bigIncrements('id'); + $table->uuid('generation'); + $table->string('platform', 32); + $table->string('platform_label', 64); + $table->string('store_name', 191); + $table->dateTime('occurred_at')->nullable(); + $table->decimal('income', 20, 4)->default(0); + $table->decimal('expense', 20, 4)->default(0); + $table->string('bill_item', 255)->nullable(); + $table->string('record_code', 191); + $table->string('source_file', 255); + $table->string('row_hash', 64); + $table->dateTime('imported_at'); + $table->unique(['generation', 'platform', 'record_code'], 'store_report_bills_generation_record_unique'); + $table->index(['generation', 'store_name', 'occurred_at'], 'store_report_bills_store_time_index'); + }); + } + + if (! $schema->hasTable('store_report_lifecycle_rows')) { + $schema->create('store_report_lifecycle_rows', function (Blueprint $table): void { + $table->bigIncrements('id'); + $table->uuid('generation'); + $table->string('business_key', 64); + $table->string('erp_store_name', 191); + $table->string('company_name', 191)->nullable(); + $table->dateTime('effective_at')->nullable(); + $table->dateTime('expires_at')->nullable(); + $table->string('source_file', 255); + $table->string('row_hash', 64); + $table->dateTime('imported_at'); + $table->unique(['generation', 'business_key'], 'store_report_lifecycle_generation_key_unique'); + $table->index(['generation', 'erp_store_name', 'effective_at'], 'store_report_lifecycle_store_time_index'); + }); + } + } + + public function down(): void + { + $schema = Schema::connection($this->connection()); + $schema->dropIfExists('store_report_lifecycle_rows'); + $schema->dropIfExists('store_report_bill_rows'); + $schema->dropIfExists('store_report_import_states'); + } +}; diff --git a/deploy/Dockerfile b/deploy/Dockerfile index 2024a960..aba499ae 100644 --- a/deploy/Dockerfile +++ b/deploy/Dockerfile @@ -72,6 +72,12 @@ RUN --mount=type=cache,target=/tmp/composer-cache \ FROM php-base AS runtime +# Keep the existing PHP/Composer cache layers when adding PostgreSQL support. +RUN apt-get -o Acquire::Retries=3 update \ + && apt-get -o Acquire::Retries=3 install -y --no-install-recommends libpq-dev \ + && docker-php-ext-install -j2 pdo_pgsql \ + && rm -rf /var/lib/apt/lists/* + ARG VCS_REF=unknown ARG BUILD_DATE=unknown diff --git a/deploy/Dockerfile.local b/deploy/Dockerfile.local index 2f1383b5..eb2307bc 100644 --- a/deploy/Dockerfile.local +++ b/deploy/Dockerfile.local @@ -83,6 +83,11 @@ RUN --mount=type=cache,target=/root/.npm \ npm ci --no-audit --no-fund --legacy-peer-deps # ---------------- 复制源码 ---------------- +RUN apt-get -o Acquire::Retries=3 update \ + && apt-get -o Acquire::Retries=3 install -y --no-install-recommends libpq-dev \ + && docker-php-ext-install -j2 pdo_pgsql \ + && rm -rf /var/lib/apt/lists/* + COPY . /var/www/html COPY deploy/apache-vhost.conf /etc/apache2/sites-available/000-default.conf COPY deploy/php.ini /usr/local/etc/php/conf.d/jiangling.ini diff --git a/deploy/local-init.sh b/deploy/local-init.sh index a01ff624..dc1bc7b3 100644 --- a/deploy/local-init.sh +++ b/deploy/local-init.sh @@ -316,6 +316,23 @@ if [ "$db_host" = "mysql" ]; then fi +# 已保存的团队密码可解锁最新端点;服务器迁移不应要求每台开发机手改 .env。 +# 无有效团队密码的首次引导保持原状,仍通过网页完成一次接入。 +echo "==> 自动对齐本机团队共享连接..." +if docker compose -f "$compose_file" run --rm --no-deps -T --entrypoint sh app -c \ + 'jiangling-entrypoint true && exec runuser -u www-data -- php artisan system-database:connect-team --no-interaction --no-ansi'; then + team_connect_status=0 +else + team_connect_status=$? +fi +if [ "$team_connect_status" -ne 0 ] && [ "$team_connect_status" -ne 13 ]; then + echo "错误:本机未完成自动接入;请先按上方提示核对连接和私有配置备份。" >&2 + exit "$team_connect_status" +fi + +db_host="$(read_env_value DB_HOST)" +app_installed="$(read_env_value APP_INSTALLED | tr '[:upper:]' '[:lower:]')" + docker compose -f "$compose_file" up -d --no-build app # 容器可能已经存在而没有重新执行 entrypoint。显式复用同一权限初始化逻辑, @@ -397,16 +414,30 @@ fi echo "" echo "=====================================================" -echo "本地测试环境已就绪:http://127.0.0.1:18088" +echo "本地网页可访问:http://127.0.0.1:18088" -db_host="$(read_env_value DB_HOST)" -if [ "$db_host" = "mysql" ]; then - echo "临时登录凭据保存在本机私有文件:$credential_file" - echo "请自行打开该文件,不要把内容提交到 Git、聊天或日志。" - echo "" - echo "下一步:登录后打开“系统设置 → 数据库管理 → 系统数据库”," - echo "点击“连接”,只填写负责人提供的系统数据库密码。" +# Docker healthy 只表示当前连接可用;共享接入必须由新 Laravel 进程实测, +# 不能根据 DB_HOST 不是 mysql 就宣称团队数据已连接。 +if docker compose -f "$compose_file" exec -T -u www-data app \ + php artisan system-database:verify --no-ansi; then + shared_status=0 else - echo "已连接团队共享系统数据库,请使用其中属于自己的正式账号登录。" + shared_status=$? fi + +case "$shared_status" in + 0) + echo "团队共享系统数据库验证通过,请使用其中属于自己的正式账号登录。" + ;; + 13) + echo "团队数据接入尚未完成;后台 worker 启动不代表数据已经同步。" + echo "临时登录凭据保存在本机私有文件:$credential_file" + echo "请登录后输入一次系统数据库密码,连接成功后改用团队正式账号登录。" + ;; + *) + echo "团队共享系统数据库未通过验证;不要将 Docker healthy 当作接入成功。" >&2 + echo "请让 Codex 根据上方失败阶段检查本机配置,保留现有配置和未提交代码。" >&2 + exit "$shared_status" + ;; +esac echo "=====================================================" diff --git a/docs/上游业务镜像清单.md b/docs/上游业务镜像清单.md index 11710982..8854bf74 100644 --- a/docs/上游业务镜像清单.md +++ b/docs/上游业务镜像清单.md @@ -20,6 +20,14 @@ ## 经营分析副本库(operations_replica) +店铺维度报表不使用订单件数代替销量。五个平台原始账单和店铺生命周期 +以完整导入代次写入 `caiwu_business_mirror` 的 `store_report_bill_rows`、 +`store_report_lifecycle_rows`,成功后原子切换,失败时继续读取上一代。 +店铺资料仍以 `rpt.fys_shopInfo` 镜像为主;两份导入数据按 ERP 店铺名称 +关联。每个生命周期行使用 `[effective_at, expires_at)`,销量固定为账单 +`income` 汇总;结算时间存在时优先使用结算时间,否则使用平台发生时间。 +所有开发机与生产页面读取同一份共享镜像,不依赖开发者本机文件。 + | 业务范围 | 上游对象 | 镜像目标 | | --- | --- | --- | | 公司资料 | `rpt.fys_companyInfo` | `ops_fys_company_info` | @@ -30,6 +38,7 @@ | 推广费用 | `rpt.ads_itemid_sales_cost_ratio_v2_di` | `ops_item_sales_cost_ratio` | | 退款退货 | `rpt.jst_returnAndRefundInfo`、`rpt.jst_returnAndRefundItem` | `ops_jst_return_refund_*` | | 聚水潭店铺 | `rpt.jst_shopInfo` | `ops_jst_shop_info` | +| 店铺维度报表 | 五个平台原始账单导出、店铺生命周期导出 | `store_report_bill_rows`、`store_report_lifecycle_rows` | 商品价格 `rpt.Bulk_Price_Limit` 继续使用已经上线的类型化同步器,不在通用原表任务中重复同步。 @@ -39,6 +48,7 @@ | --- | --- | | 营销账户 | `dbo.tmall_spread_bills`、`dbo.pdd_spread_bills`、`dbo.alibaba_bill`、`dbo.jd_spreadBills`、`dbo.jd_costBills` | | 供应商应付 | `dbo.finance_purchaseReturn`、`dbo.finance_invoiceDeduction`、`dbo.fys_supplierPaymentSummary`、`dbo.finance_advanceLoans`、`dbo.pm_unOrderDetails`、`dbo.finance_sampleDeduction`、`dbo.fys_express_courier_settlement` | +| 安装明细 | `dbo.fys_Ticketing_install_view`、`dbo.fys_Ticketing_install_item` | | 板材 | `dbo.finance_boardDeduction`、`dbo.pm_plainPlateData`、`dbo.pm_boardPurchasePrice`、`dbo.pm_boardPriceChangeImport`、`dbo.pm_boardApplyDetail` | | 封边条 | `dbo.pm_edgeBandData`、`dbo.pm_edgeBandApplyDetail`、`dbo.pm_edgeBandProfitInfo`、`dbo.finance_edgeBandDeductio`、`dbo.pm_averageUnitPrice` | | 库存与货款对账 | `dbo.pm_boardInventoryDetail_snapshot`、`dbo.fys_supplierGoods_sttlement_purchase`、`dbo.fys_supplierGoods_sttlement_sale`、`dbo.pm_purchaseIn` | diff --git a/docs/开发与部署SOP.md b/docs/开发与部署SOP.md index 146c0a6d..37511d8a 100644 --- a/docs/开发与部署SOP.md +++ b/docs/开发与部署SOP.md @@ -163,6 +163,8 @@ git pull --ff-only ## 8. 多上游财务数据库同步 +新数据库和新数据集优先使用 [通用上游数据接入](通用上游数据接入.md):来源由连接记录自动登记,已批准的声明式数据集由调度自动首次同步和增量同步。下面的固定来源、逐项 `--full` 和旧快照说明适用于既有兼容路径,不应当作为新增普通开发任务的重复审批要求。 + ### 8.1 统一接入边界 本系统按 3 个以上上游数据库设计。生产系统数据库的启动连接只存在于服务器私有 `runtime/.env`;本地新人只填写系统数据库密码,即可自动解锁两套上游的固定主机、端口、库名、账号和密码,不需要复制生产 `APP_KEY`、填写上游环境变量或等待负责人开放范围。 diff --git a/docs/通用上游数据接入.md b/docs/通用上游数据接入.md new file mode 100644 index 00000000..619c63a3 --- /dev/null +++ b/docs/通用上游数据接入.md @@ -0,0 +1,59 @@ +# 通用上游数据接入 + +## 使用边界 + +财务主库仍为团队 MariaDB/MySQL。SQL Server、PostgreSQL、MySQL/MariaDB 上游共用来源发现、字典、有限样本查询和表级镜像入口。添加同类型的第二个数据库不需要修改来源枚举、工厂或刷新器。 + +数据库连接记录未设置历史 `source_key` 时,系统使用稳定标识 `database_<记录ID>`。已有 `jiansun`、`operations_replica`、`js_data_middle_platform` 等映射保留。普通连接表单不再要求手工绑定来源或填写旧快照 scope;编辑旧连接仍保留这些历史字段。 + +原生 PostgreSQL/MySQL 连接统一要求 TLS、只读会话、非持久连接及最长 15 秒查询。SQL Server 继续复用已验证的 ODBC/DBLIB 入口。应用限制不替代上游只读账号;不提供接受用户原始 SQL 的网页接口。 + +## 开发流程(由 Codex 执行) + +1. 新人通过标准初始化启动本地环境,首次输入一次团队系统数据库密码。Docker healthy 只证明当前程序可用,`system-database:verify` 成功才表示共享接入通过。普通页面、计算逻辑和自动测试可先使用明确的隔离 fixture;不得把模拟数据当作真实财务数据。 +2. 在数据库管理页面添加并启用连接;“测试”与实际读取共用同一驱动。生产及现有字典刷新任务会发现新来源,不再要求逐库添加专用刷新器。 +3. `php artisan upstream:inspect` 列出来源标识,不输出凭据。`upstream:inspect ` 列出已有字典对象;`upstream:inspect ` 查看字段。 +4. 缺少目录时运行 `upstream:schema:sync `。此命令只读上游元数据、写入系统库目录,失败保留原目录。 +5. 需要样本时明确指定 `--columns=id,updated_at,label --limit=5`,最多返回20行。只选本任务有权读取的非敏感字段;真实样本不得进入 Git、PR 或公开日志。样本不能证明全量完整性。 +6. 确认业务键、需要的字段、增量依据和业务含义后,登记数据集。以下仅为示例,表名必须来自实际目录: + +```sh +php artisan business-mirror:register database_17 public example_orders \ + --keys=id --columns=id,updated_at,label --watermark=updated_at +``` + +登记默认不启动同步。生产首次同步涉及系统库登记和业务镜像表创建/写入,应先说明精确字段、批次、影响及备份,再取得负责人授权。授权后对同一声明追加 `--enable-sync`,生产调度即可自动完成或续传首次全量,完成后转为增量;不再为每次部署、分页、筛选重新审批或生成快照。`--disable-sync` 停止后续自动执行,保留已有镜像。 + +声明式登记不能覆盖代码维护的数据集,也不能悄悄修改已有声明的稳定键、字段和增量口径;此类变化需评估历史数据/游标兼容性。来源表没有可靠更新时间时只按主键追加,不能承诺自动发现历史修改或删除,需另行声明校准策略。 + +## 页面读取与同步状态 + +### 已有公司、店铺、商标页在新人电脑没有数据 + +`deploy/local-init.sh` 会先使用本机已经保存的团队密码,按当前代码的加密团队资料验证最新端点;验证成功后自动备份并对齐系统库和业务镜像库的私有配置。这样服务器迁移后不必每台开发机手动修改主机、端口或上游开关。当前连接已正确时不会重复改写配置;验证失败保留原配置。此自动修复只在本地开发运行,不修改生产端点,不执行迁移或业务同步。 + +已有本地容器可由 Codex 直接执行 `docker compose -f docker-compose.local.yml exec -T -u www-data app php artisan system-database:connect-team`。从未保存团队密码的机器返回“首次接入”状态,仍只需在网页输入一次系统数据库密码;自动化接入可用 `--password-stdin` 安全传入,不提供明文密码命令行参数。不能把普通测试管理员的网页登录密码当成系统数据库密码。 + +这些页面优先读取 `caiwu_business_mirror`。本机 Docker 正常、临时引导库有用户,并不代表已经连接团队共享库。Codex 应先在最新主线执行 `deploy/local-init.sh`,通过系统数据库入口完成一次共享接入和重新登录,再运行 `system-database:verify` 并检查相应 `business-mirror:status`。镜像连接默认复用系统数据库主机、端口、账号和 TLS,不要求新人另填 SQL Server 密码,也不要把 `furniture3d_prod` 填成 `operations_replica`。 + +已有镜像可读时,上游开关关闭或网络暂不可达不应阻断页面。读取失败不能解释为真实记录数为 0。只有探索尚未接入的新表时,才需要使用对应上游的只读连接;不要为了让已有资料页出数据,在每个开发者的引导库重复登记生产来源。 + +新页面在原有公司、角色和数据权限校验之后调用 `BusinessMirrorDataReader::paginate($datasetKey, $filters, $page, $perPage)`。它只读业务镜像,返回完整总数、稳定分页、实际同步状态和最近完成时间。来源暂时关闭或不可达时,已完成镜像仍可读取;无首次镜像时明确报“尚未完成首次同步”,不登记逐页快照、不从页面直接访问上游。 + +通用筛选只接受已声明字段的等值条件;复杂业务联表、金额口径、公司行级过滤继续由固定 Repository 实现。不能把此内部读取器当作绕过业务权限的公开接口。 + +`php artisan business-mirror:status [dataset] --json` 只读输出启用、首次完成、最近尝试、最近完成及错误类型。`running` 表示一个同步窗口尚未完成,不保证此刻正有进程运行。最后尝试时间不是最后数据完成时间。 + +通用调度默认每轮最多8个数据集、30秒周期预算;未完成窗口与常规数据集交替处理,单数据集默认8秒软预算,积压任务默认最多6批。预算在批次之间检查,单条查询仍受驱动超时限制;不提高上游并发。失败任务至少冷却5分钟。专用商品/价格/钉钉调度保持既有路径。 + +只对明确批准的登记数据集自动首次同步;既有代码数据集仍保留原首次授权流程。未完成的旧页面切换不等于已经上线,必须逐页验收并与正在开发的 PR 协调。 + +## 类型及验收 + +- PostgreSQL 带时区时间戳在镜像中保存为 UTC;业务展示按自己的明确时区换算。无时区日期时间保留原值。`bytea` 使用十六进制文本表示;不自动解释 JSON、数组、枚举中的业务语义。 +- 验收包含:没有修改来源枚举也能发现新库;TLS/只读保护;字段目录;只复制明确字段;首次同步、第二页、末页、总数和筛选;增量插入与更新;源不可达时读旧镜像;失败和未授权状态。 +- SQL Server 既有 Repository 与专用同步器需回归。不得把“新连接测试成功”当作全项目页面已迁移或所有业务表已同步。 + +## 系统连接保存后回滚的排查 + +截图中的“新配置未通过应用级验证”表示已尝试保存并复核,不等于未输入密码。保存系统连接必须同时统一 `DB_HOST`、读写地址,清除旧 `DATABASE_URL` 和 socket 覆盖,并处理重复赋值;验证失败时恢复完整旧配置。不要只改 `SYSTEM_DATABASE_MODE` 标志,也不要关闭失败回滚来制造“连接成功”。 diff --git a/modules/EcommerceFinance/Repositories/BusinessInformationMirrorRepository.php b/modules/EcommerceFinance/Repositories/BusinessInformationMirrorRepository.php index 26bcbbda..78a840be 100644 --- a/modules/EcommerceFinance/Repositories/BusinessInformationMirrorRepository.php +++ b/modules/EcommerceFinance/Repositories/BusinessInformationMirrorRepository.php @@ -48,6 +48,20 @@ class BusinessInformationMirrorRepository }; } + /** @return array> */ + public function allStores(): array + { + if (! $this->ready('store-information.read')) { + throw new RuntimeException('business_mirror_dataset_not_ready'); + } + + return $this->storeQuery() + ->orderByDesc('s.Id') + ->get() + ->map(static fn (object $row): array => (array) $row) + ->all(); + } + /** @return array{data: array>, total: int} */ private function companies(int $page, int $perPage): array { @@ -63,6 +77,7 @@ legalRepresentative AS legal_representative, registeredAddress AS registered_address, businessScope AS business_scope, accountSerialNumber AS basic_deposit_account_number, +registeredCapitalVaildTime AS subscribed_at, companyStatus AS company_status, financialManager AS financial_officer, supervisor AS supervisor, @@ -75,6 +90,11 @@ SQL); /** @return array{data: array>, total: int} */ private function stores(int $page, int $perPage): array + { + return $this->page($this->storeQuery(), $page, $perPage, 's.Id'); + } + + private function storeQuery() { $latestCompanies = $this->connection->table('ops_fys_company_info') ->selectRaw('unifiedSocialCreditCode, MAX(Id) AS latest_id') @@ -89,21 +109,26 @@ SQL); ->leftJoin('ops_fys_company_info AS c', 'c.Id', '=', 'company_keys.latest_id') ->selectRaw(<<<'SQL' s.Id AS id, +s.sotreId AS store_id, s.openTime AS opened_at, +s.renewalTime AS renewed_at, s.erpShopName AS erp_store_name, s.unifiedSocialCreditCode AS unified_social_credit_code, c.businessLicenseName AS company_name, +c.businessLicenseType AS company_type, s.trademarkNumber AS trademark_number, NULL AS trademark_name, s.platform AS platform, s.responsibleDepartment AS profit_center, s.shopHead AS store_manager, +s.customerDeptManager AS department_manager, s.shopState AS store_status, +s.shopBankAccount AS withdrawal_account, s.shopType AS store_type, s.channelType AS channel_mode SQL); - return $this->page($query, $page, $perPage, 's.Id'); + return $query; } /** @return array{data: array>, total: int} */ @@ -113,11 +138,16 @@ SQL); ->where('isDelete', 0) ->selectRaw(<<<'SQL' Id AS id, +trademarkNumber AS trademark_number, trademarkName AS trademark_name, internationalClassification AS trademark_type, trademarkStatus AS trademark_status, registrationDate AS registration_date, termOfValidity AS expiry_date, +approvedCommodities AS approved_goods, +transferenceUser AS final_holder, +registrationAddress AS registered_address, +custodyDepartment AS custody_department, remark AS notes SQL); diff --git a/modules/EcommerceFinance/Repositories/InstallationDetailMirrorRepository.php b/modules/EcommerceFinance/Repositories/InstallationDetailMirrorRepository.php new file mode 100644 index 00000000..81929719 --- /dev/null +++ b/modules/EcommerceFinance/Repositories/InstallationDetailMirrorRepository.php @@ -0,0 +1,232 @@ +>,total:int,ready:bool} */ + public function paginate(int $page, int $perPage, array $filters = []): array + { + $connection = $this->mirror->connection(); + $dataset = $this->datasets->get(self::DATASET); + $items = $this->datasets->get(self::ITEM_DATASET); + if (! $this->ready($dataset, $items, $connection->getName())) { + return ['data' => [], 'total' => 0, 'ready' => false]; + } + + $perPage = min(max($perPage, 1), 100); + // 主视图按商品行展开,同一安装单会出现多行;主表只保留每个 + // SourceGuid 的最新 Id,商品行通过 details() 单独展开。 + $latest = $connection->table($dataset->targetTable()) + ->selectRaw('MAX(Id) AS id') + ->groupBy('SourceGuid'); + $query = $connection->table($dataset->targetTable() . ' AS v') + ->joinSub($latest, 'latest', static fn ($join) => $join->on('latest.id', '=', 'v.Id')) + ->select('v.*'); + $this->applyFilters($query, $filters, $items->targetTable()); + $total = (clone $query)->count(); + $rows = $query->orderByDesc('v.Id')->forPage(max($page, 1), $perPage)->get(); + + return ['data' => $rows->map(fn (object $row): array => $this->map((array) $row))->all(), 'total' => $total, 'ready' => true]; + } + + /** @return array{data:list>,ready:bool} */ + public function details(string $guid): array + { + $dataset = $this->datasets->get(self::DATASET); + $items = $this->datasets->get(self::ITEM_DATASET); + $connection = $this->mirror->connection(); + if (! $this->ready($dataset, $items, $connection->getName())) { + return ['data' => [], 'ready' => false]; + } + + $mainExists = $connection->table($dataset->targetTable()) + ->where('SourceGuid', $guid) + ->exists(); + if (! $mainExists) { + return ['data' => [], 'ready' => true]; + } + + $transactionNo = $connection->table($dataset->targetTable()) + ->where('SourceGuid', $guid) + ->orderByDesc('Id')->value('OrderNum'); + $rows = $connection->table($items->targetTable()) + ->where('guid', $guid) + ->orderBy('Id') + ->limit(101) + ->get(); + + return ['data' => $rows->take(100)->map(fn (object $row): array => $this->mapItem((array) $row, $transactionNo))->all(), 'ready' => true, 'has_more' => $rows->count() > 100]; + } + + private function ready($dataset, $items, string $connection): bool + { + return $this->mirror->ready($dataset) && $this->mirror->ready($items) + && Schema::connection($connection)->hasTable($dataset->targetTable()) + && Schema::connection($connection)->hasTable($items->targetTable()); + } + + /** @return array{rows:int,requested:float,settled:float,paid:float} */ + public function summary(array $filters = []): array + { + $dataset = $this->datasets->get(self::DATASET); + $connection = $this->mirror->connection(); + if (! $this->mirror->ready($dataset) || ! Schema::connection($connection->getName())->hasTable($dataset->targetTable())) { + return ['rows' => 0, 'requested' => 0.0, 'settled' => 0.0, 'paid' => 0.0]; + } + $latest = $connection->table($dataset->targetTable()) + ->selectRaw('MAX(Id) AS id') + ->groupBy('SourceGuid'); + $query = $connection->table($dataset->targetTable() . ' AS v') + ->joinSub($latest, 'latest', static fn ($join) => $join->on('latest.id', '=', 'v.Id')); + $this->applyFilters($query, $filters, $this->datasets->get(self::ITEM_DATASET)->targetTable()); + $row = $query + ->selectRaw('COUNT(*) AS row_count, COALESCE(SUM(v.installcost),0) AS requested, COALESCE(SUM(v.settleAmount),0) AS settled, COALESCE(SUM(v.acceptAmount),0) AS paid') + ->first(); + return ['rows' => (int) $row->row_count, 'requested' => (float) $row->requested, 'settled' => (float) $row->settled, 'paid' => (float) $row->paid]; + } + + public static function requestFilters(): array + { + $rules = []; + foreach (['created', 'customer_review', 'finance_review'] as $prefix) { + $rules[$prefix . '_from'] = ['nullable', 'date_format:Y-m-d']; + $rules[$prefix . '_to'] = ['nullable', 'date_format:Y-m-d']; + if (request()->filled($prefix . '_from')) $rules[$prefix . '_to'][] = 'after_or_equal:' . $prefix . '_from'; + } + foreach (['push_status','order_no','customer_review','finance_review','quote_status','arranged_status','fee_payer','completed_status','shop_name','delete_status','installation_company','installation_no','order_type','product_id'] as $key) { + $rules[$key] = ['nullable', 'string', 'max:191']; + } + foreach (['customer_review', 'finance_review', 'arranged_status', 'completed_status', 'delete_status'] as $key) $rules[$key][] = 'in:0,1'; + return validator(request()->only(array_keys($rules)), $rules)->validate(); + } + + private function applyFilters($query, array $filters, string $itemTable): void + { + foreach (['created'=>'SubmitTime', 'customer_review'=>'kfshTime', 'finance_review'=>'cwshTime'] as $prefix=>$column) { + if (!empty($filters[$prefix.'_from'])) $query->where('v.'.$column, '>=', $filters[$prefix.'_from']); + if (!empty($filters[$prefix.'_to'])) $query->where('v.'.$column, '<', \Illuminate\Support\Carbon::parse($filters[$prefix.'_to'])->addDay()->toDateString()); + } + foreach (['push_status'=>'apistatus','order_no'=>'OrderNum','customer_review'=>'kfsh','finance_review'=>'cwsh','quote_status'=>'BidStatus','fee_payer'=>'costtaker','completed_status'=>'installComplete','shop_name'=>'shop_name','delete_status'=>'Delete','installation_company'=>'installCompanyName','installation_no'=>'ordernum_id','order_type'=>'TypeDto'] as $key=>$column) { + if (isset($filters[$key]) && $filters[$key] !== '') $query->where('v.'.$column, $filters[$key]); + } + if (isset($filters['arranged_status']) && $filters['arranged_status'] !== '') { + $query->where('v.StatusId', $filters['arranged_status'] === '1' ? '>' : '<=', 0); + } + if (!empty($filters['product_id'])) { + $query->whereExists(function ($sub) use ($itemTable, $filters) { + $sub->selectRaw('1')->from($itemTable.' AS filter_item') + ->whereColumn('filter_item.guid', 'v.SourceGuid')->where('filter_item.shopproid', $filters['product_id']); + }); + } + } + + /** @param array $row */ + private function map(array $row): array + { + $value = static function (array $row, string ...$names): mixed { + foreach ($names as $name) { + foreach ($row as $key => $value) if (strcasecmp((string) $key, $name) === 0) return $value; + } + return null; + }; + + $mapped = [ + 'id' => $value($row, 'detail_id', 'Id'), 'installation_no' => $value($row, 'ordernum_id', 'OrderNum'), + 'delete_status' => $value($row, 'Delete'), 'quote_status' => $value($row, 'BidStatus'), 'quote_deadline' => $value($row, 'BidEndTime'), + 'installation_company' => $value($row, 'installCompanyName'), 'submitted_at' => $value($row, 'SubmitTime'), + 'shipped_at' => $value($row, 'ShipTIme'), 'buyer_nickname' => $value($row, 'BuyersNick'), + 'store_name' => $value($row, 'shop_name', 'shopname'), 'installation_type' => $value($row, 'TypeDto'), + 'pre_push_verification' => $value($row, 'checkMsg'), 'push_status' => $value($row, 'apistatus'), + 'push_result' => $value($row, 'apiresult'), 'arranged_status' => $this->arranged($value($row, 'StatusId')), + 'order_no' => $value($row, 'OrderNum'), + 'item_quantity' => $value($row, 'Num'), 'buyer_name' => $value($row, 'BuyersName'), 'buyer_phone' => $value($row, 'Mobile'), + 'notes' => $value($row, 'Nots'), 'province' => $value($row, 'Province'), 'city' => $value($row, 'City'), + 'district' => $value($row, 'Area'), 'address' => $value($row, 'Address'), 'requested_amount' => $value($row, 'installcost'), + 'settled_amount' => $value($row, 'settleAmount'), 'image_urls' => $value($row, 'Picture'), + 'completed_status' => $this->yesNo($value($row, 'installComplete')), + 'customer_service_review' => $this->review($value($row, 'kfsh')), 'customer_service_reviewed_at' => $value($row, 'kfshTime'), + 'finance_review' => $this->review($value($row, 'cwsh')), 'finance_reviewed_at' => $value($row, 'cwshTime'), + 'actual_install_quantity' => $value($row, 'installNum'), 'actual_install_fee' => $value($row, 'install_Cost'), + 'actual_install_product_name' => $value($row, 'installGoodsName'), 'installer_phone' => $value($row, 'installTel'), + 'installation_company_notes' => $value($row, 'installNote'), 'installation_fee_payer' => $value($row, 'costtaker'), + 'payment_channel' => $value($row, 'costWay'), 'payment_transaction_no' => $value($row, 'actporderNum'), + 'payment_voucher' => $value($row, 'costWayVoucher'), 'payment_amount' => $value($row, 'acceptAmount'), + 'creator' => $value($row, 'Operation'), 'related_guid' => $value($row, 'detail_guid', 'SourceGuid'), + 'product_id' => $value($row, 'shopproid'), 'auxiliary_spu' => $value($row, 'Spu'), + 'product_store_code' => $value($row, 'shopskucode'), 'item_no' => $value($row, 'proCode'), + 'barcode' => $value($row, 'barCode'), 'product_name' => $value($row, 'skuName'), 'supplier_name' => $value($row, 'scmname'), + 'detail_requested_amount' => $value($row, 'applyItemAmount'), 'detail_requested_quantity' => $value($row, 'installcount'), + 'detail_settled_amount' => $value($row, 'settleItemAmount'), 'detail_received_amount' => $value($row, 'acceptItemAmount'), + 'change_reason' => $value($row, 'remark'), 'source_created_at' => $value($row, 'item_created_at'), + 'source_modified_at' => $value($row, 'item_modified_at'), + ]; + // 页面按模板的完整 61 列渲染;视图暂未提供的字段必须显式为 null, + // 不能让 Blade 读取未定义属性。 + return array_replace(array_fill_keys(array_keys(\App\Models\Business\InstallationDetail::columnLabels()), null), $mapped); + } + + /** @param array $row */ + private function mapItem(array $row, mixed $transactionNo = null): array + { + $value = static function (array $row, string ...$names): mixed { + foreach ($names as $name) { + foreach ($row as $key => $value) { + if (strcasecmp((string) $key, $name) === 0) { + return $value; + } + } + } + return null; + }; + + return [ + 'transaction_no' => $value($row, 'Sourceno', 'source_no', 'orderNum', 'ordernum') ?? $transactionNo, + 'product_id' => $value($row, 'shopproid'), + 'id_owner' => $value($row, 'id_owner', 'idOwner'), + 'department_owner' => $value($row, 'department_owner', 'departmentOwner'), + 'auxiliary_spu' => $value($row, 'Spu'), + 'product_store_code' => $value($row, 'shopskucode'), + 'item_no' => $value($row, 'proCode'), + 'barcode' => $value($row, 'barCode'), + 'product_name' => $value($row, 'skuName'), + 'supplier_name' => $value($row, 'scmname'), + 'detail_requested_amount' => $value($row, 'applyItemAmount'), + 'detail_requested_quantity' => $value($row, 'installcount'), + 'detail_settled_amount' => $value($row, 'settleItemAmount'), + 'detail_received_amount' => $value($row, 'acceptItemAmount'), + 'change_reason' => $value($row, 'remark'), + 'source_created_at' => $value($row, 'createTime'), + 'source_modified_at' => $value($row, 'modifyTime'), + ]; + } + + private function yesNo(mixed $value): ?string + { + return $value === null ? null : ((int) $value === 1 ? '是' : '否'); + } + + private function review(mixed $value): ?string + { + return $value === null ? null : ((int) $value === 1 ? '已审核' : '未审核'); + } + + private function arranged(mixed $value): ?string + { + return $value === null ? null : ((int) $value > 0 ? '已安排' : '未安排'); + } +} diff --git a/modules/EcommerceFinance/Repositories/StoreReportRepository.php b/modules/EcommerceFinance/Repositories/StoreReportRepository.php new file mode 100644 index 00000000..9a3494cb --- /dev/null +++ b/modules/EcommerceFinance/Repositories/StoreReportRepository.php @@ -0,0 +1,222 @@ +connection = $databases->connection( + (string) config('upstream_mirrors.connection', 'business_mirror'), + ); + } + + /** + * @param array> $stores + * @return array> + */ + public function rows(array $stores, ?int $reportYear = null): array + { + if ($stores === []) { + return []; + } + + [$billGeneration, $lifecycleGeneration] = $this->activeGenerations(); + $reportYear ??= (int) now()->format('Y'); + if ($reportYear < 2000 || $reportYear > 2999) { + throw new RuntimeException('店铺维度报表年份无效。'); + } + + $names = array_values(array_unique(array_filter(array_map( + static fn (array $row): string => trim((string) ($row['erp_store_name'] ?? '')), + $stores, + )))); + $lifecycles = $this->lifecycles($names, $lifecycleGeneration); + $lifecycleSales = $this->lifecycleSales($names, $billGeneration, $lifecycleGeneration); + $fallbackSales = $this->fallbackSales(array_values(array_diff($names, array_keys($lifecycles))), $billGeneration); + $rows = []; + + foreach ($stores as $store) { + $storeName = trim((string) ($store['erp_store_name'] ?? '')); + if ($storeName === '') { + continue; + } + + $periods = $lifecycles[$storeName] ?? [[ + 'company_name' => $store['company_name'] ?? null, + 'effective_at' => $store['opened_at'] ?? sprintf('%d-01-01', $reportYear), + 'expires_at' => self::DEFAULT_EXPIRES_AT, + 'lifecycle_id' => null, + ]]; + + foreach ($periods as $period) { + $saleData = $period['lifecycle_id'] === null + ? ($fallbackSales[$storeName] ?? ['period' => 0.0, 'months' => []]) + : ($lifecycleSales[(string) $period['lifecycle_id']] ?? ['period' => 0.0, 'months' => []]); + $monthly = []; + for ($month = 1; $month <= 12; $month++) { + $monthly[sprintf('sales_%02d', $month)] = (float) ( + $saleData['months'][sprintf('%d-%02d', $reportYear, $month)] ?? 0 + ); + } + + $rows[] = [ + ...$store, + 'store_id' => $store['store_id'] ?? $storeName, + 'status' => $this->status($store['store_status'] ?? null), + 'entity' => $period['company_name'] ?: ($store['company_name'] ?? null), + 'effective_at' => $this->date($period['effective_at'] ?? null, sprintf('%d-01-01', $reportYear)), + 'invalid_at' => $this->date($period['expires_at'] ?? null, self::DEFAULT_EXPIRES_AT), + 'period_sales' => (float) $saleData['period'], + ...$monthly, + ]; + } + } + + return $rows; + } + + /** @return array{0:string,1:string} */ + private function activeGenerations(): array + { + $schema = Schema::connection($this->connection->getName()); + foreach (['store_report_import_states', 'store_report_bill_rows', 'store_report_lifecycle_rows'] as $table) { + if (! $schema->hasTable($table)) { + throw new RuntimeException('店铺维度报表共享镜像尚未完成迁移。'); + } + } + + $states = $this->connection->table('store_report_import_states') + ->whereIn('dataset_key', [self::BILLS_DATASET, self::LIFECYCLES_DATASET]) + ->get(['dataset_key', 'status', 'active_generation']) + ->keyBy('dataset_key'); + + foreach ([self::BILLS_DATASET, self::LIFECYCLES_DATASET] as $dataset) { + $state = $states->get($dataset); + if ($state === null || empty($state->active_generation)) { + throw new RuntimeException('店铺维度报表共享数据尚未完成首次导入。'); + } + } + + return [ + (string) $states->get(self::BILLS_DATASET)->active_generation, + (string) $states->get(self::LIFECYCLES_DATASET)->active_generation, + ]; + } + + /** @return array>> */ + private function lifecycles(array $names, string $generation): array + { + $result = []; + foreach (array_chunk($names, 500) as $chunk) { + $rows = $this->connection->table('store_report_lifecycle_rows') + ->where('generation', $generation) + ->whereIn('erp_store_name', $chunk) + ->orderBy('erp_store_name')->orderBy('effective_at')->orderBy('id') + ->get(['id', 'erp_store_name', 'company_name', 'effective_at', 'expires_at']); + + foreach ($rows as $row) { + $result[(string) $row->erp_store_name][] = [ + 'lifecycle_id' => (int) $row->id, + 'company_name' => $row->company_name, + 'effective_at' => $row->effective_at, + 'expires_at' => $row->expires_at, + ]; + } + } + + return $result; + } + + /** @return array}> */ + private function lifecycleSales(array $names, string $billGeneration, string $lifecycleGeneration): array + { + $result = []; + $month = $this->monthExpression('b.occurred_at'); + foreach (array_chunk($names, 500) as $chunk) { + $placeholders = implode(',', array_fill(0, count($chunk), '?')); + $rows = $this->connection->select( + "SELECT l.id AS lifecycle_id, {$month} AS bill_month, + SUM(COALESCE(b.income, 0)) AS income_total + FROM store_report_lifecycle_rows AS l + INNER JOIN store_report_bill_rows AS b + ON b.generation = ? AND b.store_name = l.erp_store_name AND b.occurred_at IS NOT NULL + AND b.occurred_at >= COALESCE(l.effective_at, '1000-01-01 00:00:00') + AND b.occurred_at < COALESCE(l.expires_at, '2999-01-01 00:00:00') + WHERE l.generation = ? AND l.erp_store_name IN ($placeholders) + GROUP BY l.id, {$month}", + [$billGeneration, $lifecycleGeneration, ...$chunk], + ); + $this->collectSales($result, $rows, 'lifecycle_id'); + } + + return $result; + } + + /** @return array}> */ + private function fallbackSales(array $names, string $generation): array + { + $result = []; + $month = $this->monthExpression('occurred_at'); + foreach (array_chunk($names, 500) as $chunk) { + $placeholders = implode(',', array_fill(0, count($chunk), '?')); + $rows = $this->connection->select( + "SELECT store_name, {$month} AS bill_month, + SUM(COALESCE(income, 0)) AS income_total + FROM store_report_bill_rows + WHERE generation = ? AND store_name IN ($placeholders) AND occurred_at IS NOT NULL + GROUP BY store_name, {$month}", + [$generation, ...$chunk], + ); + $this->collectSales($result, $rows, 'store_name'); + } + + return $result; + } + + /** @param array}> $result */ + private function collectSales(array &$result, array $rows, string $keyColumn): void + { + foreach ($rows as $row) { + $key = (string) $row->{$keyColumn}; + $result[$key]['period'] = ($result[$key]['period'] ?? 0.0) + (float) $row->income_total; + $result[$key]['months'][(string) $row->bill_month] = (float) $row->income_total; + } + } + + private function monthExpression(string $column): string + { + return $this->connection->getDriverName() === 'sqlite' + ? "strftime('%Y-%m', {$column})" + : "DATE_FORMAT({$column}, '%Y-%m')"; + } + + private function date(mixed $value, string $fallback): string + { + $value = trim((string) ($value ?? '')); + return $value === '' ? $fallback : substr($value, 0, 10); + } + + private function status(mixed $value): ?string + { + return match ((string) $value) { + '0', '正常' => '开店中', + '3', '关闭中' => '关店中', + '6', '已关闭' => '关店', + default => $value === null || $value === '' ? null : (string) $value, + }; + } +} diff --git a/modules/EcommerceFinance/Services/BusinessMirrorDataReader.php b/modules/EcommerceFinance/Services/BusinessMirrorDataReader.php new file mode 100644 index 00000000..a4c4f5bc --- /dev/null +++ b/modules/EcommerceFinance/Services/BusinessMirrorDataReader.php @@ -0,0 +1,46 @@ +datasets->get($key); + if (! $this->mirror->ready($dataset)) { + throw new ServiceUnavailableHttpException(30, '该数据集尚未完成首次同步,请查看同步状态;页面不会临时访问上游。'); + } + $columns = array_column($this->schemas->columns($dataset), 'column_name'); + $query = $this->mirror->connection()->table($dataset->targetTable()); + foreach ($filters as $column => $value) { + if (! in_array($column, $columns, true) || (! is_scalar($value) && $value !== null)) { + throw new InvalidArgumentException('镜像筛选字段或值无效。'); + } + $query->where($column, $value); + } + $total = (clone $query)->count(); + foreach ($dataset->keyColumns() as $column) $query->orderBy($column); + $page = max($page, 1); + $perPage = min(max($perPage, 1), 100); + $state = $this->mirror->state($dataset); + return [ + 'rows' => $query->select($columns)->offset(($page - 1) * $perPage)->limit($perPage)->get() + ->map(fn ($row) => (array) $row)->all(), + 'total' => $total, 'page' => $page, 'per_page' => $perPage, + 'last_page' => max((int) ceil($total / $perPage), 1), + 'sync_status' => $state['status'] ?? 'unknown', + 'last_succeeded_at' => $state['last_succeeded_at'] ?? null, + ]; + } +} diff --git a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php index 6464d7a8..52029d6d 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php @@ -3,6 +3,8 @@ namespace Modules\EcommerceFinance\Services; use InvalidArgumentException; +use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Schema; use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset; class BusinessMirrorDatasetRegistry @@ -12,7 +14,7 @@ class BusinessMirrorDatasetRegistry { $datasets = []; - foreach ((array) config('upstream_mirrors.datasets', []) as $key => $definition) { + foreach ($this->definitions() as $key => $definition) { $datasets[(string) $key] = $this->make((string) $key, (array) $definition); } @@ -21,7 +23,7 @@ class BusinessMirrorDatasetRegistry public function get(string $key): BusinessMirrorDataset { - $definitions = (array) config('upstream_mirrors.datasets', []); + $definitions = $this->definitions(); $definition = (array) ($definitions[$key] ?? []); if ($definition === []) { @@ -31,6 +33,43 @@ class BusinessMirrorDatasetRegistry return $this->make($key, $definition); } + private function definitions(): array + { + $definitions = (array) config('upstream_mirrors.datasets', []); + if (Schema::hasTable('business_mirror_datasets')) { + foreach (DB::table('business_mirror_datasets')->get() as $row) { + // Existing code-owned datasets cannot be redirected by a new registration. + $definitions[$row->dataset_key] ??= json_decode($row->definition, true, 32, JSON_THROW_ON_ERROR); + } + } + return $definitions; + } + + public function automaticInitialSyncAllowed(string $key): bool + { + return Schema::hasTable('business_mirror_datasets') && DB::table('business_mirror_datasets') + ->where('dataset_key', $key)->where('sync_enabled', true)->where('initial_sync_approved', true)->exists(); + } + + public function scheduled(string $key): bool + { + $definition = $this->definitions()[$key] ?? null; + if (! $definition || ! app(UpstreamSourceRegistry::class)->enabled($definition['source_key'])) return false; + if (array_key_exists($key, (array) config('upstream_mirrors.datasets', []))) return true; + return Schema::hasTable('business_mirror_datasets') && DB::table('business_mirror_datasets') + ->where('dataset_key', $key)->where('sync_enabled', true)->exists(); + } + + public function validate(string $key, array $definition): BusinessMirrorDataset + { + return $this->make($key, $definition); + } + + public function columnSelection(string $key): ?array + { + return $this->definitions()[$key]['columns'] ?? null; + } + private function make(string $key, array $definition): BusinessMirrorDataset { $source = trim((string) ($definition['source_key'] ?? '')); @@ -45,17 +84,23 @@ class BusinessMirrorDatasetRegistry )); $indexColumns = array_merge(...($indexes ?: [[]])); - foreach ([$source, $schema, $object, $target, ...$keys, ...$indexColumns] as $identifier) { + foreach ([$source, $target] as $identifier) { if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $identifier) !== 1) { throw new InvalidArgumentException('业务镜像数据集包含不安全的标识符:' . $key); } } + foreach ([$schema, $object, ...$keys, ...$indexColumns] as $identifier) { + if ($identifier === '' || mb_strlen($identifier) > 128 || str_contains($identifier, "\0")) { + throw new InvalidArgumentException('业务镜像数据集包含无效的标识符:' . $key); + } + } + if ($keys === [] || in_array([], $indexes, true) || strlen($target) > 64) { throw new InvalidArgumentException('业务镜像数据集缺少稳定键或目标表名过长:' . $key); } - if ($watermark !== '' && preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $watermark) !== 1) { + if ($watermark !== '' && (mb_strlen($watermark) > 128 || str_contains($watermark, "\0"))) { throw new InvalidArgumentException('业务镜像数据集的增量字段不安全:' . $key); } diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php b/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php index 547b5401..7f011d71 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorSchemaManager.php @@ -6,6 +6,7 @@ use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Schema; use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset; +use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry; use RuntimeException; class BusinessMirrorSchemaManager @@ -87,6 +88,13 @@ class BusinessMirrorSchemaManager ->map(fn (object $row): array => (array) $row) ->all(); + $selected = app(BusinessMirrorDatasetRegistry::class)->columnSelection($dataset->key()); + if ($selected !== null) { + $selected = array_map('strtolower', $selected); + $rows = array_values(array_filter($rows, fn (array $row): bool => in_array(strtolower($row['column_name']), $selected, true))); + if (count($rows) !== count($selected)) throw new RuntimeException('business_mirror_selected_column_missing:' . $dataset->key()); + } + if ($rows === []) { throw new RuntimeException('business_mirror_schema_catalog_missing:' . $dataset->key()); } @@ -121,24 +129,24 @@ class BusinessMirrorSchemaManager $scale = min(max((int) ($column['numeric_scale'] ?? 0), 0), min($precision, 30)); $definition = match ($type) { - 'bit' => $table->boolean($name), + 'bit', 'boolean' => $table->boolean($name), 'tinyint' => $table->unsignedTinyInteger($name), 'smallint' => $table->smallInteger($name), - 'int' => $table->integer($name), + 'int', 'integer' => $table->integer($name), 'bigint' => $table->bigInteger($name), 'decimal', 'numeric', 'money', 'smallmoney' => $table->decimal($name, $precision, $scale), - 'float', 'real' => $table->double($name), + 'float', 'real', 'double', 'double precision' => $table->double($name), 'date' => $table->date($name), - 'datetime', 'datetime2', 'smalldatetime' => $table->dateTime($name, 6), + 'datetime', 'datetime2', 'smalldatetime', 'timestamp without time zone', 'timestamp with time zone' => $table->dateTime($name, 6), 'time' => $table->time($name, 6), - 'uniqueidentifier' => $table->char($name, 36), + 'uniqueidentifier', 'uuid' => $table->char($name, 36), 'binary', 'varbinary', 'image', 'timestamp', 'rowversion' => $table->binary($name), // MariaDB counts every inline varchar column towards the 65,535-byte // row limit. Wide legacy SQL Server tables can contain dozens of // nvarchar columns even though each individual column looks small. // Keep only genuinely short values inline; TEXT is stored off-page // and lets the mirror preserve the complete source schema. - 'char', 'nchar', 'varchar', 'nvarchar' => $length > 0 && $length <= 64 + 'char', 'nchar', 'varchar', 'nvarchar', 'character', 'character varying' => $length > 0 && $length <= 64 ? $table->string($name, $length) : $table->longText($name), default => $table->longText($name), @@ -157,8 +165,8 @@ class BusinessMirrorSchemaManager $type = strtolower((string) $column['data_type']); $length = (int) ($column['max_length'] ?? 0); - return ! in_array($type, ['text', 'ntext', 'xml', 'image', 'binary', 'varbinary'], true) - && (! in_array($type, ['char', 'nchar', 'varchar', 'nvarchar'], true) || ($length > 0 && $length <= 64)); + return ! in_array($type, ['text', 'ntext', 'xml', 'image', 'binary', 'varbinary', 'bytea', 'json', 'jsonb', 'array', 'user-defined'], true) + && (! in_array($type, ['char', 'nchar', 'varchar', 'nvarchar', 'character', 'character varying'], true) || ($length > 0 && $length <= 64)); } return false; diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php index 7577037c..673340dd 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php @@ -11,6 +11,7 @@ use Throwable; class BusinessMirrorSynchronizer { + private string $driver = 'sqlsrv'; public function __construct( private readonly BusinessMirrorDatasetRegistry $datasets, private readonly BusinessMirrorSchemaManager $schemas, @@ -23,10 +24,11 @@ class BusinessMirrorSynchronizer public function sync(string $datasetKey, bool $full = false, ?int $batchSize = null, ?int $maxBatches = null): array { $dataset = $this->datasets->get($datasetKey); + $this->driver = app(UpstreamSourceRegistry::class)->driver($dataset->sourceKey()); // A long-running scheduler may have retained a timed-out PDO instance // from the previous dataset. Each dataset starts with a fresh upstream // connection so one source failure cannot poison subsequent mirrors. - UpstreamOdbcConnection::reset($dataset->sourceKey()); + UpstreamConnection::reset($dataset->sourceKey()); $state = $this->mirror->state($dataset); if (! $full && ($state['initial_completed_at'] ?? null) === null) { @@ -60,6 +62,7 @@ class BusinessMirrorSynchronizer $batchSize = min(max($batchSize ?? (int) config('upstream_mirrors.table_sync.batch_size', 500), 1), 5000); $maxBatches = max($maxBatches ?? (int) config('upstream_mirrors.table_sync.max_batches_per_run', 2), 0); $batches = 0; + $deadline = microtime(true) + max((int) config('upstream_pipeline.sync.dataset_seconds', 8), 1); $rows = (int) ($state['last_upserted_rows'] ?? 0); try { @@ -92,7 +95,7 @@ class BusinessMirrorSynchronizer return $this->result($dataset, 'succeeded', $mode, $batches, $rows); } - if ($maxBatches > 0 && $batches >= $maxBatches) { + if ($maxBatches > 0 && ($batches >= $maxBatches || microtime(true) >= $deadline)) { return $this->result($dataset, 'running', $mode, $batches, $rows); } } @@ -120,8 +123,11 @@ class BusinessMirrorSynchronizer if ($mode === 'incremental' && $dataset->watermarkColumn() !== null) { $watermark = $this->quote($dataset->watermarkColumn()); $where[] = $watermark . ' >= ? AND ' . $watermark . ' < ?'; - $bindings[] = $this->sqlServerDateTime($windowStart); - $bindings[] = $this->sqlServerDateTime($windowEnd); + $timezoneAware = $this->driver === 'pgsql' && collect($columns)->contains(fn ($c) => + $c['column_name'] === $dataset->watermarkColumn() && $c['data_type'] === 'timestamp with time zone' + ); + $bindings[] = $timezoneAware ? $windowStart?->format('Y-m-d\TH:i:s.uP') : $this->sqlServerDateTime($windowStart); + $bindings[] = $timezoneAware ? $windowEnd?->format('Y-m-d\TH:i:s.uP') : $this->sqlServerDateTime($windowEnd); } elseif ($mode === 'incremental' && $cursor === null && $highWatermark !== []) { [$predicate, $values] = $this->keyset($dataset->keyColumns(), $highWatermark); $where[] = $predicate; @@ -134,11 +140,12 @@ class BusinessMirrorSynchronizer array_push($bindings, ...$values); } - $sql = 'SELECT TOP (' . $limit . ') ' - . implode(', ', array_map(fn (array $column): string => $this->selectExpression($column), $columns)) - . ' FROM ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject()) + $tail = 'FROM ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject()) . ($where === [] ? '' : ' WHERE (' . implode(') AND (', $where) . ')') . ' ORDER BY ' . implode(', ', array_map(fn (string $name): string => $this->quote($name), $order)); + $sql = (new UpstreamSqlDialect($this->driver))->select( + implode(', ', array_map(fn (array $column): string => $this->selectExpression($column), $columns)), $tail, $limit, + ); return $this->queries->forSource($dataset->sourceKey())->select($sql, $bindings); } @@ -149,6 +156,17 @@ class BusinessMirrorSynchronizer $name = (string) $column['column_name']; $quoted = $this->quote($name); + if ($this->driver !== 'sqlsrv') { + if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'timestamp with time zone') { + // Store timezone-aware PostgreSQL instants as UTC in the shared mirror. + return "to_char({$quoted} AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') AS {$quoted}"; + } + if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'bytea') { + return "encode({$quoted}, 'hex') AS {$quoted}"; + } + return $quoted; + } + // FreeTDS may return SQL Server date values using the connection locale // (for example "Jan 1 2026"). Convert temporal columns at the source so // every mirror receives deterministic ISO values that MariaDB accepts. @@ -244,7 +262,7 @@ class BusinessMirrorSynchronizer $type = $types[strtolower((string) $name)] ?? ''; if ($type === 'date') { $row[$name] = substr($value, 0, 10); - } elseif (in_array($type, ['datetime', 'datetime2', 'smalldatetime'], true)) { + } elseif (in_array($type, ['datetime', 'datetime2', 'smalldatetime', 'timestamp without time zone', 'timestamp with time zone'], true)) { $row[$name] = preg_replace('/(\.\d{6})\d+/', '$1', str_replace('T', ' ', $value)); } elseif (in_array($type, ['char', 'nchar', 'varchar', 'nvarchar', 'text', 'ntext', 'xml'], true)) { $row[$name] = $this->normalizeText($value); @@ -271,11 +289,7 @@ class BusinessMirrorSynchronizer private function quote(string $identifier): string { - if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $identifier) !== 1) { - throw new RuntimeException('business_mirror_invalid_identifier'); - } - - return '[' . $identifier . ']'; + return (new UpstreamSqlDialect($this->driver))->quote($identifier); } private function cursorValue(array $cursor, string $column): mixed diff --git a/modules/EcommerceFinance/Services/StoreReportDataImporter.php b/modules/EcommerceFinance/Services/StoreReportDataImporter.php new file mode 100644 index 00000000..a3cd097b --- /dev/null +++ b/modules/EcommerceFinance/Services/StoreReportDataImporter.php @@ -0,0 +1,344 @@ + ['label' => '得物', 'markers' => ['得物']], + 'douyin' => ['label' => '抖音', 'markers' => ['头条放心购', '抖音']], + 'taobao' => ['label' => '淘系', 'markers' => ['淘系']], + 'jd' => ['label' => '京东', 'markers' => ['京东商城', '京东']], + 'pinduoduo' => ['label' => '拼多多', 'markers' => ['拼多多']], + ]; + + private const TIME_HEADERS = [ + 'dewu' => ['结算时间', '支付/退款时间', '发生时间'], + 'douyin' => ['结算时间', '发生时间'], + 'taobao' => ['结算时间', '入账时间', '发生时间'], + 'jd' => ['结算时间', '费用结算时间', '发生时间'], + 'pinduoduo' => ['结算时间', '发生时间'], + ]; + + private readonly ConnectionInterface $connection; + + public function __construct(DatabaseManager $databases) + { + $this->connection = $databases->connection( + (string) config('upstream_mirrors.connection', 'business_mirror'), + ); + } + + /** @return array{dataset_key:string,row_count:int,generation:string} */ + public function importBills(string $path, int $batchSize = 500): array + { + $files = $this->xlsxFiles($path); + $generation = (string) Str::uuid(); + $dataset = StoreReportRepository::BILLS_DATASET; + $this->begin($dataset, $generation); + $batchSize = min(max($batchSize, 100), 2000); + $batch = []; + $count = 0; + + try { + foreach ($files as $file) { + [$platform, $label] = $this->platform($file); + foreach ($this->spreadsheetRows($file) as [$headers, $values]) { + $storeName = $this->text($this->value($headers, $values, ['店铺', '店铺名称'])); + $recordCode = $this->text($this->value($headers, $values, ['记录编码', '账单流水号', '业务单号'])); + if ($storeName === null || $recordCode === null) { + continue; + } + + $occurredAt = $this->dateTime($this->value($headers, $values, self::TIME_HEADERS[$platform])); + $income = $this->money($this->value($headers, $values, ['收入金额', '收入'])); + $expense = $this->money($this->value($headers, $values, ['支出金额', '支出'])); + $billItem = $this->text($this->value($headers, $values, ['账单项目', '摘要'])); + $business = [$platform, $recordCode, $storeName, $occurredAt, $income, $expense, $billItem]; + $batch[] = [ + 'generation' => $generation, + 'platform' => $platform, + 'platform_label' => $label, + 'store_name' => $storeName, + 'occurred_at' => $occurredAt, + 'income' => $income, + 'expense' => $expense, + 'bill_item' => $billItem, + 'record_code' => $recordCode, + 'source_file' => basename($file), + 'row_hash' => hash('sha256', json_encode($business, JSON_UNESCAPED_UNICODE)), + 'imported_at' => now(), + ]; + if (count($batch) >= $batchSize) { + $count += $this->storeBills($batch); + $batch = []; + } + } + } + if ($batch !== []) { + $count += $this->storeBills($batch); + } + $count = (int) $this->connection->table('store_report_bill_rows') + ->where('generation', $generation)->count(); + if ($count === 0) { + throw new RuntimeException('平台账单文件中没有可导入的数据。'); + } + + $this->activate($dataset, $generation, $count, $files); + } catch (Throwable $exception) { + $this->fail($dataset, $generation, 'store_report_bill_rows', $exception); + throw $exception; + } + + return ['dataset_key' => $dataset, 'row_count' => $count, 'generation' => $generation]; + } + + /** @return array{dataset_key:string,row_count:int,generation:string} */ + public function importLifecycles(string $path, int $batchSize = 500): array + { + $files = $this->xlsxFiles($path); + if (count($files) !== 1) { + throw new RuntimeException('店铺生命周期导入只接受一个 XLSX 文件。'); + } + $file = $files[0]; + $generation = (string) Str::uuid(); + $dataset = StoreReportRepository::LIFECYCLES_DATASET; + $this->begin($dataset, $generation); + $batchSize = min(max($batchSize, 100), 2000); + $batch = []; + $count = 0; + + try { + foreach ($this->spreadsheetRows($file) as [$headers, $values]) { + $storeName = $this->text($this->value($headers, $values, ['ERP系统店铺名称'])); + if ($storeName === null) { + continue; + } + $companyName = $this->text($this->value($headers, $values, ['公司名称'])); + $effectiveAt = $this->dateTime($this->value($headers, $values, ['生效时间'])); + $expiresAt = $this->dateTime($this->value($headers, $values, ['失效时间']), true); + $business = [$storeName, $companyName, $effectiveAt, $expiresAt]; + $batch[] = [ + 'generation' => $generation, + 'business_key' => hash('sha256', implode('|', [$storeName, $effectiveAt ?? '', $expiresAt ?? ''])), + 'erp_store_name' => $storeName, + 'company_name' => $companyName, + 'effective_at' => $effectiveAt, + 'expires_at' => $expiresAt, + 'source_file' => basename($file), + 'row_hash' => hash('sha256', json_encode($business, JSON_UNESCAPED_UNICODE)), + 'imported_at' => now(), + ]; + if (count($batch) >= $batchSize) { + $count += $this->storeLifecycles($batch); + $batch = []; + } + } + if ($batch !== []) { + $count += $this->storeLifecycles($batch); + } + $count = (int) $this->connection->table('store_report_lifecycle_rows') + ->where('generation', $generation)->count(); + if ($count === 0) { + throw new RuntimeException('店铺生命周期文件中没有可导入的数据。'); + } + + $this->activate($dataset, $generation, $count, [$file]); + } catch (Throwable $exception) { + $this->fail($dataset, $generation, 'store_report_lifecycle_rows', $exception); + throw $exception; + } + + return ['dataset_key' => $dataset, 'row_count' => $count, 'generation' => $generation]; + } + + /** @return array */ + private function xlsxFiles(string $path): array + { + $real = realpath($path); + if ($real === false) { + throw new RuntimeException('导入路径不存在。'); + } + $files = is_dir($real) ? glob($real . DIRECTORY_SEPARATOR . '*.xlsx') : [$real]; + $files = array_values(array_filter($files ?: [], static fn (string $file): bool => + is_file($file) && is_readable($file) && ! str_starts_with(basename($file), '~$') + )); + sort($files); + if ($files === []) { + throw new RuntimeException('导入路径中没有可读的 XLSX 文件。'); + } + return $files; + } + + /** @return array{0:string,1:string} */ + private function platform(string $file): array + { + foreach (self::PLATFORMS as $key => $definition) { + foreach ($definition['markers'] as $marker) { + if (str_contains(basename($file), $marker)) { + return [$key, $definition['label']]; + } + } + } + throw new RuntimeException('无法从文件名识别账单平台:' . basename($file)); + } + + /** @return iterable,1:array}> */ + private function spreadsheetRows(string $file): iterable + { + $reader = IOFactory::createReaderForFile($file); + $reader->setReadDataOnly(true); + $spreadsheet = $reader->load($file); + try { + foreach ($spreadsheet->getWorksheetIterator() as $worksheet) { + $rows = $worksheet->toArray(null, true, true, false); + $headerRow = array_shift($rows); + if (! is_array($headerRow)) { + continue; + } + $headers = []; + foreach ($headerRow as $index => $header) { + $name = $this->text($header); + if ($name !== null) { + $headers[$name] = $index; + } + } + foreach ($rows as $values) { + if (array_filter($values, static fn (mixed $value): bool => $value !== null && $value !== '') === []) { + continue; + } + yield [$headers, $values]; + } + } + } finally { + $spreadsheet->disconnectWorksheets(); + unset($spreadsheet); + } + } + + private function value(array $headers, array $values, array $candidates): mixed + { + foreach ($candidates as $candidate) { + if (array_key_exists($candidate, $headers)) { + $value = $values[$headers[$candidate]] ?? null; + if ($value !== null && $value !== '') { + return $value; + } + } + } + return null; + } + + private function text(mixed $value): ?string + { + $value = trim((string) ($value ?? '')); + return $value === '' ? null : mb_substr($value, 0, 255); + } + + private function money(mixed $value): float + { + $normalized = str_replace([',', '¥', '¥', ' '], '', (string) ($value ?? '0')); + if (! is_numeric($normalized)) { + throw new RuntimeException('账单金额无法解析。'); + } + return round((float) $normalized, 4); + } + + private function dateTime(mixed $value, bool $allowOpenEnded = false): ?string + { + if ($value === null || $value === '') { + return null; + } + if ($allowOpenEnded && in_array(trim((string) $value), ['存续中', '至今', '长期', '无'], true)) { + return null; + } + if ($value instanceof DateTimeInterface) { + return $value->format('Y-m-d H:i:s'); + } + if (is_numeric($value)) { + return ExcelDate::excelToDateTimeObject((float) $value)->format('Y-m-d H:i:s'); + } + try { + return Carbon::parse(str_replace('/', '-', trim((string) $value)))->format('Y-m-d H:i:s'); + } catch (Throwable) { + throw new RuntimeException('账单或生命周期时间无法解析。'); + } + } + + private function begin(string $dataset, string $generation): void + { + $this->assertSchemaReady(); + $now = now(); + $this->connection->table('store_report_import_states')->insertOrIgnore([ + 'dataset_key' => $dataset, 'status' => 'pending', 'row_count' => 0, + 'created_at' => $now, 'updated_at' => $now, + ]); + $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([ + 'status' => 'running', 'pending_generation' => $generation, + 'last_attempted_at' => $now, 'last_error_code' => null, 'updated_at' => $now, + ]); + } + + private function assertSchemaReady(): void + { + $schema = Schema::connection($this->connection->getName()); + foreach (['store_report_import_states', 'store_report_bill_rows', 'store_report_lifecycle_rows'] as $table) { + if (! $schema->hasTable($table)) { + throw new RuntimeException('店铺维度报表共享镜像尚未完成迁移:' . $table); + } + } + } + + private function storeBills(array $rows): int + { + $this->connection->table('store_report_bill_rows')->upsert( + $rows, ['generation', 'platform', 'record_code'], + ['platform_label', 'store_name', 'occurred_at', 'income', 'expense', 'bill_item', 'source_file', 'row_hash', 'imported_at'], + ); + return count($rows); + } + + private function storeLifecycles(array $rows): int + { + $this->connection->table('store_report_lifecycle_rows')->upsert( + $rows, ['generation', 'business_key'], + ['erp_store_name', 'company_name', 'effective_at', 'expires_at', 'source_file', 'row_hash', 'imported_at'], + ); + return count($rows); + } + + private function activate(string $dataset, string $generation, int $count, array $files): void + { + $hashes = array_map(static fn (string $file): string => hash_file('sha256', $file) ?: '', $files); + $now = now(); + $this->connection->transaction(function () use ($dataset, $generation, $count, $files, $hashes, $now): void { + $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([ + 'status' => 'succeeded', 'active_generation' => $generation, 'pending_generation' => null, + 'row_count' => $count, 'source_sha256' => hash('sha256', implode('|', $hashes)), + 'source_files_json' => json_encode(array_map('basename', $files), JSON_UNESCAPED_UNICODE), + 'last_succeeded_at' => $now, 'last_error_code' => null, 'updated_at' => $now, + ]); + }); + } + + private function fail(string $dataset, string $generation, string $table, Throwable $exception): void + { + $this->connection->table($table)->where('generation', $generation)->delete(); + $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([ + 'status' => 'failed', 'pending_generation' => null, + 'last_error_code' => class_basename($exception), 'updated_at' => now(), + ]); + } +} diff --git a/modules/EcommerceFinance/Services/UpstreamConnection.php b/modules/EcommerceFinance/Services/UpstreamConnection.php new file mode 100644 index 00000000..8ebc7f08 --- /dev/null +++ b/modules/EcommerceFinance/Services/UpstreamConnection.php @@ -0,0 +1,73 @@ + */ + private static array $native = []; + + public static function get(string $source): PDO + { + $registry = app(UpstreamSourceRegistry::class); + if ($registry->driver($source) === 'sqlsrv') { + return UpstreamOdbcConnection::get($source); + } + if (! $registry->enabled($source)) { + throw new ServiceUnavailableHttpException(30, '该数据源尚未启用。'); + } + return self::$native[$source] ??= self::openNative($registry->driver($source), $registry->value($source, 'connection', [])); + } + + public static function reset(?string $source = null): void + { + if ($source === null) self::$native = []; + else unset(self::$native[$source]); + UpstreamOdbcConnection::reset($source); + } + + public static function openNative(string $driver, #[SensitiveParameter] array $config): PDO + { + if (! in_array($driver, ['pgsql', 'mysql'], true) || ! in_array($driver, PDO::getAvailableDrivers(), true)) { + throw new ServiceUnavailableHttpException(30, '应用缺少该数据库类型的驱动,请更新应用镜像。'); + } + $timeout = min(max((int) ($config['connect_timeout'] ?? 5), 1), 30); + $seconds = min(max((int) ($config['query_timeout'] ?? 15), 1), 15); + $options = [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_PERSISTENT => false, PDO::ATTR_TIMEOUT => $timeout]; + $host = (string) $config['host']; + $database = (string) $config['database']; + if (preg_match('/[;\x00-\x1F]/', $host . $database)) { + throw new ServiceUnavailableHttpException(30, '主机或数据库名称包含不支持的字符。'); + } + $port = min(max((int) $config['port'], 1), 65535); + if ($driver === 'pgsql') { + $quote = static fn (string $s): string => "'" . str_replace(['\\', "'"], ['\\\\', "\\'"], $s) . "'"; + $dsn = 'pgsql:host=' . $quote($host) . ';port=' . $port . ';dbname=' . $quote($database) + . ';sslmode=require;connect_timeout=' . $timeout + . ";options='-c default_transaction_read_only=on -c statement_timeout=" . ($seconds * 1000) . "'"; + } else { + $dsn = "mysql:host={$host};port={$port};dbname={$database};charset=utf8mb4"; + $options[PDO::MYSQL_ATTR_SSL_CIPHER] = 'DEFAULT'; + $options[PDO::MYSQL_ATTR_SSL_VERIFY_SERVER_CERT] = false; + } + try { + $pdo = new PDO($dsn, (string) $config['username'], (string) $config['password'], $options); + if ($driver === 'mysql') { + $tls = $pdo->query("SHOW SESSION STATUS LIKE 'Ssl_cipher'")->fetch(PDO::FETCH_NUM); + if (empty($tls[1])) throw new ServiceUnavailableHttpException(30, '上游数据库未建立 TLS 加密连接。'); + $pdo->exec('SET SESSION TRANSACTION READ ONLY'); + $pdo->exec(str_contains($pdo->getAttribute(PDO::ATTR_SERVER_VERSION), 'MariaDB') + ? 'SET SESSION max_statement_time=' . $seconds + : 'SET SESSION MAX_EXECUTION_TIME=' . ($seconds * 1000)); + } + return $pdo; + } catch (PDOException) { + throw new ServiceUnavailableHttpException(30, '上游只读连接失败,请检查账号、权限、网络和 TLS 配置。'); + } + } +} diff --git a/modules/EcommerceFinance/Services/UpstreamDatabaseProbe.php b/modules/EcommerceFinance/Services/UpstreamDatabaseProbe.php new file mode 100644 index 00000000..7cd0adbe --- /dev/null +++ b/modules/EcommerceFinance/Services/UpstreamDatabaseProbe.php @@ -0,0 +1,38 @@ +connect($driver, $connection); + $statement = $pdo->query('SELECT 1'); + if ($statement === false || (int) $statement->fetchColumn() !== 1) { + throw new ServiceUnavailableHttpException(30, '只读测试未返回预期结果。'); + } + $statement->closeCursor(); + } catch (PDOException) { + throw new ServiceUnavailableHttpException(30, '上游只读连接失败,请检查账号、权限、网络和 TLS 配置。'); + } + } + + protected function connect(string $driver, #[SensitiveParameter] array $connection): PDO + { + return UpstreamConnection::openNative($driver, $connection); + } +} diff --git a/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php b/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php index 30825d79..8d2f6ea1 100644 --- a/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php +++ b/modules/EcommerceFinance/Services/UpstreamReadOnlyQuery.php @@ -27,7 +27,7 @@ class UpstreamReadOnlyQuery { self::assertReadOnlySql($sql); - $statement = UpstreamOdbcConnection::get($this->source)->prepare($sql); + $statement = UpstreamConnection::get($this->source)->prepare($sql); $statement->execute($bindings); return $statement->fetchAll(PDO::FETCH_ASSOC); @@ -45,7 +45,7 @@ class UpstreamReadOnlyQuery { self::assertReadOnlySql($sql); - $statement = UpstreamOdbcConnection::get($this->source)->prepare($sql); + $statement = UpstreamConnection::get($this->source)->prepare($sql); $statement->execute($bindings); return $statement->fetchColumn(); @@ -68,6 +68,7 @@ class UpstreamReadOnlyQuery $inspect = preg_replace("/'(?:''|[^'])*'/s", "''", $sql) ?? $sql; $inspect = preg_replace('/\[(?:\]\]|[^\]])*\]/s', '[]', $inspect) ?? $inspect; $inspect = preg_replace('/"(?:""|[^"])*"/s', '""', $inspect) ?? $inspect; + $inspect = preg_replace('/`(?:``|[^`])*`/s', '``', $inspect) ?? $inspect; $inspect = strtoupper(ltrim($inspect)); if (preg_match('/^(SELECT|WITH)\b/', $inspect) !== 1) { @@ -76,7 +77,7 @@ class UpstreamReadOnlyQuery $forbidden = '/\b(?:INSERT|UPDATE|DELETE|MERGE|DROP|ALTER|CREATE|TRUNCATE|EXEC|EXECUTE|' . 'GRANT|REVOKE|DENY|BACKUP|RESTORE|DBCC|BULK|OPENROWSET|OPENDATASOURCE|INTO|' - . 'WAITFOR|KILL|SHUTDOWN)\b|\bNEXT\s+VALUE\s+FOR\b/i'; + . 'WAITFOR|KILL|SHUTDOWN|SET_CONFIG|NEXTVAL|SETVAL|SLEEP|BENCHMARK|GET_LOCK|RELEASE_LOCK|LOAD_FILE|PG_SLEEP)\b|\bNEXT\s+VALUE\s+FOR\b/i'; if (preg_match($forbidden, $inspect) === 1) { throw new InvalidArgumentException('上游查询包含禁止的写入或高风险语法。'); diff --git a/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php b/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php index a70f0a05..62093dc6 100644 --- a/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php +++ b/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php @@ -7,7 +7,7 @@ use Illuminate\Support\Facades\Schema; use RuntimeException; /** - * 将上游 SQL Server 的表、视图和字段结构同步到共享系统数据库。 + * 将已支持类型的上游表、视图和字段结构同步到共享系统数据库。 * * 这里只读取 INFORMATION_SCHEMA 元数据,不读取业务行,也不保存连接资料。 * 获取完整结果成功后才替换单个来源的旧目录,避免网络故障清空现有字典。 @@ -28,25 +28,15 @@ class UpstreamSchemaCatalogSynchronizer throw new RuntimeException('该上游数据库尚未启用。'); } - $rows = $this->queries->forSource($source)->select( - "SELECT - c.TABLE_SCHEMA AS schema_name, - c.TABLE_NAME AS object_name, - CASE WHEN t.TABLE_TYPE = 'VIEW' THEN 'view' ELSE 'table' END AS object_type, - c.COLUMN_NAME AS column_name, - c.ORDINAL_POSITION AS ordinal_position, - c.DATA_TYPE AS data_type, - c.CHARACTER_MAXIMUM_LENGTH AS max_length, - c.NUMERIC_PRECISION AS numeric_precision, - c.NUMERIC_SCALE AS numeric_scale, - c.IS_NULLABLE AS is_nullable - FROM INFORMATION_SCHEMA.COLUMNS c - INNER JOIN INFORMATION_SCHEMA.TABLES t - ON t.TABLE_SCHEMA = c.TABLE_SCHEMA - AND t.TABLE_NAME = c.TABLE_NAME - WHERE c.TABLE_SCHEMA NOT IN ('sys', 'INFORMATION_SCHEMA') - ORDER BY c.TABLE_SCHEMA, c.TABLE_NAME, c.ORDINAL_POSITION" - ); + $driver = $this->sources->driver($source); + $rows = $this->queries->forSource($source)->select((new UpstreamSqlDialect($driver))->catalogSql()); + if ($driver === 'mysql') { + foreach ($rows as &$row) { + // SQL Server timestamp is binary; MySQL timestamp is temporal. + if (strtolower($row['data_type'] ?? '') === 'timestamp') $row['data_type'] = 'datetime'; + } + unset($row); + } if ($rows === []) { throw new RuntimeException('上游数据库没有返回可用的表字段目录。'); diff --git a/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php b/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php index d5287524..3380924f 100644 --- a/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php +++ b/modules/EcommerceFinance/Services/UpstreamSourceRegistry.php @@ -28,7 +28,11 @@ class UpstreamSourceRegistry $definition = config('upstreams.sources.' . $source); if (! is_array($definition)) { - throw new InvalidArgumentException('未登记的上游数据源:' . $source); + $connection = $this->storedConfiguration->find($source); + if ($connection === null) { + throw new InvalidArgumentException('未登记的上游数据源:' . $source); + } + $definition = ['label' => $connection->name, 'config' => 'upstream_pipeline.defaults']; } $configKey = trim((string) ($definition['config'] ?? '')); @@ -77,6 +81,25 @@ class UpstreamSourceRegistry return (bool) $this->value($source, 'enabled', false); } + public function driver(string $source): string + { + $definition = $this->definition($source); + $default = config($definition['config'] . '.connection.driver', 'sqlsrv'); + return (string) (($this->usesLocalConfiguration() + ? $default : $this->storedConfiguration->value($source, 'connection.driver', $default)) ?: 'sqlsrv'); + } + + /** @return array */ + public function all(): array + { + $keys = array_unique(array_merge(array_keys((array) config('upstreams.sources', [])), array_keys($this->storedConfiguration->all()))); + $definitions = []; + foreach ($keys as $key) { + $definitions[$key] = $this->definition($key); + } + return $definitions; + } + private function usesLocalConfiguration(): bool { return (bool) config('upstreams.local_configuration.enabled', false); @@ -85,6 +108,7 @@ class UpstreamSourceRegistry private function connection(string $source, array $configured): array { $fields = array_values(array_unique(array_merge(array_keys($configured), [ + 'driver', 'host', 'port', 'database', diff --git a/modules/EcommerceFinance/Services/UpstreamSqlDialect.php b/modules/EcommerceFinance/Services/UpstreamSqlDialect.php new file mode 100644 index 00000000..1f813875 --- /dev/null +++ b/modules/EcommerceFinance/Services/UpstreamSqlDialect.php @@ -0,0 +1,55 @@ +driver) { + 'pgsql' => '"' . str_replace('"', '""', $identifier) . '"', + 'mysql' => '`' . str_replace('`', '``', $identifier) . '`', + default => '[' . str_replace(']', ']]', $identifier) . ']', + }; + } + + public function select(string $columns, string $tail, int $limit): string + { + $limit = min(max($limit, 1), 5000); + return $this->driver === 'sqlsrv' + ? 'SELECT TOP (' . $limit . ') ' . $columns . ' ' . $tail + : 'SELECT ' . $columns . ' ' . $tail . ' LIMIT ' . $limit; + } + + public function catalogSql(): string + { + // Same normalized catalog shape for all engines; MySQL never enumerates other databases. + $schema = $this->driver === 'sqlsrv' ? 'INFORMATION_SCHEMA' : 'information_schema'; + $sql = "SELECT c.table_schema AS schema_name, c.table_name AS object_name, + CASE WHEN t.table_type = 'VIEW' THEN 'view' ELSE 'table' END AS object_type, + c.column_name AS column_name, c.ordinal_position AS ordinal_position, + c.data_type AS data_type, c.character_maximum_length AS max_length, + c.numeric_precision AS numeric_precision, c.numeric_scale AS numeric_scale, + c.is_nullable AS is_nullable + FROM {$schema}." . ($this->driver === 'sqlsrv' ? 'COLUMNS' : 'columns') . " c + INNER JOIN {$schema}." . ($this->driver === 'sqlsrv' ? 'TABLES' : 'tables') . " t + ON t.table_schema = c.table_schema AND t.table_name = c.table_name + WHERE c.table_schema NOT IN ('sys', 'INFORMATION_SCHEMA', 'information_schema', 'pg_catalog', 'mysql', 'performance_schema')"; + if ($this->driver === 'mysql') $sql .= ' AND c.table_schema = DATABASE()'; + $sql .= ' ORDER BY c.table_schema, c.table_name, c.ordinal_position'; + if ($this->driver === 'sqlsrv') { + $sql = preg_replace_callback('/\b([ct])\.([a-z_]+)\b/', fn ($m) => $m[1] . '.' . strtoupper($m[2]), $sql); + } + return $sql; + } +} diff --git a/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php b/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php index 82e5c4cd..cd9feee3 100644 --- a/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php +++ b/modules/EcommerceFinance/Services/UpstreamStoredConfiguration.php @@ -42,10 +42,13 @@ class UpstreamStoredConfiguration return $this->connections[$source] = null; } - $this->connections[$source] = DatabaseConnection::query() - ->where('company_id', self::SYSTEM_COMPANY_ID) - ->where('source_key', $source) - ->first(); + $query = DatabaseConnection::query()->where('company_id', self::SYSTEM_COMPANY_ID); + if (preg_match('/^database_([1-9][0-9]*)$/', $source, $matches)) { + $query->where('id', $matches[1])->whereNull('source_key'); + } else { + $query->where('source_key', $source); + } + $this->connections[$source] = $query->first(); return $this->connections[$source]; } @@ -55,6 +58,16 @@ class UpstreamStoredConfiguration unset($this->connections[$source]); } + /** @return array */ + public function all(): array + { + if (! $this->enabled() || ! Schema::hasTable('system_database_connections')) { + return []; + } + return DatabaseConnection::query()->where('company_id', self::SYSTEM_COMPANY_ID) + ->get()->keyBy(fn (DatabaseConnection $row) => $row->sourceIdentifier())->all(); + } + public function value(string $source, string $path, mixed $default = null): mixed { $connection = $this->find($source); @@ -68,6 +81,7 @@ class UpstreamStoredConfiguration 'allowed_scopes' => $connection->allowed_scopes ?? [], 'connection' => $this->connection($connection, $default), 'connection.host' => $connection->host, + 'connection.driver' => $connection->driver, 'connection.port' => $connection->port, 'connection.database' => $connection->database_name, 'connection.username' => $connection->username, @@ -85,6 +99,7 @@ class UpstreamStoredConfiguration $configured = is_array($default) ? $default : []; return array_replace($configured, [ + 'driver' => $connection->driver, 'host' => $connection->host, 'port' => $connection->port, 'database' => $connection->database_name, diff --git a/resources/views/business/pages/partials/directory-table.blade.php b/resources/views/business/pages/partials/directory-table.blade.php index cff0b00d..0f8a4b5c 100644 --- a/resources/views/business/pages/partials/directory-table.blade.php +++ b/resources/views/business/pages/partials/directory-table.blade.php @@ -1,6 +1,15 @@ @php $directoryRows = $rows ?? collect(); $directoryFields = $directory['fields'] ?? []; + $directoryFields = collect($directoryFields)->flatMap(function (array $field): array { + if ($field['name'] !== 'recent_three_month_sales') { + return [$field]; + } + return array_map(static function (int $monthsAgo): array { + $month = now()->startOfMonth()->subMonthsNoOverflow($monthsAgo); + return ['name' => $month->format('sales_Y_m'), 'label' => $month->format('n月销量'), 'kind' => 'number']; + }, range(3, 1)); + })->all(); $directoryIsPaginated = is_object($directoryRows) && method_exists($directoryRows, 'total'); $directoryTotal = $directoryIsPaginated ? $directoryRows->total() : (is_countable($directoryRows) ? count($directoryRows) : 0); $directoryCurrentPage = $directoryIsPaginated ? $directoryRows->currentPage() : 1; @@ -279,7 +288,7 @@
- {{ $directory['empty'] ?? '尚未同步数据。' }} + {{ !empty($upstreamError) ? '资料暂时无法读取,请按上方提示检查连接;这不代表业务数据为 0。' : ($directory['empty'] ?? '尚未同步数据。') }}
@@ -288,7 +297,7 @@ - @if ($directoryIsPaginated) + @if ($directoryIsPaginated && empty($upstreamError))