优化财务资源：减少重复扫描、隔离重型任务、约束系统日志 (#390)
This commit is contained in:
parent
eb0f246765
commit
1f4f6f48d4
@ -63,6 +63,7 @@ class JstOrderLabelImporter
|
||||
->where('ordered_at', '>=', $start)
|
||||
->where('ordered_at', '<', $end)
|
||||
->delete();
|
||||
app(PlantingOrderLabelLookup::class)->forget($companyId);
|
||||
}
|
||||
|
||||
$now = now();
|
||||
@ -71,9 +72,10 @@ class JstOrderLabelImporter
|
||||
$noLabelOrders = 0;
|
||||
$buffer = [];
|
||||
|
||||
$flush = function () use (&$buffer, &$rows, $target): void {
|
||||
$flush = function () use (&$buffer, &$rows, $target, $companyId): void {
|
||||
foreach (array_chunk($buffer, self::INSERT_CHUNK) as $chunk) {
|
||||
$target->table('jst_order_labels')->insert($chunk);
|
||||
app(PlantingOrderLabelLookup::class)->forget($companyId);
|
||||
$rows += count($chunk);
|
||||
}
|
||||
$buffer = [];
|
||||
@ -191,6 +193,7 @@ class JstOrderLabelImporter
|
||||
->where('ordered_at', '>=', $start)
|
||||
->where('ordered_at', '<', $end)
|
||||
->delete();
|
||||
app(PlantingOrderLabelLookup::class)->forget($companyId);
|
||||
};
|
||||
$cleared = ! $replace;
|
||||
|
||||
@ -203,9 +206,10 @@ class JstOrderLabelImporter
|
||||
$orderSeen = [];
|
||||
$buffer = [];
|
||||
|
||||
$flush = function () use (&$buffer, &$rows, $target): void {
|
||||
$flush = function () use (&$buffer, &$rows, $target, $companyId): void {
|
||||
foreach (array_chunk($buffer, self::INSERT_CHUNK) as $chunk) {
|
||||
$target->table('jst_order_labels')->insert($chunk);
|
||||
app(PlantingOrderLabelLookup::class)->forget($companyId);
|
||||
$rows += count($chunk);
|
||||
}
|
||||
$buffer = [];
|
||||
|
||||
@ -82,6 +82,7 @@ class JstOrderLabelPlantingRules
|
||||
public function forget(int $companyId): void
|
||||
{
|
||||
Cache::forget($this->key($companyId));
|
||||
app(PlantingOrderLabelLookup::class)->forget($companyId);
|
||||
}
|
||||
|
||||
private function key(int $companyId): string
|
||||
|
||||
68
app/Business/PlantingOrderLabelLookup.php
Normal file
68
app/Business/PlantingOrderLabelLookup.php
Normal file
@ -0,0 +1,68 @@
|
||||
<?php
|
||||
|
||||
namespace App\Business;
|
||||
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
use Illuminate\Support\Facades\DB;
|
||||
|
||||
/** Reuse label classification across monthly/platform jobs; writers invalidate it. */
|
||||
class PlantingOrderLabelLookup
|
||||
{
|
||||
/** @param list<string> $positiveLabels @return list<string> */
|
||||
public function orderNumbers(int $companyId, array $positiveLabels): array
|
||||
{
|
||||
sort($positiveLabels, SORT_STRING);
|
||||
$rules = hash('sha256', json_encode($positiveLabels, JSON_THROW_ON_ERROR));
|
||||
$key = $this->key($companyId);
|
||||
$read = function () use ($key, $rules, $companyId): ?array {
|
||||
$cached = Cache::get($key);
|
||||
|
||||
return is_array($cached) && ($cached['rules'] ?? null) === $rules
|
||||
&& ($cached['revision'] ?? null) === $this->revision($companyId)
|
||||
? $cached['orders'] : null;
|
||||
};
|
||||
if (($orders = $read()) !== null) {
|
||||
return $orders;
|
||||
}
|
||||
|
||||
return Cache::lock($key . ':build', 300)->block(305, function () use ($companyId, $positiveLabels, $key, $rules, $read): array {
|
||||
if (($orders = $read()) !== null) {
|
||||
return $orders;
|
||||
}
|
||||
$revision = $this->revision($companyId);
|
||||
$orders = [];
|
||||
foreach (DB::table('jst_order_labels')->where('company_id', $companyId)
|
||||
->where(function ($query) use ($positiveLabels): void {
|
||||
if ($positiveLabels !== []) {
|
||||
$query->whereIn('label', $positiveLabels);
|
||||
}
|
||||
$query->orWhereRaw('FIND_IN_SET(?, item_code) > 0', [JstOrderLabelPlantingRules::PLANTING_ITEM_CODE]);
|
||||
})->select('order_no')->distinct()->cursor() as $row) {
|
||||
$orders[] = (string) $row->order_no;
|
||||
}
|
||||
// An import/rule change during the scan must not publish an old result.
|
||||
if ($this->revision($companyId) === $revision) {
|
||||
Cache::put($key, ['rules' => $rules, 'revision' => $revision, 'orders' => $orders], now()->addHours(24));
|
||||
}
|
||||
|
||||
return $orders;
|
||||
});
|
||||
}
|
||||
|
||||
public function forget(int $companyId): void
|
||||
{
|
||||
$key = $this->key($companyId);
|
||||
Cache::store('upstream_shared')->forever($key . ':revision', bin2hex(random_bytes(16)));
|
||||
Cache::forget($key);
|
||||
}
|
||||
|
||||
public function revision(int $companyId): ?string
|
||||
{
|
||||
return Cache::store('upstream_shared')->get($this->key($companyId) . ':revision');
|
||||
}
|
||||
|
||||
private function key(int $companyId): string
|
||||
{
|
||||
return 'planting-order-labels:v1:' . $companyId;
|
||||
}
|
||||
}
|
||||
@ -142,7 +142,12 @@ class SyncJstOrderLabelItemCodes extends Command
|
||||
|
||||
$params = array_merge($bindings, [$companyId], $orderNos, [$start, $end]);
|
||||
|
||||
return DB::connection()->update($sql, $params);
|
||||
$updated = DB::connection()->update($sql, $params);
|
||||
if ($updated > 0) {
|
||||
app(\App\Business\PlantingOrderLabelLookup::class)->forget($companyId);
|
||||
}
|
||||
|
||||
return $updated;
|
||||
}
|
||||
|
||||
/** @return array{0:string,1:string} 起止日期（Y-m-d） */
|
||||
|
||||
@ -8,6 +8,7 @@ use Illuminate\Support\Facades\DB;
|
||||
use Illuminate\Support\Facades\Schema;
|
||||
use InvalidArgumentException;
|
||||
use App\Business\JstOrderLabelPlantingRules;
|
||||
use App\Business\PlantingOrderLabelLookup;
|
||||
use Modules\EcommerceFinance\Services\AccountLedgerRemarkMap;
|
||||
use Modules\EcommerceFinance\Services\AccountMappingStore;
|
||||
use Modules\EcommerceFinance\Services\LedgerRemarkFilter;
|
||||
@ -1297,16 +1298,7 @@ class SyncStoreReceiptOrderMatches extends Command
|
||||
|
||||
// 并集的第一部分：聚水潭订单标签明细里「是否种菜订单＝是」的订单号
|
||||
// （命中种菜标签映射，或货号含信誉订单虚拟条码 X00006，与页面同一规则）
|
||||
foreach (DB::table('jst_order_labels')
|
||||
->where('company_id', $companyId)
|
||||
->where(function ($query) use ($positiveLabels): void {
|
||||
if ($positiveLabels !== []) {
|
||||
$query->whereIn('label', $positiveLabels);
|
||||
}
|
||||
$query->orWhereRaw('FIND_IN_SET(?, item_code) > 0', [JstOrderLabelPlantingRules::PLANTING_ITEM_CODE]);
|
||||
})
|
||||
->distinct()
|
||||
->pluck('order_no') as $orderNo) {
|
||||
foreach (app(PlantingOrderLabelLookup::class)->orderNumbers($companyId, $positiveLabels) as $orderNo) {
|
||||
$core = self::orderCoreNo((string) $orderNo);
|
||||
if ($core !== '') {
|
||||
$cores[$core] = true;
|
||||
|
||||
@ -9,14 +9,14 @@ use Throwable;
|
||||
|
||||
class WarmOrderDataShopCandidates extends Command
|
||||
{
|
||||
protected $signature = 'order-data:warm-shop-candidates';
|
||||
protected $signature = 'order-data:warm-shop-candidates {--full : 全量核对店铺候选（夜间维护）}';
|
||||
|
||||
protected $description = '预热订单售后费用明细的店铺候选（要扫两张百万级订单宽表，不能放在页面请求里）';
|
||||
protected $description = '增量预热订单售后费用明细的店铺候选，夜间使用 --full 全量核对';
|
||||
|
||||
public function handle(OrderDataMirrorRepository $orders): int
|
||||
{
|
||||
try {
|
||||
$names = $orders->refreshShopNamesCache();
|
||||
$names = $orders->refreshShopNamesCache((bool) $this->option('full'));
|
||||
$this->info(sprintf('店铺候选已预热，共 %d 个店铺。', count($names)));
|
||||
|
||||
return self::SUCCESS;
|
||||
|
||||
@ -68,6 +68,7 @@ services:
|
||||
volumes:
|
||||
- caiwu_clickhouse_database:/var/lib/clickhouse
|
||||
- caiwu_clickhouse_logs:/var/log/clickhouse-server
|
||||
- ./clickhouse-resource-policy.xml:/etc/clickhouse-server/config.d/caiwu-resource-policy.xml:ro
|
||||
mem_limit: 4g
|
||||
cpus: "2.00"
|
||||
ulimits:
|
||||
@ -100,6 +101,7 @@ services:
|
||||
condition: service_healthy
|
||||
environment:
|
||||
RUN_MIGRATIONS: ${RUN_MIGRATIONS:-0}
|
||||
FINANCE_BATCH_WORKER_ENABLED: "1"
|
||||
SUPPLIER_PAYABLE_MAINTENANCE_DB_CONNECTION: business_mirror
|
||||
JST_DB_CONNECTION: business_mirror
|
||||
env_file:
|
||||
@ -121,6 +123,35 @@ services:
|
||||
retries: 10
|
||||
start_period: 45s
|
||||
|
||||
# Heavy CLI jobs use a separate cgroup, not Apache's unbounded app container.
|
||||
batch:
|
||||
image: caiwu_akaunting:${IMAGE_TAG:?IMAGE_TAG is required}
|
||||
pull_policy: never
|
||||
profiles: [batch]
|
||||
user: www-data
|
||||
entrypoint: [php, artisan]
|
||||
command: [list]
|
||||
restart: "no"
|
||||
environment:
|
||||
SUPPLIER_PAYABLE_MAINTENANCE_DB_CONNECTION: business_mirror
|
||||
JST_DB_CONNECTION: business_mirror
|
||||
env_file:
|
||||
- ./runtime/qiniu-storage.env
|
||||
- ./runtime/clickhouse.env
|
||||
volumes:
|
||||
- ./runtime/.env:/var/www/html/.env:ro
|
||||
- caiwu_storage:/var/www/html/storage
|
||||
cpus: "1.00"
|
||||
mem_limit: 3g
|
||||
memswap_limit: 3g
|
||||
healthcheck:
|
||||
disable: true
|
||||
logging:
|
||||
driver: json-file
|
||||
options:
|
||||
max-size: "10m"
|
||||
max-file: "3"
|
||||
|
||||
volumes:
|
||||
caiwu_database:
|
||||
name: caiwu_database
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
<?php
|
||||
|
||||
return [
|
||||
'batch_worker_enabled' => (bool) env('FINANCE_BATCH_WORKER_ENABLED', false),
|
||||
// Connection records supply addresses and credentials, never these defaults.
|
||||
'defaults' => ['enabled' => false, 'connection' => ['connect_timeout' => 5, 'query_timeout' => 15]],
|
||||
'sync' => [
|
||||
|
||||
@ -189,9 +189,9 @@ CRON_TZ=Asia/Shanghai
|
||||
* * * * * 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
|
||||
# 店铺经营报表按 15 分钟刷新；单轮覆盖 50000 行回看窗口，避免空闲时持续重读千万级账单。
|
||||
7,22,37,52 * * * * root /usr/bin/flock -w 600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-store-report --max-batches=12 --quiet >/dev/null 2>&1
|
||||
# 订单售后费用明细的店铺候选：完整列表要扫两张百万级订单宽表（约 2 分钟），放定时任务预热；
|
||||
# 页面冷缓存时先用毫秒级的店铺目录兜底，避免请求一直不结束、页面一直转圈。
|
||||
# 店铺候选平时仅读镜像变更；夜间全量核对历史名称。冷页面不扫描订单宽表。
|
||||
45 * * * * root /usr/bin/flock -w 600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan order-data:warm-shop-candidates --quiet >/dev/null 2>&1
|
||||
45 2 * * * root /bin/sh /opt/caiwuxitong/run-finance-batch.sh order-data:warm-shop-candidates --full --quiet >/dev/null 2>&1
|
||||
# 页面字段自动跟随上游：先刷新数据字典，再把上游新增字段补进镜像并建列，页面自动多出这一列。
|
||||
40 6 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:follow-upstream-columns --quiet >/dev/null 2>&1
|
||||
0 0 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-third-party-warehouse-storage --max-batches=0 --quiet && /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:warm-report-caches --quiet' >/dev/null 2>&1
|
||||
@ -205,10 +205,9 @@ CRON_TZ=Asia/Shanghai
|
||||
# 供应商管理每天北京时间 08:15 完整校准档案、合同、联系人与打款资料。
|
||||
15 8 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-supplier-management --quiet >/dev/null 2>&1
|
||||
*/2 * * * * root /usr/bin/flock -n /run/caiwu-esign-polling.lock /usr/bin/docker exec -u www-data caiwu_app php artisan esign:poll-signatures --quiet >/dev/null 2>&1
|
||||
# 科目映射表保存后自动重跑业绩明细。页面保存时会立即起后台进程，这里每分钟兜底一次，
|
||||
# 防止容器重启、请求进程退出或并发时遗漏队列任务。
|
||||
* * * * * root /usr/bin/flock -n /run/caiwu-account-mapping-resync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan store-receipt:auto-resync --quiet >/dev/null 2>&1
|
||||
28 1 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jushuitan:sync-shipping --days=3 >/dev/null 2>&1
|
||||
# 科目映射保存后保留自动重跑队列，每分钟在受限容器内串行消费。
|
||||
* * * * * root /usr/bin/flock -n /run/caiwu-account-mapping-resync.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh store-receipt:auto-resync --quiet >/dev/null 2>&1
|
||||
28 1 * * * root /bin/sh /opt/caiwuxitong/run-finance-batch.sh jushuitan:sync-shipping --days=3 >/dev/null 2>&1
|
||||
# 头条放心购（店铺收款账户账单 › 抖音账户）资金流水：每天 09:30 定点把该渠道所用
|
||||
# 原始账单镜像追平一次，保证上班前当日/前一日流水已进系统，而不是等每分钟轮转慢慢走。
|
||||
30 9 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables operations_replica.rpt.jst_originalBillInfo --max-batches=20 --quiet >/dev/null 2>&1
|
||||
|
||||
@ -193,9 +193,9 @@ CRON_TZ=Asia/Shanghai
|
||||
* * * * * 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
|
||||
# 店铺经营报表按 15 分钟刷新；单轮覆盖 50000 行回看窗口，避免空闲时持续重读千万级账单。
|
||||
7,22,37,52 * * * * root /usr/bin/flock -w 600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-store-report --max-batches=12 --quiet >/dev/null 2>&1
|
||||
# 订单售后费用明细的店铺候选：完整列表要扫两张百万级订单宽表（约 2 分钟），放定时任务预热；
|
||||
# 页面冷缓存时先用毫秒级的店铺目录兜底，避免请求一直不结束、页面一直转圈。
|
||||
# 店铺候选平时仅读镜像变更；夜间全量核对历史名称。冷页面不扫描订单宽表。
|
||||
45 * * * * root /usr/bin/flock -w 600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan order-data:warm-shop-candidates --quiet >/dev/null 2>&1
|
||||
45 2 * * * root /bin/sh /opt/caiwuxitong/run-finance-batch.sh order-data:warm-shop-candidates --full --quiet >/dev/null 2>&1
|
||||
# 页面字段自动跟随上游：先刷新数据字典，再把上游新增字段补进镜像并建列，页面自动多出这一列。
|
||||
40 6 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:follow-upstream-columns --quiet >/dev/null 2>&1
|
||||
0 0 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /bin/sh -c '/usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-third-party-warehouse-storage --max-batches=0 --quiet && /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:warm-report-caches --quiet' >/dev/null 2>&1
|
||||
@ -209,10 +209,9 @@ CRON_TZ=Asia/Shanghai
|
||||
# 供应商管理每天北京时间 08:15 完整校准档案、合同、联系人与打款资料。
|
||||
15 8 * * * root /usr/bin/flock -w 3600 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-supplier-management --quiet >/dev/null 2>&1
|
||||
*/2 * * * * root /usr/bin/flock -n /run/caiwu-esign-polling.lock /usr/bin/docker exec -u www-data caiwu_app php artisan esign:poll-signatures --quiet >/dev/null 2>&1
|
||||
# 科目映射表保存后自动重跑业绩明细。页面保存时会立即起后台进程，这里每分钟兜底一次，
|
||||
# 防止容器重启、请求进程退出或并发时遗漏队列任务。
|
||||
* * * * * root /usr/bin/flock -n /run/caiwu-account-mapping-resync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan store-receipt:auto-resync --quiet >/dev/null 2>&1
|
||||
28 1 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jushuitan:sync-shipping --days=3 >/dev/null 2>&1
|
||||
# 科目映射保存后保留自动重跑队列，每分钟在受限容器内串行消费。
|
||||
* * * * * root /usr/bin/flock -n /run/caiwu-account-mapping-resync.lock /bin/sh /opt/caiwuxitong/run-finance-batch.sh store-receipt:auto-resync --quiet >/dev/null 2>&1
|
||||
28 1 * * * root /bin/sh /opt/caiwuxitong/run-finance-batch.sh jushuitan:sync-shipping --days=3 >/dev/null 2>&1
|
||||
# 头条放心购（店铺收款账户账单 › 抖音账户）资金流水：每天 09:30 定点把该渠道所用
|
||||
# 原始账单镜像追平一次，保证上班前当日/前一日流水已进系统，而不是等每分钟轮转慢慢走。
|
||||
30 9 * * * root /usr/bin/flock -w 1800 /run/caiwu-upstream-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan business-mirror:sync-tables operations_replica.rpt.jst_originalBillInfo --max-batches=20 --quiet >/dev/null 2>&1
|
||||
|
||||
32
deploy/clickhouse-resource-policy.xml
Normal file
32
deploy/clickhouse-resource-policy.xml
Normal file
@ -0,0 +1,32 @@
|
||||
<clickhouse>
|
||||
<!-- Keep diagnostics bounded on the shared 8-core / 16GB host. -->
|
||||
<logger><level>information</level></logger>
|
||||
<background_schedule_pool_size>32</background_schedule_pool_size>
|
||||
<total_memory_profiler_step>67108864</total_memory_profiler_step>
|
||||
<text_log>
|
||||
<level>information</level>
|
||||
<flush_interval_milliseconds>30000</flush_interval_milliseconds>
|
||||
<max_size_rows>16384</max_size_rows>
|
||||
<reserved_size_rows>1024</reserved_size_rows>
|
||||
<buffer_size_rows_flush_threshold>8192</buffer_size_rows_flush_threshold>
|
||||
<ttl>event_date + INTERVAL 7 DAY</ttl>
|
||||
</text_log>
|
||||
<trace_log>
|
||||
<flush_interval_milliseconds>30000</flush_interval_milliseconds>
|
||||
<max_size_rows>16384</max_size_rows>
|
||||
<reserved_size_rows>1024</reserved_size_rows>
|
||||
<buffer_size_rows_flush_threshold>8192</buffer_size_rows_flush_threshold>
|
||||
<ttl>event_date + INTERVAL 7 DAY</ttl>
|
||||
</trace_log>
|
||||
<metric_log>
|
||||
<collect_interval_milliseconds>10000</collect_interval_milliseconds>
|
||||
<flush_interval_milliseconds>60000</flush_interval_milliseconds>
|
||||
<ttl>event_date + INTERVAL 7 DAY</ttl>
|
||||
</metric_log>
|
||||
<asynchronous_metric_log>
|
||||
<flush_interval_milliseconds>60000</flush_interval_milliseconds>
|
||||
<ttl>event_date + INTERVAL 7 DAY</ttl>
|
||||
</asynchronous_metric_log>
|
||||
<part_log><ttl>event_date + INTERVAL 7 DAY</ttl></part_log>
|
||||
<query_log><ttl>event_date + INTERVAL 14 DAY</ttl></query_log>
|
||||
</clickhouse>
|
||||
24
deploy/run-finance-batch.sh
Normal file
24
deploy/run-finance-batch.sh
Normal file
@ -0,0 +1,24 @@
|
||||
#!/bin/sh
|
||||
set -eu
|
||||
|
||||
REMOTE_DIR="${CAIWU_REMOTE_DIR:-/opt/caiwuxitong}"
|
||||
LOCK_FILE="${CAIWU_UPSTREAM_LOCK:-/run/caiwu-upstream-sync.lock}"
|
||||
case "${1:-}" in
|
||||
jushuitan:sync-shipping|store-receipt:order-match:sync|store-receipt:product-cost:sync|store-receipt:auto-resync|order-data:warm-shop-candidates) ;;
|
||||
*) echo 'Unsupported finance batch command' >&2; exit 64 ;;
|
||||
esac
|
||||
cd "$REMOTE_DIR"
|
||||
exec 9>"$LOCK_FILE"
|
||||
if ! flock -w "${CAIWU_BATCH_LOCK_WAIT:-1800}" 9; then
|
||||
echo 'Upstream synchronization is busy; no batch container started' >&2
|
||||
exit 75
|
||||
fi
|
||||
image="$(docker inspect caiwu_app --format '{{.Config.Image}}')"
|
||||
case "$image" in
|
||||
caiwu_akaunting:*) IMAGE_TAG="${image#*:}" ;;
|
||||
*) echo 'Unsupported active finance image; no batch container started' >&2; exit 65 ;;
|
||||
esac
|
||||
RUN_MIGRATIONS=0
|
||||
export IMAGE_TAG RUN_MIGRATIONS
|
||||
# Holds the same lock as scheduled sync and release activation for the full job.
|
||||
docker compose -f compose.jiangling.yml --profile batch run --rm -T --no-deps batch "$@"
|
||||
@ -269,6 +269,15 @@ php artisan business-mirror:sync-tables
|
||||
|
||||
首次全量、历史回补和镜像校准由任务 Codex 自主执行：先备份受影响的系统配置、镜像表及同步状态，再在生产同步节点内使用现有上游互斥锁，按稳定游标分批执行或续传 `--full`，直至完成并核对源表/镜像和实际页面。无需负责人另行授权或代跑命令；开发机驱动不支持时直接通过 SSH 使用生产容器，不以本机不能同步为由停止任务。后续无参数命令只执行或继续受控增量，不对无关数据集批量全量。
|
||||
|
||||
重型发货回填、业绩明细重算、商品成本重算和店铺候选全量核对，统一从生产节点使用受限任务入口：
|
||||
|
||||
```bash
|
||||
sh /opt/caiwuxitong/run-finance-batch.sh jushuitan:sync-shipping --from=YYYY-MM-DD --to=YYYY-MM-DD
|
||||
sh /opt/caiwuxitong/run-finance-batch.sh store-receipt:order-match:sync YYYY-MM --platform=taobao-alipay
|
||||
```
|
||||
|
||||
入口持有现有上游同步锁，使用 `batch` Profile 的独立容器（1 核、3GiB 内存、不使用 Swap、无 Web 健康检查）；正常发布不启动该 Profile。先备份再执行回补，失败时保留原有未完成状态，不重复并行启动任务。店铺候选平时按镜像同步时间增量刷新，02:45 使用 `--full` 全量核对。ClickHouse 系统日志的保留期由 `deploy/clickhouse-resource-policy.xml` 管理；业务表不受日志策略影响。
|
||||
|
||||
### 8.3 开发者需要尚未接入的上游数据时
|
||||
|
||||
系统数据库保存团队配置、数据字典和同步状态，共享业务库保存已接入业务行；二者都不是 SQL Server 网络代理。开发探索与正式页面的读取方式不同：
|
||||
|
||||
@ -5,6 +5,7 @@ namespace Modules\EcommerceFinance\Repositories;
|
||||
use Illuminate\Database\ConnectionInterface;
|
||||
use Illuminate\Database\Query\Builder;
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
use Illuminate\Support\Carbon;
|
||||
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
|
||||
use Throwable;
|
||||
|
||||
@ -13,6 +14,8 @@ class OrderDataMirrorRepository
|
||||
/** 完整店铺候选的缓存键（页面店铺多选使用）。 */
|
||||
public const SHOP_NAMES_CACHE_KEY = 'order-data.shop-candidates:v2';
|
||||
|
||||
public const SHOP_NAMES_CURSORS_KEY = 'order-data.shop-candidates:cursors:v1';
|
||||
|
||||
/** 订单号搜索一次最多解析出多少候选订单，超过就退回模糊匹配。 */
|
||||
private const ORDER_NUMBER_CANDIDATE_LIMIT = 500;
|
||||
|
||||
@ -67,21 +70,26 @@ class OrderDataMirrorRepository
|
||||
/**
|
||||
* 重新计算并缓存完整店铺候选（店铺目录 ∪ 订单镜像里出现过的店铺）。
|
||||
*
|
||||
* 只允许定时任务调用：单次要扫两张百万级宽表。
|
||||
* 按镜像同步时间增量补充；夜间 --full 对账移除已不存在的候选。
|
||||
*
|
||||
* @return array<int,string>
|
||||
*/
|
||||
public function refreshShopNamesCache(): array
|
||||
public function refreshShopNamesCache(bool $full = false): array
|
||||
{
|
||||
$cached = Cache::get(self::SHOP_NAMES_CACHE_KEY);
|
||||
$incremental = ! $full && is_array($cached);
|
||||
$cursors = $incremental ? (array) Cache::get(self::SHOP_NAMES_CURSORS_KEY, []) : [];
|
||||
[$changed, $nextCursors] = $this->computedShopNames($incremental, $cursors);
|
||||
$names = array_values(array_unique(array_merge(
|
||||
$this->computedShopNames(),
|
||||
$incremental ? $cached : [],
|
||||
$changed,
|
||||
$this->shopDirectoryNames(),
|
||||
)));
|
||||
sort($names, SORT_STRING);
|
||||
|
||||
if ($names !== []) {
|
||||
Cache::put(self::SHOP_NAMES_CACHE_KEY, $names, now()->addHours(24));
|
||||
}
|
||||
// Empty is also a valid full reconciliation result.
|
||||
Cache::put(self::SHOP_NAMES_CACHE_KEY, $names, now()->addHours(24));
|
||||
Cache::put(self::SHOP_NAMES_CURSORS_KEY, $nextCursors, now()->addHours(24));
|
||||
|
||||
return $names;
|
||||
}
|
||||
@ -184,22 +192,34 @@ class OrderDataMirrorRepository
|
||||
/**
|
||||
* 非淘系 + 淘系两个来源合并去重（淘系店铺名形如"天猫-xx/淘宝-xx"）。
|
||||
*
|
||||
* 分开各自 DISTINCT 再在 PHP 合并：UNION 会让 MariaDB 把两个上百万行的结果
|
||||
* 一起物化去重再排序，单次要两三分钟。只在定时预热时执行。
|
||||
*
|
||||
* @return array<int,string>
|
||||
* @param array<string,string|null> $cursors
|
||||
* @return array{0:list<string>,1:array<string,string|null>}
|
||||
*/
|
||||
private function computedShopNames(): array
|
||||
private function computedShopNames(bool $incremental, array $cursors): array
|
||||
{
|
||||
$names = [];
|
||||
$nextCursors = [];
|
||||
|
||||
foreach (['ops_jst_orders', 'ops_jst_qmcust_orders'] as $table) {
|
||||
$rows = $this->connection->table($table)
|
||||
$hasTimestamp = $this->connection->getSchemaBuilder()->hasColumn($table, '_mirror_synced_at');
|
||||
$high = $hasTimestamp ? $this->connection->table($table)->max('_mirror_synced_at') : null;
|
||||
$since = $cursors[$table] ?? null;
|
||||
if ($incremental && $since === null && $high !== null) {
|
||||
// Upgrade the existing 24-hour complete cache without another full scan.
|
||||
$since = Carbon::parse($high)->subDay()->format('Y-m-d H:i:s');
|
||||
}
|
||||
$query = $this->connection->table($table)
|
||||
->select('shop_name')
|
||||
->distinct()
|
||||
->whereNotNull('shop_name')
|
||||
->where('shop_name', '<>', '')
|
||||
->pluck('shop_name');
|
||||
->where('shop_name', '<>', '');
|
||||
if ($incremental && $hasTimestamp && $since !== null && $high !== null) {
|
||||
// Inclusive lower bound retains writes sharing the same timestamp.
|
||||
$query->where('_mirror_synced_at', '>=', $since)
|
||||
->where('_mirror_synced_at', '<=', $high);
|
||||
}
|
||||
$rows = $query->pluck('shop_name');
|
||||
$nextCursors[$table] = $high;
|
||||
|
||||
foreach ($this->normalizeShopNames($rows) as $name) {
|
||||
$names[$name] = true;
|
||||
@ -209,7 +229,7 @@ class OrderDataMirrorRepository
|
||||
$names = array_keys($names);
|
||||
sort($names, SORT_STRING);
|
||||
|
||||
return $names;
|
||||
return [$names, $nextCursors];
|
||||
}
|
||||
|
||||
/** @param iterable<mixed> $rows @return array<int,string> */
|
||||
|
||||
@ -100,6 +100,16 @@ class AccountMappingResyncDispatcher
|
||||
* 已有进程在跑时新进程会拿不到锁直接退出；正在跑的进程处理完当前任务后会继续取新任务。
|
||||
*/
|
||||
public function spawnWorker(): void
|
||||
{
|
||||
// Production cron consumes the durable queue in the resource-limited batch profile.
|
||||
if (config('upstream_pipeline.batch_worker_enabled', false)) {
|
||||
return;
|
||||
}
|
||||
|
||||
$this->launchWorker();
|
||||
}
|
||||
|
||||
protected function launchWorker(): void
|
||||
{
|
||||
try {
|
||||
$php = escapeshellarg(PHP_BINARY ?: 'php');
|
||||
|
||||
@ -543,6 +543,8 @@ scp_to_remote \
|
||||
"$repo_root/deploy/activate-release.sh" \
|
||||
"$repo_root/deploy/activate-incremental-release.sh" \
|
||||
"$repo_root/deploy/config-db-init.sh" \
|
||||
"$repo_root/deploy/run-finance-batch.sh" \
|
||||
"$repo_root/deploy/clickhouse-resource-policy.xml" \
|
||||
"$repo_root/deploy/update-runtime-url.sh" \
|
||||
"$repo_root/deploy/configure-dingtalk-callback.sh" \
|
||||
"$repo_root/deploy/host-nginx-8442-dingtalk.conf" \
|
||||
|
||||
@ -464,6 +464,8 @@ try {
|
||||
deploy/activate-release.sh `
|
||||
deploy/activate-incremental-release.sh `
|
||||
deploy/config-db-init.sh `
|
||||
deploy/run-finance-batch.sh `
|
||||
deploy/clickhouse-resource-policy.xml `
|
||||
deploy/update-runtime-url.sh `
|
||||
deploy/configure-dingtalk-callback.sh `
|
||||
deploy/host-nginx-8442-dingtalk.conf `
|
||||
|
||||
73
tests/Unit/Business/PlantingOrderLabelLookupTest.php
Normal file
73
tests/Unit/Business/PlantingOrderLabelLookupTest.php
Normal file
@ -0,0 +1,73 @@
|
||||
<?php
|
||||
|
||||
namespace Tests\Unit\Business;
|
||||
|
||||
use App\Business\JstOrderLabelPlantingRules;
|
||||
use App\Business\PlantingOrderLabelLookup;
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
use Illuminate\Support\Facades\DB;
|
||||
use Tests\TestCase;
|
||||
|
||||
class PlantingOrderLabelLookupTest extends TestCase
|
||||
{
|
||||
protected function setUp(): void
|
||||
{
|
||||
parent::setUp();
|
||||
DB::connection()->getPdo()->sqliteCreateFunction('FIND_IN_SET', static function ($value, $list): int {
|
||||
$index = array_search($value, explode(',', (string) $list), true);
|
||||
|
||||
return $index === false ? 0 : $index + 1;
|
||||
});
|
||||
}
|
||||
|
||||
public function test_reuses_the_exact_company_label_or_item_rule_without_scanning_again(): void
|
||||
{
|
||||
$this->rows([
|
||||
[1, '000123', 'positive', null], [1, '000123', 'positive', null],
|
||||
[1, '260901-ABC', 'other', 'A,X00006'], [1, 'NO', 'other', 'X000060'],
|
||||
[2, 'OTHER-COMPANY', 'positive', null],
|
||||
]);
|
||||
$lookup = app(PlantingOrderLabelLookup::class);
|
||||
$result = $lookup->orderNumbers(1, ['positive']);
|
||||
sort($result);
|
||||
$this->assertSame(['000123', '260901-ABC'], $result);
|
||||
DB::connection()->enableQueryLog();
|
||||
$this->assertSame($result, tap($lookup->orderNumbers(1, ['positive']), static function (&$rows): void { sort($rows); }));
|
||||
$this->assertCount(0, array_filter(DB::connection()->getQueryLog(), static fn ($q) => str_contains($q['query'], 'from "ak_jst_order_labels"')));
|
||||
}
|
||||
|
||||
public function test_shared_revision_invalidates_another_instances_local_cache(): void
|
||||
{
|
||||
$this->rows([[1, 'A', 'positive', null], [1, 'B', 'other', null]]);
|
||||
$lookup = app(PlantingOrderLabelLookup::class);
|
||||
$this->assertSame(['A'], $lookup->orderNumbers(1, ['positive']));
|
||||
DB::table('jst_order_labels')->where('order_no', 'B')->update(['item_code' => 'X00006']);
|
||||
// A different workstation changes the shared revision; local file cache remains.
|
||||
Cache::store('upstream_shared')->forever('planting-order-labels:v1:1:revision', 'other-instance');
|
||||
$result = $lookup->orderNumbers(1, ['positive']);
|
||||
sort($result);
|
||||
$this->assertSame(['A', 'B'], $result);
|
||||
}
|
||||
|
||||
public function test_rule_changes_and_item_only_classification_do_not_reuse_old_results(): void
|
||||
{
|
||||
$this->rows([[1, 'A', 'positive', null], [1, 'B', 'other', 'X00006']]);
|
||||
$lookup = app(PlantingOrderLabelLookup::class);
|
||||
$this->assertCount(2, $lookup->orderNumbers(1, ['positive']));
|
||||
$this->assertSame(['B'], $lookup->orderNumbers(1, []));
|
||||
DB::table('jst_order_labels')->where('order_no', 'B')->update(['item_code' => null]);
|
||||
$lookup->forget(1);
|
||||
$this->assertSame([], $lookup->orderNumbers(1, []));
|
||||
}
|
||||
|
||||
private function rows(array $rows): void
|
||||
{
|
||||
foreach ($rows as [$company, $number, $label, $code]) {
|
||||
DB::table('jst_order_labels')->insert([
|
||||
'company_id' => $company, 'order_no' => $number, 'label' => $label,
|
||||
'item_code' => $code, 'shop_name' => 'test', 'ordered_at' => '2026-09-01 00:00:00',
|
||||
'created_at' => now(), 'updated_at' => now(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
}
|
||||
50
tests/Unit/Deployment/FinanceBatchIsolationTest.php
Normal file
50
tests/Unit/Deployment/FinanceBatchIsolationTest.php
Normal file
@ -0,0 +1,50 @@
|
||||
<?php
|
||||
|
||||
namespace Tests\Unit\Deployment;
|
||||
|
||||
use PHPUnit\Framework\TestCase;
|
||||
use Symfony\Component\Process\Process;
|
||||
|
||||
class FinanceBatchIsolationTest extends TestCase
|
||||
{
|
||||
public function test_busy_sync_lock_prevents_starting_a_second_heavy_job(): void
|
||||
{
|
||||
$directory = sys_get_temp_dir() . '/caiwu-batch-' . bin2hex(random_bytes(6));
|
||||
mkdir($directory);
|
||||
$lock = fopen($directory . '/sync.lock', 'c');
|
||||
flock($lock, LOCK_EX);
|
||||
try {
|
||||
$process = $this->runBatch($directory, ['jushuitan:sync-shipping', '--days=3']);
|
||||
$this->assertSame(75, $process->getExitCode());
|
||||
$this->assertFileDoesNotExist($directory . '/called');
|
||||
flock($lock, LOCK_UN);
|
||||
$process = $this->runBatch($directory, ['jushuitan:sync-shipping', '--days=3']);
|
||||
$this->assertTrue($process->isSuccessful(), $process->getErrorOutput());
|
||||
$arguments = file($directory . '/called', FILE_IGNORE_NEW_LINES);
|
||||
$this->assertContains('batch', $arguments);
|
||||
$this->assertSame(['jushuitan:sync-shipping', '--days=3'], array_slice($arguments, -2));
|
||||
$this->assertContains('--no-deps', $arguments);
|
||||
$this->assertSame('image=123456789abc', $arguments[0]);
|
||||
} finally {
|
||||
fclose($lock);
|
||||
foreach (glob($directory . '/*') as $path) { unlink($path); }
|
||||
rmdir($directory);
|
||||
}
|
||||
}
|
||||
|
||||
private function runBatch(string $directory, array $arguments): Process
|
||||
{
|
||||
file_put_contents($directory . '/docker', '#!/bin/sh' . "\n"
|
||||
. 'if [ "$1" = inspect ]; then echo caiwu_akaunting:123456789abc; exit 0; fi' . "\n"
|
||||
. 'printf "%s\n" "image=$IMAGE_TAG" "$@" > "$CAIWU_CAPTURE"' . "\n");
|
||||
chmod($directory . '/docker', 0700);
|
||||
$process = new Process(array_merge(['sh', dirname(__DIR__, 3) . '/deploy/run-finance-batch.sh'], $arguments), $directory, [
|
||||
'PATH' => $directory . ':' . getenv('PATH'), 'CAIWU_REMOTE_DIR' => $directory,
|
||||
'CAIWU_UPSTREAM_LOCK' => $directory . '/sync.lock', 'CAIWU_BATCH_LOCK_WAIT' => '0.1',
|
||||
'CAIWU_CAPTURE' => $directory . '/called',
|
||||
]);
|
||||
$process->run();
|
||||
|
||||
return $process;
|
||||
}
|
||||
}
|
||||
@ -8,6 +8,36 @@ use Tests\TestCase;
|
||||
|
||||
class AccountMappingResyncDispatcherTest extends TestCase
|
||||
{
|
||||
public function test_production_leaves_queue_processing_to_the_limited_batch_worker(): void
|
||||
{
|
||||
config(['upstream_pipeline.batch_worker_enabled' => true]);
|
||||
$dispatcher = $this->recordingDispatcher();
|
||||
$dispatcher->spawnWorker();
|
||||
|
||||
$this->assertSame(0, $dispatcher->launches);
|
||||
}
|
||||
|
||||
public function test_local_development_still_launches_an_immediate_worker(): void
|
||||
{
|
||||
config(['upstream_pipeline.batch_worker_enabled' => false]);
|
||||
$dispatcher = $this->recordingDispatcher();
|
||||
$dispatcher->spawnWorker();
|
||||
|
||||
$this->assertSame(1, $dispatcher->launches);
|
||||
}
|
||||
|
||||
private function recordingDispatcher(): AccountMappingResyncDispatcher
|
||||
{
|
||||
return new class extends AccountMappingResyncDispatcher {
|
||||
public int $launches = 0;
|
||||
|
||||
protected function launchWorker(): void
|
||||
{
|
||||
$this->launches++;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
public function test_all_taobao_family_platforms_trigger_the_matching_receipt_sources(): void
|
||||
{
|
||||
$reflection = new ReflectionClass(AccountMappingResyncDispatcher::class);
|
||||
|
||||
@ -4,6 +4,7 @@ namespace Tests\Unit\EcommerceFinance;
|
||||
|
||||
use Illuminate\Database\Schema\Blueprint;
|
||||
use Illuminate\Support\Facades\DB;
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
use Illuminate\Support\Facades\Schema;
|
||||
use Mockery;
|
||||
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
|
||||
@ -160,6 +161,48 @@ class OrderDataMirrorRepositoryTest extends TestCase
|
||||
$this->assertSame(['京东-乙店', '抖音-丙店', '拼多多-甲店'], $names);
|
||||
}
|
||||
|
||||
public function test_shop_candidates_scan_only_changed_rows_and_reconcile_deleted_names_at_night(): void
|
||||
{
|
||||
$db = DB::connection('business_mirror');
|
||||
foreach (['ops_jst_orders', 'ops_jst_qmcust_orders'] as $table) {
|
||||
$db->getSchemaBuilder()->table($table, function (Blueprint $t): void {
|
||||
$t->dateTime('_mirror_synced_at')->nullable()->index();
|
||||
});
|
||||
}
|
||||
$this->insertOrder($db, 1, 'SO-A', '历史店铺');
|
||||
$db->table('ops_jst_orders')->update(['_mirror_synced_at' => '2026-10-01 00:00:00']);
|
||||
$repository = $this->repository($db);
|
||||
$this->assertSame(['历史店铺'], $repository->refreshShopNamesCache());
|
||||
$this->insertOrder($db, 2, 'SO-B', '新增店铺');
|
||||
$db->table('ops_jst_orders')->where('o_id', 2)->update(['_mirror_synced_at' => '2026-10-19 12:00:00']);
|
||||
$db->enableQueryLog();
|
||||
$this->assertSame(['历史店铺', '新增店铺'], $repository->refreshShopNamesCache());
|
||||
$scans = array_values(array_filter($db->getQueryLog(), static fn ($q) => str_contains($q['query'], 'select distinct "shop_name" from "ops_jst_orders"')));
|
||||
$this->assertCount(1, $scans);
|
||||
$this->assertStringContainsString('"_mirror_synced_at" >= ?', $scans[0]['query']);
|
||||
$this->assertSame('2026-10-01 00:00:00', $scans[0]['bindings'][1]);
|
||||
// A second write with the exact same timestamp must also be discovered.
|
||||
$this->insertOrder($db, 3, 'SO-C', '同秒新增');
|
||||
$db->table('ops_jst_orders')->where('o_id', 3)->update(['_mirror_synced_at' => '2026-10-19 12:00:00']);
|
||||
$this->assertContains('同秒新增', $repository->refreshShopNamesCache());
|
||||
$db->table('ops_jst_orders')->where('o_id', 1)->delete();
|
||||
$this->assertNotContains('历史店铺', $repository->refreshShopNamesCache(true));
|
||||
}
|
||||
|
||||
public function test_upgrade_keeps_legacy_complete_candidates_and_reads_a_day_of_changes(): void
|
||||
{
|
||||
$db = DB::connection('business_mirror');
|
||||
foreach (['ops_jst_orders', 'ops_jst_qmcust_orders'] as $table) {
|
||||
$db->getSchemaBuilder()->table($table, function (Blueprint $t): void {
|
||||
$t->dateTime('_mirror_synced_at')->nullable();
|
||||
});
|
||||
}
|
||||
Cache::put(OrderDataMirrorRepository::SHOP_NAMES_CACHE_KEY, ['历史店铺']);
|
||||
$this->insertOrder($db, 1, 'SO-A', '新增店铺');
|
||||
$db->table('ops_jst_orders')->update(['_mirror_synced_at' => '2026-10-19 12:00:00']);
|
||||
$this->assertSame(['历史店铺', '新增店铺'], $this->repository($db)->refreshShopNamesCache());
|
||||
}
|
||||
|
||||
public function test_after_sales_includes_taobao_orders_and_their_refunds(): void
|
||||
{
|
||||
$db = DB::connection('business_mirror');
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user