How to Take Bulk Screenshots in Python with a Screenshot API

1
from screenshotone import Client, TakeOptions
2
import shutil
3

4
client = Client('your-access-key', 'your-secret-key')
5

6
options = (TakeOptions.url('https://example.com')
7
    .format('png')
8
    .viewport_width(1920)
9
    .viewport_height(1080)
10
    .full_page(True))
11

12
image = client.take(options)
13

14
with open('screenshot.png', 'wb') as f:
15
    shutil.copyfileobj(image, f)

1
from screenshotone import Client, TakeOptions
2
import asyncio
3
import aiohttp
4
from pathlib import Path
5

6
class BulkScreenshotter:
7
    def __init__(self, access_key, secret_key, output_dir='screenshots'):
8
        self.client = Client(access_key, secret_key)
9
        self.output_dir = Path(output_dir)
10
        self.output_dir.mkdir(exist_ok=True)
11

12
    def _get_signed_url(self, url):
13
        """Generate a signed URL for the screenshot."""
14
        options = (TakeOptions.url(url)
15
            .format('png')
16
            .viewport_width(1920)
17
            .viewport_height(1080)
18
            .full_page(True)
19
            .block_cookie_banners(True)
20
            .block_chats(True))
21
        return self.client.generate_take_url(options)
22

23
    async def _fetch_screenshot(self, session, url, output_path):
24
        """Fetch a single screenshot."""
25
        signed_url = self._get_signed_url(url)
26
        try:
27
            async with session.get(signed_url) as response:
28
                if response.status == 200:
29
                    content = await response.read()
30
                    with open(output_path, 'wb') as f:
31
                        f.write(content)
32
                    return {'url': url, 'status': 'success', 'path': str(output_path)}
33
                else:
34
                    return {'url': url, 'status': 'failed', 'error': f'HTTP {response.status}'}
35
        except Exception as e:
36
            return {'url': url, 'status': 'failed', 'error': str(e)}
37

38
    async def process(self, urls, concurrency=5):
39
        """Process URLs with controlled concurrency."""
40
        semaphore = asyncio.Semaphore(concurrency)
41
        results = []
42

43
        async def bounded_fetch(session, url, index):
44
            async with semaphore:
45
                output_path = self.output_dir / f'{index:05d}.png'
46
                return await self._fetch_screenshot(session, url, output_path)
47

48
        async with aiohttp.ClientSession() as session:
49
            tasks = [
50
                bounded_fetch(session, url, i)
51
                for i, url in enumerate(urls)
52
            ]
53
            results = await asyncio.gather(*tasks)
54

55
        return results
56

57
# Usage
58
async def main():
59
    urls = [
60
        'https://example.com',
61
        'https://github.com',
62
        'https://stackoverflow.com',
63
    ]
64

65
    screenshotter = BulkScreenshotter('your-access-key', 'your-secret-key')
66
    results = await screenshotter.process(urls, concurrency=5)
67

68
    success = sum(1 for r in results if r['status'] == 'success')
69
    print(f"Completed: {success}/{len(results)} successful")
70

71
asyncio.run(main())

1
options = (TakeOptions.url('https://example.com')
2
    .block_cookie_banners(True))

1
options = (TakeOptions.url('https://example.com')
2
    .block_chats(True))

1
options = (TakeOptions.url('https://example.com')
2
    .dark_mode(True))

1
options = (TakeOptions.url('https://example.com')
2
    .delay(3000))  # Wait 3 seconds before capture

1
options = (TakeOptions.url('https://example.com')
2
    .selector('.main-content'))  # Wait for element

1
import csv
2
import asyncio
3
from pathlib import Path
4

5
async def process_csv(csv_path, access_key, secret_key):
6
    urls = []
7
    with open(csv_path, 'r') as f:
8
        reader = csv.DictReader(f)
9
        urls = [row['url'] for row in reader]
10

11
    screenshotter = BulkScreenshotter(access_key, secret_key)
12
    results = await screenshotter.process(urls, concurrency=10)
13

14
    # Save results
15
    with open('results.csv', 'w', newline='') as f:
16
        writer = csv.DictWriter(f, fieldnames=['url', 'status', 'path', 'error'])
17
        writer.writeheader()
18
        writer.writerows(results)
19

20
    return results
21

22
asyncio.run(process_csv('urls.csv', 'your-access-key', 'your-secret-key'))

1
import asyncio
2
import aiohttp
3
from tenacity import retry, stop_after_attempt, wait_exponential
4

5
class RobustScreenshotter:
6
    def __init__(self, access_key, secret_key):
7
        self.client = Client(access_key, secret_key)
8

9
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
10
    async def _fetch_with_retry(self, session, url, output_path):
11
        """Fetch with automatic retry on failure."""
12
        options = (TakeOptions.url(url)
13
            .format('png')
14
            .full_page(True))
15

16
        signed_url = self.client.generate_take_url(options)
17

18
        async with session.get(signed_url) as response:
19
            response.raise_for_status()
20
            content = await response.read()
21
            with open(output_path, 'wb') as f:
22
                f.write(content)
23
            return {'url': url, 'status': 'success'}
24

25
    async def process(self, urls):
26
        results = []
27
        async with aiohttp.ClientSession() as session:
28
            for i, url in enumerate(urls):
29
                try:
30
                    result = await self._fetch_with_retry(session, url, f'screenshots/{i}.png')
31
                    results.append(result)
32
                except Exception as e:
33
                    results.append({'url': url, 'status': 'failed', 'error': str(e)})
34
        return results

1
options = (TakeOptions.url('https://example.com')
2
    .webhook_url('https://your-server.com/webhook'))