-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathscrape_images.py
More file actions
84 lines (72 loc) · 2.47 KB
/
Copy pathscrape_images.py
File metadata and controls
84 lines (72 loc) · 2.47 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
import os
import re
import requests
from urllib.parse import urljoin
GAFFA_API_KEY = os.getenv("GAFFA_API_KEY")
HEADERS = {
"x-api-key": GAFFA_API_KEY,
"Content-Type": "application/json"
}
def get_sitemap_urls(site_url, max_cache_age=86400):
payload = {
"url": site_url,
"max_cache_age": max_cache_age
}
print("Retrieving sitemap URLs.")
response = requests.post("https://api.gaffa.dev/v1/site/map", json=payload, headers=HEADERS)
return response.json()["data"]["links"]
def get_dom(url):
payload = {
"url": url,
"async": False,
"settings": {
"actions": [
{"type": "wait", "selector": "img", "timeout": 20000},
{"type": "capture_dom"}
],
"time_limit": 40000
}
}
print("Capturing DOM URL.")
response = requests.post("https://api.gaffa.dev/v1/browser/requests", json=payload, headers=HEADERS)
dom_url = response.json()["data"]["actions"][1]["output"]
print("Retrieving DOM.")
dom_response = requests.get(dom_url)
return dom_response.text
def extract_image_urls(dom_content, base_url):
image_urls = []
src_pattern = r'<img[^>]+(?:src|data-src)=["\']([^"\']+)["\']'
matches = re.findall(src_pattern, dom_content)
for src in matches:
if not src.startswith(('http:', 'https:')):
src = urljoin(base_url, src)
image_urls.append(src)
return image_urls
def download_image(image_url, filename):
payload = {
"url": image_url,
"async": False,
"settings": {
"actions": [{"type": "download_file"}]
}
}
print("Retrieving download URL.")
response = requests.post("https://api.gaffa.dev/v1/browser/requests", json=payload, headers=HEADERS)
actions = response.json()["data"]["actions"]
download_url = actions[0]["output"]
download_ext = os.path.splitext(download_url)[1]
print("Downloading image.")
img_response = requests.get(download_url)
filepath = f"{filename}{download_ext}"
with open(filepath, 'wb') as f:
f.write(img_response.content)
def main():
site_url = "https://gaffa.dev"
sitemap_urls = get_sitemap_urls(site_url)[:3]
for i, url in enumerate(sitemap_urls, 1):
dom_content = get_dom(url)
image_urls = extract_image_urls(dom_content, url)
if image_urls:
download_image(image_urls[0], f"image_{i}")
if __name__ == "__main__":
main()