initial commit
This commit is contained in:
Bartal Laearsson
@@ -0,0 +1,298 @@
|
||||
# /// script
|
||||
# requires-python = ">=3.12"
|
||||
# dependencies = [
|
||||
# "requests",
|
||||
# "beautifulsoup4",
|
||||
# ]
|
||||
# ///
|
||||
"""
|
||||
Generic scraper. Saves all files flat in ./scraped_<domain>/ with sanitized filenames.
|
||||
Writes structured error log to ./scraped_<domain>/scrape_errors.jsonl.
|
||||
|
||||
Usage:
|
||||
uv run scraper.py https://www.logting.fo
|
||||
uv run scraper.py https://taks.fo
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
from collections import deque
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from urllib.parse import urljoin, urlparse
|
||||
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
DELAY = 0.1
|
||||
TIMEOUT = 60
|
||||
MAX_RETRIES = 3
|
||||
RETRY_BACKOFF = 5.0
|
||||
USER_AGENT = "web-scraper/1.0 (research)"
|
||||
|
||||
CONTENT_TYPE_EXT = {
|
||||
"text/html": ".html",
|
||||
"application/pdf": ".pdf",
|
||||
"application/json": ".json",
|
||||
"application/xml": ".xml",
|
||||
"text/xml": ".xml",
|
||||
"text/plain": ".txt",
|
||||
"text/css": ".css",
|
||||
"application/javascript": ".js",
|
||||
"text/javascript": ".js",
|
||||
"image/png": ".png",
|
||||
"image/jpeg": ".jpg",
|
||||
"image/gif": ".gif",
|
||||
"image/svg+xml": ".svg",
|
||||
"image/webp": ".webp",
|
||||
"application/vnd.ms-excel": ".xls",
|
||||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
|
||||
"application/msword": ".doc",
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
|
||||
"application/zip": ".zip",
|
||||
"application/octet-stream": ".bin",
|
||||
}
|
||||
|
||||
SEEN: set[str] = set()
|
||||
|
||||
|
||||
def log_error(
|
||||
log_path: Path,
|
||||
url: str,
|
||||
error_type: str,
|
||||
message: str,
|
||||
status_code: int | None = None,
|
||||
) -> None:
|
||||
entry = {
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"url": url,
|
||||
"error_type": error_type,
|
||||
"message": message,
|
||||
"status_code": status_code,
|
||||
}
|
||||
with open(log_path, "a", encoding="utf-8") as f:
|
||||
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
||||
print(f" LOGGED ERROR [{error_type}]: {message}")
|
||||
|
||||
|
||||
def derive_base_domain(url: str) -> str:
|
||||
parsed = urlparse(url)
|
||||
hostname = parsed.hostname
|
||||
if not hostname:
|
||||
print(f"Could not parse hostname from {url}")
|
||||
sys.exit(1)
|
||||
parts = hostname.split(".")
|
||||
if len(parts) >= 2:
|
||||
return ".".join(parts[-2:])
|
||||
return hostname
|
||||
|
||||
|
||||
def is_internal(url: str, base_domain: str) -> bool:
|
||||
parsed = urlparse(url)
|
||||
hostname = parsed.hostname
|
||||
if not hostname:
|
||||
return False
|
||||
return hostname == base_domain or hostname.endswith(f".{base_domain}")
|
||||
|
||||
|
||||
def normalize_url(url: str) -> str:
|
||||
url = url.split("#")[0].rstrip("?&")
|
||||
parsed = urlparse(url)
|
||||
scheme = parsed.scheme.lower()
|
||||
netloc = parsed.netloc.lower()
|
||||
path = parsed.path
|
||||
query = parsed.query
|
||||
if query:
|
||||
return f"{scheme}://{netloc}{path}?{query}"
|
||||
return f"{scheme}://{netloc}{path}"
|
||||
|
||||
|
||||
def get_extension(url: str, content_type: str) -> str:
|
||||
path_suffix = Path(urlparse(url).path).suffix.lower()
|
||||
if path_suffix:
|
||||
return path_suffix
|
||||
|
||||
ct = content_type.lower().split(";")[0].strip()
|
||||
if ct in CONTENT_TYPE_EXT:
|
||||
return CONTENT_TYPE_EXT[ct]
|
||||
|
||||
return ".bin"
|
||||
|
||||
|
||||
def url_to_filename(url: str, content_type: str = "") -> str:
|
||||
parsed = urlparse(url)
|
||||
|
||||
if len(url) > 200:
|
||||
hash_part = hashlib.sha256(url.encode()).hexdigest()[:16]
|
||||
ext = get_extension(url, content_type)
|
||||
return f"{hash_part}{ext}"
|
||||
|
||||
host_prefix = parsed.netloc.replace(".", "_").replace(":", "_")
|
||||
path = parsed.path.lstrip("/") or "index"
|
||||
|
||||
# Split extension from path so query hash goes before it
|
||||
path_obj = Path(path)
|
||||
stem = path_obj.stem
|
||||
ext = path_obj.suffix.lower()
|
||||
if not ext:
|
||||
ext = get_extension(url, content_type)
|
||||
|
||||
query_suffix = ""
|
||||
if parsed.query:
|
||||
query_hash = hashlib.md5(parsed.query.encode()).hexdigest()[:8]
|
||||
query_suffix = f"_{query_hash}"
|
||||
|
||||
filename = f"{host_prefix}_{stem}{query_suffix}{ext}"
|
||||
filename = "".join(c if c.isalnum() or c in "._-" else "_" for c in filename)
|
||||
|
||||
return filename
|
||||
|
||||
|
||||
def save(output_dir: Path, url: str, content: bytes, content_type: str = "") -> None:
|
||||
filename = url_to_filename(url, content_type)
|
||||
filepath = output_dir / filename
|
||||
filepath.write_bytes(content)
|
||||
print(f" saved -> {filepath.name}")
|
||||
|
||||
|
||||
def extract_links(base_url: str, html: str, base_domain: str) -> set[str]:
|
||||
soup = BeautifulSoup(html, "html.parser")
|
||||
links: set[str] = set()
|
||||
|
||||
for tag in soup.find_all("a", href=True):
|
||||
href = tag["href"].strip()
|
||||
if not href or href.startswith(("mailto:", "tel:", "javascript:")):
|
||||
continue
|
||||
full_url = urljoin(base_url, href)
|
||||
full_url = normalize_url(full_url)
|
||||
if is_internal(full_url, base_domain):
|
||||
links.add(full_url)
|
||||
|
||||
return links
|
||||
|
||||
|
||||
def fetch(url: str, log_path: Path) -> requests.Response | None:
|
||||
for attempt in range(1, MAX_RETRIES + 1):
|
||||
try:
|
||||
return requests.get(
|
||||
url,
|
||||
timeout=TIMEOUT,
|
||||
headers={"User-Agent": USER_AGENT},
|
||||
allow_redirects=True,
|
||||
)
|
||||
except requests.Timeout:
|
||||
if attempt < MAX_RETRIES:
|
||||
wait = RETRY_BACKOFF * attempt
|
||||
print(f" timeout, retry {attempt}/{MAX_RETRIES} in {wait}s...")
|
||||
time.sleep(wait)
|
||||
else:
|
||||
log_error(
|
||||
log_path,
|
||||
url,
|
||||
"timeout",
|
||||
f"Timed out after {MAX_RETRIES} retries ({TIMEOUT}s each)",
|
||||
)
|
||||
return None
|
||||
except requests.ConnectionError as e:
|
||||
if attempt < MAX_RETRIES:
|
||||
wait = RETRY_BACKOFF * attempt
|
||||
print(
|
||||
f" connection error, retry {attempt}/{MAX_RETRIES} in {wait}s..."
|
||||
)
|
||||
time.sleep(wait)
|
||||
else:
|
||||
log_error(log_path, url, "connection_error", str(e))
|
||||
return None
|
||||
except requests.RequestException as e:
|
||||
log_error(log_path, url, "request_error", str(e))
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: uv run scraper.py <start-url>")
|
||||
print("Example: uv run scraper.py https://www.logting.fo")
|
||||
sys.exit(1)
|
||||
|
||||
start_url = sys.argv[1]
|
||||
base_domain = derive_base_domain(start_url)
|
||||
output_dir = Path(f"./scraped_{base_domain.replace('.', '_')}")
|
||||
log_dir = output_dir / "logs"
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
log_path = log_dir / "scrape_errors.jsonl"
|
||||
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
log_path.unlink(missing_ok=True)
|
||||
|
||||
queue: deque[str] = deque([start_url])
|
||||
count = 0
|
||||
error_count = 0
|
||||
|
||||
print(f"Scraping: {start_url}")
|
||||
print(f"Base domain: {base_domain}")
|
||||
print(f"Output dir: {output_dir}")
|
||||
print()
|
||||
|
||||
while queue:
|
||||
url = normalize_url(queue.popleft())
|
||||
|
||||
if url in SEEN:
|
||||
continue
|
||||
SEEN.add(url)
|
||||
|
||||
print(f"[{count}] fetching: {url}")
|
||||
resp = fetch(url, log_path)
|
||||
|
||||
if resp is None:
|
||||
error_count += 1
|
||||
time.sleep(DELAY)
|
||||
continue
|
||||
|
||||
if resp.status_code != 200:
|
||||
log_error(
|
||||
log_path,
|
||||
url,
|
||||
"http_error",
|
||||
f"HTTP {resp.status_code}",
|
||||
resp.status_code,
|
||||
)
|
||||
error_count += 1
|
||||
time.sleep(DELAY)
|
||||
continue
|
||||
|
||||
final_url = normalize_url(resp.url)
|
||||
content_type = resp.headers.get("Content-Type", "")
|
||||
|
||||
if "text/html" in content_type:
|
||||
try:
|
||||
html = resp.text
|
||||
save(output_dir, final_url, html.encode("utf-8"), content_type)
|
||||
links = extract_links(final_url, html, base_domain)
|
||||
new_links = [u for u in links if u not in SEEN]
|
||||
queue.extend(new_links)
|
||||
print(f" found {len(links)} links ({len(new_links)} new)")
|
||||
except Exception as e:
|
||||
log_error(log_path, final_url, "parse_error", str(e))
|
||||
error_count += 1
|
||||
else:
|
||||
try:
|
||||
save(output_dir, final_url, resp.content, content_type)
|
||||
print(f" binary: {content_type}")
|
||||
except Exception as e:
|
||||
log_error(log_path, final_url, "save_error", str(e))
|
||||
error_count += 1
|
||||
|
||||
count += 1
|
||||
time.sleep(DELAY)
|
||||
|
||||
print(
|
||||
f"\nDone. Fetched {count} URLs ({len(SEEN)} seen total). Errors: {error_count}."
|
||||
)
|
||||
print(f"Error log: {log_path}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user