huginn/app/Console/Commands/BulkFetchCommonCrawlIndex.php

<?php

namespace App\Console\Commands;

use App\Console\Commands\Concerns\SavesCommonCrawlCaptures;
use App\Services\CommonCrawl\CommonCrawlIndexService;
use Illuminate\Console\Attributes\Description;
use Illuminate\Console\Attributes\Signature;
use Illuminate\Console\Command;

#[Signature('crawlindex:bulk-fetch
    {--collection= : Common Crawl collection id, e.g. CC-MAIN-2026-25. Defaults to the latest collection (resolved via the query API — pass this explicitly to avoid touching the API at all)}
    {--tld= : Only ingest captures whose host ends in this TLD (e.g. es). Filtered locally after download, never via the query API}
    {--shard-start=0 : Index (0-based) of the first CDX shard file to process}
    {--shard-count=1 : Number of consecutive shards to process, starting at --shard-start}
    {--all-shards : Process every remaining shard, starting at --shard-start (overrides --shard-count)}
    {--with-metadata : Also fetch each page from the WARC archive and extract title/description/keywords/content (slower, one extra request per saved page — expensive at bulk scale, consider pairing with --tld)}
')]
#[Description('Bulk-populate/refresh sites and pages by walking Common Crawl\'s CDX shard files directly from data.commoncrawl.org, bypassing the index.commoncrawl.org query API')]
class BulkFetchCommonCrawlIndex extends Command
{
    use SavesCommonCrawlCaptures;

    /** Raw shard lines are grouped and saved in batches this large, to bound memory use per shard. */
    private const CHUNK_SIZE = 5000;

    public function __construct(private readonly CommonCrawlIndexService $service)
    {
        parent::__construct();
    }

    /**
     * Execute the console command.
     */
    public function handle(): int
    {
        $tld = $this->option('tld') ? strtolower(ltrim((string) $this->option('tld'), '.')) : null;
        $shardStart = max(0, (int) $this->option('shard-start'));
        $allShards = (bool) $this->option('all-shards');
        $shardCount = $allShards ? null : max(1, (int) $this->option('shard-count'));
        $withMetadata = (bool) $this->option('with-metadata');

        $collection = $this->option('collection') ?: $this->service->resolveLatestCollection();

        if (! $collection) {
            $this->error('Could not resolve a Common Crawl collection id. Pass one explicitly with --collection.');

            return self::FAILURE;
        }

        $this->info("Listing CDX shards for collection {$collection}...");
        $shardPaths = $this->service->listCdxShardPaths($collection);

        if ($shardPaths === null) {
            $this->error('Could not download or parse cc-index.paths.gz for this collection.');

            return self::FAILURE;
        }

        $totalShards = count($shardPaths);

        if ($totalShards === 0) {
            $this->error('cc-index.paths.gz for this collection listed no CDX shards.');

            return self::FAILURE;
        }

        if ($shardStart >= $totalShards) {
            $this->error("--shard-start ({$shardStart}) is beyond the last shard index ({$totalShards} shards total, 0-".($totalShards - 1).').');

            return self::FAILURE;
        }

        $shardEnd = $allShards ? $totalShards - 1 : min($totalShards - 1, $shardStart + $shardCount - 1);

        $this->info("Collection has {$totalShards} CDX shard(s). Processing shard(s) {$shardStart}-{$shardEnd}".($tld ? " filtered to .{$tld}" : '').'.');

        $totals = ['sitesCreated' => 0, 'sitesUpdated' => 0, 'pagesCreated' => 0, 'pagesUpdated' => 0];
        $contentCache = [];

        for ($shardIndex = $shardStart; $shardIndex <= $shardEnd; $shardIndex++) {
            $path = $shardPaths[$shardIndex];
            $this->info("Fetching shard {$shardIndex}/{$shardEnd}: {$path}");

            $lines = $this->service->fetchShardLines($path);

            if ($lines === null) {
                $this->warn("  Failed to download or decompress shard {$shardIndex}, skipping.");

                continue;
            }

            if ($tld) {
                $prefix = $tld.',';
                $lines = array_values(array_filter($lines, fn (string $line) => str_starts_with($line, $prefix)));
            }

            $this->line('  '.count($lines).' matching raw record(s) in this shard.');

            foreach (array_chunk($lines, self::CHUNK_SIZE) as $chunk) {
                $jsonLines = array_values(array_filter(array_map(
                    fn (string $line) => $this->service->parseShardLine($line),
                    $chunk
                )));

                if (! empty($jsonLines)) {
                    $this->saveCaptures($jsonLines, $collection, $withMetadata, $contentCache, $totals);
                }
            }
        }

        $this->newLine();
        $this->info("Done. Processed shard(s) {$shardStart}-{$shardEnd}. Sites: {$totals['sitesCreated']} new, {$totals['sitesUpdated']} updated. Pages: {$totals['pagesCreated']} new, {$totals['pagesUpdated']} updated.");

        return self::SUCCESS;
    }
}