#!/usr/bin/env python3
"""
Refresh `` product grids in PHP pages.
1. Query Google Images (google.es with `tbm=isch`) using `stem+...+site:amazon.es`.
2. Parse amazon.es DP links when embedded in HTML.
3. Fallback to Amazon.es SERP scraping (`https://www.amazon.es/s`) when ASINs are unavailable.
Preserves PHP placeholders `'.$tagafiliadoEs.'` and `'.$rutaimagen.'`.
Appends successfully written filenames to `refresh-products-logs.txt` (see --progress-log).
On startup, skips any `.php` name already recorded so the next alphabetical files continue.
Adds a random 200-500 ms pause after each Google/Amazon HTTP response.
CLI logging uses English.
"""
from __future__ import annotations
import argparse
import html
import logging
import random
import re
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlparse, urlunparse
import requests
from bs4 import BeautifulSoup
LOGGER = logging.getLogger("refresh_amazon_products")
SECTION_OPEN_RE = re.compile(
r'',
re.IGNORECASE,
)
ASIN_RE = re.compile(
r"/(?:dp|gp/product)/([A-Z0-9]{10})(?:[/?#]|$)",
re.IGNORECASE,
)
USER_AGENT = (
"Mozilla/5.0 (Windows NT 10.0; WOW64; rv:55.0) "
"Gecko/20100101 Firefox/55.0"
)
AMAZON_ACCEPT_HEADER = (
"text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
)
SKIP_FILENAMES = frozenset({"index.php", "404.php"})
PROGRESS_LOG_FILE_NAME = "refresh-products-logs.txt"
REQUEST_JITTER_MIN_SEC = 0.2
REQUEST_JITTER_MAX_SEC = 0.5
def inter_request_pause() -> None:
"""Random delay between HTTP calls to reduce throttling."""
delay = random.uniform(REQUEST_JITTER_MIN_SEC, REQUEST_JITTER_MAX_SEC)
time.sleep(delay)
def configure_http_session_headers(session: requests.Session) -> None:
"""Browser-like defaults so Amazon.es SERP returns full HTML where possible."""
session.headers.update(
{
"User-Agent": USER_AGENT,
"Accept": AMAZON_ACCEPT_HEADER,
"Accept-Language": "es-ES,es;q=0.9,en;q=0.5",
"Upgrade-Insecure-Requests": "1",
}
)
def warmup_amazon_homepage(
session: requests.Session,
timeout: float,
retries: int,
) -> None:
"""Prime cookies/session against amazon.es once before SERP scraping."""
last_error: str | None = None
for attempt in range(retries + 1):
try:
response = session.get("https://www.amazon.es/", timeout=timeout)
inter_request_pause()
if 200 <= response.status_code < 400:
LOGGER.info("Amazon.es homepage warmup succeeded (HTTP %s)", response.status_code)
return
last_error = f"HTTP {response.status_code}"
time.sleep(1.2 * (attempt + 1))
except requests.RequestException as exc:
last_error = str(exc)
inter_request_pause()
time.sleep(1.2 * (attempt + 1))
LOGGER.warning("Amazon.es homepage warmup failed: %s", last_error)
def amazon_serp_looks_blocked(html_blob: str) -> bool:
"""Heuristic for anti-bot / captcha bodies that are not normal SERPs."""
lower = html_blob.lower()
needles = (
"api-services-support@amazon",
"enter the characters you see below",
"robot check",
"sorry, we just need to make sure you're not a robot",
"captchatable",
"glimpseamazon",
)
return any(token in lower for token in needles)
def load_progress_log(progress_path: Path) -> set[str]:
"""Return basenames (.php) that were already persisted in a prior run."""
if not progress_path.is_file():
return set()
processed: set[str] = set()
try:
text = progress_path.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
LOGGER.warning("Cannot read progress log %s: %s", progress_path, exc)
return processed
for line in text.splitlines():
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
if stripped.endswith(".php"):
processed.add(stripped)
return processed
def append_progress_entry(progress_path: Path, basename: str) -> None:
"""Append one completed filename after a successful disk write."""
progress_path.parent.mkdir(parents=True, exist_ok=True)
with progress_path.open("a", encoding="utf-8", newline="\n") as handle:
handle.write(basename.strip() + "\n")
@dataclass(frozen=True)
class AmazonHit:
"""Matched Amazon.es product extracted from Google's SERP."""
product_url_without_affiliate: str
asin: str
title: str
image_url: str
def strip_google_redirect(href: str) -> str:
parsed = urlparse(href)
host = parsed.netloc.lower()
if ("google.es" == host or host.endswith(".google.es")) and parsed.path.startswith("/url"):
query = parse_qs(parsed.query)
if query.get("q"):
return unquote(query["q"][0])
return href
def canonical_dp_url(candidate: str) -> tuple[str | None, str | None]:
raw = strip_google_redirect(candidate.strip().replace("&", "&"))
if raw.startswith("//"):
raw = "https:" + raw
if not raw.lower().startswith("http"):
return None, None
decoded = raw.replace("\\/", "/")
parsed = urlparse(decoded)
if "amazon.es" not in parsed.netloc.lower():
return None, None
path = parsed.path or ""
matched = ASIN_RE.search(path)
if not matched:
matched = ASIN_RE.search(parsed.geturl())
if not matched:
return None, None
asin_val = matched.group(1).upper()
slug = ""
if "/dp/" in path:
slug = path.split("/dp/")[0].rstrip("/")
elif "/gp/product/" in path:
slug = path.split("/gp/product/")[0].rstrip("/")
rel_path = f"{slug}/dp/{asin_val}" if slug else f"/dp/{asin_val}"
if not rel_path.startswith("/"):
rel_path = "/" + rel_path
canon_url = urlunparse(("https", "www.amazon.es", rel_path, "", "", ""))
return canon_url, asin_val
def amazon_hits_regex_fallback(html_blob: str, limit: int) -> list[AmazonHit]:
"""
Recover ASINs from raw HTML when BeautifulSoup misses search-result rows.
Preserves left-to-right order and skips duplicates.
"""
collected: list[AmazonHit] = []
seen: set[str] = set()
for match in re.finditer(r'data-asin="([A-Z0-9]{10})"', html_blob):
asin = match.group(1).upper()
if asin in seen:
continue
seen.add(asin)
product_url, _ = canonical_dp_url(f"https://www.amazon.es/dp/{asin}")
if not product_url:
continue
thumb_guess = (
f"https://m.media-amazon.com/images/P/{asin}.01._SCLZZZZZZZ_.jpg"
)
collected.append(
AmazonHit(
product_url_without_affiliate=product_url,
asin=asin,
title=f"Amazon product {asin}",
image_url=thumb_guess,
)
)
if len(collected) >= limit:
break
return collected
def extract_google_thumb_pairs(html_blob: str) -> list[tuple[str, str]]:
"""Pair thumbnail + landing page URLs emitted by Google's JSON blobs."""
pairs: list[tuple[str, str]] = []
for amazon_match in re.finditer(
r'"(?:ru|isu)":"(https:\\/\\/www\\.amazon\\.es[^\"]+)"',
html_blob,
):
segment = html_blob[max(0, amazon_match.start() - 2400) : amazon_match.start()]
thumbs = re.findall(r'"ou":"(https:\\/\\/[^\"]+)"', segment)
if not thumbs:
continue
encoded_thumb = thumbs[-1]
encoded_landing = amazon_match.group(1)
thumb_plain = encoded_thumb.replace("\\\\/", "/").replace("\\/", "/")
landing_plain = encoded_landing.replace("\\\\/", "/").replace("\\/", "/")
pairs.append((thumb_plain, strip_google_redirect(landing_plain)))
return pairs
def normalize_title(text: str, asin: str) -> str:
stripped = (
text.replace("\r", " ")
.replace("\n", " ")
.strip()
)
if not stripped:
stripped = f"Amazon product {asin}"
if len(stripped) > 200:
stripped = stripped[:197] + "..."
return stripped
def extract_amazon_hits(html_blob: str, limit: int) -> list[AmazonHit]:
"""Collect up to `limit` unique ASIN-backed hits from Google's HTML."""
merged: dict[str, AmazonHit] = {}
order: list[str] = []
def preferred_title(primary: str, secondary: str) -> str:
placeholder = lambda value: value.startswith("Amazon product ")
if placeholder(secondary) and not placeholder(primary):
return primary
if placeholder(primary) and not placeholder(secondary):
return secondary
return primary if len(primary) >= len(secondary) else secondary
def push(hit: AmazonHit | None) -> None:
nonlocal merged, order
if hit is None:
return
if hit.asin in merged:
prior = merged[hit.asin]
merged[hit.asin] = AmazonHit(
product_url_without_affiliate=prior.product_url_without_affiliate,
asin=prior.asin,
title=preferred_title(prior.title, hit.title),
image_url=prior.image_url or hit.image_url,
)
return
if len(order) >= limit:
return
merged[hit.asin] = hit
order.append(hit.asin)
soup = BeautifulSoup(html_blob, "html.parser")
for anchor in soup.find_all("a", href=True):
if len(order) >= limit:
break
href_raw = anchor["href"]
target = strip_google_redirect(href_raw)
canon, maybe_asin = canonical_dp_url(target)
if not canon or not maybe_asin:
continue
label = (
anchor.get("title")
or anchor.get_text(strip=True, separator=" ").strip()
)
img_node = anchor.find("img")
img_src = img_node.get("src") if img_node else ""
push(
AmazonHit(
product_url_without_affiliate=canon,
asin=maybe_asin,
title=normalize_title(label or "", maybe_asin),
image_url=img_src.strip() if img_src else "",
)
)
amazon_url_re = re.compile(
r"https:\\/\\/www\\.amazon\\.es(?:\\/[^\\\"\\s]+)*\\/(?:dp|gp\\/product)\\/"
r"([A-Z0-9]{10})(?:[^\s\\\"]*)",
flags=re.IGNORECASE,
)
for occurrence in amazon_url_re.finditer(html_blob):
if len(order) >= limit:
break
raw = (
occurrence.group(0)
.replace("\\/", "/")
.replace('\\\\"', '"')
)
canon, maybe_asin = canonical_dp_url(raw)
if not canon:
continue
push(
AmazonHit(
product_url_without_affiliate=canon,
asin=maybe_asin,
title=f"Amazon product {maybe_asin}",
image_url="",
)
)
plain_pat = re.compile(
r"https://www\.amazon\.es(?:\/[^?\s\"'<>]*)?/(?:dp|gp/product)/"
r"([A-Z0-9]{10})(?:[^\s\"'<>]*)",
flags=re.IGNORECASE,
)
for occurrence in plain_pat.finditer(html_blob):
if len(order) >= limit:
break
canon, maybe_asin = canonical_dp_url(occurrence.group(0))
if not canon:
continue
push(
AmazonHit(
product_url_without_affiliate=canon,
asin=maybe_asin,
title=f"Amazon product {maybe_asin}",
image_url="",
)
)
for thumb, landing in extract_google_thumb_pairs(html_blob):
if len(order) >= limit:
break
canon, maybe_asin = canonical_dp_url(landing)
if not canon:
continue
push(
AmazonHit(
product_url_without_affiliate=canon,
asin=maybe_asin,
title=f"Amazon product {maybe_asin}",
image_url=thumb,
)
)
return [merged[asin] for asin in order[:limit]]
def count_x2(section_inner: str) -> int:
return len(
re.findall(
r' tuple[int, int] | None:
match = SECTION_OPEN_RE.search(text)
if not match:
return None
start_inner = match.end()
close_idx = text.lower().find("", start_inner)
if close_idx == -1:
return None
return start_inner, close_idx
# Remove C0 controls and DEL that break HTML/XML or PHP source readability.
_CONTROL_AND_DISALLOWED_RE = re.compile(r"[\x00-\x08\x0B-\x0C\x0E-\x1F\x7F]")
def sanitize_plaintext(value: str) -> str:
if not value:
return ""
return _CONTROL_AND_DISALLOWED_RE.sub("", value)
def escape_html_double_quoted_attr(value: str) -> str:
"""Safe for title=\"...\" and alt=\"...\" inside PHP-embedded HTML."""
cleaned = sanitize_plaintext(value)
escaped = html.escape(cleaned, quote=True)
escaped = escaped.replace("'", "'")
escaped = escaped.replace("`", "`")
return escaped
def escape_url_double_quoted_attr(url: str) -> str:
"""
Safe for href/src=\"...\": HTML-escape &, <, >; percent-encode quotes;
strip line breaks that would split the attribute.
"""
cleaned = sanitize_plaintext(url.strip())
cleaned = cleaned.replace("\r", "").replace("\n", "").replace("\t", "")
escaped = html.escape(cleaned, quote=False)
escaped = escaped.replace('"', "%22")
escaped = escaped.replace("'", "%27")
escaped = escaped.replace("`", "%60")
escaped = escaped.replace("\\", "%5C")
return escaped
def escape_html_element_text(value: str) -> str:
"""Safe for ...
embedded inside PHP single-quoted strings."""
cleaned = sanitize_plaintext(value)
collapsed = " ".join(cleaned.split())
# quote=True emits ' for apostrophes so filenames like meta_parrafo='...'
# are not broken by titles such as Readyy'y.
return html.escape(collapsed, quote=True)
def escape_attr(value: str) -> str:
"""Backward-compatible alias for legacy callers (attribute context)."""
return escape_html_double_quoted_attr(value)
def php_single_quote(value: str) -> str:
return value.replace("\\", "\\\\").replace("'", "\\'")
def render_card(hit: AmazonHit) -> str:
title_attr = escape_html_double_quoted_attr(hit.title)
parrafo_text = escape_html_element_text(hit.title)
img_src = hit.image_url or (
f"https://m.media-amazon.com/images/P/{hit.asin}.01._SCLZZZZZZZ_.jpg"
)
img_attr = escape_url_double_quoted_attr(img_src)
base = hit.product_url_without_affiliate
base_attr = escape_url_double_quoted_attr(base)
lines = [
'',
(
f'
'
),
(
f'\t\t
'
f'
Comprar'
),
"
\t\t\t ",
"\t\t
",
"\t\t\t
Compartir:",
(
f'\t\t\t
'
f'
'
),
(
f'\t\t\t
'
f'
'
),
(
f'\t\t\t
'
f'
'
),
"\t\t
",
"
\t\t\t ",
f'
{parrafo_text}
',
"