科目映射表保存后自动重跑业绩明细
映射表保存、删除、恢复、导入和重置后,自动把受影响平台已存在的月份排入后台队列;后台按平台×月份逐个独立进程重跑,数据库锁防重复,生产 cron 每分钟兜底。保存响应直接提示排队任务数。
This commit is contained in:
parent
664fc05faf
commit
cbbf32bb5c
107
app/Console/Commands/RunAccountMappingResync.php
Normal file
107
app/Console/Commands/RunAccountMappingResync.php
Normal file
@ -0,0 +1,107 @@
|
||||
<?php
|
||||
|
||||
namespace App\Console\Commands;
|
||||
|
||||
use Illuminate\Console\Command;
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
use Illuminate\Support\Facades\DB;
|
||||
use Illuminate\Support\Facades\Log;
|
||||
use Modules\EcommerceFinance\Services\AccountMappingResyncDispatcher;
|
||||
use Symfony\Component\Process\Process;
|
||||
use Throwable;
|
||||
|
||||
/**
|
||||
* 科目映射表变更后的业绩明细自动重跑队列。
|
||||
*
|
||||
* 一个常驻进程依次处理队列里的「平台 × 月份」;每个实际同步放在独立 PHP 子进程中,
|
||||
* 避免大平台重跑时的内存累积。也可手动执行本命令补跑。
|
||||
*/
|
||||
class RunAccountMappingResync extends Command
|
||||
{
|
||||
protected $signature = 'store-receipt:auto-resync';
|
||||
|
||||
protected $description = '处理科目映射表变更触发的业绩明细重跑队列';
|
||||
|
||||
public function handle(AccountMappingResyncDispatcher $dispatcher): int
|
||||
{
|
||||
// 页面(Apache)与每分钟 cron 都以 www-data 运行,使用共享数据库锁,
|
||||
// 不受 storage/framework/cache 文件属主影响,也能避免多容器重复消费。
|
||||
$lock = Cache::store('upstream_shared')->lock('store-receipt:auto-resync', 86400);
|
||||
if (! $lock->get()) {
|
||||
// 已有进程在跑;本次新入队的任务会在当前进程下一轮循环或每分钟兜底时执行。
|
||||
return self::SUCCESS;
|
||||
}
|
||||
|
||||
try {
|
||||
// 进程崩溃遗留的 running 任务超过 60 分钟视为可重试。
|
||||
DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||||
->where('status', 'running')
|
||||
->where('updated_at', '<', now()->subMinutes(60))
|
||||
->update(['status' => 'pending', 'updated_at' => now()]);
|
||||
|
||||
while (true) {
|
||||
$job = DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||||
->where('status', 'pending')
|
||||
->orderBy('id')
|
||||
->first();
|
||||
if ($job === null) {
|
||||
break;
|
||||
}
|
||||
|
||||
DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||||
->where('id', $job->id)
|
||||
->where('status', 'pending')
|
||||
->update([
|
||||
'status' => 'running',
|
||||
'attempts' => DB::raw('attempts + 1'),
|
||||
'updated_at' => now(),
|
||||
]);
|
||||
|
||||
try {
|
||||
$process = Process::fromShellCommandline(
|
||||
implode(' ', [
|
||||
escapeshellarg(PHP_BINARY ?: 'php'),
|
||||
'artisan',
|
||||
'store-receipt:order-match:sync',
|
||||
escapeshellarg((string) $job->month),
|
||||
'--platform=' . escapeshellarg((string) $job->platform),
|
||||
]),
|
||||
base_path()
|
||||
);
|
||||
$process->setTimeout(7200);
|
||||
$process->run();
|
||||
$output = trim($process->getOutput() . "\n" . $process->getErrorOutput());
|
||||
|
||||
// 同步期间又保存了映射时,dispatchFor 已把状态改回 pending;
|
||||
// 这里只在仍为 running 时写 done,不能吞掉新请求。
|
||||
DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||||
->where('id', $job->id)
|
||||
->where('status', 'running')
|
||||
->update([
|
||||
'status' => $process->isSuccessful() ? 'done' : 'failed',
|
||||
'message' => mb_substr(
|
||||
$output !== '' ? $output : ($process->isSuccessful() ? 'OK' : '同步命令执行失败'),
|
||||
0,
|
||||
1000
|
||||
),
|
||||
'updated_at' => now(),
|
||||
]);
|
||||
} catch (Throwable $exception) {
|
||||
Log::error('科目映射表自动重跑失败:' . $exception->getMessage());
|
||||
DB::table(AccountMappingResyncDispatcher::JOBS_TABLE)
|
||||
->where('id', $job->id)
|
||||
->where('status', 'running')
|
||||
->update([
|
||||
'status' => 'failed',
|
||||
'message' => mb_substr($exception->getMessage(), 0, 1000),
|
||||
'updated_at' => now(),
|
||||
]);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
$lock->release();
|
||||
}
|
||||
|
||||
return self::SUCCESS;
|
||||
}
|
||||
}
|
||||
@ -12,6 +12,7 @@ use Modules\EcommerceFinance\Services\AccountMappingCoa;
|
||||
use Modules\EcommerceFinance\Services\AccountMappingStore;
|
||||
use Modules\EcommerceFinance\Services\AccountMappingValidator;
|
||||
use Modules\EcommerceFinance\Services\AccountMappingWorkbookImporter;
|
||||
use Modules\EcommerceFinance\Services\AccountMappingResyncDispatcher;
|
||||
use RuntimeException;
|
||||
use Symfony\Component\HttpFoundation\StreamedResponse;
|
||||
|
||||
@ -74,27 +75,36 @@ class AccountMappingActions extends Controller
|
||||
|
||||
$this->store->create($values, $this->userId());
|
||||
|
||||
return response()->json(['message' => '已新增 1 条规则。'] + $this->payload());
|
||||
$hint = $this->dispatchResync([(string) ($values['platform'] ?? '')]);
|
||||
|
||||
return response()->json(['message' => '已新增 1 条规则。' . $hint] + $this->payload());
|
||||
}
|
||||
|
||||
public function update(Request $request, int $id): JsonResponse
|
||||
{
|
||||
if ($this->store->find($id) === null) {
|
||||
$before = $this->store->find($id);
|
||||
if ($before === null) {
|
||||
throw ValidationException::withMessages(['id' => '这条规则已经不存在了,请刷新页面。']);
|
||||
}
|
||||
|
||||
$this->store->update($id, $this->payloadValues($request), $this->userId());
|
||||
$values = $this->payloadValues($request);
|
||||
$this->store->update($id, $values, $this->userId());
|
||||
|
||||
return response()->json(['message' => '已保存。'] + $this->payload());
|
||||
$hint = $this->dispatchResync([(string) ($before['platform'] ?? ''), (string) ($values['platform'] ?? '')]);
|
||||
|
||||
return response()->json(['message' => '已保存。' . $hint] + $this->payload());
|
||||
}
|
||||
|
||||
public function destroy(int $id): JsonResponse
|
||||
{
|
||||
$before = $this->store->find($id);
|
||||
if (! $this->store->destroy($id, $this->userId())) {
|
||||
throw ValidationException::withMessages(['id' => '这条规则已经不存在了,请刷新页面。']);
|
||||
}
|
||||
|
||||
return response()->json(['message' => '已删除 1 条规则。'] + $this->payload());
|
||||
$hint = $this->dispatchResync([(string) ($before['platform'] ?? '')]);
|
||||
|
||||
return response()->json(['message' => '已删除 1 条规则。' . $hint] + $this->payload());
|
||||
}
|
||||
|
||||
/**
|
||||
@ -181,21 +191,24 @@ class AccountMappingActions extends Controller
|
||||
}
|
||||
|
||||
$restored = $this->store->create($before, $this->userId());
|
||||
$hint = $this->dispatchResync([(string) ($restored['platform'] ?? '')]);
|
||||
|
||||
return response()->json([
|
||||
'message' => '已恢复「' . $restored['platform'] . ' · ' . $restored['item'] . '」(新记录 #' . (int) $restored['id'] . ')。',
|
||||
'message' => '已恢复「' . $restored['platform'] . ' · ' . $restored['item'] . '」(新记录 #' . (int) $restored['id'] . ')。' . $hint,
|
||||
] + $this->payload());
|
||||
}
|
||||
|
||||
if ($action === 'update') {
|
||||
if ($this->store->find($ruleId) === null) {
|
||||
$current = $this->store->find($ruleId);
|
||||
if ($current === null) {
|
||||
throw ValidationException::withMessages(['history' => '这条规则已经被删掉了,请用同一条的「删除」记录恢复。']);
|
||||
}
|
||||
|
||||
$this->store->update($ruleId, $before, $this->userId());
|
||||
$hint = $this->dispatchResync([(string) ($current['platform'] ?? ''), (string) ($before['platform'] ?? '')]);
|
||||
|
||||
return response()->json([
|
||||
'message' => '已恢复成这次修改前的内容(记录 #' . $ruleId . ')。',
|
||||
'message' => '已恢复成这次修改前的内容(记录 #' . $ruleId . ')。' . $hint,
|
||||
] + $this->payload());
|
||||
}
|
||||
|
||||
@ -227,6 +240,11 @@ class AccountMappingActions extends Controller
|
||||
$parsed = $this->importer->parse($file);
|
||||
$result = $this->store->importRows($parsed['rows'], $request->boolean('overwrite'), $this->userId());
|
||||
$message = "导入完成:识别 {$result['total']} 行,新增 {$result['created']} 条,更新 {$result['updated']} 条,跳过重复 {$result['skipped']} 条。";
|
||||
$platforms = array_values(array_unique(array_map(
|
||||
static fn (array $row): string => (string) ($row['platform'] ?? ''),
|
||||
$parsed['rows']
|
||||
)));
|
||||
$message .= $this->dispatchResync($platforms);
|
||||
|
||||
return response()->json(['message' => $message, 'result' => $result] + $this->payload());
|
||||
}
|
||||
@ -302,7 +320,7 @@ class AccountMappingActions extends Controller
|
||||
{
|
||||
$total = $this->store->reset($this->userId());
|
||||
|
||||
return response()->json(['message' => "已恢复为初始 {$total} 条规则。"] + $this->payload());
|
||||
return response()->json(['message' => "已恢复为初始 {$total} 条规则。" . $this->dispatchResyncAll()] + $this->payload());
|
||||
}
|
||||
|
||||
/** 账套科目(一级 + 各一级下的明细),供下拉使用 */
|
||||
@ -406,6 +424,44 @@ class AccountMappingActions extends Controller
|
||||
return user_id() ? (int) user_id() : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 保存映射后把受影响的平台排进后台重跑队列。
|
||||
*
|
||||
* 映射表本身是实时读取的,但业绩明细的优惠/冲减/退款金额是同步时落库的,
|
||||
* 所以保存成功后需要自动重跑对应平台和月份。队列创建失败不能把已经保存成功的规则回滚。
|
||||
*
|
||||
* @param list<string> $platforms
|
||||
*/
|
||||
private function dispatchResync(array $platforms): string
|
||||
{
|
||||
try {
|
||||
$result = app(AccountMappingResyncDispatcher::class)->dispatchFor($platforms);
|
||||
} catch (\Throwable $exception) {
|
||||
report($exception);
|
||||
|
||||
return ' 业绩明细自动重跑任务创建失败,请稍后重试保存或联系管理员。';
|
||||
}
|
||||
|
||||
return $result['jobs'] > 0
|
||||
? " 业绩明细已自动排队重跑:{$result['jobs']} 个平台×月份任务,后台依次执行。"
|
||||
: '';
|
||||
}
|
||||
|
||||
private function dispatchResyncAll(): string
|
||||
{
|
||||
try {
|
||||
$result = app(AccountMappingResyncDispatcher::class)->dispatchAll();
|
||||
} catch (\Throwable $exception) {
|
||||
report($exception);
|
||||
|
||||
return ' 业绩明细自动重跑任务创建失败,请稍后重试或联系管理员。';
|
||||
}
|
||||
|
||||
return $result['jobs'] > 0
|
||||
? " 业绩明细已自动排队重跑:{$result['jobs']} 个平台×月份任务,后台依次执行。"
|
||||
: '';
|
||||
}
|
||||
|
||||
/**
|
||||
* 原始操作记录 + 页面/CSV 直接可用的字段(变更内容与操作人只在服务端算一次)。
|
||||
*
|
||||
|
||||
@ -0,0 +1,33 @@
|
||||
<?php
|
||||
|
||||
use Illuminate\Database\Migrations\Migration;
|
||||
use Illuminate\Database\Schema\Blueprint;
|
||||
use Illuminate\Support\Facades\Schema;
|
||||
|
||||
/**
|
||||
* 科目映射表变更后,自动重跑业绩明细的待办队列。
|
||||
*
|
||||
* 一条 = 一个「同步平台 × 月份」,由后台命令 store-receipt:auto-resync 逐个消费。
|
||||
*/
|
||||
return new class extends Migration
|
||||
{
|
||||
public function up(): void
|
||||
{
|
||||
Schema::create('ec_account_mapping_resync_jobs', function (Blueprint $table): void {
|
||||
$table->bigIncrements('id');
|
||||
$table->string('platform', 64);
|
||||
$table->string('month', 7);
|
||||
$table->string('status', 16)->default('pending');
|
||||
$table->unsignedInteger('attempts')->default(0);
|
||||
$table->text('message')->nullable();
|
||||
$table->timestamps();
|
||||
|
||||
$table->unique(['platform', 'month'], 'ec_account_mapping_resync_unique');
|
||||
});
|
||||
}
|
||||
|
||||
public function down(): void
|
||||
{
|
||||
Schema::dropIfExists('ec_account_mapping_resync_jobs');
|
||||
}
|
||||
};
|
||||
@ -205,6 +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
|
||||
# 头条放心购(店铺收款账户账单 › 抖音账户)资金流水:每天 09:30 定点把该渠道所用
|
||||
# 原始账单镜像追平一次,保证上班前当日/前一日流水已进系统,而不是等每分钟轮转慢慢走。
|
||||
|
||||
@ -209,6 +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
|
||||
# 头条放心购(店铺收款账户账单 › 抖音账户)资金流水:每天 09:30 定点把该渠道所用
|
||||
# 原始账单镜像追平一次,保证上班前当日/前一日流水已进系统,而不是等每分钟轮转慢慢走。
|
||||
|
||||
@ -0,0 +1,118 @@
|
||||
<?php
|
||||
|
||||
namespace Modules\EcommerceFinance\Services;
|
||||
|
||||
use Illuminate\Support\Facades\DB;
|
||||
use Illuminate\Support\Facades\Log;
|
||||
use Symfony\Component\Process\Process;
|
||||
use Throwable;
|
||||
|
||||
/**
|
||||
* 科目映射表变更后,把受影响的「业绩明细」平台×月份放进后台重跑队列。
|
||||
*
|
||||
* 业绩明细的优惠/冲减/退款金额是同步时按映射表算好落库的,改映射必须重跑对应
|
||||
* 平台×月份才会更新;这里把重跑做成后台自动执行,页面保存后无需人工执行命令。
|
||||
*/
|
||||
class AccountMappingResyncDispatcher
|
||||
{
|
||||
public const JOBS_TABLE = 'ec_account_mapping_resync_jobs';
|
||||
|
||||
/**
|
||||
* 科目映射表的平台名 → [同步命令 platform 键, 业绩明细平台名]。
|
||||
* 只保留已接入业绩同步的平台;其它平台(如唯品会JX)没有同步任务。
|
||||
*/
|
||||
private const PLATFORM_MAP = [
|
||||
'淘宝' => [['taobao-alipay', '淘系支付宝'], ['taobao-wechat', '淘系微信']],
|
||||
'天猫' => [['taobao-alipay', '淘系支付宝'], ['taobao-wechat', '淘系微信']],
|
||||
'阿里巴巴' => [['taobao-alipay', '淘系支付宝']],
|
||||
'拼多多' => [['pinduoduo', '拼多多']],
|
||||
'京喜' => [['jingxi', '京喜账单']],
|
||||
'京东' => [['jd-transactions', '京东账单'], ['jd-wallet', '京东网银钱包']],
|
||||
'抖音' => [['douyin', '抖音']],
|
||||
'得物' => [['dewu', '得物']],
|
||||
];
|
||||
|
||||
/**
|
||||
* @param list<string> $mappingPlatforms 科目映射表里的平台名
|
||||
* @return array{jobs:int}
|
||||
*/
|
||||
public function dispatchFor(array $mappingPlatforms): array
|
||||
{
|
||||
$targets = [];
|
||||
foreach ($mappingPlatforms as $platform) {
|
||||
foreach (self::PLATFORM_MAP[trim((string) $platform)] ?? [] as [$key, $label]) {
|
||||
$targets[$key] = $label;
|
||||
}
|
||||
}
|
||||
if ($targets === []) {
|
||||
return ['jobs' => 0];
|
||||
}
|
||||
|
||||
$mirror = DB::connection((string) config('upstream_mirrors.connection', 'business_mirror'));
|
||||
$months = $mirror->table('store_receipt_order_match_rows')
|
||||
->whereIn('platform', array_values($targets))
|
||||
->selectRaw("DATE_FORMAT(report_month, '%Y-%m') as ym")
|
||||
->distinct()
|
||||
->orderBy('ym')
|
||||
->pluck('ym')
|
||||
->all();
|
||||
if ($months === []) {
|
||||
return ['jobs' => 0];
|
||||
}
|
||||
|
||||
$now = now();
|
||||
$jobs = 0;
|
||||
foreach ($targets as $key => $label) {
|
||||
foreach ($months as $month) {
|
||||
// running 时再次保存映射,会把状态改回 pending;当前进程完成时发现状态已变,
|
||||
// 不会覆盖成 done,于是会紧接着再跑一遍,避免漏掉刚保存的新规则。
|
||||
$affected = DB::table(self::JOBS_TABLE)->updateOrInsert(
|
||||
['platform' => (string) $key, 'month' => (string) $month],
|
||||
[
|
||||
'status' => 'pending',
|
||||
'message' => null,
|
||||
'updated_at' => $now,
|
||||
'created_at' => $now,
|
||||
]
|
||||
);
|
||||
if ($affected) {
|
||||
$jobs++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if ($jobs > 0) {
|
||||
$this->spawnWorker();
|
||||
}
|
||||
|
||||
return ['jobs' => $jobs];
|
||||
}
|
||||
|
||||
/** @return array{jobs:int} 所有已接入平台一起重跑(用于「恢复初始 112 条」) */
|
||||
public function dispatchAll(): array
|
||||
{
|
||||
return $this->dispatchFor(array_keys(self::PLATFORM_MAP));
|
||||
}
|
||||
|
||||
/**
|
||||
* 起一个后台进程处理队列。
|
||||
*
|
||||
* 已有进程在跑时新进程会拿不到锁直接退出;正在跑的进程处理完当前任务后会继续取新任务。
|
||||
*/
|
||||
public function spawnWorker(): void
|
||||
{
|
||||
try {
|
||||
$php = escapeshellarg(PHP_BINARY ?: 'php');
|
||||
$command = 'cd ' . escapeshellarg(base_path())
|
||||
. ' && nohup ' . $php . ' artisan store-receipt:auto-resync >/dev/null 2>&1 &';
|
||||
|
||||
Process::fromShellCommandline($command, base_path())
|
||||
->disableOutput()
|
||||
->setTimeout(10)
|
||||
->run();
|
||||
} catch (Throwable $exception) {
|
||||
// 后台进程起不来时任务仍留在队列里,可手动执行 store-receipt:auto-resync 补跑。
|
||||
Log::warning('科目映射表自动重跑进程启动失败:' . $exception->getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,33 @@
|
||||
<?php
|
||||
|
||||
namespace Tests\Unit\EcommerceFinance;
|
||||
|
||||
use Modules\EcommerceFinance\Services\AccountMappingResyncDispatcher;
|
||||
use ReflectionClass;
|
||||
use Tests\TestCase;
|
||||
|
||||
class AccountMappingResyncDispatcherTest extends TestCase
|
||||
{
|
||||
public function test_all_taobao_family_platforms_trigger_the_matching_receipt_sources(): void
|
||||
{
|
||||
$reflection = new ReflectionClass(AccountMappingResyncDispatcher::class);
|
||||
$map = $reflection->getConstant('PLATFORM_MAP');
|
||||
|
||||
$this->assertSame(
|
||||
[['taobao-alipay', '淘系支付宝'], ['taobao-wechat', '淘系微信']],
|
||||
$map['淘宝']
|
||||
);
|
||||
$this->assertSame(
|
||||
[['taobao-alipay', '淘系支付宝'], ['taobao-wechat', '淘系微信']],
|
||||
$map['天猫']
|
||||
);
|
||||
$this->assertSame([['taobao-alipay', '淘系支付宝']], $map['阿里巴巴']);
|
||||
}
|
||||
|
||||
public function test_platform_without_receipt_sync_is_a_noop(): void
|
||||
{
|
||||
$result = app(AccountMappingResyncDispatcher::class)->dispatchFor(['唯品会JX']);
|
||||
|
||||
$this->assertSame(['jobs' => 0], $result);
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user