198 lines
9.0 KiB
PHP
198 lines
9.0 KiB
PHP
<?php
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use Illuminate\Console\Command;
|
||
use Modules\EcommerceFinance\AccountBillColumns;
|
||
use Modules\EcommerceFinance\JdBillPlatform;
|
||
use Modules\EcommerceFinance\Repositories\DewuSettlementBillRepository;
|
||
use Modules\EcommerceFinance\Repositories\JdDailyBillRepository;
|
||
use Modules\EcommerceFinance\Repositories\PinduoduoSettlementBillRepository;
|
||
use Modules\EcommerceFinance\Repositories\StoreReceivingAlipayBillMirrorRepository;
|
||
use Modules\EcommerceFinance\Repositories\StoreReceivingDewuBillMirrorRepository;
|
||
use Modules\EcommerceFinance\Repositories\StoreReceivingDouyinBillMirrorRepository;
|
||
use Modules\EcommerceFinance\Repositories\StoreReceivingJdWalletBillRepository;
|
||
use Modules\EcommerceFinance\Repositories\StoreReceivingPinduoduoBillMirrorRepository;
|
||
use Modules\EcommerceFinance\Repositories\TaobaoWechatSettlementBillRepository;
|
||
use Modules\EcommerceFinance\Services\AccountBillsExportTasks;
|
||
use Modules\EcommerceFinance\Services\AccountLedgerRemarkMap;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 后台执行「店铺收款账户账单」导出任务：生成 CSV 并更新任务状态，供页面列表下载。
|
||
*/
|
||
class RunAccountBillsExportTask extends Command
|
||
{
|
||
protected $signature = 'account-bills:export-run {token : 导出任务 token}';
|
||
|
||
protected $description = '执行店铺收款账户账单导出任务（后台进程调用）';
|
||
|
||
public function handle(AccountBillsExportTasks $tasks, AccountLedgerRemarkMap $remarkMap): int
|
||
{
|
||
$token = (string) $this->argument('token');
|
||
$task = $this->findTask($tasks, $token);
|
||
if ($task === null) {
|
||
$this->error('导出任务不存在：' . $token);
|
||
|
||
return self::FAILURE;
|
||
}
|
||
|
||
$task = $tasks->patch($task, ['status' => 'running', 'started_at' => AccountBillsExportTasks::beijingNow()]);
|
||
|
||
try {
|
||
if (! empty($task['cancel'])) {
|
||
$tasks->patch($task, ['status' => 'cancelled', 'finished_at' => AccountBillsExportTasks::beijingNow()]);
|
||
$this->info('任务已取消。');
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
$columns = array_values((array) ($task['columns'] ?? []));
|
||
if ($columns === []) {
|
||
throw new \RuntimeException('导出任务没有可用字段。');
|
||
}
|
||
|
||
$csv = fopen($tasks->csvPath($task), 'w');
|
||
if ($csv === false) {
|
||
throw new \RuntimeException('无法创建导出文件。');
|
||
}
|
||
|
||
// 让「行数」列能显示真实百分比：镜像连接改为不缓冲取数，
|
||
// 行一取到就写盘，进度按 已写行数 / 总行数 实时刷新。
|
||
$this->streamRowsWithoutBuffering($remarkMap);
|
||
|
||
fputs($csv, chr(0xEF) . chr(0xBB) . chr(0xBF));
|
||
fputcsv($csv, array_map(static fn (array $column): string => (string) ($column['label'] ?? ''), $columns));
|
||
|
||
$written = 0;
|
||
foreach ($this->rows($task, $remarkMap) as $row) {
|
||
fputcsv($csv, array_map(
|
||
static fn (array $column): string => self::cellValue($row, $column),
|
||
$columns,
|
||
));
|
||
$written++;
|
||
if ($written % 2000 === 0) {
|
||
// 用户在弹窗里点了「取消」：停止导出并删掉半成品文件。
|
||
$fresh = $tasks->find((int) $task['user_id'], (string) $task['token']);
|
||
if (! empty($fresh['cancel'])) {
|
||
fclose($csv);
|
||
@unlink($tasks->csvPath($task));
|
||
$tasks->patch($task, ['status' => 'cancelled', 'rows' => $written, 'finished_at' => AccountBillsExportTasks::beijingNow(), 'size' => 0]);
|
||
$this->info('任务已取消（已写入 ' . $written . ' 行后停止）。');
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
$task = $tasks->patch($task, ['status' => 'running', 'rows' => $written]);
|
||
}
|
||
}
|
||
fclose($csv);
|
||
|
||
$task = $tasks->patch($task, [
|
||
'status' => 'done',
|
||
'rows' => $written,
|
||
'size' => (int) @filesize($tasks->csvPath($task)),
|
||
'finished_at' => AccountBillsExportTasks::beijingNow(),
|
||
'error' => null,
|
||
]);
|
||
$this->info('导出完成：' . $written . ' 行');
|
||
|
||
return self::SUCCESS;
|
||
} catch (Throwable $exception) {
|
||
$tasks->patch($task, [
|
||
'status' => 'failed',
|
||
'finished_at' => AccountBillsExportTasks::beijingNow(),
|
||
'error' => mb_substr($exception->getMessage(), 0, 200),
|
||
]);
|
||
$this->error('导出失败：' . $exception->getMessage());
|
||
|
||
return self::FAILURE;
|
||
}
|
||
}
|
||
|
||
private function findTask(AccountBillsExportTasks $tasks, string $token): ?array
|
||
{
|
||
$root = storage_path('app/account-bills-exports');
|
||
foreach (glob($root . '/*/' . $token . '.json') ?: [] as $path) {
|
||
$task = json_decode((string) file_get_contents($path), true);
|
||
if (is_array($task) && ($task['token'] ?? null) === $token) {
|
||
return $task;
|
||
}
|
||
}
|
||
|
||
return null;
|
||
}
|
||
|
||
/** @return iterable<int,array<string,mixed>> */
|
||
private function rows(array $task, AccountLedgerRemarkMap $remarkMap): iterable
|
||
{
|
||
$tab = (string) ($task['tab'] ?? '');
|
||
$filters = (array) ($task['filters'] ?? []);
|
||
// 后台导出固定按「发生时间」取数：该列上有 (occurred_at, record_code) 联合索引，
|
||
// 数据库能按索引顺序边扫边返回，首行几百毫秒内到达，进度可实时刷新；
|
||
// 按「记录编码」走主键反向扫描过滤日期，要扫完整表才出首行，进度会长时间停在 0%。
|
||
$sort = 'occurred_at';
|
||
$dir = strtolower((string) ($filters['dir'] ?? 'desc')) === 'asc' ? 'asc' : 'desc';
|
||
unset($filters['sort'], $filters['dir'], $filters['export'], $filters['token'], $filters['total'], $filters['columns']);
|
||
|
||
$rows = match ($tab) {
|
||
'taobao-alipay' => app(StoreReceivingAlipayBillMirrorRepository::class)->exportRows($filters, $sort, $dir),
|
||
'taobao-wechat' => app(TaobaoWechatSettlementBillRepository::class)->exportAccountBills($filters, $sort, $dir),
|
||
'pinduoduo' => app(PinduoduoSettlementBillRepository::class)->exportAccountBills($filters, $sort, $dir),
|
||
'dewu' => app(DewuSettlementBillRepository::class)->exportAccountBills($filters, $sort, $dir),
|
||
'jd-transactions', 'jingxi' => JdBillPlatform::isJdBillTab($tab)
|
||
? app(JdDailyBillRepository::class)->forPlatform(JdBillPlatform::platformForBillType($tab))->exportRows($filters, $sort, $dir)
|
||
: [],
|
||
'jd-wallet' => app(StoreReceivingJdWalletBillRepository::class)->exportRows($filters, $sort, $dir),
|
||
'douyin' => app(StoreReceivingDouyinBillMirrorRepository::class)->exportRows($filters, $sort, $dir),
|
||
default => throw new \RuntimeException('该页签暂不支持后台导出：' . $tab),
|
||
};
|
||
|
||
return $remarkMap->decorateAll($rows, $tab);
|
||
}
|
||
|
||
/**
|
||
* 把镜像连接切成「不缓冲」并预热总账备注映射，使取数阶段逐批返回行。
|
||
*
|
||
* Web/CLI 默认用缓冲查询：整段结果集在数据库端算完才返回，进度会长时间停在 0%，
|
||
* 最后一次性跳到 100%。切到不缓冲后行随查询推进到达，进度即真实百分比。
|
||
*/
|
||
private function streamRowsWithoutBuffering(AccountLedgerRemarkMap $remarkMap): void
|
||
{
|
||
try {
|
||
// 预热：避免逐行装饰时中途再去读缓存/文件。
|
||
$remarkMap->map();
|
||
|
||
$connection = app(\Illuminate\Database\DatabaseManager::class)
|
||
->connection((string) config('upstream_mirrors.connection', 'business_mirror'));
|
||
if (! $connection instanceof \Illuminate\Database\MySqlConnection) {
|
||
return;
|
||
}
|
||
|
||
$pdo = $connection->getPdo();
|
||
$pdo->setAttribute(\PDO::MYSQL_ATTR_USE_BUFFERED_QUERY, false);
|
||
// 读连接与写连接是两套句柄，这里统一成同一个不缓冲句柄。
|
||
$connection->setReadPdo($pdo);
|
||
} catch (Throwable) {
|
||
// 拿不到句柄时退回默认行为，不影响导出结果。
|
||
}
|
||
}
|
||
|
||
/** @param array<string,mixed> $row */
|
||
private static function cellValue(array $row, array $column): string
|
||
{
|
||
$field = (string) ($column['field'] ?? '');
|
||
$value = str_starts_with($field, 'payload.')
|
||
? data_get($row['payload'] ?? [], substr($field, 8))
|
||
: data_get($row, $field);
|
||
|
||
$text = trim((string) ($value ?? ''));
|
||
if ($text === '') {
|
||
return '';
|
||
}
|
||
if (preg_match('/^\d{12,}$/', $text) === 1) {
|
||
return '="' . $text . '"';
|
||
}
|
||
|
||
return preg_match('/^[=+\-@]/u', $text) === 1 ? "'" . $text : $text;
|
||
}
|
||
}
|