323 lines
11 KiB
PHP
323 lines
11 KiB
PHP
<?php
|
||
|
||
namespace App\Business;
|
||
|
||
use Illuminate\Database\DatabaseManager;
|
||
use Illuminate\Support\Carbon;
|
||
use InvalidArgumentException;
|
||
use RuntimeException;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 聚水潭订单标签明细的每日同步。
|
||
*
|
||
* 先核对历史日期是否逐日覆盖，再补齐缺口，最后替换刷新最近数日以吸收上游迟到修改。
|
||
*/
|
||
class JstOrderLabelDailySynchronizer
|
||
{
|
||
public const SOURCE_DATASET = 'operations_replica.rpt.jst_Orders';
|
||
|
||
public function __construct(
|
||
private readonly DatabaseManager $databases,
|
||
private readonly JstOrderLabelImporter $importer,
|
||
) {}
|
||
|
||
/**
|
||
* @return array<string,mixed>
|
||
*/
|
||
public function sync(
|
||
string $from,
|
||
string $to,
|
||
int $companyId,
|
||
int $refreshDays = 7,
|
||
bool $dryRun = false,
|
||
): array {
|
||
[$fromDate, $toDate] = $this->dateRange($from, $to);
|
||
if ($companyId <= 0) {
|
||
throw new InvalidArgumentException('公司 ID 无效。');
|
||
}
|
||
if ($refreshDays < 1 || $refreshDays > 31) {
|
||
throw new InvalidArgumentException('刷新天数必须在 1 至 31 之间。');
|
||
}
|
||
|
||
$this->assertSourceReady();
|
||
|
||
$refreshStart = $fromDate->copy();
|
||
$candidateRefreshStart = $toDate->copy()->subDays($refreshDays - 1);
|
||
if ($candidateRefreshStart->gt($refreshStart)) {
|
||
$refreshStart = $candidateRefreshStart;
|
||
}
|
||
|
||
$historyEnd = $refreshStart->copy()->subDay();
|
||
$historyCoverage = $historyEnd->lt($fromDate)
|
||
? []
|
||
: $this->coverage($fromDate, $historyEnd, $companyId);
|
||
$historyIssues = array_values(array_filter(
|
||
$historyCoverage,
|
||
static fn (array $row): bool => $row['status'] !== 'ok',
|
||
));
|
||
|
||
foreach ($historyIssues as $issue) {
|
||
if ($issue['status'] === 'source_empty') {
|
||
throw new RuntimeException(sprintf(
|
||
'历史数据检查未通过：源订单镜像在 %s 没有订单，已停止拉取新数据。',
|
||
$issue['date'],
|
||
));
|
||
}
|
||
}
|
||
|
||
$backfills = $this->contiguousRanges($historyIssues);
|
||
if ($dryRun) {
|
||
return [
|
||
'status' => $historyIssues === [] ? 'would_refresh' : 'would_backfill',
|
||
'from' => $fromDate->toDateString(),
|
||
'to' => $toDate->toDateString(),
|
||
'checked_days' => (int) $fromDate->diffInDays($toDate) + 1,
|
||
'history_issues' => $historyIssues,
|
||
'backfills' => array_map(static fn (array $range): array => [
|
||
'from' => $range['from'],
|
||
'to' => $range['to'],
|
||
], $backfills),
|
||
'refresh' => [
|
||
'from' => $refreshStart->toDateString(),
|
||
'to' => $toDate->toDateString(),
|
||
],
|
||
];
|
||
}
|
||
|
||
$backfillResults = [];
|
||
foreach ($backfills as $range) {
|
||
$backfillResults[] = $this->importer->import(
|
||
$range['from'],
|
||
$range['to'],
|
||
$companyId,
|
||
true,
|
||
'jst-order-label-daily-sync',
|
||
);
|
||
}
|
||
|
||
if ($historyIssues !== []) {
|
||
$remainingHistory = $this->coverage($fromDate, $historyEnd, $companyId);
|
||
$remainingIssues = array_values(array_filter(
|
||
$remainingHistory,
|
||
static fn (array $row): bool => $row['status'] !== 'ok',
|
||
));
|
||
if ($remainingIssues !== []) {
|
||
throw new RuntimeException('历史缺口补导后仍未通过覆盖检查：' . $this->issueSummary($remainingIssues));
|
||
}
|
||
}
|
||
|
||
$refreshResult = $this->importer->import(
|
||
$refreshStart->toDateString(),
|
||
$toDate->toDateString(),
|
||
$companyId,
|
||
true,
|
||
'jst-order-label-daily-sync',
|
||
);
|
||
|
||
$finalCoverage = $this->coverage($fromDate, $toDate, $companyId);
|
||
$finalIssues = array_values(array_filter(
|
||
$finalCoverage,
|
||
static fn (array $row): bool => $row['status'] !== 'ok',
|
||
));
|
||
if ($finalIssues !== []) {
|
||
throw new RuntimeException('同步后覆盖检查未通过：' . $this->issueSummary($finalIssues));
|
||
}
|
||
|
||
return [
|
||
'status' => 'ok',
|
||
'from' => $fromDate->toDateString(),
|
||
'to' => $toDate->toDateString(),
|
||
'checked_days' => count($finalCoverage),
|
||
'backfilled_ranges' => array_map(static fn (array $range): array => [
|
||
'from' => $range['from'],
|
||
'to' => $range['to'],
|
||
], $backfills),
|
||
'backfill_results' => $backfillResults,
|
||
'refresh' => $refreshResult,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 只读确认指定日期都已覆盖；任何缺日或订单数不一致都会阻止下游重算。
|
||
*
|
||
* @return list<array{date:string,source_orders:int,target_orders:int,target_rows:int,status:string}>
|
||
*/
|
||
public function assertCovered(string $from, string $to, int $companyId): array
|
||
{
|
||
[$fromDate, $toDate] = $this->dateRange($from, $to);
|
||
if ($companyId <= 0) {
|
||
throw new InvalidArgumentException('公司 ID 无效。');
|
||
}
|
||
|
||
$this->assertSourceReady();
|
||
$coverage = $this->coverage($fromDate, $toDate, $companyId);
|
||
$issues = array_values(array_filter(
|
||
$coverage,
|
||
static fn (array $row): bool => $row['status'] !== 'ok',
|
||
));
|
||
if ($issues !== []) {
|
||
throw new RuntimeException('历史数据检查未通过：' . $this->issueSummary($issues));
|
||
}
|
||
|
||
return $coverage;
|
||
}
|
||
|
||
/**
|
||
* 逐日比较源订单号数与标签明细订单号数。
|
||
*
|
||
* 标签拆行后同一订单会出现多行，因此不能用总行数直接比较订单完整性。
|
||
*
|
||
* @return list<array{
|
||
* date:string,source_orders:int,target_orders:int,target_rows:int,status:string
|
||
* }>
|
||
*/
|
||
public function coverage(Carbon $from, Carbon $to, int $companyId): array
|
||
{
|
||
$start = $from->copy()->startOfDay()->toDateTimeString();
|
||
$end = $to->copy()->addDay()->startOfDay()->toDateTimeString();
|
||
$mirror = $this->mirror();
|
||
$target = $this->databases->connection();
|
||
|
||
$sourceRows = $mirror->table('ops_jst_orders')
|
||
->where('order_date', '>=', $start)
|
||
->where('order_date', '<', $end)
|
||
->selectRaw('DATE(order_date) AS day, COUNT(DISTINCT so_id) AS order_count')
|
||
->groupByRaw('DATE(order_date)')
|
||
->get()
|
||
->keyBy('day');
|
||
|
||
$targetRows = $target->table('jst_order_labels')
|
||
->where('company_id', $companyId)
|
||
->where('ordered_at', '>=', $start)
|
||
->where('ordered_at', '<', $end)
|
||
->selectRaw('DATE(ordered_at) AS day, COUNT(*) AS row_count, COUNT(DISTINCT order_no) AS order_count')
|
||
->groupByRaw('DATE(ordered_at)')
|
||
->get()
|
||
->keyBy('day');
|
||
|
||
$coverage = [];
|
||
for ($day = $from->copy()->startOfDay(); $day->lte($to); $day->addDay()) {
|
||
$date = $day->toDateString();
|
||
$sourceOrders = (int) ($sourceRows[$date]->order_count ?? 0);
|
||
$targetOrderCount = (int) ($targetRows[$date]->order_count ?? 0);
|
||
$targetRowsCount = (int) ($targetRows[$date]->row_count ?? 0);
|
||
$status = match (true) {
|
||
$sourceOrders === 0 => 'source_empty',
|
||
$targetOrderCount === 0 => 'missing',
|
||
$targetOrderCount < $sourceOrders => 'incomplete',
|
||
$targetOrderCount > $sourceOrders => 'extra_orders',
|
||
default => 'ok',
|
||
};
|
||
|
||
$coverage[] = [
|
||
'date' => $date,
|
||
'source_orders' => $sourceOrders,
|
||
'target_orders' => $targetOrderCount,
|
||
'target_rows' => $targetRowsCount,
|
||
'status' => $status,
|
||
];
|
||
}
|
||
|
||
return $coverage;
|
||
}
|
||
|
||
private function assertSourceReady(): void
|
||
{
|
||
$mirror = $this->mirror();
|
||
$schema = $mirror->getSchemaBuilder();
|
||
if (! $schema->hasTable('ops_jst_orders')) {
|
||
throw new RuntimeException('聚水潭订单镜像尚未建立（ops_jst_orders）。');
|
||
}
|
||
if (! $schema->hasTable('business_mirror_table_states')) {
|
||
throw new RuntimeException('聚水潭订单镜像同步状态表尚未建立。');
|
||
}
|
||
|
||
$state = $mirror->table('business_mirror_table_states')
|
||
->where('dataset_key', self::SOURCE_DATASET)
|
||
->first(['status', 'initial_completed_at']);
|
||
if ($state === null || $state->initial_completed_at === null) {
|
||
throw new RuntimeException('聚水潭订单镜像尚未完成首次同步。');
|
||
}
|
||
if ((string) $state->status !== 'succeeded') {
|
||
throw new RuntimeException('聚水潭订单镜像最近同步未成功，已停止标签明细同步。');
|
||
}
|
||
}
|
||
|
||
private function mirror(): \Illuminate\Database\ConnectionInterface
|
||
{
|
||
return $this->databases->connection(
|
||
(string) config('upstream_mirrors.connection', 'business_mirror'),
|
||
);
|
||
}
|
||
|
||
/**
|
||
* @param list<array{date:string}> $issues
|
||
* @return list<array{from:string,to:string}>
|
||
*/
|
||
private function contiguousRanges(array $issues): array
|
||
{
|
||
usort($issues, static fn (array $left, array $right): int => $left['date'] <=> $right['date']);
|
||
$ranges = [];
|
||
$start = null;
|
||
$previous = null;
|
||
|
||
foreach ($issues as $issue) {
|
||
$day = Carbon::createFromFormat('!Y-m-d', $issue['date'], 'Asia/Shanghai');
|
||
if ($start === null) {
|
||
$start = $day;
|
||
$previous = $day;
|
||
continue;
|
||
}
|
||
|
||
if (! $day->isSameDay($previous->copy()->addDay())) {
|
||
$ranges[] = ['from' => $start->toDateString(), 'to' => $previous->toDateString()];
|
||
$start = $day;
|
||
}
|
||
$previous = $day;
|
||
}
|
||
|
||
if ($start !== null && $previous !== null) {
|
||
$ranges[] = ['from' => $start->toDateString(), 'to' => $previous->toDateString()];
|
||
}
|
||
|
||
return $ranges;
|
||
}
|
||
|
||
/** @param list<array{date:string,status:string}> $issues */
|
||
private function issueSummary(array $issues): string
|
||
{
|
||
return implode('，', array_map(
|
||
static fn (array $issue): string => $issue['date'] . '=' . $issue['status'],
|
||
$issues,
|
||
));
|
||
}
|
||
|
||
/** @return array{0:Carbon,1:Carbon} */
|
||
private function dateRange(string $from, string $to): array
|
||
{
|
||
$fromDate = $this->date($from);
|
||
$toDate = $this->date($to);
|
||
if ($fromDate === null || $toDate === null || $fromDate->gt($toDate)) {
|
||
throw new InvalidArgumentException('日期范围无效，应为 from <= to，格式 YYYY-MM-DD。');
|
||
}
|
||
|
||
return [$fromDate, $toDate];
|
||
}
|
||
|
||
private function date(string $value): ?Carbon
|
||
{
|
||
if (preg_match('/^\d{4}-\d{2}-\d{2}$/', $value) !== 1) {
|
||
return null;
|
||
}
|
||
|
||
try {
|
||
$date = Carbon::createFromFormat('!Y-m-d', $value, 'Asia/Shanghai');
|
||
} catch (Throwable) {
|
||
return null;
|
||
}
|
||
|
||
return $date->format('Y-m-d') === $value ? $date : null;
|
||
}
|
||
}
|