connection = DB::connection((string) config('upstream_mirrors.connection', 'business_mirror')); } /** * 即时推送候选：需提醒 + 未处理 + 未推送 + 指定级别。 * * @param list $levels * @return Collection */ public function instantCandidates(array $levels, bool $retryFailed, int $limit): Collection { return $this->connection->table(StoreViolationRecordRepository::TABLE) ->where('needs_alert', true) ->where('handling_status', (string) config('store_violations.default_handling_status', '未处理')) ->whereIn('alert_level', $levels) ->where(function ($query) use ($retryFailed): void { $query->where('push_status', '未推送'); if ($retryFailed) { $query->orWhere('push_status', '推送失败'); } }) ->orderByRaw("CASE alert_level WHEN 'P0' THEN 0 WHEN 'P1' THEN 1 ELSE 2 END") ->orderBy('id') ->limit(max($limit, 1)) ->get(); } /** * 每日汇总取数：各级别待处理数量与超时未认领记录。 * * @return array{pending:array,overdue:Collection,p2_pending:Collection} */ public function digestData(int $overdueLimit = 20): array { $defaultStatus = (string) config('store_violations.default_handling_status', '未处理'); $pending = []; foreach ($this->connection->table(StoreViolationRecordRepository::TABLE) ->selectRaw('alert_level, COUNT(*) AS aggregate') ->where('needs_alert', true) ->where('handling_status', $defaultStatus) ->groupBy('alert_level') ->get() as $row) { $pending[(string) $row->alert_level] = (int) $row->aggregate; } $overdue = $this->connection->table(StoreViolationRecordRepository::TABLE) ->whereIn('alert_level', array_keys((array) config('store_violations.alert_levels', []))) ->whereNotIn('handling_status', StoreViolationRecordRepository::FINAL_HANDLING_STATUSES) ->whereNotNull('handling_deadline') ->where('handling_deadline', '<', Carbon::now()) ->where('needs_alert', true) ->orderBy('handling_deadline') ->limit(max($overdueLimit, 1)) ->get(['id', 'platform', 'shop_name', 'owner_name', 'alert_level', 'violation_type', 'handling_deadline']); $p2 = $this->connection->table(StoreViolationRecordRepository::TABLE) ->where('needs_alert', true) ->where('alert_level', 'P2') ->where('handling_status', $defaultStatus) ->where('push_status', '未推送') ->orderBy('id') ->get(['id']); return ['pending' => $pending, 'overdue' => $overdue, 'p2_pending' => $p2]; } /** @param list $ids */ public function markPushed(array $ids, Carbon $at): void { if ($ids === []) { return; } foreach (array_chunk($ids, 500) as $chunk) { $this->connection->table(StoreViolationRecordRepository::TABLE) ->whereIn('id', $chunk) ->update(['push_status' => '已推送', 'last_pushed_at' => $at, 'updated_at' => $at]); } } public function markPushFailed(int $id, Carbon $at): void { $this->connection->table(StoreViolationRecordRepository::TABLE) ->where('id', $id) ->update(['push_status' => '推送失败', 'last_pushed_at' => $at, 'updated_at' => $at]); } /** @param array $entry */ public function log(array $entry): void { if (! $this->available()) { return; } $now = now(); $this->connection->table(self::LOG_TABLE)->insert($entry + [ 'created_at' => $now, 'updated_at' => $now, ]); } public function available(): bool { return \Illuminate\Support\Facades\Schema::connection($this->connection->getName()) ->hasTable(self::LOG_TABLE); } }