Cloudflare をパージしても古い壁紙が届き続けた原因は、オリジン側で JPEG を握っていた XPageSpeed

"""cloudflare_purge_urls.txt を 30 本ずつ Cloudflare の purge_cache へ送る。
   トークンは環境変数から読む（記事中に実値を書かない）。
"""
import json
import os
import sys
import time
import urllib.request
 
ZONE_ID = os.environ["CF_ZONE_ID"]
TOKEN = os.environ["CF_API_TOKEN"]          # Zone > Cache Purge 権限だけを持つトークン
ENDPOINT = f"https://api.cloudflare.com/client/v4/zones/{ZONE_ID}/purge_cache"
CHUNK = 30                                  # ファイル単位パージの上限（1 リクエストあたり）
 
 
def purge(files):
    body = json.dumps({"files": files}).encode()
    req = urllib.request.Request(
        ENDPOINT, data=body, method="POST",
        headers={"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=30) as res:
        data = json.load(res)
    if not data.get("success"):
        raise RuntimeError(data.get("errors"))
    return data["result"]["id"]
 
 
def main(path):
    with open(path, encoding="utf-8") as f:
        urls = [line.strip() for line in f if line.strip()]
    # 同じ URL を二度送らない。順序は保つ
    seen, ordered = set(), []
    for u in urls:
        if u not in seen:
            seen.add(u)
            ordered.append(u)
    total = len(ordered)
    for i in range(0, total, CHUNK):
        chunk = ordered[i:i + CHUNK]
        rid = purge(chunk)
        print(f"[{i + len(chunk):>4}/{total}] purged {len(chunk)} files (request {rid})")
        time.sleep(0.5)                     # レート制限に当てない程度の間隔
 
 
if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "cloudflare_purge_urls.txt")

# storage.example.com の配信ルート直下 .htaccess
# 壁紙は配信時に最終画質へ揃えてあるので、オリジン側の再圧縮は切る
<IfFile /var/xpagespeed/xpagespeed_ctl>
  XPagespeed off
</IfFile>
 
# 参考: 静的画像の約束はこちら。上書きするときはパージまでが 1 セット
<FilesMatch "\.(jpg|jpeg|png|webp)$">
  Header set Cache-Control "public, max-age=31536000, immutable"
</FilesMatch>

"""checkserver: ローカル／オリジン（cb 付き）／エッジの三者を突き合わせ、
   事故を種類ごとに名指しする。サンプル数は --sample、開始 ID は --id-from。
"""
import argparse
import os
import random
import urllib.request
from dataclasses import dataclass
 
BASE = "https://storage.example.com/ios/wallpaper_apps/ukiyo-e/6_5inch/wallpaper"
LOCAL_DIR = "./_out_ukiyo-e/ios/6_5inch/wallpaper"
 
 
@dataclass
class Probe:
    size: int
    cf_status: str
    age: str
    etag: str
    last_modified: str
 
 
def fetch(url):
    req = urllib.request.Request(url, headers={"User-Agent": "checkserver/1.0"})
    with urllib.request.urlopen(req, timeout=20) as res:
        body = res.read()
        h = {k.lower(): v for k, v in res.headers.items()}
    return Probe(
        size=len(body),
        cf_status=h.get("cf-cache-status", "-"),
        age=h.get("age", "-"),
        etag=h.get("etag", "-"),
        last_modified=h.get("last-modified", "-"),
    )
 
 
def classify(local_size, edge, origin):
    """判定の順番が大事。上流（ローカル）から順に疑う。"""
    if local_size is None:
        return "MISSING_LOCAL"
    if origin.etag.startswith('W/"PSA-') or origin.last_modified == "-":
        return "ORIGIN_OPTIMIZER"       # cb 付きでも古い＝オリジン側の最適化層が握っている
    if origin.size != local_size:
        return "ORIGIN_STALE"           # 上げ漏れ・merge 失敗
    if edge.size != origin.size and edge.cf_status == "HIT":
        return "EDGE_STALE"             # パージで直る
    return "OK"
 
 
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--sample", type=int, default=20)
    ap.add_argument("--id-from", type=int, default=1)
    ap.add_argument("--id-to", type=int, required=True)
    args = ap.parse_args()
 
    ids = random.sample(range(args.id_from, args.id_to + 1),
                        k=min(args.sample, args.id_to - args.id_from + 1))
    tally = {}
    for wid in sorted(ids):
        local = os.path.join(LOCAL_DIR, f"wallpaper{wid}.jpg")
        local_size = os.path.getsize(local) if os.path.exists(local) else None
        edge = fetch(f"{BASE}/wallpaper{wid}.jpg")
        origin = fetch(f"{BASE}/wallpaper{wid}.jpg?cb={random.randint(1, 10**9)}")
        verdict = classify(local_size, edge, origin)
        tally[verdict] = tally.get(verdict, 0) + 1
        if verdict != "OK":
            print(f"wallpaper{wid}: {verdict}  local={local_size} origin={origin.size} "
                  f"edge={edge.size} cf={edge.cf_status} age={edge.age} etag={origin.etag}")
    print("---")
    for k, v in sorted(tally.items()):
        print(f"{k:<18} {v}")
    if tally.get("ORIGIN_OPTIMIZER"):
        print("→ XPageSpeed が旧版を保持しています。OFF にしてから、もう一度パージしてください")
    if tally.get("EDGE_STALE"):
        print("→ エッジが古い版を保持しています。プレフィックスか URL 単位でパージしてください")
 
 
if __name__ == "__main__":
    main()