#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Script to search a Wikimedia Commons category for 0x0 PDF files using
server-side ElasticSearch, purge their thumbnails in batches with sampling verification,
and do a final full verification pass.
29 June 2026
"""
import sys
import time
import random
import pywikibot
from colorama import Fore, Style, init
init(strip=False, convert=False)
pywikibot.config.put_throttle = 0
# Batch size - MediaWiki allows up to 500, but 50 is safer for reliability
BATCH_SIZE = 50
# Number of files to sample from each batch for verification
SAMPLE_SIZE = 5
# How long to wait after batch purge before sampling
SAMPLE_WAIT = 4
def batch_purge_titles(site, titles):
"""
Purge multiple pages in a single API call.
Returns True if the request succeeded, False otherwise.
"""
if not titles:
return True
titles_str = '|'.join(titles)
payload = {
'action': 'purge',
'titles': titles_str,
'format': 'json'
}
try:
response = site.simple_request(**payload).submit()
return 'purge' in response
except Exception as e:
print(Fore.RED + f" \u2192 [Batch Purge Error] {str(e)}" + Fore.WHITE)
return False
def get_file_dimensions(file_page):
"""Get width and height from a file page, returns (width, height)."""
if hasattr(file_page, '_imageinfo'):
del file_page._imageinfo
try:
info = file_page.latest_file_info
return getattr(info, 'width', 0), getattr(info, 'height', 0)
except Exception:
return 0, 0
def sample_verify_batch(site, batch_titles, sample_size=SAMPLE_SIZE):
"""
Sample a few files from the batch and verify their dimensions.
Returns (successes, failures, sample_details)
"""
if len(batch_titles) <= sample_size:
sample_titles = batch_titles
else:
sample_titles = random.sample(batch_titles, sample_size)
successes = 0
failures = 0
details = []
for title in sample_titles:
file_page = pywikibot.FilePage(site, title)
width, height = get_file_dimensions(file_page)
if width > 0 or height > 0:
successes += 1
details.append((title, width, height, True))
else:
failures += 1
details.append((title, width, height, False))
return successes, failures, details
def check_and_purge_pdfs(category_name: str):
"""
Leverages ElasticSearch to find 0x0 files in a category, purges them in batches,
samples each batch for verification, and does a final full verification pass.
"""
site = pywikibot.Site('commons', 'commons')
clean_cat_name = category_name
if clean_cat_name.lower().startswith('category:'):
clean_cat_name = clean_cat_name[9:].strip()
search_query = f'incategory:"{clean_cat_name}" filewidth:<1'
print(Fore.CYAN + "*" * 70)
print(Fore.GREEN + f"Executing search: {search_query} ...")
print(Fore.CYAN + "*" * 70 + Fore.WHITE)
gen = site.search(search_query, namespaces=[6])
# First pass: collect all PDF titles that need purging
titles_to_purge = []
for page in gen:
if not page.title().lower().endswith('.pdf'):
continue
titles_to_purge.append(page.title())
if not titles_to_purge:
print(Fore.GREEN + f"\nNo 0x0 PDFs found in '{clean_cat_name}'. Everything looks good!" + Fore.WHITE)
return
total_files = len(titles_to_purge)
print(Fore.YELLOW + f"\nFound {total_files} 0x0 PDFs. Purging in batches of {BATCH_SIZE}..." + Fore.WHITE)
print(Fore.YELLOW + f"Sampling {SAMPLE_SIZE} files per batch for immediate verification.\n" + Fore.WHITE)
# Track batch results for summary
batch_results = []
# Second pass: batch purge with sampling
batch_num = 0
for i in range(0, total_files, BATCH_SIZE):
batch = titles_to_purge[i:i + BATCH_SIZE]
batch_num += 1
batch_range = f"{i+1}-{min(i+BATCH_SIZE, total_files)}"
print(Fore.CYAN + f"{'='*70}" + Fore.WHITE)
print(Fore.CYAN + f"Batch {batch_num} (files {batch_range}): Purging {len(batch)} files..." + Fore.WHITE)
start_time = time.time()
purge_success = batch_purge_titles(site, batch)
elapsed = time.time() - start_time
if not purge_success:
print(Fore.RED + f" \u2192 Batch purge request failed after {elapsed:.2f}s" + Fore.WHITE)
batch_results.append((batch_num, batch_range, False, 0, 0, []))
continue
print(Fore.GREEN + f" \u2192 Purge request completed in {elapsed:.2f}s" + Fore.WHITE)
# Wait for servers to process, then sample
print(Fore.YELLOW + f" \u2192 Waiting {SAMPLE_WAIT}s for server processing..." + Fore.WHITE)
time.sleep(SAMPLE_WAIT)
# Sample verification
print(Fore.BLUE + f" \u2192 Sampling {min(SAMPLE_SIZE, len(batch))} files for verification..." + Fore.WHITE)
sample_start = time.time()
successes, failures, details = sample_verify_batch(site, batch, SAMPLE_SIZE)
sample_elapsed = time.time() - sample_start
# Display sample results
for title, width, height, ok in details:
short_title = title.replace('File:', '')[:45]
if ok:
print(Fore.GREEN + f" \u2713 {short_title} -> {width}x{height}" + Fore.WHITE)
else:
print(Fore.RED + f" \u2717 {short_title} -> still 0x0" + Fore.WHITE)
print(Fore.WHITE + f" \u2192 Sample verification: {successes}/{successes+failures} passed ({sample_elapsed:.2f}s)" + Fore.WHITE)
# Store results
batch_results.append((batch_num, batch_range, True, successes, failures, details))
# If all samples failed, something might be wrong - wait longer
if failures == successes + failures and failures > 0:
print(Fore.MAGENTA + f" \u2192 [WARNING] All samples failed! Server may be lagging. Waiting 15s extra..." + Fore.WHITE)
time.sleep(15)
# Re-sample to see if it was just lag
print(Fore.BLUE + f" \u2192 Re-sampling after delay..." + Fore.WHITE)
successes2, failures2, details2 = sample_verify_batch(site, batch, SAMPLE_SIZE)
if successes2 > 0:
print(Fore.GREEN + f" \u2192 Re-sample shows {successes2}/{successes2+failures2} passed - was just server lag." + Fore.WHITE)
else:
print(Fore.RED + f" \u2192 Re-sample still shows 0/{failures2} passed - possible persistent issue." + Fore.WHITE)
# Summary of batch processing
print(Fore.CYAN + "\n" + "=" * 70)
print(Fore.CYAN + "BATCH PURGE SUMMARY" + Fore.WHITE)
print(Fore.CYAN + "=" * 70 + Fore.WHITE)
total_sampled_ok = 0
total_sampled_fail = 0
for batch_num, batch_range, purged, ok, fail, _ in batch_results:
status = Fore.GREEN + "PURGED" + Fore.WHITE if purged else Fore.RED + "FAILED" + Fore.WHITE
sample_info = f"Sample: {ok}/{ok+fail} OK" if purged else "N/A"
print(f" Batch {batch_num} ({batch_range}): {status} | {sample_info}")
total_sampled_ok += ok
total_sampled_fail += fail
total_sampled = total_sampled_ok + total_sampled_fail
if total_sampled > 0:
pct = (total_sampled_ok / total_sampled) * 100
print(Fore.WHITE + f"\n Overall sample success rate: {total_sampled_ok}/{total_sampled} ({pct:.1f}%)" + Fore.WHITE)
# Final full verification pass
print(Fore.CYAN + "\n" + "=" * 70)
print(Fore.CYAN + "FINAL FULL VERIFICATION" + Fore.WHITE)
print(Fore.CYAN + "=" * 70 + Fore.WHITE)
print(Fore.YELLOW + f"\nWaiting 5 seconds before full verification..." + Fore.WHITE)
time.sleep(5)
verified_count = 0
failed_count = 0
still_zero = []
for idx, title in enumerate(titles_to_purge, 1):
file_page = pywikibot.FilePage(site, title)
width, height = get_file_dimensions(file_page)
if width > 0 or height > 0:
verified_count += 1
# Only print every 50th success to reduce noise
if idx % 50 == 0 or idx == total_files:
print(Fore.GREEN + f" [{idx}/{total_files}] {title[:50]}... -> {width}x{height} \u2713" + Fore.WHITE)
else:
failed_count += 1
still_zero.append(title)
print(Fore.RED + f" [{idx}/{total_files}] {title[:50]}... -> still 0x0 \u2717" + Fore.WHITE)
# Final summary
print(Fore.CYAN + "\n" + "-" * 70)
print(Fore.GREEN + f"FINAL RESULTS:" + Fore.WHITE)
print(Fore.GREEN + f" Total files processed: {total_files}" + Fore.WHITE)
print(Fore.GREEN + f" Successfully restored: {verified_count} ({(verified_count/total_files)*100:.1f}%)" + Fore.WHITE)
print(Fore.RED + f" Still 0x0: {failed_count} ({(failed_count/total_files)*100:.1f}%)" + Fore.WHITE)
if still_zero:
print(Fore.MAGENTA + f"\nFiles still showing 0x0 (may need manual review or re-purging):" + Fore.WHITE)
for title in still_zero[:10]: # Show first 10
print(Fore.MAGENTA + f" - {title}" + Fore.WHITE)
if len(still_zero) > 10:
print(Fore.MAGENTA + f" ... and {len(still_zero) - 10} more" + Fore.WHITE)
print(Fore.CYAN + "-" * 70 + Fore.WHITE)
if __name__ == '__main__':
if len(sys.argv) < 2:
print(Fore.MAGENTA + "Usage: python3 purge_0x0_pdfs.py 'Category Name'" + Fore.WHITE)
print(Fore.MAGENTA + "Example: python3 purge_0x0_pdfs.py 'Old books from American Libraries'" + Fore.WHITE)
sys.exit(1)
target_category = sys.argv[1]
check_and_purge_pdfs(target_category)