CaiWuXiTong/modules/EcommerceFinance/Repositories/DingOnlineDaifaPriceMirrorRepository.php

1125 lines
45 KiB
PHP
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

<?php
namespace Modules\EcommerceFinance\Repositories;
use Illuminate\Database\ConnectionInterface;
use Illuminate\Database\Query\Builder;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Modules\EcommerceFinance\Services\DingPayableWorkflowStatus;
use RuntimeException;
/**
* 线上代发价格的统一镜像查询。
*
* 钉钉线上代发审批和聚水潭采购单先分别同步到共享业务镜像,再在这里合并。
* 页面请求只读共享镜像,不跨来源实时查询上游。
*/
class DingOnlineDaifaPriceMirrorRepository
{
public const DATASET_KEY = 'operations_replica.rpt.ding_online_daifa_price';
public const TARGET_TABLE = 'ding_online_daifa_price_mirror';
private const ORDER_DATASET_KEY = 'operations_replica.rpt.jst_Orders';
/** 聚水潭订单来源:非淘系 + 淘系(奇门),两套表结构一致。 */
private const JST_ORDER_SOURCES = [
['ops_jst_orders', 'ops_jst_order_item'],
['ops_jst_qmcust_orders', 'ops_jst_qmcust_order_items'],
];
/** 被拆分、合并、作废的原单不代表实际发货,有其它有效订单时不参与拼接。 */
private const JST_INACTIVE_ORDER_STATUSES = ['Cancelled', 'Split', 'Merged', 'Delete'];
private const JST_LOOKUP_CHUNK = 500;
private const EXPORT_HYDRATE_BATCH = 200;
/** 核查索引一次关联聚水潭订单的行数(与订单查询分块一致,减少往返次数)。 */
private const AUDIT_HYDRATE_BATCH = 500;
/** 报销成本核查结果:代发订单金额 vs Σ(实际发货子商品成本价 × 数量)。 */
public const AUDIT_OVER = '多报销';
public const AUDIT_OK = '正常';
public const AUDIT_COST_MISSING = '成本缺失';
public const AUDIT_NO_SHIPMENT = '无发货明细';
public const AUDIT_RESULTS = [self::AUDIT_OVER, self::AUDIT_OK, self::AUDIT_COST_MISSING, self::AUDIT_NO_SHIPMENT];
/** 报销金额比系统成本高出超过此值即判为多报销(元)。 */
private const AUDIT_TOLERANCE = 0.01;
private const AUDIT_CACHE_SECONDS = 600;
private readonly ConnectionInterface $connection;
/** @var array<string,array<string,mixed>>|null */
private ?array $sourceStatusCache = null;
/** @var list<array{orders:string,items:string,status:bool,labels:bool,combine:bool}>|null */
private ?array $jstOrderSourceCache = null;
public function __construct(GenericBusinessMirrorRepository $mirror)
{
$this->connection = $mirror->connection();
}
public function ready(): bool
{
return collect($this->sourceStatuses())->contains(
static fn (array $source): bool => (bool) ($source['ready'] ?? false),
);
}
/**
* 页面分页。
*
* 不再把两张来源表整表 UNION 成派生表再数数、汇总、排序(每次打开都要把两边全量
* 物化三遍,连同审批 JSON 一起落盘排序,数据一多就超时)。改为:
* - 筛选条件分别下推到钉钉、采购两个来源;
* - 总数、合计按来源各算一次再相加;
* - 两边各按同一排序只取前 (页码×每页) 条主键,合并后取当前页;
* - 只给当前页的行回表取完整字段(含审批 JSON)。
* 结果与原先整表合并后的排序、筛选、合计一致。
*
* @return array{data:list<array<string,mixed>>,total:int,status:string,last_synced_at:?string,summary:array<string,mixed>,source_statuses:array<string,array<string,mixed>>}
*/
public function paginate(array $filters, int $page, int $perPage): array
{
if (! $this->ready()) {
throw new RuntimeException('business_mirror_dataset_not_ready');
}
$page = max($page, 1);
$perPage = min(max($perPage, 1), 100);
$sources = $this->sourceStatuses();
$ding = (bool) ($sources['ding']['ready'] ?? false) ? $this->dingFiltered($filters) : null;
$purchase = (bool) ($sources['purchase']['ready'] ?? false) ? $this->purchaseFiltered($filters) : null;
$summary = [
'count' => 0,
'matched_count' => 0,
'matchable_count' => 0,
'purchase_count' => 0,
'daifa_amount' => 0.0,
'paid_amount' => 0.0,
'price_diff' => 0.0,
];
if ($ding !== null) {
$aggregate = (clone $ding)
->selectRaw(
'COUNT(*) AS row_count, '
. "SUM(CASE WHEN d.product_id IS NOT NULL AND d.product_id <> '' "
. "AND d.product_name IS NOT NULL AND d.product_name <> '' THEN 1 ELSE 0 END) AS matched_count, "
. 'COALESCE(SUM(d.daifa_amount), 0) AS daifa_amount, '
. 'COALESCE(SUM(d.paid_amount), 0) AS paid_amount, '
. 'COALESCE(SUM(d.price_diff), 0) AS price_diff'
)
->first();
$summary['matchable_count'] = (int) ($aggregate->row_count ?? 0);
$summary['matched_count'] = (int) ($aggregate->matched_count ?? 0);
$summary['daifa_amount'] = round((float) ($aggregate->daifa_amount ?? 0), 2);
$summary['paid_amount'] = round((float) ($aggregate->paid_amount ?? 0), 2);
$summary['price_diff'] = round((float) ($aggregate->price_diff ?? 0), 2);
}
if ($purchase !== null) {
$summary['purchase_count'] = (int) (clone $purchase)->count();
}
$total = $summary['matchable_count'] + $summary['purchase_count'];
$summary['count'] = $total;
$pageRows = [];
if ($total > 0 && ($filters['audit'] ?? '') === '') {
$pageRows = $this->pageRows($ding, $purchase, $page, $perPage);
}
$audit = $total > 0 ? $this->auditIndex($filters) : ['keys' => [], 'counts' => [], 'over_amount' => 0.0];
$summary['audit_counts'] = $audit['counts'];
$summary['over_amount'] = $audit['over_amount'];
if (($filters['audit'] ?? '') !== '') {
$keys = $audit['keys'][$filters['audit']] ?? [];
$total = count($keys);
$pageRows = $this->rowsForKeys(array_slice($keys, ($page - 1) * $perPage, $perPage), $filters);
}
$data = [];
foreach ($pageRows as $row) {
$data[] = $this->presentRow($row);
}
$this->attachJstOrderDetails($data);
$this->attachCostAudit($data);
$readySources = array_filter($sources, static fn (array $source): bool => (bool) ($source['ready'] ?? false));
$synced = array_values(array_filter(array_column($readySources, 'last_synced_at')));
return [
'data' => $data,
'total' => $total,
'status' => count($readySources) > 1
? 'available'
: (string) (reset($readySources)['status'] ?? 'unknown'),
'last_synced_at' => $synced === [] ? null : max($synced),
'summary' => $summary,
'source_statuses' => $sources,
];
}
/**
* 两个来源各取前 N 条主键(只带排序列,排序很轻),合并排序后截出当前页,
* 再按主键回表取完整行。排序口径与原 UNION 一致:创建时间、单号(按文本)、明细序号均倒序。
*
* @return list<object>
*/
private function pageRows(?Builder $ding, ?Builder $purchase, int $page, int $perPage): array
{
$need = $page * $perPage;
$branches = [];
if ($ding !== null) {
$branches[] = (clone $ding)
->select([
DB::raw("'ding' AS source_type"),
'd.process_instance_id',
'd.detail_index',
'd.workflow_created_at',
])
->orderByDesc('d.workflow_created_at')
->orderByDesc('d.process_instance_id')
->orderByDesc('d.detail_index')
->limit($need);
}
if ($purchase !== null) {
$created = 'COALESCE(p.document_created_at, p.purchase_updated_at)';
$branches[] = (clone $purchase)
->select([
DB::raw("'purchase' AS source_type"),
DB::raw('CAST(p.purchase_order_id AS CHAR) AS process_instance_id'),
'p.item_id as detail_index',
DB::raw($created . ' AS workflow_created_at'),
])
->orderByRaw($created . ' DESC')
->orderByRaw('CAST(p.purchase_order_id AS CHAR) DESC')
->orderByDesc('p.item_id')
->limit($need);
}
if ($branches === []) {
return [];
}
$union = array_shift($branches);
foreach ($branches as $branch) {
$union->unionAll($branch);
}
$keys = $this->connection->query()
->fromSub($union, 'online_daifa_keys')
->orderByDesc('workflow_created_at')
->orderByDesc('process_instance_id')
->orderByDesc('detail_index')
->offset(($page - 1) * $perPage)
->limit($perPage)
->get();
$dingIds = [];
$itemIds = [];
foreach ($keys as $key) {
if ((string) $key->source_type === 'purchase') {
$itemIds[(int) $key->detail_index] = true;
} else {
$dingIds[(string) $key->process_instance_id] = true;
}
}
$full = [];
if ($dingIds !== []) {
foreach ($this->dingQuery()->whereIn('d.process_instance_id', array_keys($dingIds))->get() as $row) {
$full['ding|' . $row->process_instance_id . '|' . (int) $row->detail_index] = $row;
}
}
if ($itemIds !== []) {
foreach ($this->purchaseQuery()->whereIn('p.item_id', array_keys($itemIds))->get() as $row) {
$full['purchase|' . (int) $row->detail_index] = $row;
}
}
$rows = [];
foreach ($keys as $key) {
$lookup = (string) $key->source_type === 'purchase'
? 'purchase|' . (int) $key->detail_index
: 'ding|' . $key->process_instance_id . '|' . (int) $key->detail_index;
if (isset($full[$lookup])) {
$rows[] = $full[$lookup];
}
}
return $rows;
}
/** 钉钉来源加上页面筛选;该来源被筛选条件整体排除时返回 null。 */
private function dingFiltered(array $filters): ?Builder
{
if (($filters['status'] ?? '') !== '' && strtoupper((string) $filters['status']) !== $this->dingStatusKey()) {
return null;
}
$query = $this->connection->table(self::TARGET_TABLE . ' as d');
if (($filters['product_id'] ?? '') !== '') {
$query->where('d.product_id', (string) $filters['product_id']);
}
if (($filters['shop_name'] ?? '') !== '') {
$query->where('d.shop_name', (string) $filters['shop_name']);
}
if (($filters['supplier_name'] ?? '') !== '') {
$query->whereRaw("d.supplier_name LIKE ? ESCAPE '!'", [$this->contains((string) $filters['supplier_name'])]);
}
if (($filters['material_no'] ?? '') !== '') {
$query->where('d.material_no', (string) $filters['material_no']);
}
if (($filters['keyword'] ?? '') !== '') {
$like = '%' . (string) $filters['keyword'] . '%';
$query->where(function ($q) use ($like): void {
$q->where('d.process_instance_id', 'like', $like)
->orWhere('d.daifa_trade_no', 'like', $like)
->orWhere('d.shop_trade_no', 'like', $like);
});
}
if (($filters['created_from'] ?? '') !== '') {
$query->where('d.workflow_created_at', '>=', (string) $filters['created_from']);
}
if (($filters['created_to'] ?? '') !== '') {
$query->where('d.workflow_created_at', '<', $this->createdToBound((string) $filters['created_to']));
}
return $query;
}
/** 采购单来源加上页面筛选;该来源被筛选条件整体排除时返回 null。 */
private function purchaseFiltered(array $filters): ?Builder
{
// 采购单行没有商品ID,按商品ID筛选时不会有采购单行。
if (($filters['product_id'] ?? '') !== '') {
return null;
}
$query = $this->connection->table('product_purchase_price_mirror_rows as p')
->whereIn('p.document_status', ['Confirmed', 'Finished']);
// 只有按店铺/关键字筛选时才需要关联订单表。
if (($filters['shop_name'] ?? '') !== '' || ($filters['keyword'] ?? '') !== '') {
$query->leftJoin('ops_jst_orders as o', 'o.o_id', '=', 'p.source_order_no');
}
if (($filters['shop_name'] ?? '') !== '') {
$query->where('o.shop_name', (string) $filters['shop_name']);
}
if (($filters['supplier_name'] ?? '') !== '') {
$query->whereRaw("p.supplier_name LIKE ? ESCAPE '!'", [$this->contains((string) $filters['supplier_name'])]);
}
if (($filters['material_no'] ?? '') !== '') {
$query->where('p.sku_id', (string) $filters['material_no']);
}
if (($filters['keyword'] ?? '') !== '') {
$like = '%' . (string) $filters['keyword'] . '%';
$query->where(function ($q) use ($like): void {
$q->where('p.purchase_order_id', 'like', $like)
->orWhere('o.so_id', 'like', $like)
->orWhere('p.source_order_no', 'like', $like);
});
}
if (($filters['status'] ?? '') !== '') {
$query->whereRaw('UPPER(p.document_status) = ?', [strtoupper((string) $filters['status'])]);
}
if (($filters['created_from'] ?? '') !== '') {
$query->whereRaw('COALESCE(p.document_created_at, p.purchase_updated_at) >= ?', [(string) $filters['created_from']]);
}
if (($filters['created_to'] ?? '') !== '') {
$query->whereRaw(
'COALESCE(p.document_created_at, p.purchase_updated_at) < ?',
[$this->createdToBound((string) $filters['created_to'])],
);
}
return $query;
}
private function createdToBound(string $date): string
{
return Carbon::parse($date)->addDay()->startOfDay()->format('Y-m-d H:i:s');
}
private function dingStatusKey(): string
{
return strtoupper(trim((string) config(
'upstream_mirrors.ding_online_daifa_price.only_status',
'COMPLETED',
))) ?: 'COMPLETED';
}
/** @return \Generator<int,array<string,mixed>> */
public function exportRows(array $filters): \Generator
{
if (! $this->ready()) {
throw new RuntimeException('business_mirror_dataset_not_ready');
}
if (($filters['audit'] ?? '') !== '') {
$keys = $this->auditIndex($filters)['keys'][$filters['audit']] ?? [];
foreach (array_chunk($keys, self::EXPORT_HYDRATE_BATCH) as $chunk) {
$buffer = [];
foreach ($this->rowsForKeys($chunk, $filters) as $row) {
$buffer[] = $this->presentRow($row);
}
$this->attachJstOrderDetails($buffer);
$this->attachCostAudit($buffer);
yield from $buffer;
}
return;
}
$rows = $this->filteredQuery($filters)
->orderByDesc('workflow_created_at')
->orderByDesc('process_instance_id')
->orderByDesc('detail_index')
->cursor();
$buffer = [];
foreach ($rows as $row) {
$buffer[] = $this->presentRow($row);
if (count($buffer) >= self::EXPORT_HYDRATE_BATCH) {
$this->attachJstOrderDetails($buffer);
$this->attachCostAudit($buffer);
yield from $buffer;
$buffer = [];
}
}
if ($buffer !== []) {
$this->attachJstOrderDetails($buffer);
$this->attachCostAudit($buffer);
yield from $buffer;
}
}
/**
* 按线上订单号(钉钉行)或内部单号(采购单行)批量关联聚水潭订单,补充:
* - ship_skus:实际发货商品编码(组合装拆开后的子商品编码 × 数量);
* - jst_labels:聚水潭订单界面的「标记|多标签」。
*
* 只按当前这一批行的单号做等值查询(so_id / o_id 均有索引),不扫整张订单表。
*
* @param list<array<string,mixed>> $rows
*/
private function attachJstOrderDetails(array &$rows): void
{
foreach ($rows as &$row) {
$row['ship_skus'] = '';
$row['ship_sku_items'] = [];
$row['jst_labels'] = '';
}
unset($row);
$soIds = [];
$oIds = [];
foreach ($rows as $row) {
if (($row['source_type'] ?? 'ding') === 'purchase') {
$oId = trim((string) ($row['source_order_no'] ?? ''));
if ($oId !== '' && ctype_digit($oId)) {
$oIds[(int) $oId] = true;
}
} else {
foreach ($this->splitTradeNos((string) ($row['shop_trade_no'] ?? '')) as $soId) {
$soIds[$soId] = true;
}
}
}
if ($soIds === [] && $oIds === []) {
return;
}
try {
[$ordersBySo, $ordersByOid] = $this->loadJstOrders(array_keys($soIds), array_keys($oIds));
} catch (\Throwable $exception) {
report($exception);
return;
}
foreach ($rows as &$row) {
if (($row['source_type'] ?? 'ding') === 'purchase') {
$orders = $ordersByOid[trim((string) ($row['source_order_no'] ?? ''))] ?? [];
} else {
$orders = [];
foreach ($this->splitTradeNos((string) ($row['shop_trade_no'] ?? '')) as $soId) {
foreach ($ordersBySo[$soId] ?? [] as $order) {
$orders[$order['key']] = $order;
}
}
$orders = array_values($orders);
}
if ($orders === []) {
continue;
}
$active = array_values(array_filter(
$orders,
static fn (array $order): bool => ! in_array($order['status'], self::JST_INACTIVE_ORDER_STATUSES, true),
));
if ($active === []) {
$active = $orders;
}
$labels = [];
$items = [];
foreach ($active as $order) {
foreach ($order['labels'] as $label) {
$labels[$label] = true;
}
foreach ($order['items'] as $item) {
$items[] = $item;
}
}
$shipped = $this->shippedItemsFor($items, (string) ($row['material_no'] ?? ''));
$row['ship_sku_items'] = $shipped;
$row['ship_skus'] = implode(', ', array_map(
fn (array $item): string => $item['sku_id'] . '×' . $this->formatQty($item['qty']),
$shipped,
));
$row['jst_labels'] = implode(',', array_keys($labels));
}
unset($row);
}
/**
* 同一订单里可能有多个组合装或混着普通商品:优先只取由本行「商品编码」
* 拆出来的子商品;本行编码本身就是普通商品时取它自己;都对不上再列整单。
*
* @param list<array{sku_id:string,qty:float,combine:string}> $items
* @return list<array{sku_id:string,qty:float}>
*/
private function shippedItemsFor(array $items, string $materialNo): array
{
$materialNo = trim($materialNo);
$selected = [];
if ($materialNo !== '') {
$selected = array_values(array_filter($items, static fn (array $i): bool => $i['combine'] === $materialNo));
if ($selected === []) {
$selected = array_values(array_filter(
$items,
static fn (array $i): bool => $i['combine'] === '' && $i['sku_id'] === $materialNo,
));
}
}
if ($selected === []) {
$selected = $items;
}
$merged = [];
foreach ($selected as $item) {
if ($item['sku_id'] === '') {
continue;
}
$merged[$item['sku_id']] = ($merged[$item['sku_id']] ?? 0.0) + $item['qty'];
}
$result = [];
foreach ($merged as $skuId => $qty) {
if ($qty > 0) {
$result[] = ['sku_id' => (string) $skuId, 'qty' => $qty];
}
}
return $result;
}
/**
* @param list<string> $soIds
* @param list<int> $oIds
* @return array{0:array<string,list<array<string,mixed>>>,1:array<string,list<array<string,mixed>>>}
*/
private function loadJstOrders(array $soIds, array $oIds): array
{
$bySo = [];
$byOid = [];
foreach ($this->jstOrderSources() as $source) {
$columns = ['o_id', 'so_id'];
if ($source['status']) {
$columns[] = 'status';
}
if ($source['labels']) {
$columns[] = 'labels';
}
$orders = [];
foreach (array_chunk($soIds, self::JST_LOOKUP_CHUNK) as $chunk) {
foreach ($this->connection->table($source['orders'])->whereIn('so_id', $chunk)->get($columns) as $order) {
$orders[(string) $order->o_id] = $order;
}
}
foreach (array_chunk($oIds, self::JST_LOOKUP_CHUNK) as $chunk) {
foreach ($this->connection->table($source['orders'])->whereIn('o_id', $chunk)->get($columns) as $order) {
$orders[(string) $order->o_id] = $order;
}
}
if ($orders === []) {
continue;
}
$itemsByOrder = [];
if ($source['items'] !== '') {
$itemColumns = ['o_id', 'sku_id', 'qty'];
if ($source['combine']) {
$itemColumns[] = 'src_combine_sku_id';
}
foreach (array_chunk(array_map('intval', array_keys($orders)), self::JST_LOOKUP_CHUNK) as $chunk) {
foreach ($this->connection->table($source['items'])->whereIn('o_id', $chunk)->get($itemColumns) as $item) {
$itemsByOrder[(string) $item->o_id][] = [
'sku_id' => trim((string) ($item->sku_id ?? '')),
'qty' => (float) ($item->qty ?? 0),
'combine' => trim((string) ($item->src_combine_sku_id ?? '')),
];
}
}
}
foreach ($orders as $oId => $order) {
$entry = [
'key' => $source['orders'] . ':' . $oId,
'status' => trim((string) ($order->status ?? '')),
'labels' => $this->splitLabels((string) ($order->labels ?? '')),
'items' => $itemsByOrder[$oId] ?? [],
];
$soId = trim((string) ($order->so_id ?? ''));
if ($soId !== '') {
$bySo[$soId][] = $entry;
}
$byOid[$oId][] = $entry;
}
}
return [$bySo, $byOid];
}
/** @return list<array{orders:string,items:string,status:bool,labels:bool,combine:bool}> */
private function jstOrderSources(): array
{
if ($this->jstOrderSourceCache !== null) {
return $this->jstOrderSourceCache;
}
$schema = Schema::connection($this->connection->getName());
$sources = [];
foreach (self::JST_ORDER_SOURCES as [$orders, $items]) {
if (! $schema->hasTable($orders) || ! $schema->hasColumns($orders, ['o_id', 'so_id'])) {
continue;
}
$hasItems = $schema->hasTable($items) && $schema->hasColumns($items, ['o_id', 'sku_id', 'qty']);
$sources[] = [
'orders' => $orders,
'items' => $hasItems ? $items : '',
'status' => $schema->hasColumn($orders, 'status'),
'labels' => $schema->hasColumn($orders, 'labels'),
'combine' => $hasItems && $schema->hasColumn($items, 'src_combine_sku_id'),
];
}
return $this->jstOrderSourceCache = $sources;
}
/** @return list<string> */
private function splitTradeNos(string $value): array
{
$parts = preg_split('/[\s,,;;\/]+/u', trim($value)) ?: [];
return array_values(array_unique(array_filter(array_map('trim', $parts), static fn (string $v): bool => $v !== '')));
}
/** @return list<string> */
private function splitLabels(string $value): array
{
$parts = preg_split('/[,,]+/u', trim($value)) ?: [];
return array_values(array_unique(array_filter(array_map('trim', $parts), static fn (string $v): bool => $v !== '')));
}
private function formatQty(float $qty): string
{
return abs($qty - round($qty)) < 0.0001
? (string) (int) round($qty)
: rtrim(rtrim(number_format($qty, 4, '.', ''), '0'), '.');
}
private function filteredQuery(array $filters): Builder
{
$query = $this->combinedQuery();
if (($filters['product_id'] ?? '') !== '') {
$query->where('product_id', (string) $filters['product_id']);
}
if (($filters['shop_name'] ?? '') !== '') {
$query->where('shop_name', (string) $filters['shop_name']);
}
if (($filters['supplier_name'] ?? '') !== '') {
$query->whereRaw("supplier_name LIKE ? ESCAPE '!'", [$this->contains((string) $filters['supplier_name'])]);
}
if (($filters['material_no'] ?? '') !== '') {
$query->where('material_no', (string) $filters['material_no']);
}
if (($filters['keyword'] ?? '') !== '') {
$keyword = (string) $filters['keyword'];
$query->where(function ($q) use ($keyword): void {
$q->where('process_instance_id', 'like', '%' . $keyword . '%')
->orWhere('daifa_trade_no', 'like', '%' . $keyword . '%')
->orWhere('shop_trade_no', 'like', '%' . $keyword . '%')
->orWhere('source_order_no', 'like', '%' . $keyword . '%');
});
}
if (($filters['status'] ?? '') !== '') {
$query->where('status_key', strtoupper((string) $filters['status']));
}
if (($filters['created_from'] ?? '') !== '') {
$query->where('workflow_created_at', '>=', (string) $filters['created_from']);
}
if (($filters['created_to'] ?? '') !== '') {
$query->where(
'workflow_created_at',
'<',
$this->createdToBound((string) $filters['created_to']),
);
}
return $query;
}
private function combinedQuery(): Builder
{
$statuses = $this->sourceStatuses();
$queries = [];
if ((bool) ($statuses['ding']['ready'] ?? false)) {
$queries[] = $this->dingQuery();
}
if ((bool) ($statuses['purchase']['ready'] ?? false)) {
$queries[] = $this->purchaseQuery();
}
if ($queries === []) {
throw new RuntimeException('business_mirror_dataset_not_ready');
}
$union = array_shift($queries);
foreach ($queries as $query) {
$union->unionAll($query);
}
return $this->connection->query()->fromSub($union, 'online_daifa_rows');
}
private function dingQuery(): Builder
{
$status = $this->dingStatusKey();
return $this->connection->table(self::TARGET_TABLE . ' as d')->select([
DB::raw("'ding' AS source_type"),
'd.process_instance_id',
'd.detail_index',
DB::raw('NULL AS purchase_order_id'),
DB::raw('NULL AS source_order_no'),
'd.product_id',
'd.product_name',
'd.daifa_trade_no',
'd.shop_trade_no',
'd.shop_name',
'd.payee_name',
DB::raw("CASE WHEN d.payee_name = '宁波顾家望业家居有限公司' THEN '刘萧代发' ELSE '1688代发' END AS tag"),
'd.supplier_name',
'd.material_no',
'd.ship_qty',
'd.paid_amount',
'd.daifa_amount',
'd.price_diff',
'd.workflow_payload',
'd.scheme_payload',
'd.task_payload',
DB::raw("'" . $status . "' AS status_key"),
'd.workflow_created_at',
]);
}
private function purchaseQuery(): Builder
{
return $this->connection->table('product_purchase_price_mirror_rows as p')
->leftJoin('ops_jst_orders as o', 'o.o_id', '=', 'p.source_order_no')
->whereIn('p.document_status', ['Confirmed', 'Finished'])
->select([
DB::raw("'purchase' AS source_type"),
'p.purchase_order_id as process_instance_id',
'p.item_id as detail_index',
'p.purchase_order_id',
'p.source_order_no',
DB::raw("'' AS product_id"),
DB::raw('NULL AS product_name'),
DB::raw("'' AS daifa_trade_no"),
'o.so_id as shop_trade_no',
'o.shop_name',
DB::raw("'' AS payee_name"),
DB::raw("'采购单管理' AS tag"),
'p.supplier_name',
'p.sku_id as material_no',
'p.quantity as ship_qty',
DB::raw('NULL AS paid_amount'),
'p.unit_price as daifa_amount',
DB::raw('NULL AS price_diff'),
DB::raw('NULL AS workflow_payload'),
DB::raw('NULL AS scheme_payload'),
DB::raw('NULL AS task_payload'),
DB::raw('UPPER(p.document_status) AS status_key'),
DB::raw('COALESCE(p.document_created_at, p.purchase_updated_at) AS workflow_created_at'),
]);
}
/** @return array<string,array<string,mixed>> */
private function sourceStatuses(): array
{
if ($this->sourceStatusCache !== null) {
return $this->sourceStatusCache;
}
$schema = Schema::connection($this->connection->getName());
$dingState = null;
if ($schema->hasTable(self::TARGET_TABLE) && $schema->hasTable('business_mirror_table_states')) {
$dingState = $this->connection->table('business_mirror_table_states')
->where('dataset_key', self::DATASET_KEY)
->first();
}
$dingReady = $dingState !== null && $dingState->initial_completed_at !== null;
$purchaseState = null;
$orderState = null;
$requiredPurchaseColumns = [
'source_order_no', 'supplier_name', 'quantity', 'document_created_at',
];
$purchaseSchemaReady = $schema->hasTable('product_purchase_price_mirror_rows')
&& $schema->hasTable('product_purchase_price_sync_states')
&& $schema->hasTable('ops_jst_orders')
&& $schema->hasColumns('product_purchase_price_mirror_rows', $requiredPurchaseColumns);
if ($purchaseSchemaReady) {
$purchaseState = $this->connection->table('product_purchase_price_sync_states')
->where('dataset_key', ProductPurchasePriceMirrorRepository::DATASET_KEY)
->first();
$orderState = $schema->hasTable('business_mirror_table_states')
? $this->connection->table('business_mirror_table_states')
->where('dataset_key', self::ORDER_DATASET_KEY)
->first()
: null;
}
$purchaseReady = $purchaseState !== null
&& $purchaseState->initial_completed_at !== null
&& (int) ($purchaseState->projection_version ?? 1)
>= ProductPurchasePriceMirrorRepository::CURRENT_PROJECTION_VERSION
&& $orderState !== null
&& $orderState->initial_completed_at !== null;
return $this->sourceStatusCache = [
'ding' => [
'label' => '钉钉线上代发审批',
'ready' => $dingReady,
'status' => $dingState === null ? 'pending' : (string) $dingState->status,
'last_synced_at' => $dingState?->last_succeeded_at === null
? null
: (string) $dingState->last_succeeded_at,
],
'purchase' => [
'label' => '聚水潭采购单管理',
'ready' => $purchaseReady,
'status' => $purchaseReady
? (string) $purchaseState->status
: ($purchaseSchemaReady ? 'requires_full_sync' : 'pending'),
'last_synced_at' => $purchaseState?->last_succeeded_at === null
? null
: (string) $purchaseState->last_succeeded_at,
],
];
}
/**
* 报销成本核查:只针对钉钉行。系统成本 = Σ(实际发货子商品在普通商品资料的成本价 × 发货数量),
* 组合装按聚水潭订单里由本行商品编码拆出的子商品计算;代发订单金额 − 系统成本 > 0.01 即判为多报销。
*
* @param list<array<string,mixed>> $rows
*/
private function attachCostAudit(array &$rows): void
{
$skus = [];
foreach ($rows as &$row) {
$row['system_cost'] = null;
$row['cost_diff'] = null;
$row['audit_result'] = '';
$row['audit_detail'] = '';
if (($row['source_type'] ?? 'ding') !== 'ding') {
continue;
}
foreach ((array) ($row['ship_sku_items'] ?? []) as $item) {
$skus[(string) $item['sku_id']] = true;
}
}
unset($row);
$costs = $this->normalCostPrices(array_keys($skus));
foreach ($rows as &$row) {
if (($row['source_type'] ?? 'ding') !== 'ding') {
continue;
}
$items = (array) ($row['ship_sku_items'] ?? []);
if ($items === []) {
$row['audit_result'] = self::AUDIT_NO_SHIPMENT;
continue;
}
$cents = 0;
$missing = [];
$parts = [];
foreach ($items as $item) {
$sku = (string) $item['sku_id'];
$qty = (float) $item['qty'];
if (! isset($costs[$sku])) {
$missing[] = $sku;
$parts[] = $sku . '×' . $this->formatQty($qty) . ' 成本价缺失';
continue;
}
$line = (int) round($costs[$sku] * 100 * $qty);
$cents += $line;
$parts[] = sprintf('%s×%s×%s=%s', $sku, $this->formatQty($qty), rtrim(rtrim(number_format($costs[$sku], 4, '.', ''), '0'), '.'), number_format($line / 100, 2, '.', ''));
}
$row['system_cost'] = round($cents / 100, 2);
$row['audit_detail'] = implode("\n", $parts);
if ($missing !== []) {
$row['audit_result'] = self::AUDIT_COST_MISSING;
continue;
}
if ($row['daifa_amount'] === null) {
continue;
}
$row['cost_diff'] = round((float) $row['daifa_amount'] - $row['system_cost'], 2);
$row['audit_result'] = $row['cost_diff'] > self::AUDIT_TOLERANCE ? self::AUDIT_OVER : self::AUDIT_OK;
}
unset($row);
}
/**
* 普通商品资料成本价(ops_jst_sku,sku_type=normal,同编码取最新一条;空或 0 视为缺失)。
*
* @param list<string> $skus
* @return array<string,float>
*/
private function normalCostPrices(array $skus): array
{
$skus = array_values(array_filter($skus, static fn (string $sku): bool => $sku !== ''));
if ($skus === [] || ! Schema::connection($this->connection->getName())->hasTable('ops_jst_sku')) {
return [];
}
$costs = [];
foreach (array_chunk($skus, self::JST_LOOKUP_CHUNK) as $chunk) {
$records = $this->connection->table('ops_jst_sku')
->whereIn('sku_id', $chunk)
->where('sku_type', 'normal')
->orderByDesc('autoid')
->get(['sku_id', 'cost_price']);
foreach ($records as $record) {
$sku = (string) $record->sku_id;
if (! isset($costs[$sku]) && is_numeric($record->cost_price) && (float) $record->cost_price > 0) {
$costs[$sku] = (float) $record->cost_price;
}
}
}
return $costs;
}
/**
* 当前筛选条件下全部钉钉行的核查结果索引(按页面顺序的行键),供“核查结果”筛选、汇总和导出使用。
* 按筛选条件 + 镜像同步时间缓存 10 分钟。
*
* @return array{keys:array<string,list<string>>,counts:array<string,int>,over_amount:float}
*/
private function auditIndex(array $filters): array
{
$base = $filters;
unset($base['audit']);
$synced = (string) ($this->sourceStatuses()['ding']['last_synced_at'] ?? '');
$key = 'online-daifa:audit-index:v1:' . sha1(json_encode($base) . '|' . $synced);
$cached = Cache::get($key);
if (is_array($cached)) {
return $cached;
}
$index = ['keys' => array_fill_keys(self::AUDIT_RESULTS, []), 'counts' => array_fill_keys(self::AUDIT_RESULTS, 0), 'over_amount' => 0.0];
$buffer = [];
$flush = function () use (&$buffer, &$index): void {
if ($buffer === []) {
return;
}
$this->attachJstOrderDetails($buffer);
$this->attachCostAudit($buffer);
foreach ($buffer as $row) {
$result = (string) $row['audit_result'];
if ($result === '') {
continue;
}
$index['keys'][$result][] = $row['process_instance_id'] . '|' . $row['detail_index'];
$index['counts'][$result]++;
if ($result === self::AUDIT_OVER) {
$index['over_amount'] += (float) $row['cost_diff'];
}
}
$buffer = [];
};
// 只核查钉钉行:直接查钉钉镜像,不再经过与采购单整表合并的派生表。
$ding = (bool) ($this->sourceStatuses()['ding']['ready'] ?? false) ? $this->dingFiltered($base) : null;
$rows = $ding === null ? [] : $ding
->orderByDesc('d.workflow_created_at')
->orderByDesc('d.process_instance_id')
->orderByDesc('d.detail_index')
->get(['d.process_instance_id', 'd.detail_index', 'd.shop_trade_no', 'd.material_no', 'd.daifa_amount']);
foreach ($rows as $row) {
$buffer[] = [
'source_type' => 'ding',
'process_instance_id' => (string) $row->process_instance_id,
'detail_index' => (int) $row->detail_index,
'shop_trade_no' => (string) ($row->shop_trade_no ?? ''),
'material_no' => (string) ($row->material_no ?? ''),
'daifa_amount' => $row->daifa_amount === null ? null : (float) $row->daifa_amount,
];
if (count($buffer) >= self::AUDIT_HYDRATE_BATCH) {
$flush();
}
}
$flush();
$index['over_amount'] = round($index['over_amount'], 2);
try {
Cache::put($key, $index, self::AUDIT_CACHE_SECONDS);
} catch (\Throwable $exception) {
report($exception);
}
return $index;
}
/**
* 按“流程ID|明细序号”取回行,保持传入顺序。
*
* @param list<string> $keys
* @return list<object>
*/
private function rowsForKeys(array $keys, array $filters): array
{
if ($keys === []) {
return [];
}
// 键来自当前筛选条件下的核查索引,按流程ID直接回钉钉镜像取完整行即可。
$ids = array_values(array_unique(array_map(static fn (string $key): string => explode('|', $key, 2)[0], $keys)));
$rows = [];
foreach (array_chunk($ids, self::JST_LOOKUP_CHUNK) as $chunk) {
foreach ($this->dingQuery()->whereIn('d.process_instance_id', $chunk)->get() as $row) {
$rows[] = $row;
}
}
$byKey = [];
foreach ($rows as $row) {
$byKey[$row->process_instance_id . '|' . $row->detail_index] = $row;
}
return array_values(array_filter(array_map(static fn (string $key) => $byKey[$key] ?? null, $keys)));
}
/** @return array<string,mixed> */
private function presentRow(object $row): array
{
$sourceType = (string) ($row->source_type ?? 'ding');
$purchase = $sourceType === 'purchase';
$workflow = $purchase
? [
'businessId' => (string) ($row->purchase_order_id ?? ''),
'title' => '采购单管理',
'status' => (string) ($row->status_key ?? ''),
]
: $this->decode((string) ($row->workflow_payload ?? ''), []);
$productId = (string) ($row->product_id ?? '');
$resolvedName = $this->cleanName($row->product_name ?? null);
$product = ! $purchase && $productId !== '' && $resolvedName !== null
? ['matched' => true, 'name' => $resolvedName, 'sku_id' => null, 'i_id' => null]
: ['matched' => false, 'name' => null, 'sku_id' => null, 'i_id' => null];
$statusKey = $purchase
? strtoupper((string) ($row->status_key ?? ''))
: DingPayableWorkflowStatus::key($workflow);
return [
'source_type' => $sourceType,
'process_instance_id' => (string) $row->process_instance_id,
'detail_index' => (int) $row->detail_index,
'purchase_order_id' => $row->purchase_order_id === null ? null : (int) $row->purchase_order_id,
'source_order_no' => $row->source_order_no === null ? null : (string) $row->source_order_no,
'product_id' => $productId,
'daifa_trade_no' => (string) ($row->daifa_trade_no ?? ''),
'shop_trade_no' => (string) ($row->shop_trade_no ?? ''),
'shop_name' => (string) ($row->shop_name ?? ''),
'payee_name' => (string) ($row->payee_name ?? ''),
'tag' => (string) ($row->tag ?? ''),
'supplier_name' => (string) ($row->supplier_name ?? ''),
'material_no' => (string) ($row->material_no ?? ''),
'ship_qty' => $row->ship_qty === null ? null : (float) $row->ship_qty,
'paid_amount' => $row->paid_amount === null ? null : (float) $row->paid_amount,
'daifa_amount' => $row->daifa_amount === null ? null : (float) $row->daifa_amount,
'price_diff' => $row->price_diff === null ? null : (float) $row->price_diff,
'workflow' => $workflow,
'scheme' => $purchase ? [] : $this->decode((string) ($row->scheme_payload ?? ''), []),
'tasks' => $purchase ? [] : $this->decode((string) ($row->task_payload ?? ''), []),
'status_key' => $statusKey,
'status_label' => $purchase
? match ($statusKey) {
'CONFIRMED' => '已确认',
'FINISHED' => '已完成',
default => $statusKey,
}
: DingPayableWorkflowStatus::label($workflow),
'matched_product' => $product,
'order_matched' => ! $purchase
|| ((string) ($row->shop_trade_no ?? '') !== '' && (string) ($row->shop_name ?? '') !== ''),
'workflow_created_at' => (string) ($row->workflow_created_at ?? ''),
];
}
private function contains(string $value): string
{
return '%' . str_replace(['!', '%', '_'], ['!!', '!%', '!_'], trim($value)) . '%';
}
private function cleanName(mixed $value): ?string
{
if ($value === null) {
return null;
}
$name = trim((string) $value);
return $name === '' ? null : $name;
}
private function decode(string $json, mixed $fallback): mixed
{
if ($json === '') {
return $fallback;
}
$decoded = json_decode($json, true);
return is_array($decoded) ? $decoded : $fallback;
}
}