Compare commits

...

2 Commits

32 changed files with 1653 additions and 1227 deletions

View File

@ -102,10 +102,10 @@ SYSTEM_DATABASE_MODE=shared
1. 从共享数据字典确认来源、表或视图和字段;目录缺失或过期时由 Codex 执行 `upstream:schema:sync`;
2. 在本地通过 `UpstreamReadOnlyQueryFactory` 执行一次限定字段、限定行数的真实只读探索查询,不需要先有 Repository、`scope` 或生产快照;
3. 确认业务目的、字段含义、筛选条件、排序方式、最大返回量、公司/权限范围及建议缓存时间;
4. 在任务分支中通过 `UpstreamReadOnlyQuery` 实现固定、参数化的只读查询,页面只调用 Repository;
5. 为生产查询定义稳定 `scope`,通过 `UpstreamQueryCache` 读取,并在对应 `UpstreamSnapshotRefresher` 中登记同一范围;
4. 在任务分支中实现固定、参数化的批次 Repository,声明稳定游标并写入 `caiwu_business_mirror`;
5. 正式页面只调用镜像 Repository;只有尚未迁移的历史页面才暂时保留 `UpstreamQueryCache`;
6. 完成自动测试和本地真实只读验收后合并到 `main`;
7. 生产部署后生成首份共享快照,再验证第二个应用实例复用该快照。
7. 生产中央节点完成首次镜像后,再验证第二个应用实例直接复用同一镜像。开发机不启动刷新 worker。
系统数据库中只保存受控查询的标识、参数、访问范围和结果,不保存可执行的任意 SQL。需要调整查询时必须修改代码并重新走合并、部署和验收流程。
@ -122,9 +122,6 @@ docker compose -f docker-compose.local.yml ps
# 查看应用日志
docker compose -f docker-compose.local.yml logs -f app
# 查看共享快照后台进程
docker compose -f docker-compose.local.yml logs -f snapshot_worker
# 重启应用
docker compose -f docker-compose.local.yml restart app

View File

@ -142,13 +142,7 @@ 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);
}
$rows = $this->storeReportRows();
$offset = ($page - 1) * $perPage;
return ['data' => array_slice($rows, $offset, $perPage), 'total' => count($rows)];
@ -157,52 +151,60 @@ SQL);
/** @return array{data:array<int,array<string,mixed>>,total:int} */
private function allStores(): array
{
if ($this->mirror->ready('store-information.read')) {
$rows = $this->mirror->allStores();
return [
'data' => array_map(fn (array $row): array => $this->mapStore($row), $rows),
'total' => count($rows),
];
if (! $this->mirror->ready('store-information.read')) {
throw new ServiceUnavailableHttpException(null, '店铺资料共享镜像尚未完成首次同步。');
}
if ((app(\App\Utilities\SystemDatabaseConfiguration::class)->current()['mode'] ?? '') === 'bootstrap'
&& ! config('upstreams.local_configuration.enabled', false)) {
throw new ServiceUnavailableHttpException(null, '当前仍使用本机临时引导库。请先连接团队共享数据库,然后重新登录。');
}
return $this->caches->forSource(self::SOURCE)->remember(
'store-information.read',
['all' => true],
fn () => $this->storeSnapshotAll(new UpstreamReadOnlyQuery(self::SOURCE)),
);
$rows = $this->mirror->allStores();
$rows = array_map(fn (array $row): array => $this->mapStore($row), $rows);
return ['data' => $rows, 'total' => count($rows)];
}
private function subjectOperatingReportPage(int $page, int $perPage): array
{
$payload = $this->storePage($page, $perPage);
$rows = $this->storeReportRows();
$entityTotals = [];
foreach ($rows as $row) {
$entity = (string) ($row['entity'] ?? '');
$entityTotals[$entity] = ($entityTotals[$entity] ?? 0.0) + (float) ($row['store_sales_12m'] ?? 0);
}
$rows = array_map(static function (array $row) use ($entityTotals): array {
$entity = (string) ($row['entity'] ?? '');
return [
'data' => array_map(static function (array $row): array {
return [
'city' => null,
'entity' => $row['company_name'] ?? null,
'company_type' => $row['company_type'] ?? null,
'erp_store_name' => $row['erp_store_name'] ?? null,
'profit_center' => $row['profit_center'] ?? null,
'store_manager' => $row['store_manager'] ?? null,
'closing_manager' => null,
'opened_at' => $row['opened_at'] ?? null,
'status' => $row['store_status'] ?? null,
'recent_three_month_sales' => null,
'store_sales_12m' => null,
'entity_sales_12m' => null,
'can_change_entity' => null,
'plan' => null,
'new_entity' => null,
];
}, $payload['data']),
'total' => $payload['total'],
];
return [
'city' => null,
'entity' => $row['entity'] ?? null,
'company_type' => $row['company_type'] ?? null,
'erp_store_name' => $row['erp_store_name'] ?? null,
'profit_center' => $row['profit_center'] ?? null,
'store_manager' => $row['store_manager'] ?? null,
'closing_manager' => null,
'opened_at' => $row['opened_at'] ?? null,
'status' => $row['status'] ?? null,
'recent_three_month_sales_1' => $row['recent_three_month_sales_1'] ?? 0,
'recent_three_month_sales_2' => $row['recent_three_month_sales_2'] ?? 0,
'recent_three_month_sales_3' => $row['recent_three_month_sales_3'] ?? 0,
'store_sales_12m' => $row['store_sales_12m'] ?? 0,
'entity_sales_12m' => $entityTotals[$entity] ?? 0,
'can_change_entity' => null,
'plan' => null,
'new_entity' => null,
];
}, $rows);
$offset = ($page - 1) * $perPage;
return ['data' => array_slice($rows, $offset, $perPage), 'total' => count($rows)];
}
/** @return array<int,array<string,mixed>> */
private function storeReportRows(): array
{
$stores = $this->allStores();
try {
return ($this->storeReports ?? app(StoreReportRepository::class))->rows($stores['data']);
} catch (\RuntimeException $exception) {
throw new ServiceUnavailableHttpException(null, $exception->getMessage(), $exception);
}
}
private function storeSnapshot(UpstreamReadOnlyQuery $query, int $page, int $perPage): array
@ -252,50 +254,6 @@ SQL);
];
}
/** @return array{data:array<int,array<string,mixed>>,total:int} */
private function storeSnapshotAll(UpstreamReadOnlyQuery $query): array
{
$rows = $query->select(<<<SQL
SELECT
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,
t.trademarkName 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
FROM rpt.fys_shopInfo AS s
OUTER APPLY (
SELECT TOP 1 company.businessLicenseName, company.businessLicenseType
FROM rpt.fys_companyInfo AS company
WHERE company.unifiedSocialCreditCode = NULLIF(s.unifiedSocialCreditCode, '')
ORDER BY company.Id DESC
) AS c
OUTER APPLY (
SELECT TOP 1 trademark.trademarkName
FROM rpt.fys_TrademarkInfo AS trademark
WHERE trademark.trademarkNumber = NULLIF(s.trademarkNumber, '')
AND trademark.isDelete = 0
ORDER BY trademark.Id DESC
) AS t
ORDER BY s.Id DESC
SQL);
$rows = array_map(fn (array $row): array => $this->mapStore($row), $rows);
return ['data' => $rows, 'total' => count($rows)];
}
private function trademarkPage(int $page, int $perPage): array
{
return $this->page(

View File

@ -1,44 +0,0 @@
<?php
namespace App\Console\Commands;
use Illuminate\Console\Command;
use Modules\EcommerceFinance\Services\StoreReportDataImporter;
use Throwable;
class ImportStoreReportData extends Command
{
protected $signature = 'store-report:import-data
{--bills= : 五个平台原始账单 XLSX 文件或目录}
{--lifecycles= : 店铺生命周期 XLSX 文件}
{--batch=500 : 单批写入共享业务镜像的行数}';
protected $description = '将平台原始账单和店铺生命周期原子导入团队共享业务镜像';
public function handle(StoreReportDataImporter $importer): int
{
$bills = trim((string) $this->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;
}
}

View File

@ -0,0 +1,46 @@
<?php
namespace App\Console\Commands;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Log;
use Modules\EcommerceFinance\Services\StoreReportMirrorSynchronizer;
use Throwable;
class SyncStoreReportMirror extends Command
{
protected $signature = 'business-mirror:sync-store-report
{--batch= : 每批最多读取账单行数}
{--max-batches= : 本轮最多处理账单批次数}';
protected $description = '自动初始化并增量同步店铺维度报表共享镜像';
public function handle(StoreReportMirrorSynchronizer $synchronizer): int
{
if (! (bool) config('upstream_mirrors.refresh_enabled', false)
|| ! (bool) config('upstream_mirrors.store_report.enabled', true)) {
return self::SUCCESS;
}
try {
$result = $synchronizer->sync(
$this->option('batch') === null ? null : (int) $this->option('batch'),
$this->option('max-batches') === null ? null : (int) $this->option('max-batches'),
);
} catch (Throwable $exception) {
Log::error('店铺维度报表共享镜像同步失败。', ['exception' => $exception::class]);
$this->error('店铺维度报表共享镜像同步失败,页面继续读取上一版数据。');
return self::FAILURE;
}
$this->info(sprintf(
'店铺报表镜像:生命周期 %s/%d 行,账单 %s/%d 行。',
$result['lifecycles']['status'],
$result['lifecycles']['source_rows'],
$result['bills']['status'],
$result['bills']['source_rows'],
));
return self::SUCCESS;
}
}

View File

@ -67,7 +67,6 @@ services:
- "127.0.0.1:18088:18088"
volumes:
- ./runtime/.env:/var/www/html/.env
- ./runtime/bills.sqlite:/var/www/html/runtime/bills.sqlite:ro
- caiwu_storage:/var/www/html/storage
healthcheck:
test:

View File

@ -159,9 +159,9 @@ return [
'title' => '店铺维度报表',
'directory' => [
'icon' => 'query_stats',
'status' => '已接入真实账单与生命周期',
'description' => '店铺资料来自经营分析副本库;主体生效区间和月度销量来自本地 bills.sqlite。',
'empty' => '暂无可匹配的店铺生命周期或账单数据。',
'status' => '共享镜像',
'description' => '店铺资料、主体生效区间和平台账单统一读取团队共享业务镜像;销量按账单收入汇总。',
'empty' => '店铺维度报表尚未完成共享数据导入。',
'fields' => [
['name' => 'erp_store_name', 'label' => 'ERP系统店铺名称'],
['name' => 'store_id', 'label' => '店铺ID(唯一值)'],

View File

@ -71,17 +71,6 @@ return [
'foreign_key_constraints' => env('DB_FOREIGN_KEYS', true),
],
// 店铺维度报表的本地只读账单与生命周期数据源。
'bills_sqlite' => [
'driver' => 'sqlite',
'database' => env('BILLS_SQLITE_PATH', base_path('runtime/bills.sqlite')),
'prefix' => '',
'foreign_key_constraints' => null,
'options' => [
PDO::SQLITE_ATTR_OPEN_FLAGS => PDO::SQLITE_OPEN_READONLY,
],
],
'mysql' => [
'driver' => 'mysql',
'url' => env('DATABASE_URL'),

View File

@ -20,6 +20,20 @@ return [
env('APP_ENV', 'production') === 'production'
),
// 店铺经营报表由生产刷新节点自动完成首次同步并持续增量;开发机只读。
'store_report' => [
'enabled' => filter_var(env('STORE_REPORT_MIRROR_ENABLED', true), FILTER_VALIDATE_BOOL),
'source_key' => 'jiansun',
'bill_source_object' => 'dbo.jst_originalBillInfo',
'lifecycle_source_object' => 'dbo.fys_shopInfo_flow',
'batch_size' => (int) env('STORE_REPORT_MIRROR_BATCH_SIZE', 5000),
'max_batches_per_run' => (int) env('STORE_REPORT_MIRROR_MAX_BATCHES', 4),
'lifecycle_batch_size' => (int) env('STORE_REPORT_LIFECYCLE_BATCH_SIZE', 1000),
'lifecycle_max_batches_per_run' => (int) env('STORE_REPORT_LIFECYCLE_MAX_BATCHES', 5),
// 原始账单以 Id 追加为主;回看最近行覆盖迟到修正,不扫描 1400 万行更新时间。
'overlap_rows' => (int) env('STORE_REPORT_MIRROR_OVERLAP_ROWS', 50000),
],
'product_prices' => [
'dataset_key' => 'supplier-payables.product-prices',
'source_key' => 'operations_replica',
@ -96,6 +110,7 @@ return [
'watermark' => 'updateTime',
'indexes' => [
['o_id', 'Id'],
['order_date', 'o_id', 'Id'],
],
],
'operations_replica.rpt.jst_OrderItem' => [
@ -106,6 +121,7 @@ return [
'keys' => ['Id'],
'indexes' => [
['o_id', 'Id'],
['o_id', 'outer_oi_id', 'Id'],
],
],
'operations_replica.rpt.jst_sku' => [
@ -123,9 +139,9 @@ return [
['sku_type', 'stock_disabled', 'autoid'],
],
],
'operations_replica.rpt.jst_warehouseInfo' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_warehouseInfo', 'target' => 'ops_jst_warehouse_info', 'keys' => ['Id'], 'watermark' => 'updateTime'],
'operations_replica.rpt.jst_skumap' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_skumap', 'target' => 'ops_jst_sku_map', 'keys' => ['shop_id', 'shop_i_id', 'shop_sku_id'], 'watermark' => 'modifiedTime'],
'operations_replica.rpt.ads_itemid_sales_cost_ratio_v2_di' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'ads_itemid_sales_cost_ratio_v2_di', 'target' => 'ops_item_sales_cost_ratio', 'keys' => ['stat_date', 'item_id'], 'watermark' => 'item_id_update'],
'operations_replica.rpt.jst_warehouseInfo' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_warehouseInfo', 'target' => 'ops_jst_warehouse_info', 'keys' => ['Id'], 'watermark' => 'updateTime', 'indexes' => [['wms_co_id', 'Id']]],
'operations_replica.rpt.jst_skumap' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'jst_skumap', 'target' => 'ops_jst_sku_map', 'keys' => ['shop_id', 'shop_i_id', 'shop_sku_id'], 'watermark' => 'modifiedTime', 'indexes' => [['shop_i_id', 'modified']]],
'operations_replica.rpt.ads_itemid_sales_cost_ratio_v2_di' => ['source_key' => 'operations_replica', 'schema' => 'rpt', 'object' => 'ads_itemid_sales_cost_ratio_v2_di', 'target' => 'ops_item_sales_cost_ratio', 'keys' => ['stat_date', 'item_id'], 'watermark' => 'item_id_update', 'indexes' => [['stat_date', 'push_fee', 'item_id']]],
'operations_replica.rpt.jst_returnAndRefundInfo' => [
'source_key' => 'operations_replica',
'schema' => 'rpt',

View File

@ -0,0 +1,80 @@
<?php
use Illuminate\Database\Migrations\Migration;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\Schema;
return new class extends Migration
{
private function connection(): string
{
return (string) config('upstream_mirrors.connection', 'business_mirror');
}
public function up(): void
{
$schema = Schema::connection($this->connection());
$this->addColumn('store_report_import_states', 'run_mode', fn (Blueprint $table) =>
$table->string('run_mode', 24)->nullable()->after('status'));
$this->addColumn('store_report_import_states', 'cursor_id', fn (Blueprint $table) =>
$table->unsignedBigInteger('cursor_id')->nullable()->after('pending_generation'));
$this->addColumn('store_report_import_states', 'initial_completed_at', fn (Blueprint $table) =>
$table->dateTime('initial_completed_at')->nullable()->after('source_files_json'));
$this->addColumn('store_report_bill_rows', 'source_id', fn (Blueprint $table) =>
$table->unsignedBigInteger('source_id')->nullable()->after('generation'));
$this->addColumn('store_report_bill_rows', 'shop_id', fn (Blueprint $table) =>
$table->unsignedBigInteger('shop_id')->nullable()->after('source_id'));
$this->addColumn('store_report_bill_rows', 'source_updated_at', fn (Blueprint $table) =>
$table->dateTime('source_updated_at')->nullable()->after('row_hash'));
if (! $schema->hasIndex('store_report_bill_rows', 'store_report_bills_generation_source_unique')) {
$schema->table('store_report_bill_rows', function (Blueprint $table): void {
$table->unique(['generation', 'source_id'], 'store_report_bills_generation_source_unique');
});
}
$this->addColumn('store_report_lifecycle_rows', 'source_id', fn (Blueprint $table) =>
$table->unsignedBigInteger('source_id')->nullable()->after('generation'));
$this->addColumn('store_report_lifecycle_rows', 'unified_social_credit_code', fn (Blueprint $table) =>
$table->string('unified_social_credit_code', 64)->nullable()->after('company_name'));
if (! $schema->hasIndex('store_report_lifecycle_rows', 'store_report_lifecycle_generation_source_unique')) {
$schema->table('store_report_lifecycle_rows', function (Blueprint $table): void {
$table->unique(['generation', 'source_id'], 'store_report_lifecycle_generation_source_unique');
});
}
}
public function down(): void
{
$schema = Schema::connection($this->connection());
if ($schema->hasColumn('store_report_lifecycle_rows', 'source_id')) {
$schema->table('store_report_lifecycle_rows', function (Blueprint $table): void {
$table->dropUnique('store_report_lifecycle_generation_source_unique');
$table->dropColumn(['source_id', 'unified_social_credit_code']);
});
}
if ($schema->hasColumn('store_report_bill_rows', 'source_id')) {
$schema->table('store_report_bill_rows', function (Blueprint $table): void {
$table->dropUnique('store_report_bills_generation_source_unique');
$table->dropColumn(['source_id', 'shop_id', 'source_updated_at']);
});
}
if ($schema->hasColumn('store_report_import_states', 'run_mode')) {
$schema->table('store_report_import_states', function (Blueprint $table): void {
$table->dropColumn(['run_mode', 'cursor_id', 'initial_completed_at']);
});
}
}
private function addColumn(string $table, string $column, callable $definition): void
{
$schema = Schema::connection($this->connection());
if ($schema->hasColumn($table, $column)) {
return;
}
$schema->table($table, function (Blueprint $blueprint) use ($definition): void {
$definition($blueprint);
});
}
};

View File

@ -129,7 +129,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha"
# 账单提醒等既有任务。同一把锁保证快照与业务镜像不会并发施压上游。
# 普通商品优先高频增量,并在配置的每周低峰窗口自动继续全量校准。
cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF'
* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1
* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-store-report --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1
EOF
chmod 644 /etc/cron.d/caiwu-jiansun-snapshots

View File

@ -133,7 +133,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha"
# 账单提醒等既有任务。同一把锁保证快照与业务镜像不会并发施压上游。
# 普通商品优先高频增量,并在配置的每周低峰窗口自动继续全量校准。
cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF'
* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1
* * * * * root /usr/bin/flock -n /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-catalog --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-purchase-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-store-report --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan upstream:snapshots:refresh-all --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-product-prices --quiet; /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables --quiet' >/dev/null 2>&1
EOF
chmod 644 /etc/cron.d/caiwu-jiansun-snapshots

View File

@ -244,9 +244,12 @@ else
echo "==> [1/5] 本地开发镜像与当前源码一致,直接复用。"
fi
# 共享快照只由生产中央节点刷新。先清理旧版 worker,避免 Compose 报孤儿
# 容器,也避免不同分支继续重复访问上游。
docker rm -f jiangling_snapshot_worker >/dev/null 2>&1 || true
echo "==> [2/5] 启动本地容器..."
# 首次引导时 app 在安装前不会通过健康检查,snapshot_worker 又依赖该健康
# 状态;先只启动数据库并确认 bootstrap 状态,再启动网页容器,避免旧的
# 先只启动数据库并确认 bootstrap 状态,再启动网页容器,避免旧的
# APP_INSTALLED=true 让 entrypoint 与首次安装命令并发迁移空数据库。
docker compose -f "$compose_file" up -d --no-build mysql
@ -326,8 +329,14 @@ else
team_connect_status=$?
fi
if [ "$team_connect_status" -ne 0 ] && [ "$team_connect_status" -ne 13 ]; then
echo "错误:本机未完成自动接入;请先按上方提示核对连接和私有配置备份。" >&2
exit "$team_connect_status"
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:verify --no-interaction --no-ansi'; then
echo " 当前共享连接仍可用,保留原配置并继续初始化。"
else
echo "错误:最新端点和当前共享连接均不可用;原配置已保留,请核对网络或首次输入团队数据库密码。" >&2
exit "$team_connect_status"
fi
fi
db_host="$(read_env_value DB_HOST)"
@ -396,14 +405,6 @@ if [ "$ready" -ne 1 ]; then
exit 1
fi
docker compose -f "$compose_file" up -d --no-build snapshot_worker
if ! docker compose -f "$compose_file" ps --status running --services \
| grep -qx 'snapshot_worker'; then
echo "错误:上游快照 worker 未能启动。" >&2
exit 1
fi
http_code="$(curl -sS -o /dev/null -w '%{http_code}' --max-time 15 \
http://127.0.0.1:18088/auth/login || true)"

View File

@ -24,7 +24,6 @@ services:
- "127.0.0.1:18088:18088"
volumes:
- ./runtime/.env:/var/www/html/.env
- ./runtime/bills.sqlite:/var/www/html/runtime/bills.sqlite:ro
- jiangling_storage:/var/www/html/storage
- ./app:/var/www/html/app
- ./config:/var/www/html/config
@ -42,30 +41,6 @@ services:
retries: 10
start_period: 60s
snapshot_worker:
image: jiangling_akaunting:local
container_name: jiangling_snapshot_worker
user: www-data
restart: unless-stopped
depends_on:
app:
condition: service_healthy
command:
- sh
- -lc
- while true; do php artisan upstream:snapshots:refresh-all --quiet || true; sleep 60; done
volumes:
- ./runtime/.env:/var/www/html/.env
- ./runtime/bills.sqlite:/var/www/html/runtime/bills.sqlite:ro
- jiangling_storage:/var/www/html/storage
- ./app:/var/www/html/app
- ./config:/var/www/html/config
- ./database:/var/www/html/database
- ./modules:/var/www/html/modules
- ./routes:/var/www/html/routes
networks:
- jiangling_net
mysql:
image: mysql:8.0
container_name: jiangling_mysql

View File

@ -17,16 +17,16 @@
| 本次接入 | 工作平台快递资料页面读取 `wp_express` 主表,并预留网点 `wp_express_expNet`、打款账户 `wp_express_account` 标签;三张表均登记为独立镜像数据集,首次同步完成前页面明确显示同步状态。 |
| 本次建立 | `config/upstream_mirrors.php` 中登记全部当前代码引用的原表;通用表级同步框架负责建表、稳定游标、断点续传、增量重叠和全量校准。 |
| 已开始逐域切换 | 公司、店铺、商标页面在对应镜像完成首次全量后自动改读完整镜像表;首次全量前保留旧快照作为兼容回滚。其他业务域仍需逐项切换,不得宣称全项目已经完成。 |
| 中央同步接入 | 店铺维度报表从 `jiansun.dbo.jst_originalBillInfo` 和 `dbo.fys_shopInfo_flow` 自动、可断点同步;不再依赖开发者电脑、SQLite 或人工 XLSX。 |
## 经营分析副本库(operations_replica)
店铺维度报表不使用订单件数代替销量。五个平台原始账单和店铺生命周期
以完整导入代次写入 `caiwu_business_mirror` 的 `store_report_bill_rows`、
`store_report_lifecycle_rows`,成功后原子切换,失败时继续读取上一代。
店铺资料仍以 `rpt.fys_shopInfo` 镜像为主;两份导入数据按 ERP 店铺名称
关联。每个生命周期行使用 `[effective_at, expires_at)`,销量固定为账单
`income` 汇总;结算时间存在时优先使用结算时间,否则使用平台发生时间。
所有开发机与生产页面读取同一份共享镜像,不依赖开发者本机文件。
店铺维度报表不使用订单件数代替销量。中央刷新节点按上游主键 `Id`
分批读取 `jiansun.dbo.jst_originalBillInfo`,以 `inout_type = 1` 的
`amount` 作为收入,并从 `dbo.jst_shopInfo` 取得店铺名称;主体变更区间来自
`dbo.fys_shopInfo_flow`,按 `[applyTime, 下一次 applyTime)` 归属。
首次同步自动小批量续传,完成前不激活半份数据;失败时继续读取上一代。
所有开发机与生产页面只读同一份共享镜像,不依赖本机文件。
| 业务范围 | 上游对象 | 镜像目标 |
| --- | --- | --- |
@ -38,7 +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` |
| 店铺维度报表 | `jiansun/dbo.jst_originalBillInfo + dbo.jst_shopInfo + dbo.fys_shopInfo_flow` | `store_report_bill_rows`、`store_report_lifecycle_rows` |
商品价格 `rpt.Bulk_Price_Limit` 继续使用已经上线的类型化同步器,不在通用原表任务中重复同步。
@ -67,6 +67,7 @@
6. 页面切换必须以“总记录数、第一页、第二页、末页、筛选、汇总、空数据和更新时间”作为验收,不能只看首页有数据。
7. 普通商品的采购单价同步以采购主表 `updateTime` 加明细 `Id` 作为稳定游标;页面按商品编码选择最近完成的 1688 一键代发价格。同步保存全部状态,使完成单后续取消时也能撤回旧价格;首次全量前只显示普通商品自身采购价。
8. 聚水潭官方接口尚未接入时,普通商品历史残留排除由负责人提供的最新完整 XLSX 导出刷新。只排除“导出生成前已创建、但导出中缺失”的镜像行,导出生成后的新增商品保持可见。导入使用新同步代次,只有完整解析成功才原子替换排除表;中断或错误保留上一版。后续接入聚水潭官方“普通商品资料查询”及删除/恢复记录后,由官方接口取代人工导出校准。
9. 店铺维度报表是已批准的中央专用数据集:生产节点每分钟最多读取 4 批、每批 5000 行,按 `Id` 断点推进并回看最近 50000 行;开发机不运行刷新 worker。
## 后续切换顺序

View File

@ -245,7 +245,7 @@ php artisan business-mirror:sync-tables
完整数据集、来源对象、镜像表和分阶段切换状态见 `docs/上游业务镜像清单.md`。不得把“已登记同步目录”误报成“页面已经切换完成”;每个业务域仍需在首份全量核对通过后,把对应 Repository 改为只读镜像。
首次 `--full` 属于生产数据写入,必须先备份 `caiwu_system_config` 和 `caiwu_business_mirror`,并由负责人明确授权。后续无参数命令只执行或继续受控增量。
首次 `--full` 属于生产数据写入,必须先备份 `caiwu_system_config` 和 `caiwu_business_mirror`,并由负责人明确授权。已经在发布说明中批准自动初始化的中央专用数据集,由生产节点在迁移授权后小批量续传;普通开发者不配置、不启动也不审批这些任务。后续无参数命令只执行或继续受控增量。
### 8.3 开发者需要尚未接入的上游数据时
@ -296,7 +296,7 @@ php artisan upstream:schema:sync operations_replica
1. 页面优先读取共享系统数据库中的 `upstream_query_cache`;相同来源、数据范围、查询条件和公司/权限范围复用同一结果,与系统账户和开发机数量无关;
2. 列表和汇总默认保持 2 分钟新鲜,下拉选项等低频数据缓存 60 分钟;
3. 新鲜快照直接返回;过期快照也立即返回,同时登记为紧急后台刷新,页面不等待上游;共享系统数据库中从未存在该查询时,页面只登记需求并明确提示正在准备,绝不从 HTTP 请求直接连接上游;
4. `/etc/cron.d/caiwu-jiansun-snapshots` 和本地 `snapshot_worker` 每分钟调用总刷新入口,只处理最近 6 小时实际使用且已经到期的查询;
4. 只有生产服务器的 `/etc/cron.d/caiwu-jiansun-snapshots` 每分钟调用总刷新入口;开发机不再运行 `snapshot_worker`,避免不同分支重复访问上游;
5. 相同查询使用独立单飞锁,不同查询最多占用 2 个上游槽位,因此大量缓存命中不会被全局串行;
6. 每轮默认最多刷新 2 个活跃查询,每分钟默认最多产生 6 个实际上游查询组;列表优先于汇总,汇总优先于选项;
7. 连接 5 秒未建立或单次查询超过 15 秒即失败;慢查询和失败查询进入 5 分钟冷却,连接失败同时触发该来源熔断,不连续重试;

View File

@ -1,54 +0,0 @@
# 本地账单 SQLite
`scripts/import_bills_sqlite.py` 将 `/Users/dodo/AI系统/账单数据` 下的五个平台 Excel 导入 `runtime/bills.sqlite`。数据库位于 `runtime/`,不会进入 Git。
平台和主时间字段固定为:
| 平台 | `type` | 主时间列 |
| --- | --- | --- |
| 得物 | 得物 | 支付/退款时间 |
| 头条放心购(抖音) | 抖音 | 结算时间 |
| 淘系 | 淘系 | 入账时间 |
| 京东 | 京东 | 费用结算时间 |
| 拼多多 | 拼多多 | 发生时间 |
核心表 `bills` 至少包含 `store_name`(店铺名称)、`type`(平台类型)、`occurred_at`(统一为 `YYYY-MM-DD HH:MM:SS`)、`income`(收入)、`expense`(支出),同时保留 `bill_item`、`record_code`、来源文件和完整原始行 JSON。`(platform, record_code)` 唯一,重复执行会更新而不是重复插入。
首次导入或源文件更新后重新导入:
```bash
PY=/Users/dodo/.cache/codex-runtimes/codex-primary-runtime/dependencies/python/bin/python3
"$PY" scripts/import_bills_sqlite.py --reset
```
后续按店铺和日期汇总可直接查询视图:
```sql
SELECT * FROM bill_daily_summary
WHERE store_name = '得物-米客米家家居旗舰店'
AND bill_date BETWEEN '2026-08-01' AND '2026-08-31'
ORDER BY bill_date;
```
如需按精确时间范围匹配,查询 `bills` 的 `occurred_at >= ? AND occurred_at < ?`,并同时限定 `store_name`;金额字段使用 `income` 和 `expense`,不要从原始 JSON 再解析。
## 店铺存续时间
`scripts/import_store_lifecycle_sqlite.py` 将 `店铺存续时间.xlsx` 导入 `store_lifecycles` 表。原表四个字段全部保留:`ERP系统店铺名称`、`公司名称`、`生效时间`、`失效时间`,其中时间字段转换为 SQLite 可比较的 `YYYY-MM-DD HH:MM:SS` 文本,同时保留 `source_file` 和 `raw_json`。
导入命令:
```bash
PY=/Users/dodo/.cache/codex-runtimes/codex-primary-runtime/dependencies/python/bin/python3
"$PY" scripts/import_store_lifecycle_sqlite.py \
/Users/dodo/Documents/店铺存续时间.xlsx --db runtime/bills.sqlite --reset
```
匹配店铺在指定日期是否存续时,可使用:
```sql
SELECT * FROM store_lifecycles
WHERE erp_store_name = ?
AND effective_at <= ?
AND (expires_at IS NULL OR expires_at >= ?);
```

View File

@ -0,0 +1,354 @@
<?php
namespace Modules\EcommerceFinance\Repositories;
use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\Query\Builder;
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
class OrderDataMirrorRepository
{
private const DATASETS = [
'operations_replica.rpt.jst_Orders',
'operations_replica.rpt.jst_OrderItem',
'operations_replica.rpt.jst_sku',
'operations_replica.rpt.jst_warehouseInfo',
'operations_replica.rpt.jst_returnAndRefundInfo',
'operations_replica.rpt.jst_returnAndRefundItem',
];
private const PROMOTION_DATASETS = [
'operations_replica.rpt.ads_itemid_sales_cost_ratio_v2_di',
'operations_replica.rpt.jst_skumap',
'operations_replica.rpt.jst_shopInfo',
];
private readonly ConnectionInterface $connection;
public function __construct(
private readonly BusinessMirrorDatasetRegistry $datasets,
private readonly GenericBusinessMirrorRepository $mirror,
) {
$this->connection = $mirror->connection();
}
public function ready(): bool
{
return $this->datasetsReady(self::DATASETS);
}
public function promotionReady(): bool
{
return $this->datasetsReady(self::PROMOTION_DATASETS);
}
private function datasetsReady(array $datasets): bool
{
try {
foreach ($datasets as $dataset) {
if (! $this->mirror->ready($this->datasets->get($dataset))) {
return false;
}
}
return true;
} catch (\Throwable) {
return false;
}
}
/** @return array{rows:array<int,array<string,mixed>>} */
public function afterSales(array $filters, int $page, int $limit): array
{
$limit = min(max($limit, 1), 100);
$query = $this->baseQuery();
$this->applyFilters($query, $filters);
$rows = $query
->select($this->columns())
->orderByDesc('o.order_date')->orderByDesc('o.o_id')->orderByDesc('i.oi_id')
->offset((max($page, 1) - 1) * $limit)
->limit($limit + 1)
->get()->map(static fn (object $row): array => (array) $row)->all();
$rows = array_map(fn (array $row): array => $this->baseRow($row), $rows);
$this->hydrateRefunds($rows);
return ['rows' => $rows];
}
/** @return array{rows:array<int,array<string,mixed>>,date:?string} */
public function promotionExpenses(array $filters, int $page, int $limit): array
{
$limit = min(max($limit, 1), 100);
$date = trim((string) ($filters['date'] ?? ''));
if ($date === '') {
$date = (string) ($this->connection->table('ops_item_sales_cost_ratio')->max('stat_date') ?? '');
$date = substr($date, 0, 10);
}
if ($date === '') {
return ['rows' => [], 'date' => null];
}
$query = $this->connection->table('ops_item_sales_cost_ratio as c')
->where('c.stat_date', '>=', $date . ' 00:00:00')
->where('c.stat_date', '<', date('Y-m-d 00:00:00', strtotime($date . ' +1 day')))
->where('c.push_fee', '<>', 0);
if (($filters['shop_name'] ?? '') !== '') {
$query->whereRaw("c.shop_name LIKE ? ESCAPE '!'", [$this->contains((string) $filters['shop_name'])]);
}
if (($filters['sku_id'] ?? '') !== '') {
$needle = $this->contains((string) $filters['sku_id']);
$query->where(function (Builder $items) use ($needle): void {
$items->whereRaw("c.item_id LIKE ? ESCAPE '!'", [$needle])
->orWhereExists(function (Builder $mapping) use ($needle): void {
$mapping->selectRaw('1')->from('ops_jst_sku_map as filter_map')
->whereColumn('filter_map.shop_i_id', 'c.item_id')
->where(function (Builder $codes) use ($needle): void {
$codes->whereRaw("filter_map.sku_id LIKE ? ESCAPE '!'", [$needle])
->orWhereRaw("filter_map.i_id LIKE ? ESCAPE '!'", [$needle])
->orWhereRaw("filter_map.shop_sku_id LIKE ? ESCAPE '!'", [$needle]);
});
});
});
}
$rows = $query->select(['c.shop_name', 'c.item_id as shop_style_code', 'c.push_fee as promotion_fee'])
->orderByDesc('c.push_fee')->orderBy('c.shop_name')->orderBy('c.item_id')
->offset((max($page, 1) - 1) * $limit)->limit($limit + 1)
->get()->map(static fn (object $row): array => (array) $row)->all();
$itemIds = array_values(array_unique(array_filter(array_column($rows, 'shop_style_code'))));
$mappings = $itemIds === [] ? collect() : $this->connection->table('ops_jst_sku_map as m')
->leftJoin('ops_jst_shop_info as shop', 'shop.shop_id', '=', 'm.shop_id')
->whereIn('m.shop_i_id', $itemIds)->orderByDesc('m.modified')
->get(['m.shop_i_id', 'm.sku_id', 'm.i_id', 'm.shop_sku_id', 'm.properties_value', 'shop.shop_name'])
->groupBy('shop_i_id');
foreach ($rows as &$row) {
$candidates = $mappings->get($row['shop_style_code'], collect());
$mapping = $candidates->firstWhere('shop_name', $row['shop_name']) ?? $candidates->first();
$row['product_code'] = $mapping->sku_id ?? null;
$row['style_code'] = $mapping->i_id ?? null;
$row['shop_product_code'] = $mapping->shop_sku_id ?? null;
$row['online_color_spec'] = $mapping->properties_value ?? null;
$row['promotion_fee'] = round((float) $row['promotion_fee'], 2);
}
unset($row);
return ['rows' => $rows, 'date' => $date];
}
private function baseQuery(): Builder
{
$latestSkus = $this->connection->table('ops_jst_sku')
->selectRaw('sku_id, MAX(autoid) AS latest_autoid')
->where('sku_type', 'normal')->groupBy('sku_id');
return $this->connection->table('ops_jst_orders as o')
->join('ops_jst_order_item as i', 'i.o_id', '=', 'o.o_id')
->leftJoinSub($latestSkus, 'sku_keys', 'sku_keys.sku_id', '=', 'i.sku_id')
->leftJoin('ops_jst_sku as s', 's.autoid', '=', 'sku_keys.latest_autoid')
->leftJoin('ops_jst_warehouse_info as wh', function ($join): void {
$join->on('wh.wms_co_id', '=', $this->connection->raw('COALESCE(i.send_warehouse, o.wms_co_id)'));
});
}
private function applyFilters(Builder $query, array $filters): void
{
if (($filters['from'] ?? '') !== '') {
$query->where('o.order_date', '>=', $filters['from'] . ' 00:00:00');
}
if (($filters['to'] ?? '') !== '') {
$query->where('o.order_date', '<', date('Y-m-d 00:00:00', strtotime($filters['to'] . ' +1 day')));
}
if (($filters['order_no'] ?? '') !== '') {
$needle = $this->contains((string) $filters['order_no']);
$query->where(function (Builder $order) use ($needle): void {
$order->whereRaw("CAST(o.o_id AS CHAR) LIKE ? ESCAPE '!'", [$needle])
->orWhereRaw("o.so_id LIKE ? ESCAPE '!'", [$needle])
->orWhereRaw("i.outer_oi_id LIKE ? ESCAPE '!'", [$needle]);
});
}
if (($filters['shop_name'] ?? '') !== '') {
$query->whereRaw("o.shop_name LIKE ? ESCAPE '!'", [$this->contains((string) $filters['shop_name'])]);
}
if (($filters['sku_id'] ?? '') !== '') {
$needle = $this->contains((string) $filters['sku_id']);
$query->where(function (Builder $sku) use ($needle): void {
$sku->whereRaw("i.sku_id LIKE ? ESCAPE '!'", [$needle])
->orWhereRaw("i.shop_sku_id LIKE ? ESCAPE '!'", [$needle])
->orWhereRaw("i.shop_i_id LIKE ? ESCAPE '!'", [$needle]);
});
}
}
/** @return list<string> */
private function columns(): array
{
return [
'o.o_id as internal_order_no', 'o.so_id as online_order_no',
'i.outer_oi_id as sub_order_no', 'o.pay_date as pay_time',
'o.end_time as order_complete_time', 'o.send_date as send_time',
'o.order_date as order_time', 'o.shop_name', 'o.shop_id as shop_code',
'i.sku_id as product_code', 'i.i_id as style_code',
'i.shop_sku_id as shop_product_code', 'i.shop_i_id as shop_style_code',
'o.shop_site', 's.brand', 's.category', 's.vc_name as virtual_category',
's.cost_price as unit_cost', 'i.properties_value as color_spec',
'i.name as online_product_name', 's.name as sku_name', 'i.qty as product_quantity',
'i.amount as product_amount', 'i.buyer_paid_amount as payment_amount',
'i.seller_income_amount', 'i.is_gift', 'i.src_combine_sku_id',
'o.type as order_type', 'o.labels as order_tags',
'o.free_amount as merchant_discount', 'o.freight as freight_income',
'i.send_warehouse', 'o.wms_co_id', 'o.l_id as express_no',
'o.logistics_company as express_company', 'o.receiver_state as province',
'o.receiver_city as city', 'o.creator_name as salesperson',
'wh.name as shipping_warehouse', 'wh.remark1 as operation_cloud_warehouse',
's.supplier_id as supplier_code', 's.supplier_name',
];
}
private function baseRow(array $row): array
{
$quantity = (float) ($row['product_quantity'] ?? 0);
$amount = (float) ($row['product_amount'] ?? 0);
$cost = (float) ($row['unit_cost'] ?? 0);
$payment = (float) ($row['payment_amount'] ?? $amount);
$sellerIncome = (float) ($row['seller_income_amount'] ?? 0);
$row += [
'after_sales_no' => null, 'refund_type' => null, 'question_type' => null,
'detail_return_type' => null, 'after_sales_tags' => null,
];
$row['product_name'] = ($row['sku_name'] ?? '') !== '' ? $row['sku_name'] : ($row['color_spec'] ?? null);
$row['product_quantity'] = $quantity;
$row['product_amount'] = $amount;
$row['product_cost'] = $cost * $quantity;
$row['reversal_order_cost'] = $amount < 0 ? $cost * abs($quantity) : 0.0;
$row['merchant_discount'] = (float) ($row['merchant_discount'] ?? 0);
$row['freight_income'] = (float) ($row['freight_income'] ?? 0);
$row['payment_amount'] = $payment;
$row['platform_discount'] = $sellerIncome > $payment ? $sellerIncome - $payment : 0.0;
$row['reversal_order_amount'] = $amount < 0 ? abs($amount) : 0.0;
$row['actual_shipped_amount'] = $amount;
$row['actual_shipped_quantity'] = $quantity;
$row['actual_shipped_cost'] = $cost * $quantity;
foreach (['pre', 'post'] as $phase) {
foreach (['application_quantity', 'application_amount', 'application_cost', 'return_quantity',
'return_amount', 'return_cost', 'refund_amount', 'refund_cost', 'platform_subsidy'] as $field) {
$row[$phase . '_' . $field] = 0.0;
}
}
$row['defective_quantity'] = 0.0;
$row['is_gift'] = (int) ($row['is_gift'] ?? 0) === 1 ? '是' : '否';
$row['combine_type'] = empty($row['src_combine_sku_id']) ? '非组合' : '组合装';
$row['shipping_warehouse_code'] = $row['send_warehouse'] ?: ($row['wms_co_id'] ?? null);
$row['product_link'] = $this->productLink(
(string) ($row['shop_site'] ?? ''), (string) ($row['shop_name'] ?? ''),
(string) ($row['shop_style_code'] ?? ''),
);
foreach (['offline_customer_code', 'offline_customer_name', 'unique_code', 'registration_time',
'confirmation_time', 'warehouse_entry_time', 'classification_order',
'receiving_warehouse_code', 'receiving_warehouse'] as $field) {
$row[$field] = null;
}
foreach ($this->expenseFields() as $field) {
$row[$field] = null;
}
unset($row['shop_site'], $row['unit_cost'], $row['sku_name'], $row['seller_income_amount'],
$row['src_combine_sku_id'], $row['send_warehouse'], $row['wms_co_id']);
return $row;
}
/** @param array<int,array<string,mixed>> $rows */
private function hydrateRefunds(array &$rows): void
{
$orderIds = array_values(array_unique(array_filter(array_column($rows, 'internal_order_no'))));
if ($orderIds === []) {
return;
}
$refunds = $this->connection->table('ops_jst_return_refund_info as a')
->join('ops_jst_return_refund_item as ri', 'ri.as_id', '=', 'a.as_id')
->whereIn('a.o_id', $orderIds)
->orderByDesc('a.as_date')->orderByDesc('a.as_id')
->get([
'a.o_id as internal_order_no', 'a.as_id as after_sales_id',
'a.outer_as_id as outside_after_sales_id', 'a.as_date as after_sales_time',
'a.type as refund_type', 'a.question_type', 'a.labels as after_sales_tags',
'a.refund as refund_amount', 'a.free_amount as platform_subsidy',
'ri.outer_oi_id as sub_order_no', 'ri.sku_id as product_code',
'ri.type as detail_return_type', 'ri.qty as application_quantity',
'ri.amount as application_amount', 'ri.r_qty as return_quantity',
'ri.shop_amount as return_amount', 'ri.defective_qty as defective_quantity',
])->groupBy('internal_order_no');
foreach ($rows as &$row) {
$matches = collect($refunds->get($row['internal_order_no'], collect()))->filter(
static function (object $refund) use ($row): bool {
return (string) ($refund->sub_order_no ?? '') !== ''
? (string) $refund->sub_order_no === (string) ($row['sub_order_no'] ?? '')
: (string) ($refund->product_code ?? '') === (string) ($row['product_code'] ?? '');
},
)->values();
$first = $matches->first();
if ($first !== null) {
$row['after_sales_no'] = $first->outside_after_sales_id ?: $first->after_sales_id;
$row['refund_type'] = $first->refund_type;
$row['question_type'] = $first->question_type;
$row['detail_return_type'] = $first->detail_return_type;
$row['after_sales_tags'] = $first->after_sales_tags;
}
$unitCost = $row['product_quantity'] == 0 ? 0 : $row['product_cost'] / $row['product_quantity'];
$preReturned = 0.0;
foreach ($matches as $refund) {
$phase = empty($row['send_time']) || (string) $refund->after_sales_time < (string) $row['send_time']
? 'pre' : 'post';
$application = (float) ($refund->application_quantity ?? 0);
$returned = (float) ($refund->return_quantity ?? 0);
$row[$phase . '_application_quantity'] += $application;
$row[$phase . '_application_amount'] += (float) ($refund->application_amount ?? 0);
$row[$phase . '_application_cost'] += $application * $unitCost;
$row[$phase . '_return_quantity'] += $returned;
$row[$phase . '_return_amount'] += (float) ($refund->return_amount ?? 0);
$row[$phase . '_return_cost'] += $returned * $unitCost;
$row[$phase . '_refund_amount'] += (float) ($refund->refund_amount ?? 0);
$row[$phase . '_refund_cost'] += ($application - $returned) * $unitCost;
$row[$phase . '_platform_subsidy'] += (float) ($refund->platform_subsidy ?? 0);
$row['defective_quantity'] += (float) ($refund->defective_quantity ?? 0);
if ($phase === 'pre') {
$preReturned += $returned;
}
}
$row['actual_shipped_quantity'] = $row['product_quantity'] - $preReturned;
$row['actual_shipped_amount'] = $row['product_amount'] - $row['pre_return_amount'];
$row['actual_shipped_cost'] = $row['actual_shipped_quantity'] * $unitCost;
}
unset($row);
}
private function productLink(string $site, string $shopName, string $itemId): ?string
{
if ($itemId === '') {
return null;
}
$platform = mb_strtolower($site . ' ' . $shopName);
if (str_contains($platform, '京东') || str_contains($platform, '京喜') || str_contains($platform, 'jd')) {
return 'https://item.jd.com/' . rawurlencode($itemId) . '.html';
}
if (str_contains($platform, '抖音')) {
return 'https://haohuo.jinritemai.com/ecommerce/trade/detail/index.html?id=' . rawurlencode($itemId);
}
return null;
}
/** @return list<string> */
private function expenseFields(): array
{
return [
'account_500112', 'account_60031', 'account_800101', 'account_800103',
'account_800104', 'account_800105', 'account_800106', 'account_800107',
'account_800108', 'account_800109', 'account_800110', 'account_800114',
'account_8001201', 'account_8001202', 'account_800122', 'account_800125',
'account_800127', 'account_800129', 'account_800130', 'account_800132',
'account_800304', 'account_800501', 'account_800503',
];
}
private function contains(string $value): string
{
return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], trim($value)) . '%';
}
}

View File

@ -4,47 +4,49 @@ namespace Modules\EcommerceFinance\Repositories;
use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\DatabaseManager;
use Illuminate\Support\Facades\Schema;
use RuntimeException;
/**
* 将经营分析副本库中的店铺资料与本地账单 SQLite 组合为店铺维度报表。
*
* SQLite 只读保存店铺生命周期和平台账单;生命周期区间统一使用
* [effective_at, expires_at),避免主体切换日的账单被重复统计。
*/
/** 店铺维度报表只读取团队共享业务镜像。 */
class StoreReportRepository
{
// 保留导入命令使用的数据集键;报表页面本身读取本地 SQLite。
public const BILLS_DATASET = 'store-report.platform-bills';
public const LIFECYCLES_DATASET = 'store-report.store-lifecycles';
private const DEFAULT_EFFECTIVE_AT = '2026-05-01';
private const DEFAULT_EXPIRES_AT = '2999-01-01';
private const REPORT_YEAR = '2026';
private ?ConnectionInterface $connection = null;
private readonly ConnectionInterface $connection;
public function __construct(private readonly DatabaseManager $database) {}
public function __construct(DatabaseManager $databases)
{
$this->connection = $databases->connection(
(string) config('upstream_mirrors.connection', 'business_mirror'),
);
}
/**
* @param array<int,array<string,mixed>> $stores
* @return array<int,array<string,mixed>>
*/
public function rows(array $stores): array
public function rows(array $stores, ?int $reportYear = null): array
{
if ($stores === []) {
return [];
}
$this->assertDatabaseFile();
$this->connection ??= $this->database->connection('bills_sqlite');
[$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);
$lifecycleSales = $this->lifecycleSales($names);
$fallbackSales = $this->fallbackSales(array_values(array_diff($names, array_keys($lifecycles))));
$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) {
@ -55,7 +57,7 @@ class StoreReportRepository
$periods = $lifecycles[$storeName] ?? [[
'company_name' => $store['company_name'] ?? null,
'effective_at' => self::DEFAULT_EFFECTIVE_AT,
'effective_at' => $store['opened_at'] ?? sprintf('%d-01-01', $reportYear),
'expires_at' => self::DEFAULT_EXPIRES_AT,
'lifecycle_id' => null,
]];
@ -63,22 +65,35 @@ class StoreReportRepository
foreach ($periods as $period) {
$saleData = $period['lifecycle_id'] === null
? ($fallbackSales[$storeName] ?? ['period' => 0.0, 'months' => []])
: ($lifecycleSales[(string) $period['lifecycle_id']] ?? ['period' => 0.0, 'months' => []]);
: $this->combinedSales($lifecycleSales, (array) ($period['lifecycle_ids'] ?? [$period['lifecycle_id']]));
$monthly = [];
for ($month = 1; $month <= 12; $month++) {
$monthly[sprintf('sales_%02d', $month)] = (float) (
$saleData['months'][sprintf('%s-%02d', self::REPORT_YEAR, $month)] ?? 0
$saleData['months'][sprintf('%d-%02d', $reportYear, $month)] ?? 0
);
}
$rollingMonths = collect(range(12, 1))->map(
static fn (int $monthsAgo): string => now()->startOfMonth()->subMonthsNoOverflow($monthsAgo)->format('Y-m'),
)->all();
$recentMonths = collect(range(3, 1))->map(
static fn (int $monthsAgo): string => now()->startOfMonth()->subMonthsNoOverflow($monthsAgo)->format('Y-m'),
)->values()->all();
$rows[] = [
...$store,
'store_id' => $storeName,
'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, self::DEFAULT_EFFECTIVE_AT),
'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'],
'store_sales_12m' => array_sum(array_map(
static fn (string $month): float => (float) ($saleData['months'][$month] ?? 0),
$rollingMonths,
)),
'recent_three_month_sales_1' => (float) ($saleData['months'][$recentMonths[0]] ?? 0),
'recent_three_month_sales_2' => (float) ($saleData['months'][$recentMonths[1]] ?? 0),
'recent_three_month_sales_3' => (float) ($saleData['months'][$recentMonths[2]] ?? 0),
...$monthly,
];
}
@ -87,24 +102,43 @@ class StoreReportRepository
return $rows;
}
private function assertDatabaseFile(): void
/** @return array{0:string,1:string} */
private function activeGenerations(): array
{
$path = (string) config('database.connections.bills_sqlite.database');
if ($path === '' || ! is_file($path)) {
throw new RuntimeException('店铺账单 SQLite 尚未挂载:' . $path);
$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', 'row_count'])
->keyBy('dataset_key');
foreach ([self::BILLS_DATASET, self::LIFECYCLES_DATASET] as $dataset) {
$state = $states->get($dataset);
if ($state === null || empty($state->active_generation)) {
$label = $dataset === self::BILLS_DATASET ? '账单' : '主体历史';
$rows = number_format((int) ($state->row_count ?? 0));
throw new RuntimeException("店铺维度报表中央同步中:{$label}已写入 {$rows} 行,完整数据就绪后自动显示。");
}
}
return [
(string) $states->get(self::BILLS_DATASET)->active_generation,
(string) $states->get(self::LIFECYCLES_DATASET)->active_generation,
];
}
/** @return array<string,array<int,array<string,mixed>>> */
private function lifecycles(array $names): array
private function lifecycles(array $names, string $generation): array
{
if (! $this->hasTable('store_lifecycles')) {
return [];
}
$result = [];
foreach (array_chunk($names, 900) as $chunk) {
$rows = $this->connection->table('store_lifecycles')
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']);
@ -119,31 +153,55 @@ class StoreReportRepository
}
}
foreach ($result as $storeName => $periods) {
$normalized = [];
foreach ($periods as $period) {
$last = array_key_last($normalized);
if ($last !== null
&& (string) ($normalized[$last]['company_name'] ?? '') === (string) ($period['company_name'] ?? '')) {
$normalized[$last]['lifecycle_ids'][] = $period['lifecycle_id'];
continue;
}
$period['lifecycle_ids'] = [$period['lifecycle_id']];
$normalized[] = $period;
}
foreach ($normalized as $index => &$period) {
$period['expires_at'] = $period['expires_at']
?: ($normalized[$index + 1]['effective_at'] ?? self::DEFAULT_EXPIRES_AT);
}
unset($period);
$result[$storeName] = $normalized;
}
return $result;
}
/** @return array<string,array{period:float,months:array<string,float>}> */
private function lifecycleSales(array $names): array
private function lifecycleSales(array $names, string $billGeneration, string $lifecycleGeneration): array
{
if ($names === [] || ! $this->hasTable('store_lifecycles') || ! $this->hasTable('bills')) {
return [];
}
$result = [];
foreach (array_chunk($names, 900) as $chunk) {
$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, substr(b.occurred_at, 1, 7) AS bill_month,
"WITH lifecycle_periods AS (
SELECT id, erp_store_name, effective_at,
COALESCE(expires_at,
LEAD(effective_at) OVER (PARTITION BY erp_store_name ORDER BY effective_at, id),
'2999-01-01 00:00:00') AS computed_expires_at
FROM store_report_lifecycle_rows
WHERE generation = ?
)
SELECT l.id AS lifecycle_id, {$month} AS bill_month,
SUM(COALESCE(b.income, 0)) AS income_total
FROM store_lifecycles AS l
INNER JOIN bills AS b
ON b.store_name = l.erp_store_name
AND b.occurred_at IS NOT NULL AND b.occurred_at <> ''
AND b.occurred_at >= COALESCE(NULLIF(l.effective_at, ''), ?)
AND b.occurred_at < COALESCE(NULLIF(l.expires_at, ''), ?)
FROM lifecycle_periods 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 < l.computed_expires_at
WHERE l.erp_store_name IN ($placeholders)
GROUP BY l.id, substr(b.occurred_at, 1, 7)",
[self::DEFAULT_EFFECTIVE_AT, self::DEFAULT_EXPIRES_AT, ...$chunk],
GROUP BY l.id, {$month}",
[$lifecycleGeneration, $billGeneration, ...$chunk],
);
$this->collectSales($result, $rows, 'lifecycle_id');
}
@ -152,24 +210,19 @@ class StoreReportRepository
}
/** @return array<string,array{period:float,months:array<string,float>}> */
private function fallbackSales(array $names): array
private function fallbackSales(array $names, string $generation): array
{
if ($names === [] || ! $this->hasTable('bills')) {
return [];
}
$result = [];
foreach (array_chunk($names, 900) as $chunk) {
$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, substr(occurred_at, 1, 7) AS bill_month,
"SELECT store_name, {$month} AS bill_month,
SUM(COALESCE(income, 0)) AS income_total
FROM bills
WHERE store_name IN ($placeholders)
AND occurred_at IS NOT NULL AND occurred_at <> ''
AND occurred_at >= ? AND occurred_at < ?
GROUP BY store_name, substr(occurred_at, 1, 7)",
[...$chunk, self::DEFAULT_EFFECTIVE_AT, self::DEFAULT_EXPIRES_AT],
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');
}
@ -187,13 +240,25 @@ class StoreReportRepository
}
}
/** SQLite Schema Builder 在只读/WAL 挂载下可能尝试写入,因此使用普通查询。 */
private function hasTable(string $table): bool
/** @param array<string,array{period:float,months:array<string,float>}> $sales */
private function combinedSales(array $sales, array $ids): array
{
return $this->connection->select(
"SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1",
[$table],
) !== [];
$combined = ['period' => 0.0, 'months' => []];
foreach ($ids as $id) {
$row = $sales[(string) $id] ?? ['period' => 0.0, 'months' => []];
$combined['period'] += (float) $row['period'];
foreach ($row['months'] as $month => $amount) {
$combined['months'][$month] = ($combined['months'][$month] ?? 0.0) + (float) $amount;
}
}
return $combined;
}
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

View File

@ -0,0 +1,71 @@
<?php
namespace Modules\EcommerceFinance\Repositories;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
class StoreReportUpstreamRepository
{
private const SOURCE = 'jiansun';
private ?UpstreamReadOnlyQuery $query = null;
public function __construct(private readonly UpstreamReadOnlyQueryFactory $queries)
{
}
/** @return array<int,array<string,mixed>> */
public function billBatch(int $afterId, int $limit): array
{
$limit = min(max($limit, 1), 10000);
return $this->query()->select(<<<SQL
SELECT TOP {$limit}
bill.Id AS source_id,
bill.shop_id,
bill.channel,
bill.fee_time,
bill.del_flag,
bill.primary_key,
bill.updateTime AS source_updated_at,
bill.fee_name,
bill.inout_type,
bill.amount,
shop.shop_name
FROM dbo.jst_originalBillInfo AS bill
LEFT JOIN dbo.jst_shopInfo AS shop ON shop.shop_id = bill.shop_id
WHERE bill.Id > ?
ORDER BY bill.Id
SQL, [$afterId]);
}
/** @return array<int,array<string,mixed>> */
public function lifecycleBatch(int $afterId, int $limit): array
{
$limit = min(max($limit, 1), 5000);
return $this->query()->select(<<<SQL
SELECT TOP {$limit}
flow.Id AS source_id,
flow.erpShopName AS erp_store_name,
flow.unifiedSocialCreditCode AS unified_social_credit_code,
flow.applyTime AS effective_at,
company.businessLicenseName AS company_name
FROM dbo.fys_shopInfo_flow AS flow
OUTER APPLY (
SELECT TOP 1 info.businessLicenseName
FROM dbo.fys_companyInfo AS info
WHERE info.unifiedSocialCreditCode = NULLIF(flow.unifiedSocialCreditCode, '')
ORDER BY info.Id DESC
) AS company
WHERE flow.Id > ?
ORDER BY flow.Id
SQL, [$afterId]);
}
private function query(): UpstreamReadOnlyQuery
{
return $this->query ??= $this->queries->forSource(self::SOURCE);
}
}

View File

@ -1,344 +0,0 @@
<?php
namespace Modules\EcommerceFinance\Services;
use DateTimeInterface;
use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\DatabaseManager;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Schema;
use Illuminate\Support\Str;
use Modules\EcommerceFinance\Repositories\StoreReportRepository;
use PhpOffice\PhpSpreadsheet\IOFactory;
use PhpOffice\PhpSpreadsheet\Shared\Date as ExcelDate;
use RuntimeException;
use Throwable;
class StoreReportDataImporter
{
private const PLATFORMS = [
'dewu' => ['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<int,string> */
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<array{0:array<string,int>,1:array<int,mixed>}> */
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(),
]);
}
}

View File

@ -0,0 +1,357 @@
<?php
namespace Modules\EcommerceFinance\Services;
use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\DatabaseManager;
use Illuminate\Support\Facades\Schema;
use Illuminate\Support\Str;
use Modules\EcommerceFinance\Repositories\StoreReportRepository;
use Modules\EcommerceFinance\Repositories\StoreReportUpstreamRepository;
use RuntimeException;
use Throwable;
class StoreReportMirrorSynchronizer
{
private readonly ConnectionInterface $connection;
public function __construct(
private readonly StoreReportUpstreamRepository $upstream,
DatabaseManager $databases,
) {
$this->connection = $databases->connection(
(string) config('upstream_mirrors.connection', 'business_mirror'),
);
}
/** @return array{lifecycles:array<string,mixed>,bills:array<string,mixed>} */
public function sync(?int $batchSize = null, ?int $maxBatches = null): array
{
$this->assertSchemaReady();
$lifecycleBatch = min(max((int) config('upstream_mirrors.store_report.lifecycle_batch_size', 1000), 100), 5000);
$lifecycleMax = max((int) config('upstream_mirrors.store_report.lifecycle_max_batches_per_run', 5), 1);
$billBatch = min(max($batchSize ?? (int) config('upstream_mirrors.store_report.batch_size', 5000), 100), 10000);
$billMax = max($maxBatches ?? (int) config('upstream_mirrors.store_report.max_batches_per_run', 4), 1);
return [
'lifecycles' => $this->syncDataset(
StoreReportRepository::LIFECYCLES_DATASET,
'store_report_lifecycle_rows',
$lifecycleBatch,
$lifecycleMax,
fn (int $cursor, int $limit): array => $this->upstream->lifecycleBatch($cursor, $limit),
fn (array $rows, string $generation): array => $this->mapLifecycles($rows, $generation),
),
'bills' => $this->syncDataset(
StoreReportRepository::BILLS_DATASET,
'store_report_bill_rows',
$billBatch,
$billMax,
fn (int $cursor, int $limit): array => $this->upstream->billBatch($cursor, $limit),
fn (array $rows, string $generation): array => $this->mapBills($rows, $generation),
),
];
}
/**
* @param callable(int,int):array<int,array<string,mixed>> $fetch
* @param callable(array<int,array<string,mixed>>,string):array{upserts:array<int,array<string,mixed>>,deletes:array<int,int>} $map
* @return array{status:string,mode:string,batches:int,source_rows:int,stored_rows:int,cursor_id:int}
*/
private function syncDataset(
string $dataset,
string $table,
int $batchSize,
int $maxBatches,
callable $fetch,
callable $map,
): array {
$state = $this->state($dataset);
$initial = empty($state->active_generation);
$mode = $initial ? 'full' : 'incremental';
$resume = in_array((string) $state->status, ['running', 'failed'], true)
&& (string) ($state->run_mode ?? '') === $mode;
$generation = $initial
? ($resume && ! empty($state->pending_generation)
? (string) $state->pending_generation
: (string) Str::uuid())
: (string) $state->active_generation;
$cursor = $resume
? (int) ($state->cursor_id ?? 0)
: ($initial
? 0
: max((int) ($state->cursor_id ?? 0)
- max((int) config('upstream_mirrors.store_report.overlap_rows', 50000), 0), 0));
$storedRows = $initial && ! $resume ? 0 : (int) $state->row_count;
$sourceRows = 0;
$batches = 0;
$this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([
'status' => 'running',
'run_mode' => $mode,
'pending_generation' => $initial ? $generation : null,
'cursor_id' => $cursor,
'last_attempted_at' => now(),
'last_error_code' => null,
'updated_at' => now(),
]);
try {
while ($batches < $maxBatches) {
$rows = $fetch($cursor, $batchSize);
if ($rows === []) {
return $this->complete($dataset, $generation, $mode, $cursor, $storedRows, $batches, $sourceRows);
}
$nextCursor = max(array_map(static fn (array $row): int => (int) ($row['source_id'] ?? 0), $rows));
if ($nextCursor <= $cursor) {
throw new RuntimeException('store_report_cursor_did_not_advance');
}
$mapped = $map($rows, $generation);
$this->storeBatch($dataset, $table, $generation, $nextCursor, $mapped, $initial, $storedRows);
$cursor = $nextCursor;
$sourceRows += count($rows);
$storedRows += $initial ? count($mapped['upserts']) : 0;
$batches++;
if (count($rows) < $batchSize) {
return $this->complete($dataset, $generation, $mode, $cursor, $storedRows, $batches, $sourceRows);
}
}
return $this->result('running', $mode, $batches, $sourceRows, $storedRows, $cursor);
} catch (Throwable $exception) {
$this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([
'status' => 'failed',
'last_error_code' => class_basename($exception),
'updated_at' => now(),
]);
throw $exception;
}
}
private function state(string $dataset): object
{
$now = now();
$this->connection->table('store_report_import_states')->insertOrIgnore([
'dataset_key' => $dataset,
'status' => 'pending',
'row_count' => 0,
'source_files_json' => json_encode(['jiansun.dbo.jst_originalBillInfo', 'jiansun.dbo.fys_shopInfo_flow']),
'created_at' => $now,
'updated_at' => $now,
]);
return $this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->first();
}
/** @param array{upserts:array<int,array<string,mixed>>,deletes:array<int,int>} $mapped */
private function storeBatch(
string $dataset,
string $table,
string $generation,
int $cursor,
array $mapped,
bool $initial,
int $storedRows,
): void {
$this->connection->transaction(function () use (
$dataset, $table, $generation, $cursor, $mapped, $initial, $storedRows,
): void {
if ($mapped['deletes'] !== []) {
$this->connection->table($table)
->where('generation', $generation)
->whereIn('source_id', $mapped['deletes'])
->delete();
}
if ($mapped['upserts'] !== []) {
$updates = array_values(array_diff(array_keys($mapped['upserts'][0]), ['generation', 'source_id']));
// MySQL/MariaDB 对单条预处理语句的占位符数量有限,不能把
// 数千行、十余列一次塞进同一条 upsert。
$columnsPerRow = max(count($mapped['upserts'][0]), 1);
$chunkSize = max(1, min(2000, intdiv(50000, $columnsPerRow)));
foreach (array_chunk($mapped['upserts'], $chunkSize) as $chunk) {
$this->connection->table($table)->upsert(
$chunk, ['generation', 'source_id'], $updates,
);
}
}
$this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update([
'cursor_id' => $cursor,
'row_count' => $initial ? $storedRows + count($mapped['upserts']) : $storedRows,
'updated_at' => now(),
]);
});
}
/** @return array{status:string,mode:string,batches:int,source_rows:int,stored_rows:int,cursor_id:int} */
private function complete(
string $dataset,
string $generation,
string $mode,
int $cursor,
int $storedRows,
int $batches,
int $sourceRows,
): array {
$now = now();
$actualRows = (int) $this->connection->table(
$dataset === StoreReportRepository::BILLS_DATASET
? 'store_report_bill_rows'
: 'store_report_lifecycle_rows',
)->where('generation', $generation)->count();
$values = [
'status' => 'succeeded',
'run_mode' => null,
'cursor_id' => $cursor,
'last_succeeded_at' => $now,
'last_error_code' => null,
'row_count' => $actualRows,
'updated_at' => $now,
];
if ($mode === 'full') {
$values += [
'active_generation' => $generation,
'pending_generation' => null,
'initial_completed_at' => $now,
];
}
$this->connection->table('store_report_import_states')->where('dataset_key', $dataset)->update($values);
return $this->result('succeeded', $mode, $batches, $sourceRows, $actualRows, $cursor);
}
/** @return array{status:string,mode:string,batches:int,source_rows:int,stored_rows:int,cursor_id:int} */
private function result(string $status, string $mode, int $batches, int $sourceRows, int $storedRows, int $cursor): array
{
return [
'status' => $status,
'mode' => $mode,
'batches' => $batches,
'source_rows' => $sourceRows,
'stored_rows' => $storedRows,
'cursor_id' => $cursor,
];
}
/** @return array{upserts:array<int,array<string,mixed>>,deletes:array<int,int>} */
private function mapBills(array $rows, string $generation): array
{
$upserts = [];
$deletes = [];
foreach ($rows as $row) {
$sourceId = (int) ($row['source_id'] ?? 0);
$storeName = $this->text($row['shop_name'] ?? null, 191);
$occurredAt = $this->dateTime($row['fee_time'] ?? null);
$direction = (string) ($row['inout_type'] ?? '');
if ($sourceId < 1 || (int) ($row['del_flag'] ?? 0) !== 0 || $storeName === null
|| $occurredAt === null || ! in_array($direction, ['1', '-1'], true)) {
if ($sourceId > 0) {
$deletes[] = $sourceId;
}
continue;
}
$amount = abs((float) ($row['amount'] ?? 0));
$platform = $this->text($row['channel'] ?? null, 32) ?? 'unknown';
$business = [$sourceId, $platform, $storeName, $occurredAt, $direction, $amount];
$upserts[] = [
'generation' => $generation,
'source_id' => $sourceId,
'shop_id' => (int) ($row['shop_id'] ?? 0) ?: null,
'platform' => $platform,
'platform_label' => $this->platformLabel($platform),
'store_name' => $storeName,
'occurred_at' => $occurredAt,
'income' => $direction === '1' ? $amount : 0,
'expense' => $direction === '-1' ? $amount : 0,
'bill_item' => $this->text($row['fee_name'] ?? null, 255),
'record_code' => (string) $sourceId,
'source_file' => 'jiansun.dbo.jst_originalBillInfo',
'row_hash' => hash('sha256', json_encode($business, JSON_UNESCAPED_UNICODE)),
'source_updated_at' => $this->dateTime($row['source_updated_at'] ?? null),
'imported_at' => now(),
];
}
return ['upserts' => $upserts, 'deletes' => $deletes];
}
/** @return array{upserts:array<int,array<string,mixed>>,deletes:array<int,int>} */
private function mapLifecycles(array $rows, string $generation): array
{
$upserts = [];
$deletes = [];
foreach ($rows as $row) {
$sourceId = (int) ($row['source_id'] ?? 0);
$storeName = $this->text($row['erp_store_name'] ?? null, 191);
$creditCode = $this->text($row['unified_social_credit_code'] ?? null, 64);
$effectiveAt = $this->dateTime($row['effective_at'] ?? null);
if ($sourceId < 1 || $storeName === null || $creditCode === null || $effectiveAt === null) {
if ($sourceId > 0) {
$deletes[] = $sourceId;
}
continue;
}
$business = [$sourceId, $storeName, $creditCode, $effectiveAt];
$upserts[] = [
'generation' => $generation,
'source_id' => $sourceId,
'business_key' => hash('sha256', (string) $sourceId),
'erp_store_name' => $storeName,
'company_name' => $this->text($row['company_name'] ?? null, 191),
'unified_social_credit_code' => $creditCode,
'effective_at' => $effectiveAt,
'expires_at' => null,
'source_file' => 'jiansun.dbo.fys_shopInfo_flow',
'row_hash' => hash('sha256', json_encode($business, JSON_UNESCAPED_UNICODE)),
'imported_at' => now(),
];
}
return ['upserts' => $upserts, 'deletes' => $deletes];
}
private function platformLabel(string $channel): string
{
return match ($channel) {
'alipayCSV' => '淘系',
'TouTiaoFXGZJ' => '抖音',
'PinduoduoImport' => '拼多多',
'Jos' => '京东',
default => $channel,
};
}
private function text(mixed $value, int $length): ?string
{
$value = trim((string) ($value ?? ''));
if ($value === '') {
return null;
}
if (! mb_check_encoding($value, 'UTF-8')) {
$value = mb_convert_encoding($value, 'UTF-8', 'GB18030');
}
return mb_substr($value, 0, $length);
}
private function dateTime(mixed $value): ?string
{
$value = trim((string) ($value ?? ''));
return $value === '' ? null : str_replace('T', ' ', substr($value, 0, 26));
}
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('store_report_mirror_schema_missing:' . $table);
}
}
foreach (['run_mode', 'cursor_id', 'initial_completed_at'] as $column) {
if (! $schema->hasColumn('store_report_import_states', $column)) {
throw new RuntimeException('store_report_mirror_upgrade_missing:' . $column);
}
}
}
}

View File

@ -5,6 +5,7 @@ namespace Modules\EcommerceFinance\View;
use Illuminate\Pagination\Paginator;
use Illuminate\Support\Collection;
use Modules\EcommerceFinance\Repositories\OrderDataRepository;
use Modules\EcommerceFinance\Repositories\OrderDataMirrorRepository;
use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
use Throwable;
@ -16,6 +17,7 @@ class OrderDataPageData
public function __construct(
private readonly OrderDataRepository $orders,
private readonly UpstreamQueryCacheFactory $caches,
private readonly ?OrderDataMirrorRepository $mirrorOrders = null,
) {
}
@ -29,9 +31,21 @@ class OrderDataPageData
'sku_id' => $this->text(request('sku_id')),
];
[$page, $limit] = $this->pageSettings();
$cache = $this->caches->forSource(self::SOURCE);
try {
$mirror = $this->mirrorOrders ?? app(OrderDataMirrorRepository::class);
if ($mirror->ready()) {
$result = $mirror->afterSales($filters, $page, $limit);
return [
'orderDataRows' => $this->paginator((array) ($result['rows'] ?? []), $page, $limit),
'orderDataFilters' => $filters,
'orderDataStatus' => ['X-Caiwu-Data-Status' => 'mirror'],
'orderDataError' => null,
];
}
$cache = $this->caches->forSource(self::SOURCE);
$result = $cache->remember(
'order-data.after-sales.list',
compact('filters', 'page', 'limit'),
@ -61,9 +75,21 @@ class OrderDataPageData
'sku_id' => $this->text(request('sku_id')),
];
[$page, $limit] = $this->pageSettings();
$cache = $this->caches->forSource(self::SOURCE);
try {
$mirror = $this->mirrorOrders ?? app(OrderDataMirrorRepository::class);
if ($mirror->promotionReady()) {
$result = $mirror->promotionExpenses($filters, $page, $limit);
return [
'orderDataRows' => $this->paginator((array) ($result['rows'] ?? []), $page, $limit),
'orderDataFilters' => array_replace($filters, ['date' => $result['date'] ?? $filters['date']]),
'orderDataStatus' => ['X-Caiwu-Data-Status' => 'mirror'],
'orderDataError' => null,
];
}
$cache = $this->caches->forSource(self::SOURCE);
$result = $cache->remember(
'order-data.promotion.list',
compact('filters', 'page', 'limit'),

View File

@ -1,208 +0,0 @@
#!/usr/bin/env python3
"""将平台账单 Excel 导入本地 SQLite,供后续按店铺和时间汇总。
示例(仓库根目录):
python3 scripts/import_bills_sqlite.py
python3 scripts/import_bills_sqlite.py --input-dir /path/to/账单数据 \
--db runtime/bills.sqlite --reset
运行环境需要 openpyxl;项目桌面运行时可使用 Codex bundled Python。
"""
from __future__ import annotations
import argparse
import json
import re
import sqlite3
import sys
from datetime import date, datetime, time, timedelta
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any, Iterable
PLATFORMS: tuple[tuple[str, str, tuple[str, ...]], ...] = (
("dewu", "得物", ("得物",)),
("douyin", "抖音", ("头条放心购", "抖音")),
("taobao", "淘系", ("淘系",)),
("jd", "京东", ("京东商城", "京东")),
("pinduoduo", "拼多多", ("拼多多",)),
)
TIME_HEADERS = {
"dewu": "支付/退款时间",
"douyin": "结算时间",
"taobao": "入账时间",
"jd": "费用结算时间",
"pinduoduo": "发生时间",
}
def platform_for_file(path: Path) -> tuple[str, str]:
for key, label, markers in PLATFORMS:
if any(marker in path.name for marker in markers):
return key, label
raise ValueError(f"无法从文件名识别平台:{path.name}")
def clean(value: Any) -> str | None:
if value is None:
return None
text = str(value).strip()
return text or None
def money(value: Any) -> str:
if value is None or value == "":
return "0.00"
text = str(value).replace(",", "").replace("¥", "").replace("¥", "").strip()
try:
return format(Decimal(text), ".2f")
except InvalidOperation as exc:
raise ValueError(f"金额无法解析:{value!r}") from exc
def iso_time(value: Any) -> str | None:
if value is None or value == "":
return None
if isinstance(value, datetime):
return value.strftime("%Y-%m-%d %H:%M:%S")
if isinstance(value, date):
return datetime.combine(value, time()).strftime("%Y-%m-%d %H:%M:%S")
if isinstance(value, (int, float)):
# Excel 1900 date system; openpyxl normally converts typed date cells,
# but this covers numeric exports as well.
converted = datetime(1899, 12, 30) + timedelta(days=float(value))
return converted.strftime("%Y-%m-%d %H:%M:%S")
text = str(value).strip().replace("/", "-")
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"):
try:
return datetime.strptime(text, fmt).strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
continue
# 得物偶尔出现 2026/8/13 14:11:19 这种非零填充形式。
match = re.fullmatch(r"(\d{4})-(\d{1,2})-(\d{1,2})(?:\s+|T)(\d{1,2}):(\d{2})(?::(\d{2}))?", text)
if match:
y, m, d, hh, mm, ss = (int(x or 0) for x in match.groups())
return f"{y:04d}-{m:02d}-{d:02d} {hh:02d}:{mm:02d}:{ss:02d}"
raise ValueError(f"时间无法解析:{value!r}")
def json_default(value: Any) -> str:
if isinstance(value, (datetime, date)):
return value.isoformat(sep=" ")
return str(value)
def init_db(connection: sqlite3.Connection) -> None:
connection.executescript(
"""
PRAGMA journal_mode = WAL;
CREATE TABLE IF NOT EXISTS bills (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
type TEXT NOT NULL,
store_name TEXT NOT NULL,
occurred_at TEXT,
income NUMERIC NOT NULL DEFAULT 0,
expense NUMERIC NOT NULL DEFAULT 0,
bill_item TEXT,
record_code TEXT NOT NULL,
source_file TEXT NOT NULL,
raw_json TEXT NOT NULL,
imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(platform, record_code)
);
CREATE INDEX IF NOT EXISTS idx_bills_store_time
ON bills(store_name, occurred_at);
CREATE INDEX IF NOT EXISTS idx_bills_platform_store_time
ON bills(platform, store_name, occurred_at);
CREATE VIEW IF NOT EXISTS bill_daily_summary AS
SELECT platform, type, store_name, substr(occurred_at, 1, 10) AS bill_date,
ROUND(SUM(income), 2) AS income, ROUND(SUM(expense), 2) AS expense,
COUNT(*) AS row_count
FROM bills
WHERE occurred_at IS NOT NULL AND occurred_at <> ''
GROUP BY platform, type, store_name, substr(occurred_at, 1, 10);
"""
)
def rows_from_file(path: Path, platform: str, label: str) -> Iterable[tuple[Any, ...]]:
try:
import openpyxl
except ImportError as exc:
raise RuntimeError("缺少 openpyxl,请使用项目提供的 bundled Python 运行此脚本。") from exc
workbook = openpyxl.load_workbook(path, read_only=True, data_only=True)
try:
for sheet in workbook.worksheets:
iterator = sheet.iter_rows(values_only=True)
header_row = next(iterator, None)
if not header_row:
continue
headers = [clean(item) or "" for item in header_row]
required = {"店铺", "记录编码", TIME_HEADERS[platform], "收入金额" if platform != "douyin" else "收入", "支出金额" if platform != "douyin" else "支出"}
missing = required.difference(headers)
if missing:
raise ValueError(f"{path.name}/{sheet.title} 缺少列:{', '.join(sorted(missing))}")
index = {name: position for position, name in enumerate(headers)}
income_header = "收入" if platform == "douyin" else "收入金额"
expense_header = "支出" if platform == "douyin" else "支出金额"
for row in iterator:
if not row or all(item in (None, "") for item in row):
continue
get = lambda name: row[index[name]] if index[name] < len(row) else None
record_code = clean(get("记录编码"))
store_name = clean(get("店铺"))
if not record_code or not store_name:
continue
raw = {headers[i]: row[i] for i in range(min(len(headers), len(row))) if row[i] not in (None, "")}
yield (platform, label, store_name, iso_time(get(TIME_HEADERS[platform])), money(get(income_header)), money(get(expense_header)), clean(get("账单项目")), record_code, path.name, json.dumps(raw, ensure_ascii=False, default=json_default))
finally:
workbook.close()
def main() -> int:
parser = argparse.ArgumentParser(description="导入得物/抖音/淘系/京东/拼多多账单到 SQLite")
parser.add_argument("--input-dir", type=Path, default=Path("/Users/dodo/AI系统/账单数据"))
parser.add_argument("--db", type=Path, default=Path("runtime/bills.sqlite"))
parser.add_argument("--reset", action="store_true", help="清空已有 bills 后重新导入")
args = parser.parse_args()
files = sorted(args.input_dir.glob("*.xlsx"))
if not files:
print(f"未找到 Excel 文件:{args.input_dir}", file=sys.stderr)
return 2
args.db.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(args.db)
init_db(connection)
if args.reset:
connection.execute("DELETE FROM bills")
counts: dict[str, int] = {}
try:
for path in files:
platform, label = platform_for_file(path)
count = 0
for row in rows_from_file(path, platform, label):
connection.execute(
"""INSERT INTO bills(platform,type,store_name,occurred_at,income,expense,bill_item,record_code,source_file,raw_json)
VALUES(?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(platform,record_code) DO UPDATE SET
type=excluded.type, store_name=excluded.store_name, occurred_at=excluded.occurred_at,
income=excluded.income, expense=excluded.expense, bill_item=excluded.bill_item,
source_file=excluded.source_file, raw_json=excluded.raw_json, imported_at=CURRENT_TIMESTAMP""",
row,
)
count += 1
counts[platform] = counts.get(platform, 0) + count
connection.commit()
finally:
connection.close()
print(f"导入完成:{args.db}")
for platform, count in counts.items():
print(f" {platform}: {count} 条")
return 0
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -1,115 +0,0 @@
#!/usr/bin/env python3
"""导入店铺存续时间表到 runtime/bills.sqlite。"""
from __future__ import annotations
import argparse
import json
import sqlite3
from datetime import date, datetime, time, timedelta
from pathlib import Path
from typing import Any
def text(value: Any) -> str | None:
if value is None:
return None
value = str(value).strip()
return value or None
def sqlite_time(value: Any) -> str | None:
if value in (None, ""):
return None
if isinstance(value, datetime):
return value.strftime("%Y-%m-%d %H:%M:%S")
if isinstance(value, date):
return datetime.combine(value, time()).strftime("%Y-%m-%d %H:%M:%S")
if isinstance(value, (int, float)):
value = datetime(1899, 12, 30) + timedelta(days=float(value))
return value.strftime("%Y-%m-%d %H:%M:%S")
value = str(value).strip().replace("/", "-")
if value in {"存续中", "至今", "长期", "无"}:
return None
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
try:
return datetime.strptime(value, fmt).strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
pass
raise ValueError(f"时间无法解析:{value!r}")
def main() -> int:
parser = argparse.ArgumentParser(description="导入店铺存续时间 Excel 到 SQLite")
parser.add_argument("file", type=Path, default=Path("/Users/dodo/Documents/店铺存续时间.xlsx"), nargs="?")
parser.add_argument("--db", type=Path, default=Path("runtime/bills.sqlite"))
parser.add_argument("--reset", action="store_true", help="清空已有店铺存续时间后重新导入")
args = parser.parse_args()
try:
import openpyxl
except ImportError as exc:
raise SystemExit("缺少 openpyxl,请使用项目提供的 bundled Python 运行此脚本。") from exc
if not args.file.is_file():
raise SystemExit(f"文件不存在:{args.file}")
args.db.parent.mkdir(parents=True, exist_ok=True)
db = sqlite3.connect(args.db)
db.executescript("""
CREATE TABLE IF NOT EXISTS store_lifecycles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
erp_store_name TEXT NOT NULL,
company_name TEXT,
effective_at TEXT,
expires_at TEXT,
source_file TEXT NOT NULL,
raw_json TEXT NOT NULL,
imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE(erp_store_name, effective_at, expires_at)
);
CREATE INDEX IF NOT EXISTS idx_store_lifecycles_lookup
ON store_lifecycles(erp_store_name, effective_at, expires_at);
""")
if args.reset:
db.execute("DELETE FROM store_lifecycles")
workbook = openpyxl.load_workbook(args.file, read_only=True, data_only=True)
inserted = 0
try:
for sheet in workbook.worksheets:
rows = sheet.iter_rows(values_only=True)
header = next(rows, None)
if not header or all(value in (None, "") for value in header):
continue
headers = [text(value) or "" for value in header]
required = ["ERP系统店铺名称", "公司名称", "生效时间", "失效时间"]
missing = [name for name in required if name not in headers]
if missing:
raise SystemExit(f"{args.file.name}/{sheet.title} 缺少列:{', '.join(missing)}")
index = {name: headers.index(name) for name in required}
for row in rows:
if not row or all(value in (None, "") for value in row):
continue
get = lambda name: row[index[name]] if index[name] < len(row) else None
store = text(get("ERP系统店铺名称"))
if not store:
continue
raw = {headers[i]: row[i] for i in range(min(len(headers), len(row))) if row[i] not in (None, "")}
db.execute("""INSERT INTO store_lifecycles
(erp_store_name,company_name,effective_at,expires_at,source_file,raw_json)
VALUES (?,?,?,?,?,?)
ON CONFLICT(erp_store_name,effective_at,expires_at) DO UPDATE SET
company_name=excluded.company_name, source_file=excluded.source_file,
raw_json=excluded.raw_json, imported_at=CURRENT_TIMESTAMP""",
(store, text(get("公司名称")), sqlite_time(get("生效时间")),
sqlite_time(get("失效时间")), args.file.name,
json.dumps(raw, ensure_ascii=False, default=str)))
inserted += 1
finally:
workbook.close()
db.commit()
total = db.execute("SELECT COUNT(*) FROM store_lifecycles").fetchone()[0]
db.close()
print(f"导入完成:{args.db},本次处理 {inserted} 条,表内共 {total} 条")
return 0
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -201,7 +201,7 @@ class ManagementCrudTest extends FeatureTestCase
->assertSeeText('2999-01-01')
->assertSeeText('25,916.80')
->assertSeeText('关店中')
->assertSeeText('已接入真实账单与生命周期');
->assertSeeText('共享镜像');
preg_match_all('/<th(?:\s|>)/', $response->getContent(), $headers);
$this->assertCount(26, $headers[0]);

View File

@ -4,6 +4,7 @@ namespace Tests\Feature\Business;
use App\Models\Auth\Permission;
use Mockery;
use Modules\EcommerceFinance\Repositories\OrderDataMirrorRepository;
use Modules\EcommerceFinance\Services\UpstreamQueryCache;
use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
@ -11,6 +12,38 @@ use Tests\Feature\FeatureTestCase;
class OrderDataPagesUpstreamTest extends FeatureTestCase
{
protected function setUp(): void
{
parent::setUp();
$mirror = Mockery::mock(OrderDataMirrorRepository::class);
$mirror->shouldReceive('ready')->zeroOrMoreTimes()->andReturnFalse();
$mirror->shouldReceive('promotionReady')->zeroOrMoreTimes()->andReturnFalse();
$this->app->instance(OrderDataMirrorRepository::class, $mirror);
}
public function test_after_sales_prefers_complete_business_mirror_without_registering_a_snapshot(): void
{
$this->allowOrderDataPages();
$mirror = Mockery::mock(OrderDataMirrorRepository::class);
$mirror->shouldReceive('ready')->once()->andReturnTrue();
$mirror->shouldReceive('afterSales')->once()->andReturn(['rows' => [[
'internal_order_no' => 'MIRROR-ORDER-1',
'shop_name' => '共享镜像店铺',
'product_amount' => 88.5,
]] ]);
$this->app->instance(OrderDataMirrorRepository::class, $mirror);
$factory = Mockery::mock(UpstreamQueryCacheFactory::class);
$factory->shouldNotReceive('forSource');
$this->app->instance(UpstreamQueryCacheFactory::class, $factory);
$this->loginAs()->get(route('business.pages.show', [
'section' => 'order-data', 'page' => 'raw-data', 'tab' => 'after-sales-expenses',
]))->assertOk()
->assertSeeText('MIRROR-ORDER-1')
->assertSeeText('共享镜像店铺')
->assertSeeText('MIRROR');
}
public function test_order_data_pages_render_cached_upstream_rows_and_filters(): void
{
$this->allowOrderDataPages();

View File

@ -1,153 +0,0 @@
<?php
namespace Tests\Feature\Console;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Repositories\StoreReportRepository;
use Modules\EcommerceFinance\Services\StoreReportDataImporter;
use PhpOffice\PhpSpreadsheet\Spreadsheet;
use PhpOffice\PhpSpreadsheet\Writer\Xlsx;
use RuntimeException;
use Tests\TestCase;
class StoreReportDataImportTest extends TestCase
{
/** @var string[] */
private array $temporaryFiles = [];
protected function setUp(): void
{
parent::setUp();
config([
'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''],
'upstream_mirrors.connection' => 'business_mirror',
]);
DB::purge('business_mirror');
$this->createTables();
}
protected function tearDown(): void
{
foreach ($this->temporaryFiles as $file) {
@unlink($file);
}
parent::tearDown();
}
public function test_it_imports_platform_income_and_lifecycles_into_shared_generations(): void
{
$bills = $this->xlsx('抖音平台原始账单', [
['店铺', '记录编码', '结算时间', '发生时间', '收入', '支出', '账单项目'],
['店铺A', 'BILL-1', '2026-08-02 10:00:00', '2026-08-01 09:00:00', '100.25', '2.5', '订单收入'],
['店铺A', 'BILL-2', null, '2026-08-03 11:00:00', '50', '0', '订单收入'],
]);
$lifecycles = $this->xlsx('店铺生命周期', [
['ERP系统店铺名称', '公司名称', '生效时间', '失效时间'],
['店铺A', '主体甲', '2026-01-01', '存续中'],
]);
$importer = app(StoreReportDataImporter::class);
$billResult = $importer->importBills($bills, 100);
$lifecycleResult = $importer->importLifecycles($lifecycles, 100);
$this->assertSame(2, $billResult['row_count']);
$this->assertSame(1, $lifecycleResult['row_count']);
$this->assertDatabaseHas('store_report_bill_rows', [
'record_code' => 'BILL-1', 'occurred_at' => '2026-08-02 10:00:00', 'income' => 100.25,
], 'business_mirror');
$this->assertDatabaseHas('store_report_bill_rows', [
'record_code' => 'BILL-2', 'occurred_at' => '2026-08-03 11:00:00', 'income' => 50,
], 'business_mirror');
$this->assertDatabaseHas('store_report_import_states', [
'dataset_key' => StoreReportRepository::BILLS_DATASET, 'status' => 'succeeded', 'row_count' => 2,
], 'business_mirror');
$this->assertDatabaseHas('store_report_lifecycle_rows', [
'erp_store_name' => '店铺A', 'company_name' => '主体甲', 'expires_at' => null,
], 'business_mirror');
}
public function test_failed_reimport_keeps_the_previous_active_generation(): void
{
$valid = $this->xlsx('拼多多平台原始账单', [
['店铺', '记录编码', '发生时间', '收入金额', '支出金额'],
['店铺A', 'PDD-1', '2026-08-01', '10', '0'],
]);
$importer = app(StoreReportDataImporter::class);
$first = $importer->importBills($valid, 100);
$invalid = $this->xlsx('拼多多错误账单', [
['店铺', '记录编码', '发生时间', '收入金额'],
['店铺A', 'PDD-2', '不是时间', '20'],
]);
try {
$importer->importBills($invalid, 100);
$this->fail('无效时间应导致导入失败。');
} catch (RuntimeException) {
$state = DB::connection('business_mirror')->table('store_report_import_states')
->where('dataset_key', StoreReportRepository::BILLS_DATASET)->first();
$this->assertSame('failed', $state->status);
$this->assertSame($first['generation'], $state->active_generation);
$this->assertSame(1, DB::connection('business_mirror')->table('store_report_bill_rows')->count());
}
}
private function createTables(): void
{
$schema = Schema::connection('business_mirror');
$schema->create('store_report_import_states', function (Blueprint $table): void {
$table->string('dataset_key')->primary();
$table->string('status')->default('pending');
$table->string('active_generation')->nullable();
$table->string('pending_generation')->nullable();
$table->unsignedBigInteger('row_count')->default(0);
$table->string('source_sha256')->nullable();
$table->text('source_files_json')->nullable();
$table->dateTime('last_attempted_at')->nullable();
$table->dateTime('last_succeeded_at')->nullable();
$table->string('last_error_code')->nullable();
$table->timestamps();
});
$schema->create('store_report_bill_rows', function (Blueprint $table): void {
$table->increments('id');
$table->string('generation');
$table->string('platform');
$table->string('platform_label');
$table->string('store_name');
$table->dateTime('occurred_at')->nullable();
$table->decimal('income', 20, 4);
$table->decimal('expense', 20, 4);
$table->string('bill_item')->nullable();
$table->string('record_code');
$table->string('source_file');
$table->string('row_hash');
$table->dateTime('imported_at');
$table->unique(['generation', 'platform', 'record_code']);
});
$schema->create('store_report_lifecycle_rows', function (Blueprint $table): void {
$table->increments('id');
$table->string('generation');
$table->string('business_key');
$table->string('erp_store_name');
$table->string('company_name')->nullable();
$table->dateTime('effective_at')->nullable();
$table->dateTime('expires_at')->nullable();
$table->string('source_file');
$table->string('row_hash');
$table->dateTime('imported_at');
$table->unique(['generation', 'business_key']);
});
}
private function xlsx(string $name, array $rows): string
{
$path = sys_get_temp_dir() . DIRECTORY_SEPARATOR . $name . '-' . uniqid() . '.xlsx';
$spreadsheet = new Spreadsheet();
$spreadsheet->getActiveSheet()->fromArray($rows);
(new Xlsx($spreadsheet))->save($path);
$spreadsheet->disconnectWorksheets();
$this->temporaryFiles[] = $path;
return $path;
}
}

View File

@ -0,0 +1,157 @@
<?php
namespace Tests\Feature\Console;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Repositories\StoreReportRepository;
use Modules\EcommerceFinance\Repositories\StoreReportUpstreamRepository;
use Modules\EcommerceFinance\Services\StoreReportMirrorSynchronizer;
use Tests\TestCase;
class StoreReportMirrorSyncTest extends TestCase
{
protected function setUp(): void
{
parent::setUp();
config([
'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''],
'upstream_mirrors.connection' => 'business_mirror',
'upstream_mirrors.store_report.batch_size' => 100,
'upstream_mirrors.store_report.max_batches_per_run' => 2,
'upstream_mirrors.store_report.lifecycle_batch_size' => 100,
'upstream_mirrors.store_report.lifecycle_max_batches_per_run' => 2,
'upstream_mirrors.store_report.overlap_rows' => 50,
]);
DB::purge('business_mirror');
$this->createTables();
}
public function test_it_automatically_completes_first_sync_and_maps_income_direction(): void
{
$upstream = $this->createMock(StoreReportUpstreamRepository::class);
$upstream->expects($this->once())->method('lifecycleBatch')->with(0, 100)->willReturn([
['source_id' => 1, 'erp_store_name' => '店铺A', 'unified_social_credit_code' => 'CREDIT-A', 'effective_at' => '2026-01-01', 'company_name' => '主体甲'],
]);
$upstream->expects($this->once())->method('billBatch')->with(0, 100)->willReturn([
$this->bill(10, '1', 120.5),
$this->bill(11, '-1', 8.2),
]);
$result = (new StoreReportMirrorSynchronizer($upstream, app('db')))->sync(100, 2);
$this->assertSame('succeeded', $result['lifecycles']['status']);
$this->assertSame('succeeded', $result['bills']['status']);
$this->assertDatabaseHas('store_report_bill_rows', [
'source_id' => 10, 'store_name' => '店铺A', 'income' => 120.5, 'expense' => 0,
], 'business_mirror');
$this->assertDatabaseHas('store_report_bill_rows', [
'source_id' => 11, 'income' => 0, 'expense' => 8.2,
], 'business_mirror');
$this->assertDatabaseHas('store_report_import_states', [
'dataset_key' => StoreReportRepository::BILLS_DATASET,
'status' => 'succeeded',
'cursor_id' => 11,
'row_count' => 2,
], 'business_mirror');
$this->assertNotNull(DB::connection('business_mirror')->table('store_report_import_states')
->where('dataset_key', StoreReportRepository::BILLS_DATASET)->value('active_generation'));
}
public function test_incremental_sync_reuses_active_generation_and_overlaps_recent_ids(): void
{
$this->activeState(StoreReportRepository::LIFECYCLES_DATASET, 'lifecycle-live', 20, 1);
$this->activeState(StoreReportRepository::BILLS_DATASET, 'bill-live', 100, 10);
$upstream = $this->createMock(StoreReportUpstreamRepository::class);
$upstream->expects($this->once())->method('lifecycleBatch')->with(0, 100)->willReturn([]);
$upstream->expects($this->once())->method('billBatch')->with(50, 100)->willReturn([
$this->bill(101, '1', 33.0),
]);
(new StoreReportMirrorSynchronizer($upstream, app('db')))->sync(100, 2);
$this->assertDatabaseHas('store_report_bill_rows', [
'generation' => 'bill-live', 'source_id' => 101, 'income' => 33,
], 'business_mirror');
$this->assertDatabaseHas('store_report_import_states', [
'dataset_key' => StoreReportRepository::BILLS_DATASET,
'active_generation' => 'bill-live',
'cursor_id' => 101,
'row_count' => 1,
], 'business_mirror');
}
private function bill(int $id, string $direction, float $amount): array
{
return [
'source_id' => $id, 'shop_id' => 88, 'channel' => 'TouTiaoFXGZJ',
'fee_time' => '2026-08-01 10:00:00', 'del_flag' => 0,
'primary_key' => 'PK-' . $id, 'source_updated_at' => '2026-08-01 11:00:00',
'fee_name' => '订单收入', 'inout_type' => $direction, 'amount' => $amount,
'shop_name' => '店铺A',
];
}
private function activeState(string $dataset, string $generation, int $cursor, int $rows): void
{
DB::connection('business_mirror')->table('store_report_import_states')->insert([
'dataset_key' => $dataset, 'status' => 'succeeded', 'active_generation' => $generation,
'cursor_id' => $cursor, 'row_count' => $rows,
]);
}
private function createTables(): void
{
$schema = Schema::connection('business_mirror');
$schema->create('store_report_import_states', function (Blueprint $table): void {
$table->string('dataset_key')->primary();
$table->string('status')->default('pending');
$table->string('run_mode')->nullable();
$table->string('active_generation')->nullable();
$table->string('pending_generation')->nullable();
$table->unsignedBigInteger('cursor_id')->nullable();
$table->unsignedBigInteger('row_count')->default(0);
$table->text('source_files_json')->nullable();
$table->dateTime('last_attempted_at')->nullable();
$table->dateTime('last_succeeded_at')->nullable();
$table->dateTime('initial_completed_at')->nullable();
$table->string('last_error_code')->nullable();
$table->timestamps();
});
$schema->create('store_report_bill_rows', function (Blueprint $table): void {
$table->increments('id');
$table->string('generation');
$table->unsignedBigInteger('source_id');
$table->unsignedBigInteger('shop_id')->nullable();
$table->string('platform');
$table->string('platform_label');
$table->string('store_name');
$table->dateTime('occurred_at')->nullable();
$table->decimal('income', 20, 4);
$table->decimal('expense', 20, 4);
$table->string('bill_item')->nullable();
$table->string('record_code');
$table->string('source_file');
$table->string('row_hash');
$table->dateTime('source_updated_at')->nullable();
$table->dateTime('imported_at');
$table->unique(['generation', 'source_id']);
});
$schema->create('store_report_lifecycle_rows', function (Blueprint $table): void {
$table->increments('id');
$table->string('generation');
$table->unsignedBigInteger('source_id');
$table->string('business_key');
$table->string('erp_store_name');
$table->string('company_name')->nullable();
$table->string('unified_social_credit_code')->nullable();
$table->dateTime('effective_at')->nullable();
$table->dateTime('expires_at')->nullable();
$table->string('source_file');
$table->string('row_hash');
$table->dateTime('imported_at');
$table->unique(['generation', 'source_id']);
});
}
}

View File

@ -12,22 +12,26 @@ class BusinessMirrorScheduleTest extends TestCase
$contents = file_get_contents(dirname(__DIR__, 3) . '/deploy/' . $script);
$catalogPosition = strpos($contents, 'business-mirror:sync-product-catalog --quiet');
$purchasePricePosition = strpos($contents, 'business-mirror:sync-product-purchase-prices --quiet');
$storeReportPosition = strpos($contents, 'business-mirror:sync-store-report --quiet');
$snapshotPosition = strpos($contents, 'upstream:snapshots:refresh-all --quiet');
$pricePosition = strpos($contents, 'business-mirror:sync-product-prices --quiet');
$genericPosition = strpos($contents, 'business-mirror:sync-tables --quiet');
$this->assertNotFalse($catalogPosition, $script);
$this->assertNotFalse($purchasePricePosition, $script);
$this->assertNotFalse($storeReportPosition, $script);
$this->assertNotFalse($snapshotPosition, $script);
$this->assertNotFalse($pricePosition, $script);
$this->assertNotFalse($genericPosition, $script);
$this->assertLessThan($snapshotPosition, $catalogPosition, $script);
$this->assertLessThan($snapshotPosition, $purchasePricePosition, $script);
$this->assertLessThan($snapshotPosition, $storeReportPosition, $script);
$this->assertLessThan($purchasePricePosition, $catalogPosition, $script);
$this->assertLessThan($pricePosition, $catalogPosition, $script);
$this->assertLessThan($genericPosition, $catalogPosition, $script);
$this->assertSame(1, substr_count($contents, 'business-mirror:sync-product-catalog --quiet'), $script);
$this->assertSame(1, substr_count($contents, 'business-mirror:sync-product-purchase-prices --quiet'), $script);
$this->assertSame(1, substr_count($contents, 'business-mirror:sync-store-report --quiet'), $script);
$this->assertStringContainsString('docker exec -u www-data caiwu_app', $contents, $script);
$this->assertStringContainsString('/run/caiwu-upstream-sync.lock', $contents, $script);
}

View File

@ -33,10 +33,7 @@ class WebProcessOwnershipTest extends TestCase
'test: ["CMD", "jiangling-healthcheck"]',
$compose
);
$this->assertMatchesRegularExpression(
'/snapshot_worker:.*?user: www-data/s',
$compose
);
$this->assertStringNotContainsString('snapshot_worker:', $compose);
$this->assertIsString($initializer);
$this->assertStringContainsString(
@ -47,6 +44,7 @@ class WebProcessOwnershipTest extends TestCase
'php artisan local-bootstrap:update --no-ansi',
$initializer
);
$this->assertStringContainsString('docker rm -f jiangling_snapshot_worker', $initializer);
$this->assertDoesNotMatchRegularExpression(
'/exec -T app\s+(?:\\\\\s*)?php artisan/',
$initializer

View File

@ -0,0 +1,160 @@
<?php
namespace Tests\Unit\EcommerceFinance;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Mockery;
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
use Modules\EcommerceFinance\Repositories\OrderDataMirrorRepository;
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
use Tests\TestCase;
class OrderDataMirrorRepositoryTest extends TestCase
{
protected function setUp(): void
{
parent::setUp();
config(['database.connections.business_mirror' => [
'driver' => 'sqlite', 'database' => ':memory:', 'prefix' => '',
]]);
DB::purge('business_mirror');
$this->createTables();
}
public function test_after_sales_reads_complete_rows_from_shared_mirror(): void
{
$db = DB::connection('business_mirror');
$db->table('ops_jst_orders')->insert([
'o_id' => 1, 'so_id' => 'SO-1', 'pay_date' => '2026-08-01 08:00:00',
'end_time' => '2026-08-02 12:00:00', 'send_date' => '2026-08-02 10:00:00',
'order_date' => '2026-08-01 09:00:00', 'shop_name' => '京东店铺',
'shop_id' => 'SHOP-1', 'shop_site' => 'jd', 'type' => '普通订单',
'labels' => '正常', 'wms_co_id' => 'WH-1', 'l_id' => 'EXP-1',
'logistics_company' => '物流', 'receiver_state' => '江苏', 'receiver_city' => '常州',
'creator_name' => '销售员', 'free_amount' => 2, 'freight' => 5,
]);
$db->table('ops_jst_order_item')->insert([
'oi_id' => 11, 'o_id' => 1, 'outer_oi_id' => 'SUB-1', 'sku_id' => 'SKU-1',
'i_id' => 'STYLE-1', 'shop_sku_id' => 'SHOP-SKU-1', 'shop_i_id' => '1001',
'properties_value' => '黑色', 'name' => '线上商品', 'qty' => 2,
'amount' => 100, 'buyer_paid_amount' => 95, 'seller_income_amount' => 98,
'is_gift' => 0, 'src_combine_sku_id' => null, 'send_warehouse' => 'WH-1',
]);
$db->table('ops_jst_sku')->insert([
'autoid' => 1, 'sku_type' => 'normal', 'sku_id' => 'SKU-1', 'brand' => '品牌A', 'category' => '家具', 'vc_name' => '桌子',
'cost_price' => 20, 'name' => '镜像商品', 'supplier_id' => 'SUP-1', 'supplier_name' => '供应商A',
]);
$db->table('ops_jst_warehouse_info')->insert([
'wms_co_id' => 'WH-1', 'name' => '主仓', 'remark1' => '运营云仓',
]);
$db->table('ops_jst_return_refund_info')->insert([
'as_id' => 21, 'outer_as_id' => 'AF-1', 'o_id' => 1,
'as_date' => '2026-08-02 12:00:00', 'type' => '退货退款',
'question_type' => '质量问题', 'labels' => '售后', 'refund' => 50, 'free_amount' => 3,
]);
$db->table('ops_jst_return_refund_item')->insert([
'as_id' => 21, 'outer_oi_id' => 'SUB-1', 'sku_id' => 'SKU-1', 'type' => '退货',
'qty' => 1, 'amount' => 50, 'r_qty' => 1, 'shop_amount' => 50, 'defective_qty' => 1,
]);
$mirror = Mockery::mock(GenericBusinessMirrorRepository::class);
$mirror->shouldReceive('connection')->andReturn($db);
$repository = new OrderDataMirrorRepository(
Mockery::mock(BusinessMirrorDatasetRegistry::class), $mirror,
);
$result = $repository->afterSales([
'from' => '2026-08-01', 'to' => '2026-08-02',
'order_no' => '', 'shop_name' => '', 'sku_id' => '',
], 1, 25);
$this->assertCount(1, $result['rows']);
$row = $result['rows'][0];
$this->assertSame('SO-1', $row['online_order_no']);
$this->assertSame('AF-1', $row['after_sales_no']);
$this->assertSame(40.0, $row['product_cost']);
$this->assertSame(1.0, $row['post_return_quantity']);
$this->assertSame('https://item.jd.com/1001.html', $row['product_link']);
$this->assertNull($row['account_8001201']);
}
public function test_promotion_expenses_read_the_shared_statistics_and_sku_mapping(): void
{
$db = DB::connection('business_mirror');
$db->table('ops_item_sales_cost_ratio')->insert([
'stat_date' => '2026-08-09 00:00:00', 'shop_name' => '京东店铺',
'item_id' => 'ITEM-1', 'push_fee' => 9.05,
]);
$db->table('ops_jst_shop_info')->insert(['shop_id' => 8, 'shop_name' => '京东店铺']);
$db->table('ops_jst_sku_map')->insert([
'shop_i_id' => 'ITEM-1', 'shop_id' => 8, 'sku_id' => 'SKU-1',
'i_id' => 'STYLE-1', 'shop_sku_id' => 'SHOP-SKU-1',
'properties_value' => '黑色', 'modified' => '2026-08-09 10:00:00',
]);
$mirror = Mockery::mock(GenericBusinessMirrorRepository::class);
$mirror->shouldReceive('connection')->andReturn($db);
$repository = new OrderDataMirrorRepository(
Mockery::mock(BusinessMirrorDatasetRegistry::class), $mirror,
);
$result = $repository->promotionExpenses([
'date' => '2026-08-09', 'shop_name' => '', 'sku_id' => '',
], 1, 25);
$this->assertSame('2026-08-09', $result['date']);
$this->assertSame('SKU-1', $result['rows'][0]['product_code']);
$this->assertSame(9.05, $result['rows'][0]['promotion_fee']);
}
private function createTables(): void
{
$schema = Schema::connection('business_mirror');
$schema->create('ops_jst_orders', function (Blueprint $t): void {
$t->integer('o_id'); $t->string('so_id')->nullable(); $t->dateTime('pay_date')->nullable();
$t->dateTime('end_time')->nullable(); $t->dateTime('send_date')->nullable(); $t->dateTime('order_date');
$t->string('shop_name'); $t->string('shop_id')->nullable(); $t->string('shop_site')->nullable();
$t->string('type')->nullable(); $t->string('labels')->nullable(); $t->string('wms_co_id')->nullable();
$t->string('l_id')->nullable(); $t->string('logistics_company')->nullable();
$t->string('receiver_state')->nullable(); $t->string('receiver_city')->nullable();
$t->string('creator_name')->nullable(); $t->decimal('free_amount')->nullable(); $t->decimal('freight')->nullable();
});
$schema->create('ops_jst_order_item', function (Blueprint $t): void {
$t->integer('oi_id'); $t->integer('o_id'); $t->string('outer_oi_id')->nullable();
$t->string('sku_id'); $t->string('i_id')->nullable(); $t->string('shop_sku_id')->nullable();
$t->string('shop_i_id')->nullable(); $t->string('properties_value')->nullable(); $t->string('name')->nullable();
$t->decimal('qty')->nullable(); $t->decimal('amount')->nullable(); $t->decimal('buyer_paid_amount')->nullable();
$t->decimal('seller_income_amount')->nullable(); $t->integer('is_gift')->nullable();
$t->string('src_combine_sku_id')->nullable(); $t->string('send_warehouse')->nullable();
});
$schema->create('ops_jst_sku', function (Blueprint $t): void {
$t->integer('autoid'); $t->string('sku_type'); $t->string('sku_id'); $t->string('brand')->nullable(); $t->string('category')->nullable();
$t->string('vc_name')->nullable(); $t->decimal('cost_price')->nullable(); $t->string('name')->nullable();
$t->string('supplier_id')->nullable(); $t->string('supplier_name')->nullable();
});
$schema->create('ops_jst_warehouse_info', function (Blueprint $t): void {
$t->string('wms_co_id'); $t->string('name')->nullable(); $t->string('remark1')->nullable();
});
$schema->create('ops_jst_return_refund_info', function (Blueprint $t): void {
$t->integer('as_id'); $t->string('outer_as_id')->nullable(); $t->integer('o_id');
$t->dateTime('as_date'); $t->string('type')->nullable(); $t->string('question_type')->nullable();
$t->string('labels')->nullable(); $t->decimal('refund')->nullable(); $t->decimal('free_amount')->nullable();
});
$schema->create('ops_jst_return_refund_item', function (Blueprint $t): void {
$t->integer('as_id'); $t->string('outer_oi_id')->nullable(); $t->string('sku_id')->nullable();
$t->string('type')->nullable(); $t->decimal('qty')->nullable(); $t->decimal('amount')->nullable();
$t->decimal('r_qty')->nullable(); $t->decimal('shop_amount')->nullable(); $t->decimal('defective_qty')->nullable();
});
$schema->create('ops_item_sales_cost_ratio', function (Blueprint $t): void {
$t->dateTime('stat_date'); $t->string('shop_name'); $t->string('item_id'); $t->decimal('push_fee');
});
$schema->create('ops_jst_sku_map', function (Blueprint $t): void {
$t->string('shop_i_id'); $t->integer('shop_id'); $t->string('sku_id')->nullable();
$t->string('i_id')->nullable(); $t->string('shop_sku_id')->nullable();
$t->string('properties_value')->nullable(); $t->dateTime('modified')->nullable();
});
$schema->create('ops_jst_shop_info', function (Blueprint $t): void {
$t->integer('shop_id'); $t->string('shop_name');
});
}
}

View File

@ -4,6 +4,7 @@ namespace Tests\Unit\EcommerceFinance;
use App\Business\UpstreamBusinessInformation;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Mockery;
@ -11,119 +12,99 @@ use Modules\EcommerceFinance\Repositories\BusinessInformationMirrorRepository;
use Modules\EcommerceFinance\Repositories\InstallationDetailMirrorRepository;
use Modules\EcommerceFinance\Repositories\StoreReportRepository;
use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
use RuntimeException;
use Tests\TestCase;
class StoreReportRepositoryTest extends TestCase
{
private string $databasePath;
private StoreReportRepository $repository;
protected function setUp(): void
{
parent::setUp();
$this->databasePath = tempnam(sys_get_temp_dir(), 'store-report-');
config(['database.connections.bills_sqlite' => [
'driver' => 'sqlite',
'database' => $this->databasePath,
'prefix' => '',
'foreign_key_constraints' => false,
]]);
DB::purge('bills_sqlite');
$schema = Schema::connection('bills_sqlite');
$schema->create('store_lifecycles', function (Blueprint $table): void {
$table->increments('id');
$table->string('erp_store_name');
$table->string('company_name')->nullable();
$table->string('effective_at')->nullable();
$table->string('expires_at')->nullable();
});
$schema->create('bills', function (Blueprint $table): void {
$table->increments('id');
$table->string('store_name');
$table->string('occurred_at')->nullable();
$table->decimal('income', 18, 2)->default(0);
});
config([
'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''],
'upstream_mirrors.connection' => 'business_mirror',
]);
DB::purge('business_mirror');
$this->createTables();
DB::connection('business_mirror')->table('store_report_import_states')->insert([
['dataset_key' => StoreReportRepository::BILLS_DATASET, 'status' => 'succeeded', 'active_generation' => 'bills-v1'],
['dataset_key' => StoreReportRepository::LIFECYCLES_DATASET, 'status' => 'succeeded', 'active_generation' => 'lifecycles-v1'],
]);
$this->repository = new StoreReportRepository(app('db'));
}
protected function tearDown(): void
{
DB::purge('bills_sqlite');
if (is_file($this->databasePath)) {
unlink($this->databasePath);
}
Carbon::setTestNow();
parent::tearDown();
}
public function test_lifecycle_boundaries_split_monthly_and_period_income_without_double_counting(): void
public function test_lifecycle_boundaries_split_monthly_income_without_double_counting(): void
{
$db = DB::connection('bills_sqlite');
$db->table('store_lifecycles')->insert([
['erp_store_name' => '店铺A', 'company_name' => '主体甲', 'effective_at' => '2026-05-01', 'expires_at' => '2026-07-01'],
['erp_store_name' => '店铺A', 'company_name' => '主体乙', 'effective_at' => '2026-07-01', 'expires_at' => null],
$db = DB::connection('business_mirror');
$db->table('store_report_lifecycle_rows')->insert([
$this->lifecycle(1, '店铺A', '主体甲', '2026-05-01', '2026-07-01'),
$this->lifecycle(2, '店铺A', '主体乙', '2026-07-01', null),
]);
$db->table('bills')->insert([
['store_name' => '店铺A', 'occurred_at' => '2026-04-30 23:59:59', 'income' => 9],
['store_name' => '店铺A', 'occurred_at' => '2026-05-01 00:00:00', 'income' => 100],
['store_name' => '店铺A', 'occurred_at' => '2026-06-30 23:59:59', 'income' => 60],
['store_name' => '店铺A', 'occurred_at' => '2026-07-01 00:00:00', 'income' => 70],
['store_name' => '店铺A', 'occurred_at' => '2026-08-01 00:00:00', 'income' => 80],
['store_name' => '店铺A', 'occurred_at' => '2027-01-01 00:00:00', 'income' => 5],
$db->table('store_report_bill_rows')->insert([
$this->bill(1, '店铺A', '2026-04-30 23:59:59', 9),
$this->bill(2, '店铺A', '2026-05-01 00:00:00', 100),
$this->bill(3, '店铺A', '2026-06-30 23:59:59', 60),
$this->bill(4, '店铺A', '2026-07-01 00:00:00', 70),
$this->bill(5, '店铺A', '2026-08-01 00:00:00', 80),
$this->bill(6, '店铺A', '2027-01-01 00:00:00', 5),
]);
$rows = $this->repository->rows([
['erp_store_name' => '店铺A', 'company_name' => '旧主体', 'store_status' => '3'],
]);
['erp_store_name' => '店铺A', 'store_id' => 'STORE-001', 'company_name' => '旧主体', 'store_status' => '3'],
], 2026);
$this->assertCount(2, $rows);
$this->assertSame(['主体甲', '主体乙'], array_column($rows, 'entity'));
$this->assertSame(['2026-05-01', '2026-07-01'], array_column($rows, 'effective_at'));
$this->assertSame(['2026-07-01', '2999-01-01'], array_column($rows, 'invalid_at'));
$this->assertSame('STORE-001', $rows[0]['store_id']);
$this->assertSame([160.0, 155.0], array_column($rows, 'period_sales'));
$this->assertSame([100.0, 0.0], array_column($rows, 'sales_05'));
$this->assertSame([0.0, 70.0], array_column($rows, 'sales_07'));
$this->assertSame(['关店中', '关店中'], array_column($rows, 'status'));
}
public function test_missing_lifecycle_uses_upstream_company_and_default_period(): void
public function test_failed_refresh_keeps_the_previous_active_generation_readable(): void
{
DB::connection('bills_sqlite')->table('bills')->insert([
['store_name' => '无生命周期店铺', 'occurred_at' => '2026-05-03 12:00:00', 'income' => 25.5],
]);
DB::connection('business_mirror')->table('store_report_import_states')
->where('dataset_key', StoreReportRepository::BILLS_DATASET)
->update(['status' => 'failed', 'pending_generation' => null]);
DB::connection('business_mirror')->table('store_report_bill_rows')->insert(
$this->bill(1, '店铺A', '2026-05-03 12:00:00', 25.5),
);
$rows = $this->repository->rows([
['erp_store_name' => '无生命周期店铺', 'company_name' => '公司表主体', 'store_status' => '0'],
]);
['erp_store_name' => '店铺A', 'company_name' => '公司表主体', 'opened_at' => '2026-05-01'],
], 2026);
$this->assertSame('公司表主体', $rows[0]['entity']);
$this->assertSame('2026-05-01', $rows[0]['effective_at']);
$this->assertSame('2999-01-01', $rows[0]['invalid_at']);
$this->assertSame(25.5, $rows[0]['period_sales']);
$this->assertSame('开店中', $rows[0]['status']);
$this->assertSame(25.5, $rows[0]['sales_05']);
$this->assertSame('公司表主体', $rows[0]['entity']);
}
public function test_shop_state_mapping_keeps_opening_closing_and_closed_labels(): void
public function test_missing_first_import_is_reported_instead_of_returning_misleading_zeroes(): void
{
$rows = $this->repository->rows([
['erp_store_name' => '开店中店铺', 'store_status' => '0'],
['erp_store_name' => '关店中店铺', 'store_status' => '3'],
['erp_store_name' => '已关店店铺', 'store_status' => '6'],
]);
DB::connection('business_mirror')->table('store_report_import_states')
->where('dataset_key', StoreReportRepository::BILLS_DATASET)->delete();
$this->assertSame(['开店中', '关店中', '关店'], array_column($rows, 'status'));
$this->expectException(RuntimeException::class);
$this->expectExceptionMessage('中央同步中');
$this->repository->rows([['erp_store_name' => '店铺A']], 2026);
}
public function test_upstream_business_information_exposes_expanded_rows_for_store_report(): void
public function test_upstream_business_information_exposes_shared_store_report_rows(): void
{
DB::connection('bills_sqlite')->table('store_lifecycles')->insert([
['erp_store_name' => '店铺A', 'company_name' => '主体甲', 'effective_at' => '2026-05-01', 'expires_at' => '2026-07-01'],
['erp_store_name' => '店铺A', 'company_name' => '主体乙', 'effective_at' => '2026-07-01', 'expires_at' => null],
DB::connection('business_mirror')->table('store_report_lifecycle_rows')->insert([
$this->lifecycle(1, '店铺A', '主体甲', '2026-05-01', null),
]);
DB::connection('bills_sqlite')->table('bills')->insert([
['store_name' => '店铺A', 'occurred_at' => '2026-06-30 23:59:59', 'income' => 60],
['store_name' => '店铺A', 'occurred_at' => '2026-07-01 00:00:00', 'income' => 70],
DB::connection('business_mirror')->table('store_report_bill_rows')->insert([
$this->bill(1, '店铺A', '2026-06-30 23:59:59', 60),
]);
$mirror = Mockery::mock(BusinessInformationMirrorRepository::class);
$mirror->shouldReceive('ready')->with('store-information.read')->andReturnTrue();
$mirror->shouldReceive('allStores')->once()->andReturn([[
@ -138,8 +119,84 @@ class StoreReportRepositoryTest extends TestCase
$paginator = $service->paginate('store-management', 'store-report', 50);
$this->assertSame(2, $paginator->total());
$this->assertSame(['主体甲', '主体乙'], array_map(static fn (object $row): string => $row->entity, $paginator->items()));
$this->assertSame([60.0, 70.0], array_map(static fn (object $row): float => $row->period_sales, $paginator->items()));
$this->assertSame(1, $paginator->total());
$this->assertSame('主体甲', $paginator->items()[0]->entity);
$this->assertSame(60.0, $paginator->items()[0]->period_sales);
}
public function test_subject_operating_report_uses_real_recent_and_entity_sales(): void
{
Carbon::setTestNow('2026-09-01 10:00:00');
DB::connection('business_mirror')->table('store_report_lifecycle_rows')->insert([
$this->lifecycle(1, '店铺A', '主体甲', '2026-01-01', null),
$this->lifecycle(2, '店铺B', '主体甲', '2026-01-01', null),
]);
DB::connection('business_mirror')->table('store_report_bill_rows')->insert([
$this->bill(1, '店铺A', '2026-06-10 10:00:00', 10),
$this->bill(2, '店铺A', '2026-07-10 10:00:00', 20),
$this->bill(3, '店铺A', '2026-08-10 10:00:00', 30),
$this->bill(4, '店铺B', '2026-08-11 10:00:00', 40),
]);
$mirror = Mockery::mock(BusinessInformationMirrorRepository::class);
$mirror->shouldReceive('ready')->with('store-information.read')->andReturnTrue();
$mirror->shouldReceive('allStores')->once()->andReturn([
['erp_store_name' => '店铺A', 'company_name' => '主体甲', 'store_status' => '0'],
['erp_store_name' => '店铺B', 'company_name' => '主体甲', 'store_status' => '0'],
]);
$service = new UpstreamBusinessInformation(
Mockery::mock(UpstreamQueryCacheFactory::class),
$mirror,
Mockery::mock(InstallationDetailMirrorRepository::class),
$this->repository,
);
$rows = $service->paginate('company-management', 'subject-operating-report', 50)->items();
$this->assertCount(2, $rows);
$this->assertSame(10.0, $rows[0]->recent_three_month_sales_1);
$this->assertSame(20.0, $rows[0]->recent_three_month_sales_2);
$this->assertSame(30.0, $rows[0]->recent_three_month_sales_3);
$this->assertSame(60.0, $rows[0]->store_sales_12m);
$this->assertSame(100.0, $rows[0]->entity_sales_12m);
$this->assertSame(100.0, $rows[1]->entity_sales_12m);
}
private function createTables(): void
{
$schema = Schema::connection('business_mirror');
$schema->create('store_report_import_states', function (Blueprint $table): void {
$table->string('dataset_key')->primary();
$table->string('status');
$table->string('active_generation')->nullable();
$table->string('pending_generation')->nullable();
});
$schema->create('store_report_bill_rows', function (Blueprint $table): void {
$table->increments('id');
$table->string('generation');
$table->string('store_name');
$table->dateTime('occurred_at')->nullable();
$table->decimal('income', 20, 4)->default(0);
});
$schema->create('store_report_lifecycle_rows', function (Blueprint $table): void {
$table->increments('id');
$table->string('generation');
$table->string('erp_store_name');
$table->string('company_name')->nullable();
$table->dateTime('effective_at')->nullable();
$table->dateTime('expires_at')->nullable();
});
}
private function bill(int $id, string $store, string $time, float $income): array
{
return ['id' => $id, 'generation' => 'bills-v1', 'store_name' => $store, 'occurred_at' => $time, 'income' => $income];
}
private function lifecycle(int $id, string $store, string $company, ?string $from, ?string $to): array
{
return [
'id' => $id, 'generation' => 'lifecycles-v1', 'erp_store_name' => $store,
'company_name' => $company, 'effective_at' => $from, 'expires_at' => $to,
];
}
}