A few days ago, I wrote about using two different artificial-intelligence systems in the same family-tree project.
Grok had produced scripts that could identify the historical registers returned by a search on Italy’s Portale Antenati and then download the images from those books. ChatGPT was better at the next stage: transcribing difficult handwriting, translating the records, comparing names and relationships, and organising the evidence for entry into MacFamilyTree.
The first result felt almost magical. I described what I wanted, received a script and watched it work on the first attempt. Something that might have taken me several days to investigate and write was running within minutes.
Then reality arrived in the form of repeated HTTP 403 responses.
The first solution met the WAF
An HTTP 403 response means that the server understood the request but refused to fulfil it. In this case, the refusals were associated with the protection surrounding the Antenati website and its image infrastructure.
The earlier downloader opened the register, found its IIIF manifest and requested the individual images programmatically. Technically, that was a sensible design. The IIIF manifest describes the pages in their correct order and supplies the information needed to locate each image.
Operationally, however, downloading page after page is also exactly the kind of behaviour that a Web Application Firewall—or WAF—may treat with suspicion. A normal person pauses, reads, moves between pages and retains a browser session. A script can make many highly regular requests with identical timing and none of the surrounding behaviour of an ordinary browser.
A script can be logically correct and still fail because the system around the data evaluates how the requests are being made, not merely whether the requested URLs exist.
The failures did not necessarily happen at the first request. That made the problem more frustrating. A download could start correctly and then encounter a series of 403 responses. Retrying too aggressively risked making the pattern appear even less human and could extend the interruption rather than solve it.
I returned to Grok and asked it to correct the problem. It produced several versions of the downloader, but the 403 errors continued. The tool that had been remarkably effective at creating the original code was not able to turn it into a reliable long-running solution under the website’s protective controls.
That was an interesting reversal of the conclusion in my previous article.
Giving the problem to ChatGPT
I then gave the problem to ChatGPT.
The result was downloadAntenati.sh. It is not simply the old script with a longer sleep command. It changes the architecture of the download process.
Instead of behaving as a detached HTTP client, the new script creates a dedicated, visible Chrome session on Windows. That browser has its own persistent profile and remains separate from my ordinary browsing and from other automated browser tasks. The Bash script runs under WSL, while Playwright connects to the Windows Chrome instance through the Chrome DevTools Protocol.
Chrome itself opens the Antenati register before Playwright attaches to it. This matters because the initial navigation and any protective checks occur within a real browser session. If Antenati presents a verification step, the browser remains visible and I can complete it manually. The script waits; it does not attempt to automate or bypass a CAPTCHA.
Once the register page is ready, the helper finds the IIIF manifest and retrieves both the manifest and the missing images through that same browser context. It uses the Antenati site root as the HTTP referrer and preserves the session state established by Chrome.
Antenati register URL
│
▼
Visible Windows Chrome
│
▼
Persistent browser/WAF session
│
▼
Playwright attaches through CDP
│
▼
IIIF manifest and ordered canvases
│
▼
Sequential, delayed image downloads
Deliberately slower—and much better
The most important practical change is that the new downloader is intentionally patient.
Before downloading an image, it waits five seconds. After saving that image and before moving to the next one, it waits another five seconds. Registers supplied in a URL file are processed sequentially, with another pause between books. If a request still fails with a response such as 403 or 429, the helper retries only a limited number of times and uses an increasing delay.
This is slower than firing requests as quickly as the connection permits. That is precisely the point.
The objective is not to extract the maximum number of bytes per second. It is to retrieve public archival material reliably without producing an unnecessarily aggressive request pattern. A historical register has already survived for a century or two; waiting a few additional seconds for each page is hardly the limiting factor.
Good automation is not always the fastest possible automation. It is automation that respects the behaviour and constraints of the system on which it depends.
The script also tries hard not to make unnecessary requests. Existing page files are skipped. A local semicolon-delimited log records the canonical register URL, archive, locality, year and expected image count. Before starting Chrome or contacting the server, the script can compare that log with the local files. If every expected page is already present, it exits without making a single network request.
If a download was interrupted, it resumes with the missing files. Images are written to temporary files and then moved atomically into place, reducing the possibility that an incomplete response will masquerade as a valid page. A process lock prevents two copies of the downloader from running simultaneously and doubling the traffic accidentally.
In batch mode, the script processes the input strictly in sequence and stops at the first failure. That may sound conservative, but it prevents one WAF problem from cascading through hundreds of remaining register URLs.
The larger lesson about AI-generated code
The experience also makes the comparison between Grok and ChatGPT more interesting than the simple conclusion I reached after the first script.
Grok was extremely effective at producing the initial downloader. It understood the Antenati structure quickly and created working code with almost no effort from me. When the problem changed from obtaining the images to operating reliably in the presence of WAF protection, however, repeated revisions did not solve it.
ChatGPT approached the second problem more successfully. The resulting script is larger and more complicated, but that complexity corresponds to real operational concerns: Windows and WSL integration, a persistent browser profile, Playwright installation, port forwarding, session reuse, manual verification, controlled retries, atomic writes, resumability, URL logging, local completion checks and safe batch processing.
This does not prove that one model is universally better at programming. It demonstrates something more useful: model performance depends on the shape and stage of the problem. The AI that produces the best first implementation may not be the one that performs the best operational diagnosis. A different agent, given the failure evidence and the freedom to rethink the architecture, may find a more durable answer.
AI-assisted development still requires engineering judgement: observe the failure, reconsider the assumptions, change tools when useful and validate the result under real operating conditions.
The first script was not a wasted effort. It proved that the archive could be traversed through its IIIF structure and established the naming and organisation I needed. Its failures supplied the evidence required to design the next version properly.
That is ordinary software engineering, even when AI writes most of the code. A prototype reveals the environment. The environment breaks some assumptions. The implementation evolves.
The difference is the speed. I could describe the failure, compare alternative approaches and receive a substantial new implementation without spending days building the browser integration myself. I still had to run it, observe it and decide whether its behaviour was acceptable. AI shortened the construction work; it did not remove responsibility for the result.
The replacement script
Below is the complete downloadAntenati.sh script produced with ChatGPT. It is shown as used in my WSL and Windows environment. The script contains its own usage information, version history and maintenance guidance.
#!/bin/bash
# BEGIN SCRIPT MAINTENANCE HEADER
#---------------------------------------------------------------------------------------------------
# ATTENTION CHATGPT / CODEX:
# Read the maintenance guidance at the bottom of this script before changing it.
# Preserve unrelated user changes and append the next sequential VERSION LOG entry
# with the actual change date for every revision.
#
# VERSION LOG
# 001 | 2026-09-20 | Initial Antenati downloader. Added an isolated Windows Chrome/
# | | Playwright session, AWS-WAF-aware gallery loading, IIIF manifest
# | | discovery, sequential full-resolution JPEG downloads, atomic
# | | file replacement, and resumable page_0001.jpg naming.
# 002 | 2026-09-20 | Use the Antenati site root as the HTTP referer for IIIF manifest
# | | and image requests, matching the live Cloudflare policy.
# 003 | 2026-09-20 | Store registers under the script's media directory using the
# | | existing numeric code-year folder convention.
# 004 | 2026-09-20 | Retrieve manifests and missing images through the dedicated
# | | Chrome session. Added mandatory five-second pauses before
# | | each download and before advancing to the next image; any
# | | existing destination image bypasses all download activity.
# 005 | 2026-09-20 | Add atomic antenati_url.log upserts keyed by canonical register
# | | URL, including archive, locality, register year, and image count.
# 006 | 2026-09-20 | Add an offline completed-register preflight. A logged URL whose
# | | complete page sequence is already local now exits before Chrome,
# | | Playwright, the manifest, or any server request is used.
# 007 | 2026-09-20 | Add sequential URL-file batch mode. Only lines beginning with
# | | http are considered; completed registers stay offline, a safety
# | | pause separates new registers, and the batch stops on failure.
# 008 | 2026-09-20 | Isolate URL-file reading on a dedicated descriptor and give child
# | | downloads /dev/null as standard input, preventing browser/Windows
# | | subprocesses from consuming the remaining batch entries.
#---------------------------------------------------------------------------------------------------
# END SCRIPT MAINTENANCE HEADER
set -Eeuo pipefail
declare -r VENV_ANTENATI="${HOME}/.venv/antenati"
declare -r FETCHER_ANTENATI="${HOME}/.local/bin/antenati-fetch.py"
#---------------------------------------------------------------------------------------------------
# Dedicated Chrome instance used only by this Antenati downloader.
# These ports and this profile are intentionally different from the torrent script.
#
# Windows Chrome CDP: 127.0.0.1:9234
# WSL port proxy: 0.0.0.0:9235 -> 127.0.0.1:9234
#---------------------------------------------------------------------------------------------------
declare -r CHROME_EXE_ANTENATI='C:\Program Files\Google\Chrome\Application\chrome.exe'
declare -r CHROME_PROFILE_ANTENATI='C:\Temp\antenati-chrome-profile'
declare -r CHROME_DEBUG_PORT_ANTENATI=9234
declare -r CHROME_PROXY_PORT_ANTENATI=9235
declare -r LOCK_FILE_ANTENATI='/tmp/antenati-download.lock'
declare -r BETWEEN_REGISTER_DELAY_ANTENATI=5
#---------------------------------------------------------------------------------------------------
# Write a timestamped informational message.
#---------------------------------------------------------------------------------------------------
function Log() {
printf '%s|%s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$*"
}
#---------------------------------------------------------------------------------------------------
# Write a timestamped warning message.
#---------------------------------------------------------------------------------------------------
function Warn() {
printf '%s| [WARNING] %s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$*" >&2
}
#---------------------------------------------------------------------------------------------------
# Write a timestamped caller-aware error message.
#---------------------------------------------------------------------------------------------------
function Error() {
printf '%s|❌ [ERROR:%s:%s] %s\n' \
"$(date '+%Y-%m-%d %H:%M:%S')" \
"${FUNCNAME[1]:-MAIN}" \
"${BASH_LINENO[0]:-0}" \
"$*" >&2
}
#---------------------------------------------------------------------------------------------------
# Show command usage.
#
# Arguments:
# $1 = Antenati register URL, or a readable local file containing URLs.
# $2 = optional output root; defaults to the media directory beside this script.
#
# Output directory:
# <output-root>/<numeric-code>-<year>/page_0001.jpg
#
# Example:
# ./downloadAntenati.sh \
# 'https://antenati.cultura.gov.it/ark:/12657/an_ua37943211'
#---------------------------------------------------------------------------------------------------
function Usage() {
cat <<'USAGE_EOF'
Usage:
downloadAntenati.sh ANTENATI_URL_OR_FILE [OUTPUT_ROOT]
Single-URL example:
./downloadAntenati.sh \
'https://antenati.cultura.gov.it/ark:/12657/an_ua37943211'
URL-file example:
./downloadAntenati.sh urls.txt
In URL-file mode, only lines whose first characters are http are processed. Blank
lines, comments, and every other line are ignored. URLs are processed sequentially.
The result for Registro: 1624 is stored as:
<SCRIPT_DIRECTORY>/media/37943211-1624/page_0001.jpg
If the script itself is located inside media, that directory is used directly;
a second nested media directory is not created. OUTPUT_ROOT can still be supplied
explicitly as the second argument.
Existing page JPEG files are skipped, so an interrupted download can be resumed
by running the same command again.
To reduce the risk of triggering the server WAF, missing images are retrieved by
the dedicated Chrome session with a five-second pause before each download and a
second five-second pause before advancing to the next image.
Each processed URL creates or updates this semicolon-delimited log beside the script:
antenati_url.log
Log fields:
URL;Conservato da;Comune/Località;registro;recorded images
If the log already contains the URL and every recorded page exists locally, the
script exits successfully without opening Chrome or contacting Antenati.
USAGE_EOF
}
#---------------------------------------------------------------------------------------------------
# Validate an Antenati register URL before it is passed to the browser helper.
#
# Accepts:
# https://antenati.cultura.gov.it/ark:/12657/an_ua37943211
# https://antenati.cultura.gov.it/ark:/12657/an_ua37943211/apXJZBk
#---------------------------------------------------------------------------------------------------
function ValidateUrl_Antenati() {
local url="${1:-}"
[[ "${url}" =~ ^https://antenati\.cultura\.gov\.it/ark:/12657/an_[A-Za-z0-9_]+(/[A-Za-z0-9._~-]+)?/?$ ]]
}
#---------------------------------------------------------------------------------------------------
# Decide whether a register is already complete using only the local URL log and image files.
#
# Arguments:
# $1 = Antenati register URL, optionally including the random viewer suffix.
# $2 = antenati_url.log path.
# $3 = output root containing <numeric-code>-<year> directories.
#
# Returns:
# 0 = the canonical URL is logged and every page_####.jpg through the recorded count exists.
# 1 = the URL is absent, the row is unusable, or at least one page is missing.
#
# No browser, network, or external helper is used by this function.
#---------------------------------------------------------------------------------------------------
function RegisterAlreadyComplete_Antenati() {
local url="${1:-}"
local logFile="${2:-}"
local outputRoot="${3:-}"
local canonicalUrl
local archiveCode
local folderCode
local loggedUrl
local archiveName
local locality
local registerYear
local imageCount
local extraField
local outputDirectory
local pageName
local pageNumber
[[ -f "${logFile}" ]] || return 1
if [[ "${url}" =~ ^(https://antenati\.cultura\.gov\.it/ark:/12657/(an_[A-Za-z0-9_]+))(/[A-Za-z0-9._~-]+)?/?$ ]]; then
canonicalUrl="${BASH_REMATCH[1]}"
archiveCode="${BASH_REMATCH[2]}"
else
return 1
fi
folderCode="${archiveCode#an_ua}"
[[ -n "${folderCode}" ]] || return 1
while IFS=';' read -r \
loggedUrl \
archiveName \
locality \
registerYear \
imageCount \
extraField; do
[[ "${loggedUrl}" == "${canonicalUrl}" ]] || continue
if [[ -n "${extraField}" || ! "${registerYear}" =~ ^[12][0-9]{3}$ || ! "${imageCount}" =~ ^[1-9][0-9]*$ ]]; then
Warn "Ignoring malformed URL-log row for: '${canonicalUrl}'"
return 1
fi
outputDirectory="${outputRoot}/${folderCode}-${registerYear}"
[[ -d "${outputDirectory}" ]] || return 1
for (( pageNumber=1; pageNumber<=imageCount; pageNumber++ )); do
printf -v pageName 'page_%04d.jpg' "${pageNumber}"
[[ -f "${outputDirectory}/${pageName}" ]] || return 1
done
Log "Register is already complete locally; no server request is required: '${canonicalUrl}'"
Log " ${outputDirectory} (${imageCount} images)"
return 0
done <"${logFile}"
return 1
}
#---------------------------------------------------------------------------------------------------
# Ensure that the Windows port proxy required by WSL/Playwright exists.
#
# Required mapping:
# Windows/WSL :9235 -> Windows localhost :9234
#
# A conflicting mapping is never overwritten automatically. If the mapping does
# not exist, Windows PowerShell requests the normal Administrator/UAC approval.
#---------------------------------------------------------------------------------------------------
function EnsurePortProxy_Antenati() {
local rc
if powershell.exe -NoProfile -Command "
\$entry = netsh interface portproxy show v4tov4 |
Select-String -Pattern '0\.0\.0\.0\s+${CHROME_PROXY_PORT_ANTENATI}\s+127\.0\.0\.1\s+${CHROME_DEBUG_PORT_ANTENATI}'
if (\$entry) { exit 0 }
exit 1
" >/dev/null 2>&1; then
return 0
fi
if powershell.exe -NoProfile -Command "
\$entry = netsh interface portproxy show v4tov4 |
Select-String -Pattern '\s${CHROME_PROXY_PORT_ANTENATI}\s'
if (\$entry) { exit 0 }
exit 1
" >/dev/null 2>&1; then
Error "Port ${CHROME_PROXY_PORT_ANTENATI} already has a different Windows portproxy mapping."
return 1
fi
Warn "Windows portproxy ${CHROME_PROXY_PORT_ANTENATI} -> ${CHROME_DEBUG_PORT_ANTENATI} is not configured. Requesting Administrator approval..."
powershell.exe -NoProfile -Command "
\$process = Start-Process \
-FilePath 'powershell.exe' \
-Verb RunAs \
-Wait \
-PassThru \
-ArgumentList @(
'-NoProfile',
'-Command',
'netsh interface portproxy add v4tov4 listenaddress=0.0.0.0 listenport=${CHROME_PROXY_PORT_ANTENATI} connectaddress=127.0.0.1 connectport=${CHROME_DEBUG_PORT_ANTENATI}'
)
exit \$process.ExitCode
"
rc=${?}
if [[ ${rc} -ne 0 ]]; then
Error "Could not create Windows portproxy ${CHROME_PROXY_PORT_ANTENATI} -> ${CHROME_DEBUG_PORT_ANTENATI}."
return "${rc}"
fi
if ! powershell.exe -NoProfile -Command "
\$entry = netsh interface portproxy show v4tov4 |
Select-String -Pattern '0\.0\.0\.0\s+${CHROME_PROXY_PORT_ANTENATI}\s+127\.0\.0\.1\s+${CHROME_DEBUG_PORT_ANTENATI}'
if (\$entry) { exit 0 }
exit 1
" >/dev/null 2>&1; then
Error "The required Windows portproxy mapping was not found after creation."
return 1
fi
Log "Windows portproxy ${CHROME_PROXY_PORT_ANTENATI} -> ${CHROME_DEBUG_PORT_ANTENATI} configured."
}
#---------------------------------------------------------------------------------------------------
# Wait until the dedicated Windows Chrome debugging endpoint is ready.
#---------------------------------------------------------------------------------------------------
function WaitForChrome_Antenati() {
local -i counter=0
local -r -i maxWait=30
while (( counter < maxWait )); do
if powershell.exe -NoProfile -Command "
try {
Invoke-WebRequest \
-UseBasicParsing \
-TimeoutSec 2 \
'http://127.0.0.1:${CHROME_DEBUG_PORT_ANTENATI}/json/version' |
Out-Null
exit 0
}
catch {
exit 1
}
" >/dev/null 2>&1; then
return 0
fi
counter=$(( counter + 1 ))
sleep 1
done
Error "Antenati Chrome did not become ready after ${maxWait} seconds."
return 1
}
#---------------------------------------------------------------------------------------------------
# Start the dedicated visible Windows Chrome instance used by Antenati.
#
# The browser remains open after a download so its WAF/session state can be reused
# by later runs. Chrome opens the register before Playwright attaches.
#---------------------------------------------------------------------------------------------------
function StartChrome_Antenati() {
local url="${1:-https://antenati.cultura.gov.it/}"
powershell.exe -NoProfile -Command "
Start-Process \
-FilePath '${CHROME_EXE_ANTENATI}' \
-ArgumentList @(
'--remote-debugging-address=127.0.0.1',
'--remote-debugging-port=${CHROME_DEBUG_PORT_ANTENATI}',
'--user-data-dir=${CHROME_PROFILE_ANTENATI}',
'--no-first-run',
'--no-default-browser-check',
'${url}'
)
"
local rc=${?}
if [[ ${rc} -ne 0 ]]; then
Error "Could not start the dedicated Antenati Chrome instance."
return "${rc}"
fi
WaitForChrome_Antenati
}
#---------------------------------------------------------------------------------------------------
# Ask the already-running dedicated Chrome process to open a register URL.
#
# Opening the URL before Playwright attaches avoids an Antenati WAF rejection
# seen when an automated page itself initiates the navigation.
#---------------------------------------------------------------------------------------------------
function OpenUrlInChrome_Antenati() {
local url="${1:-}"
powershell.exe -NoProfile -Command "
Start-Process \
-FilePath '${CHROME_EXE_ANTENATI}' \
-ArgumentList @(
'--remote-debugging-address=127.0.0.1',
'--remote-debugging-port=${CHROME_DEBUG_PORT_ANTENATI}',
'--user-data-dir=${CHROME_PROFILE_ANTENATI}',
'${url}'
)
"
local rc=${?}
if [[ ${rc} -ne 0 ]]; then
Error "Could not open the Antenati register in the dedicated Chrome instance."
return "${rc}"
fi
}
#---------------------------------------------------------------------------------------------------
# Start the dedicated Antenati Chrome only when it is not already running.
#---------------------------------------------------------------------------------------------------
function StartChromeIfNecessary_Antenati() {
local url="${1:-https://antenati.cultura.gov.it/}"
local rc
if powershell.exe -NoProfile -Command "
\$process = Get-CimInstance Win32_Process |
Where-Object {
\$_.Name -eq 'chrome.exe' -and
\$_.CommandLine -like '*${CHROME_PROFILE_ANTENATI}*'
} |
Select-Object -First 1
if (\$null -ne \$process) { exit 0 }
exit 1
" >/dev/null 2>&1; then
rc=0
else
rc=${?}
fi
case ${rc} in
0)
WaitForChrome_Antenati
OpenUrlInChrome_Antenati "${url}"
;;
1)
Log "Starting dedicated Antenati Chrome..."
StartChrome_Antenati "${url}"
;;
*)
Error "Could not determine whether the dedicated Antenati Chrome instance is running."
return "${rc}"
;;
esac
}
#---------------------------------------------------------------------------------------------------
# Ensure that a dedicated Python virtual environment and Playwright package exist.
#
# This environment is deliberately independent from ~/.venv/1337.
#---------------------------------------------------------------------------------------------------
function InstallPlaywrightIfNecessary_Antenati() {
if ! command -v python3 >/dev/null 2>&1; then
Error "Python 3 is required but was not found."
return 1
fi
if ! python3 -m venv --help >/dev/null 2>&1; then
Error "Python venv support is required but was not found. Install python3-venv."
return 1
fi
if [[ ! -d "${VENV_ANTENATI}" ]]; then
Log "Creating Antenati Python virtual environment: ${VENV_ANTENATI}"
python3 -m venv "${VENV_ANTENATI}" || {
Error "Could not create the Antenati Python virtual environment."
return 1
}
fi
if ! "${VENV_ANTENATI}/bin/python3" -c 'import playwright' >/dev/null 2>&1; then
Log "Installing Playwright in the Antenati virtual environment..."
"${VENV_ANTENATI}/bin/python3" -m pip install --upgrade pip || {
Error "Could not upgrade pip in the Antenati virtual environment."
return 1
}
"${VENV_ANTENATI}/bin/python3" -m pip install playwright || {
Error "Could not install Playwright in the Antenati virtual environment."
return 1
}
fi
}
#---------------------------------------------------------------------------------------------------
# Create or update the Python helper embedded in this Bash script.
#
# The helper:
# - connects to the dedicated Windows Chrome over CDP after Chrome itself
# has opened the register URL;
# - waits for the user-visible page/WAF flow to finish;
# - extracts the IIIF manifest URL and register year;
# - downloads every manifest canvas sequentially at full resolution;
# - saves atomically as page_0001.jpg, page_0002.jpg, ...;
# - skips already-downloaded valid JPEGs for safe resume behavior.
#---------------------------------------------------------------------------------------------------
function CreateFetcherIfNecessary_Antenati() {
local dir
local bashScript
dir="$(dirname "${FETCHER_ANTENATI}")"
bashScript="$(readlink -f "${BASH_SOURCE[0]}")"
mkdir -p -- "${dir}" || {
Error "Could not create helper directory: '${dir}'"
return 1
}
if [[ ! -f "${FETCHER_ANTENATI}" || "${bashScript}" -nt "${FETCHER_ANTENATI}" ]]; then
Log "Creating/updating Antenati browser helper: ${FETCHER_ANTENATI}"
cat >"${FETCHER_ANTENATI}" <<'PYTHON_EOF'
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import html as html_module
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import time
from typing import Any
from urllib.parse import urlsplit, urlunsplit
from playwright.sync_api import sync_playwright
CDP_PROXY_PORT = 9235
PAGE_READY_WAIT_SECONDS = 600
REQUEST_TIMEOUT_MS = 180_000
MAX_DOWNLOAD_ATTEMPTS = 5
WAF_DELAY_SECONDS = 5
ARCHIVE_URL_PATTERN = re.compile(
r"^https://antenati\.cultura\.gov\.it/ark:/12657/"
r"(?P<code>an_[A-Za-z0-9_]+)(?:/[A-Za-z0-9._~-]+)?/?$"
)
MANIFEST_PATTERN = re.compile(
r"manifestId\s*[:=]\s*(['\"])(https?://[^'\"]+)\1",
re.IGNORECASE,
)
REGISTER_YEAR_PATTERN = re.compile(
r"\bRegistro\s*:\s*([12][0-9]{3})\b",
re.IGNORECASE,
)
def fail(message: str, exit_code: int = 1) -> None:
print(f"ERROR: {message}", file=sys.stderr, flush=True)
raise SystemExit(exit_code)
def get_windows_host_ip() -> str:
try:
result = subprocess.run(
["ip", "route"],
capture_output=True,
text=True,
timeout=10,
check=False,
)
except (OSError, subprocess.SubprocessError) as exc:
fail(f"Could not inspect the WSL route table: {exc}")
if result.returncode != 0:
fail("Could not determine the Windows host address from the WSL route table.")
for line in result.stdout.splitlines():
fields = line.split()
if fields and fields[0] == "default" and "via" in fields:
address = fields[fields.index("via") + 1]
if address:
return address
fail("The Windows host address is absent from the WSL route table.")
def validate_official_url(url: str, purpose: str) -> None:
parsed = urlsplit(url)
host = (parsed.hostname or "").lower()
if parsed.scheme != "https" or not (
host == "cultura.gov.it" or host.endswith(".cultura.gov.it")
):
fail(f"Refusing non-Antenati {purpose} URL: {url}")
def manifest_url_from_html(page_html: str) -> str | None:
match = MANIFEST_PATTERN.search(page_html)
if not match:
return None
return html_module.unescape(match.group(2))
def metadata_value(manifest: dict[str, Any], requested_label: str) -> str | None:
for entry in manifest.get("metadata", []):
if not isinstance(entry, dict):
continue
label = entry.get("label")
value = entry.get("value")
if isinstance(label, dict):
label = next(
(
item
for values in label.values()
if isinstance(values, list)
for item in values
if isinstance(item, str)
),
"",
)
if isinstance(value, dict):
value = next(
(
item
for values in value.values()
if isinstance(values, list)
for item in values
if isinstance(item, str)
),
"",
)
if str(label).strip().casefold() == requested_label.casefold():
return str(value)
return None
def determine_year(page_text: str, manifest: dict[str, Any]) -> str:
match = REGISTER_YEAR_PATTERN.search(page_text)
if match:
return match.group(1)
for label in ("Titolo", "Registro"):
value = metadata_value(manifest, label)
if not value:
continue
match = re.search(r"\b([12][0-9]{3})\b", value)
if match:
return match.group(1)
fail("Could not determine the four-digit year after 'Registro:' or from the IIIF metadata.")
def extract_labeled_value(page_text: str, label: str) -> str:
"""Extract one visible metadata value that follows a labeled page-text field."""
lines = [re.sub(r"\s+", " ", line).strip() for line in page_text.splitlines()]
normalized_label = label.casefold()
for index, line in enumerate(lines):
if not line:
continue
if line.casefold().startswith(normalized_label):
value = line[len(label):].strip()
if value:
return value
for following_line in lines[index + 1:]:
if following_line:
return following_line
return ""
return ""
def sanitize_log_field(value: str) -> str:
"""Keep one log value on one line without introducing delimiter characters."""
return re.sub(r"\s+", " ", value).strip().replace(";", ",")
def update_url_log(
log_file: Path,
canonical_url: str,
archive_name: str,
locality: str,
register_year: str,
image_count: int,
) -> None:
"""Atomically insert or replace the row identified by canonical_url."""
fields = [
canonical_url,
archive_name,
locality,
register_year,
str(image_count),
]
row = ";".join(sanitize_log_field(field) for field in fields)
key = sanitize_log_field(canonical_url)
retained_lines: list[str] = []
try:
if log_file.exists():
for existing_line in log_file.read_text(encoding="utf-8-sig").splitlines():
if existing_line.split(";", 1)[0].strip() != key:
retained_lines.append(existing_line)
log_file.parent.mkdir(parents=True, exist_ok=True)
temporary = log_file.with_name(f".{log_file.name}.{os.getpid()}.tmp")
try:
with temporary.open("w", encoding="utf-8", newline="\n") as log_handle:
for existing_line in retained_lines:
log_handle.write(existing_line + "\n")
log_handle.write(row + "\n")
log_handle.flush()
os.fsync(log_handle.fileno())
os.replace(temporary, log_file)
finally:
try:
temporary.unlink(missing_ok=True)
except OSError:
pass
except OSError as exc:
fail(f"Could not update URL log {log_file}: {exc}")
def manifest_canvases(manifest: dict[str, Any]) -> list[dict[str, Any]]:
try:
canvases = manifest["sequences"][0]["canvases"]
if isinstance(canvases, list) and canvases:
return canvases
except (KeyError, IndexError, TypeError):
pass
canvases = manifest.get("items")
if isinstance(canvases, list) and canvases:
return canvases
fail("The IIIF manifest contains no image canvases.")
def first_string_id(value: Any) -> str | None:
if isinstance(value, str):
return value
if isinstance(value, dict):
candidate = value.get("id") or value.get("@id")
if isinstance(candidate, str):
return candidate
if isinstance(value, list):
for item in value:
candidate = first_string_id(item)
if candidate:
return candidate
return None
def canvas_image_url(canvas: dict[str, Any]) -> str:
try:
resource = canvas["images"][0]["resource"]
candidate = first_string_id(resource)
if candidate:
return candidate
service = resource.get("service") if isinstance(resource, dict) else None
service_id = first_string_id(service)
if service_id:
return service_id.rstrip("/") + "/full/max/0/default.jpg"
except (KeyError, IndexError, TypeError):
pass
try:
body = canvas["items"][0]["items"][0]["body"]
candidate = first_string_id(body)
if candidate:
return candidate
service = body.get("service") if isinstance(body, dict) else None
service_id = first_string_id(service)
if service_id:
return service_id.rstrip("/") + "/full/max/0/default.jpg"
except (KeyError, IndexError, TypeError):
pass
fail("A IIIF canvas does not contain a usable image URL.")
def full_resolution_jpeg_url(url: str) -> str:
parsed = urlsplit(url)
parts = parsed.path.split("/")
if parts and parts[-1] == "info.json":
path = "/".join(parts[:-1]) + "/full/max/0/default.jpg"
return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))
if len(parts) >= 7 and parts[-2] in {"0", "!0"}:
parts[-3] = "max"
parts[-1] = "default.jpg"
return urlunsplit(
(parsed.scheme, parsed.netloc, "/".join(parts), parsed.query, parsed.fragment)
)
return url
def valid_jpeg(path: Path) -> bool:
try:
if path.stat().st_size < 4:
return False
with path.open("rb") as image_file:
return image_file.read(3) == b"\xff\xd8\xff"
except OSError:
return False
def waf_delay(message: str) -> None:
"""Wait before a WAF-sensitive browser action."""
print(f" {message}; waiting {WAF_DELAY_SECONDS} seconds...", flush=True)
time.sleep(WAF_DELAY_SECONDS)
def browser_request_bytes(resource_page, url: str, referer: str, purpose: str) -> bytes:
"""Retrieve one official resource through the attached, visible Chrome session."""
validate_official_url(url, purpose)
last_error = "unknown error"
for attempt in range(1, MAX_DOWNLOAD_ATTEMPTS + 1):
try:
response = resource_page.goto(
url,
referer=referer,
wait_until="load",
timeout=REQUEST_TIMEOUT_MS,
)
if response is None:
last_error = "Chrome returned no response"
else:
validate_official_url(resource_page.url, f"redirected {purpose}")
status = response.status
if 200 <= status < 300:
return response.body()
last_error = f"HTTP {status}"
if status not in {403, 408, 425, 429, 500, 502, 503, 504}:
break
except Exception as exc:
last_error = str(exc)
if resource_page.is_closed():
break
if attempt < MAX_DOWNLOAD_ATTEMPTS:
delay = min(WAF_DELAY_SECONDS * attempt, 30)
print(
f" {purpose} attempt {attempt}/{MAX_DOWNLOAD_ATTEMPTS} failed "
f"({last_error}); retrying in {delay}s...",
file=sys.stderr,
flush=True,
)
time.sleep(delay)
fail(f"Could not retrieve {purpose} from {url}: {last_error}")
def load_manifest(resource_page, manifest_url: str, referer: str) -> dict[str, Any]:
payload = browser_request_bytes(resource_page, manifest_url, referer, "IIIF manifest")
try:
manifest = json.loads(payload.decode("utf-8-sig"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
fail(f"The IIIF manifest is not valid UTF-8 JSON: {exc}")
if not isinstance(manifest, dict):
fail("The IIIF manifest root is not a JSON object.")
return manifest
def save_image(resource_page, url: str, referer: str, destination: Path) -> int:
payload = browser_request_bytes(resource_page, url, referer, "image")
if len(payload) < 4 or payload[:3] != b"\xff\xd8\xff":
fail(f"The response for {destination.name} is not a JPEG image.")
temporary = destination.with_name(f".{destination.name}.part")
try:
with temporary.open("wb") as image_file:
image_file.write(payload)
image_file.flush()
os.fsync(image_file.fileno())
os.replace(temporary, destination)
finally:
try:
temporary.unlink(missing_ok=True)
except OSError:
pass
return len(payload)
def wait_for_gallery(context, url: str, wait_seconds: int):
archive_path = urlsplit(url).path.rstrip("/")
message_shown = False
deadline = time.monotonic() + wait_seconds
while time.monotonic() < deadline:
matching_pages = []
for candidate in reversed(context.pages):
try:
candidate_path = urlsplit(candidate.url).path.rstrip("/")
if candidate_path == archive_path or candidate_path.startswith(archive_path + "/"):
matching_pages.append(candidate)
except Exception:
continue
for page in matching_pages:
if page.is_closed():
continue
try:
page_html = page.content()
manifest_url = manifest_url_from_html(page_html)
if manifest_url:
try:
page_text = page.locator("body").inner_text(timeout=5_000)
except Exception:
page_text = ""
try:
page.bring_to_front()
except Exception:
pass
return page, manifest_url, page_text
except Exception:
continue
if not message_shown:
print(file=sys.stderr)
print("======================================================================", file=sys.stderr)
print("Antenati has not exposed the register manifest yet.", file=sys.stderr)
print("Use the dedicated Windows Chrome window if verification is shown.", file=sys.stderr)
print("If the tab shows 403, wait briefly and refresh that tab.", file=sys.stderr)
print(f"The script will wait up to {wait_seconds} seconds.", file=sys.stderr)
print("======================================================================", file=sys.stderr)
print(file=sys.stderr)
message_shown = True
time.sleep(1)
fail(f"Timed out after {wait_seconds} seconds waiting for the Antenati register page.")
def parse_arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--url", required=True)
parser.add_argument("--output-root", required=True)
parser.add_argument("--log-file", required=True)
parser.add_argument("--wait-seconds", type=int, default=PAGE_READY_WAIT_SECONDS)
return parser.parse_args()
def main() -> int:
args = parse_arguments()
url_match = ARCHIVE_URL_PATTERN.fullmatch(args.url)
if not url_match:
fail(f"Invalid Antenati register URL: {args.url}", 2)
if args.wait_seconds < 1:
fail("--wait-seconds must be a positive integer.", 2)
archive_code = url_match.group("code")
canonical_url = f"https://antenati.cultura.gov.it/ark:/12657/{archive_code}"
folder_code = archive_code.removeprefix("an_ua")
if not folder_code:
fail(f"Could not derive a folder code from archive code: {archive_code}", 2)
output_root = Path(args.output_root).expanduser().resolve()
output_root.mkdir(parents=True, exist_ok=True)
log_file = Path(args.log_file).expanduser().resolve()
windows_host = get_windows_host_ip()
cdp_url = f"http://{windows_host}:{CDP_PROXY_PORT}"
playwright = sync_playwright().start()
page = None
resource_page = None
try:
try:
browser = playwright.chromium.connect_over_cdp(cdp_url, timeout=15_000)
except Exception as exc:
fail(f"Could not connect to the dedicated Windows Chrome at {cdp_url}: {exc}")
if not browser.contexts:
fail("Connected to Chrome, but no browser context exists.")
context = browser.contexts[0]
page, manifest_url, page_text = wait_for_gallery(
context,
args.url,
args.wait_seconds,
)
validate_official_url(manifest_url, "IIIF manifest")
http_referer = "https://antenati.cultura.gov.it/"
resource_page = context.new_page()
waf_delay("Preparing to load the IIIF manifest in Chrome")
manifest = load_manifest(resource_page, manifest_url, http_referer)
year = determine_year(page_text, manifest)
canvases = manifest_canvases(manifest)
archive_name = extract_labeled_value(page_text, "Conservato da:")
locality = extract_labeled_value(page_text, "Comune/Località:")
if not archive_name:
print("WARNING: Conservato da was not found on the register page.", file=sys.stderr)
if not locality:
print("WARNING: Comune/Località was not found on the register page.", file=sys.stderr)
output_directory = output_root / f"{folder_code}-{year}"
output_directory.mkdir(parents=True, exist_ok=True)
update_url_log(
log_file,
canonical_url,
archive_name,
locality,
year,
len(canvases),
)
print(f"Archive code: {archive_code}", flush=True)
print(f"Canonical URL: {canonical_url}", flush=True)
print(f"Conservato da: {archive_name}", flush=True)
print(f"Comune/Località: {locality}", flush=True)
print(f"Folder code: {folder_code}", flush=True)
print(f"Register year: {year}", flush=True)
print(f"Resolved page: {page.url}", flush=True)
print(f"IIIF manifest: {manifest_url}", flush=True)
print(f"Images: {len(canvases)}", flush=True)
print(f"Output directory: {output_directory}", flush=True)
print(f"URL log: {log_file}", flush=True)
downloaded = 0
skipped = 0
total_bytes = 0
for index, canvas in enumerate(canvases, start=1):
destination = output_directory / f"page_{index:04d}.jpg"
if destination.exists():
if not destination.is_file():
fail(f"The destination exists but is not a file: {destination}")
skipped += 1
print(
f"[{index:04d}/{len(canvases):04d}] already present: {destination.name}",
flush=True,
)
continue
image_url = full_resolution_jpeg_url(canvas_image_url(canvas))
waf_delay(f"Preparing to download {destination.name} in Chrome")
size = save_image(resource_page, image_url, http_referer, destination)
downloaded += 1
total_bytes += size
print(
f"[{index:04d}/{len(canvases):04d}] saved: {destination.name} "
f"({size / (1024 * 1024):.1f} MiB)",
flush=True,
)
if index < len(canvases):
waf_delay("Preparing to advance to the next image")
print(
f"Complete: {downloaded} downloaded, {skipped} already present, "
f"{total_bytes / (1024 * 1024):.1f} MiB transferred.",
flush=True,
)
return 0
finally:
if resource_page is not None:
try:
if not resource_page.is_closed():
resource_page.close()
except Exception:
pass
if page is not None:
try:
if not page.is_closed():
page.close()
except Exception:
pass
try:
playwright.stop()
except Exception:
pass
if __name__ == "__main__":
raise SystemExit(main())
PYTHON_EOF
chmod 700 -- "${FETCHER_ANTENATI}" || {
Error "Could not make the Antenati helper executable."
return 1
}
fi
"${VENV_ANTENATI}/bin/python3" -m py_compile "${FETCHER_ANTENATI}" || {
Error "Python syntax validation failed for: '${FETCHER_ANTENATI}'"
return 1
}
}
#---------------------------------------------------------------------------------------------------
# Initialize the isolated Antenati Python/browser support.
#---------------------------------------------------------------------------------------------------
function InitializeBrowserSupport_Antenati() {
InstallPlaywrightIfNecessary_Antenati || return 1
CreateFetcherIfNecessary_Antenati || return 1
}
#---------------------------------------------------------------------------------------------------
# Acquire the single-process Antenati lock.
#
# The descriptor remains open until the Bash process exits, at which point the
# operating system releases the lock automatically.
#---------------------------------------------------------------------------------------------------
function Lock_Antenati() {
exec {g_AntenatiLockFd}>"${LOCK_FILE_ANTENATI}" || {
Error "Could not open Antenati lock file: '${LOCK_FILE_ANTENATI}'"
return 1
}
if ! flock -w 10 "${g_AntenatiLockFd}"; then
Error "Another Antenati download still holds: '${LOCK_FILE_ANTENATI}'"
return 1
fi
printf '%s\n' "$$" >&"${g_AntenatiLockFd}"
}
#---------------------------------------------------------------------------------------------------
# Process a local file containing Antenati URLs, one line and one register at a time.
#
# Arguments:
# $1 = readable input file.
# $2 = absolute path to this script, used for isolated single-URL child runs.
# $3 = resolved output root.
# $4 = antenati_url.log path.
#
# Lines are considered only when their first characters are "http". Completed registers are
# checked locally and never start a child/browser request. New or incomplete registers run
# sequentially, followed by a five-second inter-register pause. The first invalid URL or failed
# register stops the batch so a WAF failure cannot cascade through the remaining input. The input
# file uses a private descriptor; child processes never inherit it as standard input.
#---------------------------------------------------------------------------------------------------
function ProcessUrlFile_Antenati() {
local inputFile="${1:-}"
local scriptPath="${2:-}"
local outputRoot="${3:-}"
local logFile="${4:-}"
local line
local url
local lineNumber=0
local candidateCount=0
local ignoredCount=0
local completedCount=0
local processedCount=0
local rc
local inputFd
exec {inputFd}<"${inputFile}" || {
Error "Could not open URL input file: '${inputFile}'"
return 1
}
while IFS= read -r -u "${inputFd}" line || [[ -n "${line}" ]]; do
(( lineNumber += 1 ))
line="${line%$'\r'}"
if [[ "${line}" != http* ]]; then
(( ignoredCount += 1 ))
continue
fi
url="${line}"
(( candidateCount += 1 ))
if ! ValidateUrl_Antenati "${url}"; then
Error "Invalid Antenati URL at '${inputFile}:${lineNumber}': '${url}'"
exec {inputFd}<&-
return 2
fi
Log "Batch URL ${candidateCount} from line ${lineNumber}: '${url}'"
if RegisterAlreadyComplete_Antenati "${url}" "${logFile}" "${outputRoot}"; then
(( completedCount += 1 ))
continue
fi
"${scriptPath}" "${url}" "${outputRoot}" </dev/null || {
rc=$?
Error "Batch stopped after URL ${candidateCount} at line ${lineNumber}: '${url}'"
exec {inputFd}<&-
return "${rc}"
}
(( processedCount += 1 ))
Log "Waiting ${BETWEEN_REGISTER_DELAY_ANTENATI} seconds before considering the next URL."
sleep "${BETWEEN_REGISTER_DELAY_ANTENATI}"
done
exec {inputFd}<&-
if (( candidateCount == 0 )); then
Error "The input file contains no lines beginning with http: '${inputFile}'"
return 2
fi
Log "URL-file processing complete: ${candidateCount} URL(s), ${processedCount} processed, "\
"${completedCount} already complete, ${ignoredCount} ignored line(s)."
}
#---------------------------------------------------------------------------------------------------
# Main program.
#---------------------------------------------------------------------------------------------------
function Main() {
if [[ "${1:-}" == '-h' || "${1:-}" == '--help' ]]; then
Usage
return 0
fi
if (( $# < 1 || $# > 2 )); then
Usage >&2
return 2
fi
local input="$1"
local url
local inputFile
local scriptPath
local scriptDirectory
local defaultOutputRoot
local outputRoot
local logFile
scriptPath="$(readlink -f -- "${BASH_SOURCE[0]}")" || {
Error "Could not resolve the script path."
return 1
}
scriptDirectory="$(dirname -- "${scriptPath}")"
if [[ "${scriptDirectory##*/}" == 'media' ]]; then
defaultOutputRoot="${scriptDirectory}"
else
defaultOutputRoot="${scriptDirectory}/media"
fi
outputRoot="${2:-${defaultOutputRoot}}"
logFile="${scriptDirectory}/antenati_url.log"
mkdir -p -- "${outputRoot}" || {
Error "Could not create output root: '${outputRoot}'"
return 1
}
outputRoot="$(readlink -f -- "${outputRoot}")"
if [[ -e "${input}" ]]; then
if [[ ! -f "${input}" || ! -r "${input}" ]]; then
Error "The input path is not a readable regular file: '${input}'"
return 2
fi
inputFile="$(readlink -f -- "${input}")" || {
Error "Could not resolve the input file: '${input}'"
return 1
}
Log "Processing Antenati URL file: '${inputFile}'"
ProcessUrlFile_Antenati "${inputFile}" "${scriptPath}" "${outputRoot}" "${logFile}"
return $?
fi
url="${input}"
if ! ValidateUrl_Antenati "${url}"; then
Error "Invalid Antenati register URL or unreadable input file: '${input}'"
return 2
fi
if RegisterAlreadyComplete_Antenati "${url}" "${logFile}" "${outputRoot}"; then
return 0
fi
Lock_Antenati || return 1
InitializeBrowserSupport_Antenati || return 1
EnsurePortProxy_Antenati || return 1
StartChromeIfNecessary_Antenati "${url}" || return 1
Log "Downloading Antenati register: '${url}'"
Log "Output root: '${outputRoot}'"
Log "URL log: '${logFile}'"
"${VENV_ANTENATI}/bin/python3" -u "${FETCHER_ANTENATI}" \
--url "${url}" \
--output-root "${outputRoot}" \
--log-file "${logFile}"
}
if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
Main "$@"
fi
#---------------------------------------------------------------------------------------------------
# CHATGPT / CODEX CODE-GENERATION GUIDANCE
#
# - Always inspect this current on-disk file before changing it; the user also edits it.
# - Append the next sequential VERSION LOG entry with the actual date for every revision.
# - Keep the Antenati Chrome profile, ports, virtual environment, helper, and lock independent
# from download1337.sh and all torrent browser sessions.
# - Every function must retain a meaningful comment block describing purpose, inputs, outputs,
# side effects, and important return behavior.
# - Do not automate CAPTCHA completion. Keep Chrome visible and let the user complete any
# verification presented by Antenati while the helper waits.
# - Preserve manifest order and page_0001.jpg numbering. Keep downloads resumable and atomic.
# - Keep the default output naming as media/<numeric-code>-<year> so existing downloads resume.
# - Keep all manifest/image transfers inside the dedicated Chrome session. Never remove the
# five-second pre-download and pre-next-image WAF delays without an explicit user request.
# - Keep antenati_url.log beside the script, semicolon-delimited and atomically upserted by
# canonical register URL with no random viewer-page suffix.
# - Preserve the offline completed-register preflight before every browser/helper/network action.
# - In URL-file mode, process only lines beginning with http, stay strictly sequential, retain the
# inter-register delay, and stop the batch on the first processing failure to protect the WAF.
# - Keep URL-file reads on their dedicated descriptor and child standard input on /dev/null; never
# let browser, Python, PowerShell, or other child processes consume the batch file descriptor.
# - Do not weaken the official cultura.gov.it URL validation without an explicit user request.
# - After edits, validate both Bash syntax and the embedded Python source.
#---------------------------------------------------------------------------------------------------
AI Assistance Disclosure: This article is based on my original experience, observations, decisions and conclusions.
Artificial intelligence tools were used directly in the work described. Grok generated the original Antenati downloading scripts, while ChatGPT designed the replacement browser-based downloader and assisted in organising, expanding and refining this follow-up article. AI was also used to clarify the technical distinctions between direct HTTP retrieval, browser-session automation, IIIF processing and WAF-aware request handling.
The operational results described here are based on my use of the scripts in my own environment. The final selection, testing, interpretation, opinions and conclusions remain my own.
The operational results described here are based on my use of the scripts in my own environment. The final selection, testing, interpretation, opinions and conclusions remain my own.
Comments
Post a Comment