qiuxiang
13a9697447
- 所有账户页签在「店铺」右侧新增「总账备注」下拉：按（平台+账单项目）翻译成账单项目集合过滤，页面/列设置/导出同口径；淘系支付宝、淘系微信同页签混多平台时按店铺名前缀细分 - 每日账单脚本：胜算超出同步导出上限（url=null+taskId）时自动从「异步导出下载管理」取回本次任务全部分片并合并成一份 CSV，导入前按页面笔数核对（不一致则跳过导入） - 实测：京东 7 月异步导出两片 219158+235846=455004 笔 / 收入 17695094.84 / 支出 8035128.21，与已导入口径一致
975 lines
40 KiB
PHP
975 lines
40 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Repositories;
|
||
|
||
use Illuminate\Contracts\Cache\LockTimeoutException;
|
||
use Illuminate\Database\Query\Builder;
|
||
use Illuminate\Support\Collection;
|
||
use Illuminate\Support\Facades\Cache;
|
||
use Illuminate\Support\Facades\DB;
|
||
use Illuminate\Support\Facades\Schema;
|
||
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
|
||
use Modules\EcommerceFinance\Services\LedgerRemarkFilter;
|
||
use RuntimeException;
|
||
|
||
/** 店铺收款账单·淘系支付宝账户只读经营分析副本镜像。 */
|
||
class StoreReceivingAlipayBillMirrorRepository
|
||
{
|
||
private const PAGE_QUERY_TIMEOUT_SECONDS = 30;
|
||
|
||
private const AGGREGATE_CACHE_SECONDS = 900;
|
||
|
||
private const AGGREGATE_CACHE_LOCK_SECONDS = 45;
|
||
|
||
private const AGGREGATE_CACHE_WAIT_SECONDS = 35;
|
||
|
||
private const FILTER_OPTIONS_CACHE_SECONDS = 900;
|
||
|
||
/** 淘系支付宝页面不展示其他平台店铺；头条放心购是镜像中的抖音平台名称。 */
|
||
private const EXCLUDED_TAOBAO_SHOP_SITE_PREFIXES = [
|
||
'拼多多', '京东', '抖音', '头条放心购', '得物', '唯品会',
|
||
];
|
||
|
||
public const DATASET = 'operations_replica.rpt.jst_originalBillInfo';
|
||
public const SHOP_DATASET = 'operations_replica.rpt.jst_shopInfo';
|
||
public const SKU_DATASET = 'operations_replica.rpt.jst_sku';
|
||
|
||
private ?TaobaoAlipaySettlementBillRepository $authoritative = null;
|
||
|
||
public function __construct(
|
||
private readonly BusinessMirrorDatasetRegistry $datasets,
|
||
private readonly GenericBusinessMirrorRepository $mirror,
|
||
) {}
|
||
|
||
/** @return array{data:array<int,array<string,mixed>>,total:int,current_page:int,last_page:int,per_page:int,summary:array{income:float,expense:float,balance:float,count:int},sync:array<string,mixed>} */
|
||
public function paginate(array $filters, string $sort, string $direction, int $page, int $perPage): array
|
||
{
|
||
if ($this->authoritative()->ready()) {
|
||
return $this->authoritative()->paginate($filters, $sort, $direction, $page, $perPage);
|
||
}
|
||
|
||
return $this->paginateChannel($filters, $sort, $direction, $page, $perPage, 'alipayCSV', 'taobao', self::DATASET, true);
|
||
}
|
||
|
||
/**
|
||
* Read one upstream platform channel from the shared original-bill mirror.
|
||
* The mirror is shared by several account pages; callers must provide a
|
||
* fixed source channel rather than accepting arbitrary SQL/filter input.
|
||
*
|
||
* @return array{data:array<int,array<string,mixed>>,total:int,current_page:int,last_page:int,per_page:int,summary:array{income:float,expense:float,balance:float,count:int},sync:array<string,mixed>}
|
||
*/
|
||
public function paginateChannel(
|
||
array $filters,
|
||
string $sort,
|
||
string $direction,
|
||
int $page,
|
||
int $perPage,
|
||
string|array|null $channel,
|
||
string $mapping = 'taobao',
|
||
string $datasetKey = self::DATASET,
|
||
bool $excludeNonTaobaoPlatforms = false,
|
||
): array
|
||
{
|
||
$dataset = $this->datasets->get($datasetKey);
|
||
if (! $this->mirror->ready($dataset)
|
||
|| ! $this->mirror->ready($this->datasets->get(self::SHOP_DATASET))) {
|
||
throw new RuntimeException('store_receiving_alipay_bill_mirror_not_ready');
|
||
}
|
||
|
||
$page = max($page, 1);
|
||
$perPage = min(max($perPage, 1), 100);
|
||
$direction = $direction === 'asc' ? 'asc' : 'desc';
|
||
$sortColumn = [
|
||
'record_code' => in_array($mapping, ['pinduoduo', 'dewu'], true) ? 'b.auto_id' : ($mapping === 'taobao' ? 'b.p_auto_id' : 'b.Id'),
|
||
'occurred_at' => 'b.fee_time',
|
||
'income' => 'b.amount',
|
||
'expense' => 'b.amount',
|
||
'balance' => 'b.amount',
|
||
][$sort] ?? 'b.fee_time';
|
||
|
||
$targetTable = $dataset->targetTable();
|
||
$query = $this->billQuery($targetTable);
|
||
$this->applyChannel($query, $channel, $dataset->fixedFilters());
|
||
if ($excludeNonTaobaoPlatforms) {
|
||
$this->applyTaobaoPlatformScope($query);
|
||
}
|
||
$this->applyFilters($query, $filters);
|
||
$state = $this->mirror->state($dataset) ?? [];
|
||
// Completed runs persist an exact COUNT. New incremental batches keep
|
||
// that count exact in their upsert transaction, including inserts.
|
||
$hasFilters = collect($filters)->contains(static fn (mixed $value): bool => $value !== null && trim((string) $value) !== '');
|
||
$recordedCount = $state['mirror_row_count'] ?? null;
|
||
$useRecordedCount = ! $excludeNonTaobaoPlatforms
|
||
&& $this->channelCoversDataset($channel, $dataset->fixedFilters()) && ! $hasFilters
|
||
&& (($state['status'] ?? null) === 'succeeded' || (bool) ($state['mirror_row_count_is_exact'] ?? false))
|
||
&& $recordedCount !== null && ctype_digit((string) $recordedCount);
|
||
$summary = $this->rememberAggregateSummary(
|
||
$this->aggregateCacheKey(
|
||
$datasetKey,
|
||
$filters,
|
||
$channel,
|
||
$excludeNonTaobaoPlatforms,
|
||
),
|
||
fn (): array => $this->summary($query, $filters, $channel, ! $useRecordedCount),
|
||
);
|
||
$total = $useRecordedCount
|
||
? (int) $recordedCount
|
||
: ($summary['count'] ?? $this->boundedCount(clone $query));
|
||
|
||
$billRows = $this->pageRows(
|
||
$query, $sortColumn, $direction, $page, $perPage, $filters, $channel, $targetTable,
|
||
);
|
||
$shopNames = $this->shopNames($billRows);
|
||
$productNames = $mapping === 'taobao' ? $this->productNames($billRows) : [];
|
||
$importedRows = $mapping === 'taobao' ? $this->importedRows($billRows) : [];
|
||
$rows = collect($billRows)
|
||
->map(fn (object $row): array => $this->mapRowFor($row, $shopNames[(string) $row->shop_id] ?? null, $mapping, $productNames, $importedRows))
|
||
->all();
|
||
$income = $summary['income'];
|
||
$expense = $summary['expense'];
|
||
|
||
return [
|
||
'data' => $rows,
|
||
'total' => $total,
|
||
'current_page' => $page,
|
||
'last_page' => max((int) ceil($total / $perPage), 1),
|
||
'per_page' => $perPage,
|
||
'summary' => [
|
||
'income' => $income,
|
||
'expense' => $expense,
|
||
'balance' => round($income - $expense, 2),
|
||
'count' => $total,
|
||
],
|
||
'sync' => [
|
||
'status' => $state['status'] ?? 'unknown',
|
||
'last_succeeded_at' => $state['last_succeeded_at'] ?? null,
|
||
],
|
||
];
|
||
}
|
||
|
||
/** @return iterable<int,array<string,mixed>> */
|
||
public function exportRows(array $filters, string $sort = 'occurred_at', string $direction = 'asc'): iterable
|
||
{
|
||
if ($this->authoritative()->ready()) {
|
||
return $this->authoritative()->exportRows($filters, $sort, $direction);
|
||
}
|
||
|
||
return $this->exportChannelRows($filters, $sort, $direction, 'alipayCSV', 'taobao', self::DATASET, true);
|
||
}
|
||
|
||
/** @return iterable<int,array<string,mixed>> */
|
||
public function exportChannelRows(
|
||
array $filters,
|
||
string $sort = 'occurred_at',
|
||
string $direction = 'asc',
|
||
string|array|null $channel = null,
|
||
string $mapping = 'taobao',
|
||
string $datasetKey = self::DATASET,
|
||
bool $excludeNonTaobaoPlatforms = false,
|
||
): iterable
|
||
{
|
||
$dataset = $this->datasets->get($datasetKey);
|
||
if (! $this->mirror->ready($dataset)
|
||
|| ! $this->mirror->ready($this->datasets->get(self::SHOP_DATASET))) {
|
||
throw new RuntimeException('store_receiving_alipay_bill_mirror_not_ready');
|
||
}
|
||
|
||
$targetTable = $dataset->targetTable();
|
||
$query = $this->billQuery($targetTable);
|
||
$this->applyChannel($query, $channel, $dataset->fixedFilters());
|
||
if ($excludeNonTaobaoPlatforms) {
|
||
$this->applyTaobaoPlatformScope($query);
|
||
}
|
||
$this->applyFilters($query, $filters);
|
||
$shopNames = $this->allShopNames();
|
||
$direction = $direction === 'desc' ? 'desc' : 'asc';
|
||
$sortColumn = [
|
||
'record_code' => in_array($mapping, ['pinduoduo', 'dewu'], true) ? 'b.auto_id' : ($mapping === 'taobao' ? 'b.p_auto_id' : 'b.Id'),
|
||
'occurred_at' => 'b.fee_time',
|
||
'income' => 'b.amount',
|
||
'expense' => 'b.amount',
|
||
'balance' => 'b.amount',
|
||
][$sort] ?? 'b.fee_time';
|
||
|
||
return $this->exportBatches($query, $sortColumn, $direction, $shopNames, $mapping, $targetTable);
|
||
}
|
||
|
||
/** @return array{shops:array<int,string>,bill_items:array<int,string>} */
|
||
public function filterOptionsForChannel(string|array $channel, string $datasetKey = self::DATASET, bool $excludeNonTaobaoPlatforms = false): array
|
||
{
|
||
if ($channel === 'alipayCSV' && $datasetKey === self::DATASET && $excludeNonTaobaoPlatforms && $this->authoritative()->ready()) {
|
||
return $this->authoritative()->accountBillFilterOptions();
|
||
}
|
||
|
||
$dataset = $this->datasets->get($datasetKey);
|
||
if (! $this->mirror->ready($dataset)
|
||
|| ! $this->mirror->ready($this->datasets->get(self::SHOP_DATASET))) {
|
||
throw new RuntimeException('store_receiving_alipay_bill_mirror_not_ready');
|
||
}
|
||
|
||
return $this->rememberFilterOptions(
|
||
$this->filterOptionsCacheKey($datasetKey, $channel, $excludeNonTaobaoPlatforms),
|
||
function () use ($dataset, $channel, $excludeNonTaobaoPlatforms): array {
|
||
$channelQuery = $this->billQuery($dataset->targetTable());
|
||
$this->applyChannel($channelQuery, $channel, $dataset->fixedFilters());
|
||
if ($excludeNonTaobaoPlatforms) {
|
||
$this->applyTaobaoPlatformScope($channelQuery);
|
||
}
|
||
|
||
$shopIds = (clone $channelQuery)
|
||
->whereNotNull('b.shop_id')
|
||
->distinct()
|
||
->pluck('b.shop_id')
|
||
->filter(static fn (mixed $value): bool => $value !== null && $value !== '')
|
||
->all();
|
||
|
||
$shops = $shopIds === []
|
||
? collect()
|
||
: $this->mirror->connection()->table('ops_jst_shop_info')
|
||
->whereIn('shop_id', $shopIds)
|
||
->whereNotNull('shop_name')
|
||
->orderBy('shop_name')
|
||
->pluck('shop_name');
|
||
|
||
$billItems = (clone $channelQuery)
|
||
->whereNotNull('b.fee_name')
|
||
->where('b.fee_name', '<>', '')
|
||
->select('b.fee_name')
|
||
->distinct()
|
||
->orderBy('b.fee_name')
|
||
->pluck('b.fee_name');
|
||
|
||
$normalize = static fn ($values): array => collect($values)
|
||
->map(static fn (mixed $value): string => trim((string) $value))
|
||
->filter()
|
||
->unique()
|
||
->values()
|
||
->all();
|
||
|
||
return [
|
||
'shops' => $normalize($shops),
|
||
'bill_items' => $normalize($billItems),
|
||
];
|
||
},
|
||
);
|
||
}
|
||
|
||
/** Read bounded pages so a full export does not sort/buffer millions of rows at once. */
|
||
private function exportBatches(Builder $query, string $sortColumn, string $direction, array $shopNames, string $mapping, string $targetTable): iterable
|
||
{
|
||
$operator = $direction === 'asc' ? '>' : '<';
|
||
$sortField = substr($sortColumn, 2);
|
||
$last = null;
|
||
$maximumId = $this->billQuery($targetTable)->max('b.Id');
|
||
if ($maximumId === null) return;
|
||
|
||
do {
|
||
$batch = (clone $query)->where('b.Id', '<=', $maximumId);
|
||
if ($last !== null) {
|
||
if ($sortColumn === 'b.Id') {
|
||
$batch->where('b.Id', $operator, $last->Id);
|
||
} else {
|
||
$value = $last->{$sortField};
|
||
$batch->where(function (Builder $after) use ($sortColumn, $direction, $operator, $last, $value): void {
|
||
if ($value === null) {
|
||
$after->where(function (Builder $ties) use ($sortColumn, $operator, $last): void {
|
||
$ties->whereNull($sortColumn)->where('b.Id', $operator, $last->Id);
|
||
});
|
||
if ($direction === 'asc') $after->orWhereNotNull($sortColumn);
|
||
} else {
|
||
$after->where($sortColumn, $operator, $value)
|
||
->orWhere(function (Builder $ties) use ($sortColumn, $operator, $last, $value): void {
|
||
$ties->where($sortColumn, $value)->where('b.Id', $operator, $last->Id);
|
||
});
|
||
if ($direction === 'desc') $after->orWhereNull($sortColumn);
|
||
}
|
||
});
|
||
}
|
||
}
|
||
$rows = $batch->select('b.*')->orderBy($sortColumn, $direction)
|
||
->orderBy('b.Id', $direction)->limit(1000)->get();
|
||
$productNames = $mapping === 'taobao' ? $this->productNames($rows->all()) : [];
|
||
$importedRows = $mapping === 'taobao' ? $this->importedRows($rows->all()) : [];
|
||
foreach ($rows as $row) {
|
||
yield $this->mapRowFor($row, $shopNames[(string) $row->shop_id] ?? null, $mapping, $productNames, $importedRows);
|
||
}
|
||
$last = $rows->last();
|
||
} while ($rows->count() === 1000);
|
||
}
|
||
|
||
private function billQuery(string $targetTable): Builder
|
||
{
|
||
return $this->mirror->connection()->table($targetTable . ' as b');
|
||
}
|
||
|
||
private function authoritative(): TaobaoAlipaySettlementBillRepository
|
||
{
|
||
return $this->authoritative ??= app(TaobaoAlipaySettlementBillRepository::class);
|
||
}
|
||
|
||
/** @return array<int,object> */
|
||
private function pageRows(
|
||
Builder $query,
|
||
string $sortColumn,
|
||
string $direction,
|
||
int $page,
|
||
int $perPage,
|
||
array $filters,
|
||
string|array|null $channel,
|
||
string $targetTable,
|
||
): array {
|
||
if ($channel === null || ($filters['bill_item'] ?? '') === '') {
|
||
$pageQuery = $query->select('b.*')
|
||
->orderBy($sortColumn, $direction)
|
||
->orderBy('b.Id', $direction)
|
||
->forPage($page, $perPage);
|
||
|
||
return $this->boundedGet($pageQuery)->all();
|
||
}
|
||
|
||
$idQuery = clone $query;
|
||
if ($targetTable === 'ops_jst_original_bill_info' && $idQuery->getConnection()->getDriverName() === 'mysql') {
|
||
$idQuery->forceIndex('idx_original_bill_channel_shop');
|
||
}
|
||
$idQuery->select('b.Id')
|
||
->orderBy($sortColumn, $direction)
|
||
->orderBy('b.Id', $direction)
|
||
->forPage($page, $perPage);
|
||
$ids = $this->boundedGet($idQuery)
|
||
->pluck('Id')
|
||
->all();
|
||
if ($ids === []) {
|
||
return [];
|
||
}
|
||
|
||
$rowsById = $this->boundedGet(
|
||
$this->billQuery($targetTable)->whereIn('b.Id', $ids)->select('b.*'),
|
||
)
|
||
->keyBy(static fn (object $row): string => (string) $row->Id);
|
||
|
||
return collect($ids)
|
||
->map(static fn (mixed $id): ?object => $rowsById->get((string) $id))
|
||
->filter()
|
||
->values()
|
||
->all();
|
||
}
|
||
|
||
/** @return array{income:float,expense:float,count:?int} */
|
||
private function summary(
|
||
Builder $billQuery,
|
||
array $filters,
|
||
string|array|null $channel = null,
|
||
bool $includeCount = true,
|
||
): array
|
||
{
|
||
if ($channel === null && $this->dailySummarySupports($filters)) {
|
||
$summary = $this->dailySummary($filters);
|
||
if ($summary !== null) {
|
||
return $summary + ['count' => null];
|
||
}
|
||
}
|
||
|
||
$aggregate = (clone $billQuery)->selectRaw(
|
||
($includeCount ? 'COUNT(*) AS aggregate_count, ' : '')
|
||
. "COALESCE(SUM(CASE WHEN b.inout_type = '1' THEN ABS(COALESCE(b.amount, 0)) ELSE 0 END), 0) AS income, "
|
||
. "COALESCE(SUM(CASE WHEN b.inout_type = '-1' THEN ABS(COALESCE(b.amount, 0)) ELSE 0 END), 0) AS expense"
|
||
);
|
||
$row = $this->boundedGet($aggregate)->first();
|
||
|
||
return [
|
||
'income' => round((float) ($row->income ?? 0), 2),
|
||
'expense' => round((float) ($row->expense ?? 0), 2),
|
||
'count' => $includeCount ? (int) ($row->aggregate_count ?? 0) : null,
|
||
];
|
||
}
|
||
|
||
private function dailySummarySupports(array $filters): bool
|
||
{
|
||
return ($filters['source_channel'] ?? '') === ''
|
||
&& ($filters['bill_item'] ?? '') === ''
|
||
&& ($filters['keyword'] ?? '') === '';
|
||
}
|
||
|
||
/** @return array{income:float,expense:float}|null */
|
||
private function dailySummary(array $filters): ?array
|
||
{
|
||
$connection = $this->mirror->connection();
|
||
$schema = Schema::connection($connection->getName());
|
||
if (! $schema->hasTable('store_report_import_states') || ! $schema->hasTable('store_report_daily_rows')) {
|
||
return null;
|
||
}
|
||
|
||
$state = $connection->table('store_report_import_states')
|
||
->where('dataset_key', 'store-report.daily-sales')->first();
|
||
$generation = trim((string) ($state->active_generation ?? ''));
|
||
if ($generation === '' || empty($state->initial_completed_at)) {
|
||
return null;
|
||
}
|
||
|
||
$query = $connection->table('store_report_daily_rows')->where('generation', $generation);
|
||
if (($filters['store_name'] ?? '') !== '') {
|
||
$query->where('store_name', $filters['store_name']);
|
||
}
|
||
if (($filters['start'] ?? '') !== '') {
|
||
$query->where('occurred_on', '>=', $filters['start']);
|
||
}
|
||
if (($filters['end'] ?? '') !== '') {
|
||
$query->where('occurred_on', '<=', $filters['end']);
|
||
}
|
||
$row = $query->selectRaw('COALESCE(SUM(income), 0) AS income, COALESCE(SUM(expense), 0) AS expense')->first();
|
||
$income = round((float) ($row->income ?? 0), 2);
|
||
$expense = round((float) ($row->expense ?? 0), 2);
|
||
|
||
return match ($this->directionCode((string) ($filters['business_type'] ?? ''))) {
|
||
'1' => ['income' => $income, 'expense' => 0.0],
|
||
'-1' => ['income' => 0.0, 'expense' => $expense],
|
||
default => ['income' => $income, 'expense' => $expense],
|
||
};
|
||
}
|
||
|
||
private function applyFilters(Builder $query, array $filters): void
|
||
{
|
||
if (($filters['store_name'] ?? '') !== '') {
|
||
$shopIds = $this->mirror->connection()->table('ops_jst_shop_info')
|
||
->where('shop_name', 'like', '%' . $this->escapeLike((string) $filters['store_name']) . '%')
|
||
->pluck('shop_id')
|
||
->filter(static fn (mixed $value): bool => $value !== null && $value !== '')
|
||
->all();
|
||
if ($shopIds === []) {
|
||
$query->whereRaw('1 = 0');
|
||
} else {
|
||
$query->whereIn('b.shop_id', $shopIds);
|
||
}
|
||
}
|
||
if (($filters['source_channel'] ?? '') !== '') {
|
||
$query->where('b.channel', 'like', '%' . $this->escapeLike((string) $filters['source_channel']) . '%');
|
||
}
|
||
if (($filters['bill_item'] ?? '') !== '') {
|
||
$query->where('b.fee_name', 'like', '%' . $this->escapeLike((string) $filters['bill_item']) . '%');
|
||
}
|
||
// 「总账备注」：共享原始账单镜像只有 shop_id，平台归属用店铺资料表的店铺名前缀判定。
|
||
$groups = is_array($filters['bill_item_groups'] ?? null) ? $filters['bill_item_groups'] : [];
|
||
if ($groups !== []) {
|
||
LedgerRemarkFilter::applyGroups(
|
||
$query,
|
||
$groups,
|
||
'b.fee_name',
|
||
fn (Builder $sub, string $prefix) => $sub->whereIn('b.shop_id', $this->shopIdsLike($prefix)),
|
||
fn (Builder $sub, string $prefix) => $sub->whereNotIn('b.shop_id', $this->shopIdsLike($prefix)),
|
||
);
|
||
} else {
|
||
LedgerRemarkFilter::itemsWhereIn($query, (array) ($filters['bill_items'] ?? []), 'b.fee_name');
|
||
}
|
||
if (($filters['business_type'] ?? '') !== '') {
|
||
$query->where('b.inout_type', $this->directionCode((string) $filters['business_type']));
|
||
}
|
||
if (($filters['start'] ?? '') !== '') {
|
||
$query->where('b.fee_time', '>=', $filters['start'] . ' 00:00:00');
|
||
}
|
||
if (($filters['end'] ?? '') !== '') {
|
||
$query->where('b.fee_time', '<', date('Y-m-d', strtotime($filters['end'] . ' +1 day')) . ' 00:00:00');
|
||
}
|
||
if (($filters['keyword'] ?? '') !== '') {
|
||
if (($filters['keyword_match'] ?? '') === 'exact-record') {
|
||
$keyword = (string) $filters['keyword'];
|
||
if (! preg_match('/^(0|[1-9]\d{0,9})$/', $keyword) || (float) $keyword > 2147483647) {
|
||
$query->whereRaw('1 = 0');
|
||
} else {
|
||
$query->where('b.Id', $keyword);
|
||
}
|
||
return;
|
||
}
|
||
$value = '%' . $this->escapeLike((string) $filters['keyword']) . '%';
|
||
$query->where(function (Builder $query) use ($value): void {
|
||
foreach (['b.Id', 'b.auto_id', 'b.fee_name', 'b.primary_key', 'b.p_auto_id', 'b.so_id', 'b.sku_id', 'b.shop_sku_id', 'b.shop_i_id', 'b.channel'] as $column) {
|
||
$query->orWhere($column, 'like', $value);
|
||
}
|
||
});
|
||
}
|
||
}
|
||
|
||
/** @param array<string,scalar|list<scalar>> $fixedFilters */
|
||
private function applyChannel(Builder $query, string|array|null $channel, array $fixedFilters): void
|
||
{
|
||
if ($channel === null || $channel === '') return;
|
||
if (is_string($channel)) {
|
||
if (! $this->channelCoversDataset($channel, $fixedFilters)) {
|
||
$query->where('b.channel', $channel);
|
||
}
|
||
return;
|
||
}
|
||
|
||
$channels = array_values($channel);
|
||
if ($channels === []) {
|
||
$query->whereRaw('1 = 0');
|
||
return;
|
||
}
|
||
|
||
$query->whereIn('b.channel', $channels);
|
||
}
|
||
|
||
private function applyTaobaoPlatformScope(Builder $query): void
|
||
{
|
||
$excludedShopIds = $this->mirror->connection()->table('ops_jst_shop_info')
|
||
->where(function (Builder $scope): void {
|
||
foreach (self::EXCLUDED_TAOBAO_SHOP_SITE_PREFIXES as $prefix) {
|
||
$scope->orWhere('shop_site', 'like', $prefix . '%');
|
||
}
|
||
})->select('shop_id');
|
||
// Unknown shop metadata is retained: without a platform label it cannot
|
||
// be identified as one of the explicitly excluded platforms.
|
||
$query->whereNotIn('b.shop_id', $excludedShopIds);
|
||
}
|
||
|
||
/** 店铺名前缀 → shop_id 子查询（共享原始账单镜像按店铺资料表判定平台）。 */
|
||
private function shopIdsLike(string $prefix): Builder
|
||
{
|
||
return $this->mirror->connection()->table('ops_jst_shop_info')
|
||
->where('shop_name', 'like', $this->escapeLike($prefix) . '%')
|
||
->select('shop_id');
|
||
}
|
||
|
||
/** @param array<string,scalar|list<scalar>> $fixedFilters */
|
||
private function channelCoversDataset(string|array|null $channel, array $fixedFilters): bool
|
||
{
|
||
if ($channel === null || $channel === '') {
|
||
return true;
|
||
}
|
||
|
||
foreach ($fixedFilters as $column => $value) {
|
||
if (strcasecmp($column, 'channel') !== 0) continue;
|
||
|
||
$requested = is_array($channel) ? array_map('strval', $channel) : [(string) $channel];
|
||
$configured = is_array($value) ? array_map('strval', $value) : [(string) $value];
|
||
sort($requested);
|
||
sort($configured);
|
||
|
||
return $requested === $configured;
|
||
}
|
||
|
||
return false;
|
||
}
|
||
|
||
/** @param array<int,object> $rows @return array<string,string> */
|
||
private function shopNames(array $rows): array
|
||
{
|
||
$shopIds = collect($rows)->pluck('shop_id')->filter()->unique()->values()->all();
|
||
if ($shopIds === []) {
|
||
return [];
|
||
}
|
||
|
||
return $this->mirror->connection()->table('ops_jst_shop_info')
|
||
->whereIn('shop_id', $shopIds)->pluck('shop_name', 'shop_id')
|
||
->mapWithKeys(static fn (mixed $name, mixed $id): array => [(string) $id => (string) $name])
|
||
->all();
|
||
}
|
||
|
||
/** @return array<string,string> */
|
||
private function allShopNames(): array
|
||
{
|
||
return $this->mirror->connection()->table('ops_jst_shop_info')
|
||
->whereNotNull('shop_id')->pluck('shop_name', 'shop_id')
|
||
->mapWithKeys(static fn (mixed $name, mixed $id): array => [(string) $id => (string) $name])
|
||
->all();
|
||
}
|
||
|
||
/** @param array<int,object> $rows @return array<string,string> */
|
||
private function productNames(array $rows): array
|
||
{
|
||
$skuIds = collect($rows)->map(fn (object $row): ?string => $row->sku_id ?: $row->shop_sku_id)
|
||
->filter()->unique()->values()->all();
|
||
if ($skuIds === []) return [];
|
||
|
||
try {
|
||
$dataset = $this->datasets->get(self::SKU_DATASET);
|
||
} catch (\Throwable) {
|
||
return [];
|
||
}
|
||
if (! $this->mirror->ready($dataset) || ! Schema::connection($this->mirror->connection()->getName())->hasTable('ops_jst_sku')) {
|
||
return [];
|
||
}
|
||
|
||
return $this->mirror->connection()->table('ops_jst_sku')
|
||
->whereIn('sku_id', $skuIds)->pluck('name', 'sku_id')
|
||
->mapWithKeys(static fn (mixed $name, mixed $id): array => [(string) $id => (string) $name])
|
||
->all();
|
||
}
|
||
|
||
/** @param array<int,object> $rows @return array<string,object> */
|
||
private function importedRows(array $rows): array
|
||
{
|
||
if (! Schema::hasTable('ec_alipay_bills')) {
|
||
return [];
|
||
}
|
||
|
||
$recordCodes = collect($rows)->pluck('p_auto_id')->filter()->unique()->values()->all();
|
||
if ($recordCodes === []) {
|
||
return [];
|
||
}
|
||
|
||
return DB::table('ec_alipay_bills')
|
||
->where('company_id', (int) (company_id() ?: 1))
|
||
->whereNull('deleted_at')
|
||
->whereIn('record_code', $recordCodes)
|
||
->get()->keyBy(fn (object $row): string => (string) $row->record_code)->all();
|
||
}
|
||
|
||
/** @param array<int,object> $rows */
|
||
public function mapDouyinRows(array $rows): array
|
||
{
|
||
$shops = $this->shopNames($rows);
|
||
return array_map(fn (object $row): array => $this->mapRowFor($row, $shops[(string) $row->shop_id] ?? null, 'douyin'), $rows);
|
||
}
|
||
|
||
private function mapRow(object $row, ?string $shopName, array $productNames = [], array $importedRows = []): array
|
||
{
|
||
$amount = abs((float) ($row->amount ?? 0));
|
||
$income = (string) ($row->inout_type ?? '') === '1' ? $amount : 0.0;
|
||
$expense = (string) ($row->inout_type ?? '') === '-1' ? $amount : 0.0;
|
||
|
||
$mapped = [
|
||
'id' => (string) $row->Id,
|
||
'record_code' => (string) ($row->p_auto_id ?: $row->Id),
|
||
'store_name' => $shopName,
|
||
'custom_remark' => null,
|
||
'bill_item' => $row->fee_name,
|
||
'alipay_account' => null,
|
||
'account_order_no' => null,
|
||
'business_serial_no' => null,
|
||
'merchant_order_no' => null,
|
||
'business_base_order_no' => null,
|
||
'goods_name' => $productNames[(string) ($row->sku_id ?: $row->shop_sku_id)] ?? null,
|
||
'counterparty_account' => null,
|
||
'occurred_at' => $this->dateTime($row->fee_time),
|
||
'income' => $income,
|
||
'expense' => $expense,
|
||
'balance' => null,
|
||
'business_type' => null,
|
||
'remark' => null,
|
||
'online_order_no' => $row->so_id,
|
||
'online_real_store' => $shopName,
|
||
'business_description' => null,
|
||
'business_bill_source' => null,
|
||
'business_order_no' => null,
|
||
'created_at' => $this->dateTime($row->createTime),
|
||
'updated_at' => $this->dateTime($row->updateTime),
|
||
'modifier' => '',
|
||
];
|
||
|
||
$imported = $importedRows[$mapped['record_code']] ?? null;
|
||
if ($imported === null) {
|
||
return $mapped;
|
||
}
|
||
|
||
foreach ([
|
||
'custom_remark', 'alipay_account', 'account_order_no', 'business_serial_no',
|
||
'merchant_order_no', 'business_base_order_no', 'counterparty_account', 'balance',
|
||
'business_type', 'remark', 'business_description', 'business_bill_source',
|
||
'business_order_no', 'modifier',
|
||
] as $field) {
|
||
$mapped[$field] = $imported->{$field} ?? $mapped[$field];
|
||
}
|
||
if ($mapped['goods_name'] === null || $mapped['goods_name'] === '') {
|
||
$mapped['goods_name'] = $imported->goods_name ?? null;
|
||
}
|
||
|
||
return $mapped;
|
||
}
|
||
|
||
/** @return array<string,mixed> */
|
||
private function mapRowFor(object $row, ?string $shopName, string $mapping, array $productNames = [], array $importedRows = []): array
|
||
{
|
||
if ($mapping === 'pinduoduo') {
|
||
return $this->mapPinduoduoRow($row, $shopName);
|
||
}
|
||
|
||
if ($mapping === 'dewu') {
|
||
return $this->mapDewuRow($row, $shopName);
|
||
}
|
||
|
||
if ($mapping !== 'douyin') {
|
||
return $this->mapRow($row, $shopName, $productNames, $importedRows);
|
||
}
|
||
|
||
$amount = abs((float) ($row->amount ?? 0));
|
||
$income = (string) ($row->inout_type ?? '') === '1' ? $amount : 0.0;
|
||
$expense = (string) ($row->inout_type ?? '') === '-1' ? $amount : 0.0;
|
||
$direction = $this->directionName($row->inout_type);
|
||
|
||
return [
|
||
'id' => (string) $row->Id,
|
||
'record_code' => (string) $row->Id,
|
||
'store_name' => $shopName,
|
||
'finance_remark' => null,
|
||
'order_no' => $row->so_id ?: $row->p_auto_id,
|
||
'amount' => $amount,
|
||
'custom_remark' => null,
|
||
'bill_item' => $row->fee_name,
|
||
'occurred_at' => $this->dateTime($row->fee_time),
|
||
'income' => $income,
|
||
'expense' => $expense,
|
||
'balance' => round($income - $expense, 2),
|
||
'remark' => null,
|
||
'modifier' => '聚水潭',
|
||
'created_at' => $this->dateTime($row->createTime),
|
||
'updated_at' => $this->dateTime($row->updateTime),
|
||
'payload' => [
|
||
'merchant_name' => $shopName,
|
||
'move_serial_no' => $row->auto_id ?: $row->primary_key,
|
||
'order_no_2' => $row->p_auto_id ?: $row->so_id,
|
||
'online_order_no' => $row->so_id,
|
||
'sub_order_no' => $row->shop_i_id,
|
||
'goods_id' => $row->sku_id ?: $row->shop_sku_id,
|
||
'account_type' => $direction,
|
||
'move_summary' => $row->fee_name,
|
||
'order_placed_at' => $this->dateTime($row->createTime),
|
||
'business_type' => $row->combine_type,
|
||
'settle_account' => $row->channel,
|
||
'move_scene' => '头条放心购',
|
||
'talent_id' => null,
|
||
'talent_name' => null,
|
||
],
|
||
];
|
||
}
|
||
|
||
/** @return array<string,mixed> */
|
||
private function mapPinduoduoRow(object $row, ?string $shopName): array
|
||
{
|
||
$amount = abs((float) ($row->amount ?? 0));
|
||
$income = (string) ($row->inout_type ?? '') === '1' ? $amount : 0.0;
|
||
$expense = (string) ($row->inout_type ?? '') === '-1' ? $amount : 0.0;
|
||
|
||
return [
|
||
'id' => (string) $row->Id,
|
||
'record_code' => (string) ($row->auto_id ?: $row->Id),
|
||
'store_name' => $shopName,
|
||
'finance_remark' => null,
|
||
'order_no' => $row->so_id ?: $row->p_auto_id,
|
||
'amount' => $amount,
|
||
'custom_remark' => null,
|
||
'bill_item' => $row->fee_name,
|
||
'occurred_at' => $this->dateTime($row->fee_time),
|
||
'income' => $income,
|
||
'expense' => $expense,
|
||
'balance' => round($income - $expense, 2),
|
||
'remark' => null,
|
||
'updated_at' => $this->dateTime($row->updateTime),
|
||
'modifier' => 'RPA',
|
||
'payload' => [
|
||
'merchant_order_no' => $row->so_id ?: $row->p_auto_id,
|
||
'online_order_no' => $row->so_id,
|
||
'finance_type' => $row->fee_name,
|
||
'type' => '拼多多货款账户',
|
||
],
|
||
];
|
||
}
|
||
|
||
/** @return array<string,mixed> */
|
||
private function mapDewuRow(object $row, ?string $shopName): array
|
||
{
|
||
$amount = abs((float) ($row->amount ?? 0));
|
||
$income = (string) ($row->inout_type ?? '') === '1' ? $amount : 0.0;
|
||
$expense = (string) ($row->inout_type ?? '') === '-1' ? $amount : 0.0;
|
||
$orderNo = $row->so_id ?: $row->p_auto_id;
|
||
$goodsNo = $row->sku_id ?: $row->shop_sku_id ?: $row->shop_i_id;
|
||
|
||
return [
|
||
'id' => (string) $row->Id,
|
||
'record_code' => (string) ($row->auto_id ?: $row->Id),
|
||
'store_name' => $shopName,
|
||
'finance_remark' => null,
|
||
'order_no' => $orderNo,
|
||
'amount' => $amount,
|
||
'custom_remark' => null,
|
||
'bill_item' => $row->fee_name,
|
||
'occurred_at' => $this->dateTime($row->fee_time),
|
||
'income' => $income,
|
||
'expense' => $expense,
|
||
'balance' => round($income - $expense, 2),
|
||
'remark' => null,
|
||
'updated_at' => $this->dateTime($row->updateTime),
|
||
'modifier' => '聚水潭',
|
||
'payload' => [
|
||
'online_order_no' => $orderNo,
|
||
'store_2' => $shopName,
|
||
'order_type' => null,
|
||
'goods_name' => null,
|
||
'goods_no' => $goodsNo,
|
||
'quantity' => null,
|
||
'channel' => match ((string) $row->channel) {
|
||
'Dewu' => '得物',
|
||
'DewuTK' => '退货退款订单',
|
||
'DewuKJ' => '扣减其他费用明细',
|
||
default => (string) $row->channel,
|
||
},
|
||
'source_channel' => $row->channel,
|
||
'business_type' => $this->directionName($row->inout_type),
|
||
'online_real_store' => $shopName,
|
||
'order_created_at' => null,
|
||
'ship_return_at' => null,
|
||
'settle_status' => null,
|
||
'pay_period' => null,
|
||
'pulled_at' => $this->dateTime($row->createTime),
|
||
],
|
||
];
|
||
}
|
||
|
||
private function directionCode(string $value): string
|
||
{
|
||
return match ($value) {
|
||
'收入', 'IN', 'in' => '1',
|
||
'支出', 'OUT', 'out' => '-1',
|
||
default => $value,
|
||
};
|
||
}
|
||
|
||
private function directionName(mixed $value): string
|
||
{
|
||
return match ((string) $value) {
|
||
'1' => '收入',
|
||
'-1' => '支出',
|
||
default => (string) ($value ?? ''),
|
||
};
|
||
}
|
||
|
||
private function dateTime(mixed $value): string
|
||
{
|
||
return $value === null || $value === '' ? '' : date('Y-m-d H:i:s', strtotime((string) $value));
|
||
}
|
||
|
||
/** @return Collection<int,object> */
|
||
private function boundedGet(Builder $query): Collection
|
||
{
|
||
$connection = $query->getConnection();
|
||
if ($connection->getDriverName() !== 'mysql') {
|
||
return $query->get();
|
||
}
|
||
|
||
return collect($connection->select(
|
||
'SET STATEMENT max_statement_time=' . self::PAGE_QUERY_TIMEOUT_SECONDS . ' FOR ' . $query->toSql(),
|
||
$query->getBindings(),
|
||
));
|
||
}
|
||
|
||
private function boundedCount(Builder $query): int
|
||
{
|
||
$count = $query->getConnection()->query()
|
||
->fromSub($query, 'bounded_count')
|
||
->selectRaw('COUNT(*) AS aggregate');
|
||
|
||
return (int) ($this->boundedGet($count)->first()->aggregate ?? 0);
|
||
}
|
||
|
||
/** @param array<string,mixed> $filters */
|
||
private function aggregateCacheKey(
|
||
string $datasetKey,
|
||
array $filters,
|
||
string|array|null $channel,
|
||
bool $excludeNonTaobaoPlatforms,
|
||
): string {
|
||
$normalizedFilters = array_filter(
|
||
array_map(
|
||
static fn (mixed $value): string => trim((string) $value),
|
||
$filters,
|
||
),
|
||
static fn (string $value): bool => $value !== '',
|
||
);
|
||
ksort($normalizedFilters);
|
||
$normalizedChannel = is_array($channel) ? array_map('strval', $channel) : $channel;
|
||
if (is_array($normalizedChannel)) {
|
||
sort($normalizedChannel);
|
||
}
|
||
|
||
return 'store-receiving-bills:aggregate:v3:' . hash('sha256', (string) json_encode([
|
||
'dataset' => $datasetKey,
|
||
'filters' => $normalizedFilters,
|
||
'channel' => $normalizedChannel,
|
||
'exclude_non_taobao' => $excludeNonTaobaoPlatforms,
|
||
], JSON_UNESCAPED_UNICODE));
|
||
}
|
||
|
||
/** @param callable():array{income:float,expense:float,count:?int} $resolver */
|
||
private function rememberAggregateSummary(string $key, callable $resolver): array
|
||
{
|
||
$cached = Cache::get($key);
|
||
if (is_array($cached)) {
|
||
return $cached;
|
||
}
|
||
|
||
try {
|
||
return Cache::lock($key . ':lock', self::AGGREGATE_CACHE_LOCK_SECONDS)
|
||
->block(self::AGGREGATE_CACHE_WAIT_SECONDS, function () use ($key, $resolver): array {
|
||
$cached = Cache::get($key);
|
||
if (is_array($cached)) {
|
||
return $cached;
|
||
}
|
||
|
||
$resolved = $resolver();
|
||
Cache::put($key, $resolved, self::AGGREGATE_CACHE_SECONDS);
|
||
|
||
return $resolved;
|
||
});
|
||
} catch (LockTimeoutException $exception) {
|
||
$cached = Cache::get($key);
|
||
if (is_array($cached)) {
|
||
return $cached;
|
||
}
|
||
|
||
throw new RuntimeException('store_receiving_bill_aggregate_busy', 0, $exception);
|
||
}
|
||
}
|
||
|
||
private function filterOptionsCacheKey(
|
||
string $datasetKey,
|
||
string|array $channel,
|
||
bool $excludeNonTaobaoPlatforms,
|
||
): string {
|
||
$normalizedChannel = is_array($channel) ? array_map('strval', $channel) : $channel;
|
||
if (is_array($normalizedChannel)) {
|
||
sort($normalizedChannel);
|
||
}
|
||
|
||
return 'store-receiving-bills:filter-options:v1:' . hash('sha256', (string) json_encode([
|
||
'dataset' => $datasetKey,
|
||
'channel' => $normalizedChannel,
|
||
'exclude_non_taobao' => $excludeNonTaobaoPlatforms,
|
||
], JSON_UNESCAPED_UNICODE));
|
||
}
|
||
|
||
/** @param callable():array{shops:array<int,string>,bill_items:array<int,string>} $resolver */
|
||
private function rememberFilterOptions(string $key, callable $resolver): array
|
||
{
|
||
$cached = Cache::get($key);
|
||
if (is_array($cached)) {
|
||
return $cached;
|
||
}
|
||
|
||
try {
|
||
return Cache::lock($key . ':lock', self::AGGREGATE_CACHE_LOCK_SECONDS)
|
||
->block(self::AGGREGATE_CACHE_WAIT_SECONDS, function () use ($key, $resolver): array {
|
||
$cached = Cache::get($key);
|
||
if (is_array($cached)) {
|
||
return $cached;
|
||
}
|
||
|
||
$resolved = $resolver();
|
||
Cache::put($key, $resolved, self::FILTER_OPTIONS_CACHE_SECONDS);
|
||
|
||
return $resolved;
|
||
});
|
||
} catch (LockTimeoutException $exception) {
|
||
$cached = Cache::get($key);
|
||
if (is_array($cached)) {
|
||
return $cached;
|
||
}
|
||
|
||
throw new RuntimeException('store_receiving_bill_filter_options_busy', 0, $exception);
|
||
}
|
||
}
|
||
|
||
private function escapeLike(string $value): string
|
||
{
|
||
return str_replace(['\\', '%', '_'], ['\\\\', '\\%', '\\_'], $value);
|
||
}
|
||
}
|