From 4dc9b84eb8a20d69b83464f7e6d38cf02885ce0b Mon Sep 17 00:00:00 2001 From: m105 Date: Mon, 20 Jul 2026 00:26:27 +0300 Subject: [PATCH] Add bulk CDX shard ingestion command as an API-independent fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit crawlindex:bulk-fetch walks Common Crawl's CDX shard files directly on data.commoncrawl.org (via cc-index.paths.gz), instead of the index.commoncrawl.org query API — usable for bulk backfill/refresh or as a fallback when that API is down. Shares the group/save pipeline with crawlindex:fetch via a new SavesCommonCrawlCaptures trait. Also fixes a real bug this surfaced under load: site_pages.urlkey was a varchar(255), which is too narrow for real-world captured URLs (long query strings on spam/malformed pages overflow it). Widened to text and switched the uniqueness constraint to a sha1 hash column, since MySQL can't fully index a TEXT column. --- .../Commands/BulkFetchCommonCrawlIndex.php | 117 ++++++++++++++++++ .../Concerns/SavesCommonCrawlCaptures.php | 34 +++++ .../Commands/FetchCommonCrawlIndex.php | 26 +--- app/Models/SitePage.php | 1 + .../CommonCrawl/CommonCrawlIndexService.php | 86 ++++++++++++- ...01940_widen_urlkey_on_site_pages_table.php | 84 +++++++++++++ 6 files changed, 324 insertions(+), 24 deletions(-) create mode 100644 app/Console/Commands/BulkFetchCommonCrawlIndex.php create mode 100644 app/Console/Commands/Concerns/SavesCommonCrawlCaptures.php create mode 100644 database/migrations/2026_07_20_001940_widen_urlkey_on_site_pages_table.php diff --git a/app/Console/Commands/BulkFetchCommonCrawlIndex.php b/app/Console/Commands/BulkFetchCommonCrawlIndex.php new file mode 100644 index 0000000..d116ffa --- /dev/null +++ b/app/Console/Commands/BulkFetchCommonCrawlIndex.php @@ -0,0 +1,117 @@ +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; + } +} diff --git a/app/Console/Commands/Concerns/SavesCommonCrawlCaptures.php b/app/Console/Commands/Concerns/SavesCommonCrawlCaptures.php new file mode 100644 index 0000000..1d8716c --- /dev/null +++ b/app/Console/Commands/Concerns/SavesCommonCrawlCaptures.php @@ -0,0 +1,34 @@ +service->groupIntoSites($lines, $collection); + $siteGroups = $grouped['sites']; + $totalPages = $siteGroups->sum(fn (array $group) => $group['pages']->count()); + + $this->line(" {$grouped['captureCount']} capture(s) covering {$siteGroups->count()} site(s) and {$totalPages} unique page(s)."); + + if ($withContent) { + $this->comment(" Fetching page content from the WARC archive for up to {$totalPages} page(s), this will be slower."); + } + + $this->withProgressBar($siteGroups, function (array $group) use ($collection, $withContent, &$contentCache, &$totals) { + $result = $this->service->saveSiteGroup($group, $collection, $withContent, $contentCache); + + $result['wasNewSite'] ? $totals['sitesCreated']++ : $totals['sitesUpdated']++; + $totals['pagesCreated'] += $result['pagesCreated']; + $totals['pagesUpdated'] += $result['pagesUpdated']; + }); + + $this->newLine(); + } +} diff --git a/app/Console/Commands/FetchCommonCrawlIndex.php b/app/Console/Commands/FetchCommonCrawlIndex.php index 48b1720..cfbd870 100644 --- a/app/Console/Commands/FetchCommonCrawlIndex.php +++ b/app/Console/Commands/FetchCommonCrawlIndex.php @@ -2,6 +2,7 @@ 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; @@ -22,6 +23,8 @@ use Illuminate\Console\Command; #[Description('Fetch Common Crawl URL index records and save them as sites with their pages')] class FetchCommonCrawlIndex extends Command { + use SavesCommonCrawlCaptures; + private const VALID_MATCH_TYPES = ['exact', 'prefix', 'host', 'domain']; public function __construct(private readonly CommonCrawlIndexService $service) @@ -135,27 +138,4 @@ class FetchCommonCrawlIndex extends Command return self::SUCCESS; } - - private function saveCaptures(array $lines, string $collection, bool $withContent, array &$contentCache, array &$totals): void - { - $grouped = $this->service->groupIntoSites($lines, $collection); - $siteGroups = $grouped['sites']; - $totalPages = $siteGroups->sum(fn (array $group) => $group['pages']->count()); - - $this->line(" {$grouped['captureCount']} capture(s) covering {$siteGroups->count()} site(s) and {$totalPages} unique page(s)."); - - if ($withContent) { - $this->comment(" --with-metadata is set: fetching page content from the WARC archive for up to {$totalPages} page(s), this will be slower."); - } - - $this->withProgressBar($siteGroups, function (array $group) use ($collection, $withContent, &$contentCache, &$totals) { - $result = $this->service->saveSiteGroup($group, $collection, $withContent, $contentCache); - - $result['wasNewSite'] ? $totals['sitesCreated']++ : $totals['sitesUpdated']++; - $totals['pagesCreated'] += $result['pagesCreated']; - $totals['pagesUpdated'] += $result['pagesUpdated']; - }); - - $this->newLine(); - } } diff --git a/app/Models/SitePage.php b/app/Models/SitePage.php index 4acbd3a..3af342b 100644 --- a/app/Models/SitePage.php +++ b/app/Models/SitePage.php @@ -10,6 +10,7 @@ class SitePage extends Model protected $fillable = [ 'site_id', 'urlkey', + 'urlkey_hash', 'url', 'timestamp', 'mime', diff --git a/app/Services/CommonCrawl/CommonCrawlIndexService.php b/app/Services/CommonCrawl/CommonCrawlIndexService.php index 8d42005..ad47ef2 100644 --- a/app/Services/CommonCrawl/CommonCrawlIndexService.php +++ b/app/Services/CommonCrawl/CommonCrawlIndexService.php @@ -17,6 +17,8 @@ class CommonCrawlIndexService { private const COLLINFO_URL = 'https://index.commoncrawl.org/collinfo.json'; + private const DATA_BASE_URL = 'https://data.commoncrawl.org/'; + public function __construct(private readonly WarcRecordFetcher $warcFetcher) {} public function resolveLatestCollection(): ?string @@ -134,13 +136,14 @@ class CommonCrawlIndexService foreach ($pages as $page) { $attributes = $page; unset($attributes['collection'], $attributes['domain']); + $attributes['urlkey_hash'] = sha1($page['urlkey']); if ($withContent) { $attributes = array_merge($attributes, $this->fetchContent($page, $contentCache)); } $result = SitePage::updateOrCreate( - ['site_id' => $site->id, 'urlkey' => $page['urlkey']], + ['site_id' => $site->id, 'urlkey_hash' => $attributes['urlkey_hash']], $attributes, ); @@ -155,6 +158,87 @@ class CommonCrawlIndexService ]; } + /** + * List every CDX shard file for a collection, straight from the bulk data storage + * (data.commoncrawl.org) rather than the index.commoncrawl.org query API — this is + * what lets bulk ingestion work even when that API is offline. Returns paths + * relative to DATA_BASE_URL, or null if the paths file couldn't be fetched/parsed. + */ + public function listCdxShardPaths(string $collection): ?array + { + $response = Http::withUserAgent(WarcRecordFetcher::DEFAULT_USER_AGENT) + ->timeout(60) + ->retry(3, 2000) + ->get(self::DATA_BASE_URL."crawl-data/{$collection}/cc-index.paths.gz"); + + if (! $response->successful()) { + return null; + } + + $decompressed = @gzdecode($response->body()); + + if ($decompressed === false) { + return null; + } + + $paths = array_filter(explode("\n", trim($decompressed))); + + return array_values(array_filter( + $paths, + fn (string $path) => str_contains($path, '/indexes/cdx-') && str_ends_with($path, '.gz') + )); + } + + /** + * Download and decompress one CDX shard file, returning its raw lines (each + * " ", not yet parsed — see parseShardLine()). The whole + * shard is pulled into memory at once, which can be sizeable (shards run up to a + * couple hundred MB compressed); null on download/decompress failure. + */ + public function fetchShardLines(string $path): ?array + { + $response = Http::withUserAgent(WarcRecordFetcher::DEFAULT_USER_AGENT) + ->timeout(300) + ->retry(2, 3000) + ->get(self::DATA_BASE_URL.$path); + + if (! $response->successful()) { + return null; + } + + $decompressed = @gzdecode($response->body()); + + if ($decompressed === false) { + return null; + } + + return array_values(array_filter(explode("\n", trim($decompressed)))); + } + + /** + * A raw CDX shard line is " ", with urlkey/timestamp + * outside the JSON blob — unlike the CDX API's output, which is one self-contained + * JSON object per line. Re-shape it into that same self-contained form so it can + * flow through groupIntoSites() unchanged. Returns null for malformed lines. + */ + public function parseShardLine(string $line): ?string + { + if (! preg_match('/^(\S+) (\d{14}) (\{.*\})$/', $line, $matches)) { + return null; + } + + $record = json_decode($matches[3], true); + + if (! is_array($record)) { + return null; + } + + $record['urlkey'] = $matches[1]; + $record['timestamp'] = $matches[2]; + + return json_encode($record); + } + private function fetchContent(array $page, array &$contentCache): array { $contentCache[$page['record_hash']] ??= $this->warcFetcher->fetchMetadata($page); diff --git a/database/migrations/2026_07_20_001940_widen_urlkey_on_site_pages_table.php b/database/migrations/2026_07_20_001940_widen_urlkey_on_site_pages_table.php new file mode 100644 index 0000000..b044735 --- /dev/null +++ b/database/migrations/2026_07_20_001940_widen_urlkey_on_site_pages_table.php @@ -0,0 +1,84 @@ +index('site_id', 'site_pages_site_id_temp_index'); + }); + + Schema::table('site_pages', function (Blueprint $table) { + $table->dropUnique(['site_id', 'urlkey']); + }); + + // urlkey can't be widened to text while any index still covers its full length + // (MySQL requires a key length for TEXT/BLOB columns in an index), so this can + // only happen once the drop above has gone through. + Schema::table('site_pages', function (Blueprint $table) { + $table->text('urlkey')->change(); + $table->char('urlkey_hash', 40)->nullable()->after('urlkey'); + }); + + // Hashed in PHP, chunked, rather than a SQL SHA1()/MD5() call — not every MySQL + // build ships those (this one doesn't), and this guarantees the backfilled hash + // is byte-for-byte what CommonCrawlIndexService computes for new rows. + DB::table('site_pages')->orderBy('id')->chunkById(500, function ($rows) { + foreach ($rows as $row) { + DB::table('site_pages')->where('id', $row->id)->update(['urlkey_hash' => sha1($row->urlkey)]); + } + }); + + Schema::table('site_pages', function (Blueprint $table) { + $table->char('urlkey_hash', 40)->nullable(false)->change(); + $table->unique(['site_id', 'urlkey_hash']); + }); + + Schema::table('site_pages', function (Blueprint $table) { + $table->dropIndex('site_pages_site_id_temp_index'); + }); + } + + /** + * Reverse the migrations. + */ + public function down(): void + { + Schema::table('site_pages', function (Blueprint $table) { + $table->index('site_id', 'site_pages_site_id_temp_index'); + }); + + Schema::table('site_pages', function (Blueprint $table) { + $table->dropUnique(['site_id', 'urlkey_hash']); + }); + + Schema::table('site_pages', function (Blueprint $table) { + $table->string('urlkey', 255)->change(); + $table->dropColumn('urlkey_hash'); + }); + + Schema::table('site_pages', function (Blueprint $table) { + $table->unique(['site_id', 'urlkey']); + }); + + Schema::table('site_pages', function (Blueprint $table) { + $table->dropIndex('site_pages_site_id_temp_index'); + }); + } +};