wx.sstbc.com/app/Services/CompanyNews/NewsCollection.php

<?php
namespace App\Services\CompanyNews;

use App\Models\Company;
use App\Repositories\YuanheRepository;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Str;
use InvalidArgumentException;
use RuntimeException;
use Throwable;

class NewsCollection
{
    public function configurationError(): ?string
    {
        $missing = [];
        foreach (['base_url' => 'YUANHE_BASE_URL', 'customer_id' => 'YUANHE_CUSTOMERID', 'auth_key' => 'YUANHE_AUTHKEY'] as $key => $label) {
            if (trim((string) config('yuanhe.'.$key)) === '') $missing[] = $label;
        }
        if ($missing) return '元禾接口尚未配置：'.implode('、', $missing).'。请填写后端 .env 并重启采集进程。';
        $url = config('yuanhe.base_url');
        if (!filter_var($url, FILTER_VALIDATE_URL) || !in_array(strtolower(parse_url($url, PHP_URL_SCHEME) ?? ''), ['https', 'http'], true)) {
            return 'YUANHE_BASE_URL 必须是完整的 HTTP/HTTPS 地址';
        }
        return null;
    }

    public function start(?int $actor, ?int $retryOf = null): object
    {
        if ($error = $this->configurationError()) throw new RuntimeException($error);
        return DB::transaction(function () use ($actor, $retryOf) {
            $control = DB::table('company_news_control')->where('id', 1)->lockForUpdate()->first();
            if ($control->active_batch_id) {
                throw new RuntimeException('已有采集批次运行，请等待完成后再获取更新');
            }
            $query = Company::query()->whereHas('qccAccount', fn ($q) => $q->whereIn('status', ['selected', 'occupied']))->orderBy('id');
            if ($retryOf !== null) {
                if (!DB::table('company_news_batches')->where('id', $retryOf)->exists()) {
                    throw new RuntimeException('原采集批次不存在');
                }
                $query->whereIn('id', DB::table('company_news_tasks')->where('batch_id', $retryOf)->where('status', 'failed')->select('company_id'));
            }
            $companies = $query->get(['id', 'company_name', 'credit_code']);
            $now = now();
            $id = DB::table('company_news_batches')->insertGetId([
                'created_by' => $actor, 'retry_of' => $retryOf, 'total' => $companies->count(),
                'status' => $companies->isEmpty() ? 'completed' : 'running',
                'finished_at' => $companies->isEmpty() ? $now : null, 'created_at' => $now, 'updated_at' => $now,
            ]);
            foreach ($companies->chunk(200) as $chunk) {
                DB::table('company_news_tasks')->insert($chunk->map(fn ($c) => [
                    'batch_id' => $id, 'company_id' => $c->id, 'company_name' => $c->company_name,
                    'credit_code' => $c->credit_code, 'status' => 'pending', 'created_at' => $now, 'updated_at' => $now,
                ])->all());
            }
            DB::table('company_news_control')->where('id', 1)->update(['active_batch_id' => $companies->isEmpty() ? null : $id]);
            return $this->summary($id);
        });
    }

    public function summary(int $id): object
    {
        $batch = DB::table('company_news_batches')->find($id);
        if (!$batch) {
            throw new RuntimeException('批次不存在');
        }
        $tasks = DB::table('company_news_tasks')->where('batch_id', $id);
        $counts = (clone $tasks)->selectRaw('status, COUNT(*) as count')->groupBy('status')->pluck('count', 'status');
        $batch->success = (int) ($counts['success'] ?? 0);
        $batch->failed = (int) ($counts['failed'] ?? 0);
        $batch->processed = $batch->success + $batch->failed;
        $batch->inserted = (int) (clone $tasks)->sum('inserted_count');
        $batch->invalid = (int) (clone $tasks)->sum('invalid_count');
        $batch->progress = $batch->total ? (int) floor($batch->processed * 100 / $batch->total) : 100;
        $batch->current_company = (clone $tasks)->where('status', 'processing')->value('company_name');
        return $batch;
    }

    public function runOne(): bool
    {
        $claim = DB::transaction(function () {
            $control = DB::table('company_news_control')->where('id', 1)->lockForUpdate()->first();
            if (!$control->active_batch_id || ($control->lease_until && $control->lease_until > now()->toDateTimeString())) {
                return null;
            }
            $batchId = $control->active_batch_id;
            // A dead worker may already have called the provider. Do not silently repeat that request.
            DB::table('company_news_tasks')->where('batch_id', $batchId)->where('status', 'processing')->update([
                'status' => 'failed', 'error' => '执行进程中断，外部请求结果不确定；请核查后手动重试',
                'finished_at' => now(), 'updated_at' => now(),
            ]);
            $task = DB::table('company_news_tasks')->where('batch_id', $batchId)->where('status', 'pending')->orderBy('company_id')->first();
            if (!$task) {
                $this->finishBatch($batchId);
                return null;
            }
            $token = (string) Str::uuid();
            DB::table('company_news_control')->where('id', 1)->update(['worker_token' => $token, 'lease_until' => now()->addMinutes(5)]);
            DB::table('company_news_tasks')->where('id', $task->id)->update([
                'status' => 'processing', 'started_at' => now(), 'request_ref' => 'news-'.$task->id.'-'.$token, 'updated_at' => now(),
            ]);
            return [$task, $token];
        });
        if (!$claim) {
            return false;
        }
        [$task, $token] = $claim;
        $response = null;
        $error = null;
        try {
            $company = Company::find($task->company_id);
            if (!$company || !$company->qccAccount || !in_array($company->qccAccount->status, ['selected', 'occupied'], true)) {
                throw new RuntimeException('企业已不属于候选或已占额企业，本次跳过请求');
            }
            if (!trim((string) $company->credit_code)) {
                throw new RuntimeException('企业缺少统一社会信用代码');
            }
            // Deliberately bypass QccCallService: this operation must NEVER change account status.
            $response = app(YuanheRepository::class)->enterprisePackInfo([
                'creditCode' => $company->credit_code, 'enterpriseName' => $company->company_name,
            ], 60);
            if (!in_array($response['code'] ?? null, [200, '200'], true)) {
                throw new RuntimeException('元禾业务失败：'.mb_substr(json_encode($response['msg'] ?? $response['code'] ?? '缺少 code', JSON_UNESCAPED_UNICODE), 0, 1000));
            }
            if (!isset($response['data']) || !is_array($response['data']) || !array_key_exists('news', $response['data'])) {
                throw new RuntimeException('响应缺少 data.news');
            }
            $news = $response['data']['news'];
            if ($news !== null && (!is_array($news) || array_values($news) !== $news)) {
                throw new RuntimeException('data.news 必须为数组或 null');
            }
        } catch (Throwable $e) {
            $error = mb_substr($e->getMessage(), 0, 2000);
        }
        try {
            $this->persist($task, $token, $response, $error);
        } catch (Throwable $e) {
            // The import transaction rolled back; record a failure without calling the provider again.
            $this->persist($task, $token, $response, '本地保存失败：'.mb_substr($e->getMessage(), 0, 1800));
        }
        return true;
    }

    private function persist(object $task, string $token, ?array $response, ?string $error): void
    {
        DB::transaction(function () use ($task, $token, $response, $error) {
            $control = DB::table('company_news_control')->where('id', 1)->lockForUpdate()->first();
            if ($control->worker_token !== $token) {
                return; // A stale worker must not commit after another worker reclaimed its lease.
            }
            $inserted = 0;
            $invalid = 0;
            $issues = [];
            $items = $error ? [] : ($response['data']['news'] ?? []);
            foreach ($items as $index => $item) {
                try {
                    if (!is_array($item)) {
                        throw new InvalidArgumentException('新闻不是对象');
                    }
                    $inserted += (int) app(NewsImporter::class)->store($task, $item);
                } catch (InvalidArgumentException $e) {
                    $invalid++;
                    if (count($issues) < 10) {
                        $issues[] = '第'.($index + 1).'条：'.$e->getMessage();
                    }
                }
            }
            $code = $response['code'] ?? null;
            DB::table('company_news_tasks')->where('id', $task->id)->update([
                'status' => ($error || $invalid) ? 'failed' : 'success',
                'response_code' => is_scalar($code) ? mb_substr((string) $code, 0, 40) : null,
                'news_count' => count($items), 'inserted_count' => $inserted, 'invalid_count' => $invalid,
                'error' => $error ?: ($invalid ? implode('；', $issues) : null),
                'finished_at' => now(), 'updated_at' => now(),
            ]);
            DB::table('company_news_control')->where('id', 1)->update(['worker_token' => null, 'lease_until' => null]);
            if (!DB::table('company_news_tasks')->where('batch_id', $task->batch_id)->where('status', 'pending')->exists()) {
                $this->finishBatch($task->batch_id);
            }
        });
    }

    private function finishBatch(int $batchId): void
    {
        DB::table('company_news_batches')->where('id', $batchId)->update(['status' => 'completed', 'finished_at' => now(), 'updated_at' => now()]);
        DB::table('company_news_control')->where('id', 1)->update(['active_batch_id' => null, 'worker_token' => null, 'lease_until' => null]);
    }
}