How to Check a Discount Against the EU 30-Day Lowest Price

pip install site-shot

link,country,struck
https://shop.example/products/sweatpants,DE,Statt:
https://store.example/p/cpu,DE,.strike-price
https://brand.example/products/kettle,FR,

"""Keep a daily record of each sale page on products.csv for the EU 30-day prior-price rule.

python price_record.py   (once a day)
Captures every link from its country, records today's price and the struck-through price the page
shows, and compares that struck price with the lowest price recorded in the 30 days before the
reduction started. Writes price-record/record.csv and price-record/report.html.
"""
import base64
import csv
import html
import json
import re
import time
from datetime import date, timedelta
from decimal import Decimal, InvalidOperation
from html.parser import HTMLParser
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlsplit

from site_shot import SiteShot, SiteShotError

client = SiteShot()  # reads SITESHOT_API_KEY from the environment
OUT = Path("price-record")
RECORD = OUT / "record.csv"
FIELDS = ["date", "link", "country", "price", "currency", "struck", "landed", "image"]
SETTINGS = dict(
    width=1280, height=1600,
    no_cookie_popup=True, no_ads=True,
    fresh_session=True, source_code=True,
    strict_country=True,
)
# Writes the address the browser ended up on, after any redirect, into the page source.
FINAL_URL = "document.documentElement.setAttribute('data-final-url', location.href);"
WINDOW = 30  # days before the reduction: Article 6a(2) of Directive 98/6/EC
TODAY = date(*time.gmtime()[:3])


def same(a, b):
    """Whether two addresses name the same page, ignoring the case of scheme and host."""
    def parts(url):
        p = urlsplit(url if "//" in url else "https://" + url)
        return p.scheme.lower(), p.netloc.lower(), unquote(p.path or "/"), p.query
    try:
        return parts(a) == parts(b)
    except ValueError:  # a malformed address, such as "https://[bad"
        return a == b


def amount(value):
    """A price as a Decimal with two places, or None."""
    text = str(value).strip()
    if re.fullmatch(r"\d{1,3}(\.\d{3})*,\d{1,2}|\d+,\d{1,2}", text):  # 39,99, 1,5 or 1.299,00
        text = text.replace(".", "").replace(",", ".")
    try:
        found = Decimal(text.replace(",", ""))
        return found.quantize(Decimal("0.01")) if found.is_finite() else None  # not NaN
    except (InvalidOperation, ValueError):
        return None


def first_amount(text):
    """The first price in page text: "€ 333,00", ": 39,99€", "€ 1.299,-" or "1 299,00 €"."""
    match = re.search(r"\d{1,3}(?:[ \u00a0\u202f.,]\d{3})+(?:[.,]\d{1,2})?|\d+(?:[.,]\d{1,2})?", text or "")
    if not match:
        return None
    found = re.sub(r"[ \u00a0\u202f]", "", match.group(0))  # spaces that group thousands
    if re.fullmatch(r"\d{1,3}(\.\d{3})+", found):  # 1.299 written for 1,299
        found = found.replace(".", "")
    return amount(found)


def kinds(data):
    kind = data.get("@type") or []
    return {str(k) for k in (kind if isinstance(kind, list) else [kind])}


class Page(HTMLParser):
    """Today's price and the struck-through price a sale page shows, from its rendered HTML."""

    def __init__(self, struck=""):
        super().__init__(convert_charrefs=True)
        self.struck = struck.strip()
        self.final_url, self.offers, self.meta, self.blocks, self.text = None, [], {}, [], []
        self.hidden = 0  # inside a script, style or noscript element
        self.inside, self.capture, self.found = False, None, []

    def wanted(self, tag, attrs):
        if self.struck.startswith("."):
            return self.struck[1:] in (attrs.get("class") or "").split()
        return self.struck.lower() in ("s", "del", "strike") and tag == self.struck.lower()

    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)
        key = (attrs.get("property") or "").lower()
        if tag == "html":
            self.final_url = attrs.get("data-final-url")
        elif tag == "meta" and key in ("og:price:amount", "og:price:currency",
                                       "product:price:amount", "product:price:currency"):
            self.meta.setdefault(key, (attrs.get("content") or "").strip())
        elif tag == "script" and (attrs.get("type") or "").strip().lower() == "application/ld+json":
            self.inside = True
            self.blocks.append("")
        if tag in ("script", "style", "noscript"):
            self.hidden += 1
        if self.capture:
            if tag == self.capture[0]:
                self.capture[1] += 1
        elif self.struck and self.wanted(tag, attrs):
            self.capture = [tag, 1]
            self.found.append("")

    def handle_data(self, data):
        if self.inside:
            self.blocks[-1] += data
        elif not self.hidden:
            self.text.append(data)
            if self.capture:
                self.found[-1] += data

    def handle_endtag(self, tag):
        if tag == "script" and self.inside:
            self.inside = False
            try:
                self.walk(json.loads(self.blocks[-1], strict=False))  # allows raw line breaks in text
            except ValueError:
                pass  # an unreadable block declares nothing
        if tag in ("script", "style", "noscript") and self.hidden:
            self.hidden -= 1
        if self.capture and tag == self.capture[0]:
            self.capture[1] -= 1
            if not self.capture[1]:
                self.capture = None

    def walk(self, data):
        if isinstance(data, dict):
            if {"Offer", "AggregateOffer"} & kinds(data):
                specs = data.get("priceSpecification") or []
                specs = [s for s in (specs if isinstance(specs, list) else [specs]) if isinstance(s, dict)]
                struck = [amount(s.get("price")) for s in specs
                          if str(s.get("priceType") or "").endswith(("StrikethroughPrice", "ListPrice"))]
                # An offer can carry its price only in a specification without a priceType.
                plain = [s for s in specs if not s.get("priceType") and not s.get("validForMemberTier")]
                source = data if "price" in data or "lowPrice" in data or not plain else plain[0]
                currency = source.get("priceCurrency") or data.get("priceCurrency") or ""
                self.offers.append({
                    "price": amount(source.get("price", source.get("lowPrice"))),
                    "currency": str(currency).upper() or None,
                    "struck": next((s for s in struck if s is not None), None),
                    "url": str(data.get("url") or ""),
                })
            data = list(data.values())
        if isinstance(data, list):
            for item in data:
                self.walk(item)

    def struck_price(self):
        """The struck-through price: the reader's selector or label, else the structured data's."""
        if self.struck.startswith(".") or self.struck.lower() in ("s", "del", "strike"):
            return next((a for a in map(first_amount, self.found) if a is not None), None)
        if self.struck:
            text = " ".join(" ".join(self.text).split())
            label = r"(?<![^\W\d_])%s(?![^\W\d_])" % re.escape(self.struck)  # not inside a word
            for match in re.finditer(label, text):
                found = first_amount(text[match.end():match.end() + 30])
                if found is not None:
                    return found
            return None
        return None


def variant(url):
    """The variant id a Shopify-style product link selects, if any."""
    return parse_qs(urlsplit(url).query).get("variant", [None])[0]


def capture(row, image):
    """Capture one product page, save the screenshot as OUT/image and return today's record row."""
    options = dict(SETTINGS, javascript_code=FINAL_URL)
    if row["country"]:
        options["country"] = row["country"]
    reply = client.capture_json(row["link"], **options)
    (OUT / image).write_bytes(base64.b64decode(reply["image"].split(",", 1)[1]))
    page = Page(row["struck"])
    page.feed(reply.get("source_code") or "")
    picked = [o for o in page.offers if variant(row["link"]) and variant(o["url"]) == variant(row["link"])]
    offer = (picked or page.offers or [{"price": None, "currency": None, "struck": None}])[0]
    meta = page.meta.get("og:price:amount") or page.meta.get("product:price:amount")
    if offer["price"] is None and meta:  # no price in the structured data: the price meta tag
        currency = page.meta.get("og:price:currency") or page.meta.get("product:price:currency") or ""
        offer = dict(offer, price=amount(meta), currency=currency.upper() or None)
    struck = page.struck_price() if row["struck"] else offer["struck"]
    return {
        "date": TODAY, "link": row["link"], "country": row["country"],
        "price": offer["price"], "currency": offer["currency"], "struck": struck,
        "landed": page.final_url or row["link"], "image": image,
        "status": reply.get("response", {}).get("status_code"),
        "source": reply.get("source_code_status"),
    }


def check(history, today):
    """What to look at in today's row, against the rows recorded before it for the same page."""
    notes = [] if today["status"] == 200 else ["status %s" % today["status"]]
    if today["source"] != "ok":
        return notes + ["no page source (%s)" % today["source"]], {}
    if not same(today["landed"], today["link"]):
        notes.append("lands on %s" % today["landed"])
    if today["price"] is None:
        return notes + ["no price in the structured data"], {}
    if today["struck"] is None:
        return notes + ["no struck price"], {}
    start = today["date"]
    for row in reversed(history):  # the reduction started on the first day of its run of struck prices
        if row["price"] is None:
            continue  # a capture that read no price, such as a timed-out one, neither starts nor ends it
        if row["struck"] is None or row["currency"] != today["currency"]:
            break
        start = row["date"]
    window = [row for row in history if start - timedelta(days=WINDOW) <= row["date"] < start
              and row["price"] is not None and row["currency"] == today["currency"]]
    days = len({row["date"] for row in window})
    facts = {"start": start, "days": days}
    if days < WINDOW:
        notes.append("the record covers %d of the %d days before %s" % (days, WINDOW, start))
    if window:
        low = min(window, key=lambda row: (row["price"], row["date"]))
        facts.update(low=low["price"], low_date=low["date"])
        if today["struck"] > low["price"]:
            notes.append("struck %s, lowest recorded %s on %s" % (today["struck"], low["price"], low["date"]))
    return notes, facts


def off(base, price):
    """The reduction from base to price, as a whole percentage."""
    if base and base > price:
        return "%d%% off" % ((base - price) * 100 / base).to_integral_value()
    return "not below"


def products():
    """One dict per row of products.csv: link, country and how to find the struck price."""
    with open("products.csv", newline="", encoding="utf-8-sig") as f:
        first = f.readline()
        f.seek(0)
        reader = csv.DictReader(f, delimiter=";" if ";" in first and "," not in first else ",")  # Excel
        reader.fieldnames = [(name or "").strip().lower() for name in reader.fieldnames or []]
        if "link" not in reader.fieldnames:
            raise SystemExit("products.csv needs a link column (country and struck are optional)")
        found = []
        for n, row in enumerate(reader, 1):
            link = (row.get("link") or "").strip()
            if not link:
                continue  # a blank line
            country = (row.get("country") or "").strip().upper()
            if country and not re.fullmatch(r"[A-Z]{2}", country):
                raise SystemExit("row %d: write the country as a two-letter code, such as DE" % n)
            found.append({"n": n, "link": link, "country": country,
                          "struck": (row.get("struck") or "").strip()})
        return found


def load():
    """The rows recorded on earlier days."""
    if not RECORD.exists():
        return []
    with open(RECORD, newline="", encoding="utf-8") as f:
        rows = list(csv.DictReader(f))
    for row in rows:
        row["date"] = date.fromisoformat(row["date"])
        row["price"], row["struck"] = amount(row["price"] or "x"), amount(row["struck"] or "x")
        row["currency"] = row["currency"] or None
    return rows


def main():
    OUT.mkdir(exist_ok=True)
    (OUT / TODAY.isoformat()).mkdir(exist_ok=True)
    loaded = load()
    rows = [row for row in loaded if row["date"] != TODAY]  # a second run on one day replaces the first
    earlier = {(row["link"], row["country"]): row for row in loaded if row["date"] == TODAY}
    cells = []
    for product in products():
        page = (product["link"], product["country"])
        history = sorted((row for row in rows if (row["link"], row["country"]) == page),
                         key=lambda row: row["date"])
        try:
            today = capture(product, "%s/%02d.png" % (TODAY.isoformat(), product["n"]))
        except SiteShotError as error:
            print(product["n"], "capture failed: %s" % error)
            if page in earlier:
                rows.append(earlier[page])  # a failed second run keeps the day's first row
            continue
        notes, facts = check(history, today)
        rows.append({key: today[key] for key in FIELDS})
        print(product["n"], "%s: %s" % (product["country"] or "default", "; ".join(notes) or "OK"))
        lines = "".join("<tr><td>%s</td><td>%s</td><td>%s</td></tr>" % (
            row["date"], "no price" if row["price"] is None else row["price"], row["struck"] or "")
            for row in (history + [today])[-WINDOW - 1:])
        said = "no price" if today["price"] is None else (
            "%s %s" % (today["price"], today["currency"] or "")).strip()
        said += ", struck %s" % (today["struck"] or "none")
        if facts.get("low"):
            said += "; %s the struck price, %s the lowest recorded" % (
                off(today["struck"], today["price"]), off(facts["low"], today["price"]))
        cells.append("<h2>%d. <a href='%s'>%s</a> %s</h2><p><b>%s</b><br>%s<br>landed on %s</p>"
                     "<p><a href='%s'><img src='%s' width='400'></a></p><table border='1' cellpadding='4'>"
                     "<tr><th>Date</th><th>Price</th><th>Struck</th></tr>%s</table>" % (
                         product["n"], html.escape(product["link"]), html.escape(product["link"]),
                         product["country"], html.escape("; ".join(notes) or "OK"), html.escape(said),
                         html.escape(today["landed"]), today["image"], today["image"], lines))
    with open(RECORD, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        for row in sorted(rows, key=lambda row: (row["link"], row["country"], row["date"])):
            writer.writerow({key: "" if row[key] is None else row[key] for key in FIELDS})
    (OUT / "report.html").write_text("<meta charset='utf-8'><p>Checked %s UTC</p>%s" % (
        time.strftime("%Y-%m-%d %H:%M", time.gmtime()), "".join(cells)), encoding="utf-8")


main()

# /srv/price-record/key.env (chmod 600)
export SITESHOT_API_KEY=your-key-here

# crontab -e: every day at 07:17 server time
17 7 * * * cd /srv/price-record && . ./key.env && python3 price_record.py >> run.log 2>&1