Walmartの在庫確認方法｜店舗在庫の調べ方とPythonで作る再入荷通知

{
  "observed_at": "2026-10-13T01:03:00Z",
  "ingested_at": "2026-10-13T01:03:01Z",
  "market": "US",
  "store_id": "store-001",
  "zip_code": "75201",
  "item_id": "demo-123",
  "seller": "example-seller",
  "variant": "blue-64gb",
  "fulfillment": "pickup",
  "raw_status": "IN_STOCK",
  "normalized_status": "IN_STOCK",
  "price_minor": 2997,
  "currency": "USD",
  "source": "authorized-local-fixture",
  "proxy_session_id": "not-applicable"
}

(market, store_id, item_id, seller, variant, fulfillment)

python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install "requests==2.32.5"

"""Bounded local inventory monitor used by the Rola IP editorial tutorial.

The script starts an in-process HTTP fixture, performs four checks, validates
and normalizes each response, persists valid observations, emits one restock
alert, and exits. It makes no request to Walmart or Rola IP.
"""

from __future__ import annotations

import json
import tempfile
import threading
import time
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from typing import Any, Dict, Tuple

import requests


BASE_FIXTURE: Dict[str, Any] = {
    "market": "US",
    "store_id": "store-001",
    "zip_code": "75201",
    "item_id": "demo-123",
    "seller": "example-seller",
    "variant": "blue-64gb",
    "fulfillment": "pickup",
    "price": "29.97",
    "currency": "USD",
    "source": "authorized-local-fixture",
    "proxy_session_id": "not-applicable",
}

FIXTURES = [
    {**BASE_FIXTURE, "observed_at": "2026-10-13T01:00:00Z", "raw_status": "OUT_OF_STOCK"},
    {**BASE_FIXTURE, "observed_at": "2026-10-13T01:01:00Z", "raw_status": "MYSTERY"},
    {**BASE_FIXTURE, "observed_at": "2026-10-13T01:02:00Z", "raw_status": "OUT_OF_STOCK"},
    {**BASE_FIXTURE, "observed_at": "2026-10-13T01:03:00Z", "raw_status": "IN_STOCK"},
]

STATUS_MAP = {
    "IN_STOCK": "IN_STOCK",
    "AVAILABLE": "IN_STOCK",
    "OUT_OF_STOCK": "OUT_OF_STOCK",
    "UNAVAILABLE": "OUT_OF_STOCK",
}

REQUIRED_FIELDS = {
    "observed_at",
    "market",
    "store_id",
    "item_id",
    "seller",
    "variant",
    "fulfillment",
    "raw_status",
    "price",
    "currency",
    "source",
}


class FixtureHandler(BaseHTTPRequestHandler):
    request_index = 0

    def do_GET(self) -> None:  # noqa: N802
        if self.path != "/authorized-inventory":
            self.send_error(404)
            return
        index = min(self.request_index, len(FIXTURES) - 1)
        body = json.dumps(FIXTURES[index]).encode("utf-8")
        FixtureHandler.request_index += 1
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, _format: str, *_args: object) -> None:
        return


@dataclass(frozen=True)
class Observation:
    observed_at: str
    ingested_at: str
    market: str
    store_id: str
    zip_code: str
    item_id: str
    seller: str
    variant: str
    fulfillment: str
    raw_status: str
    normalized_status: str
    price_minor: int
    currency: str
    source: str
    proxy_session_id: str

    @property
    def comparison_key(self) -> Tuple[str, str, str, str, str, str]:
        return (
            self.market,
            self.store_id,
            self.item_id,
            self.seller,
            self.variant,
            self.fulfillment,
        )


def parse_observation(payload: Dict[str, Any]) -> Observation:
    missing = sorted(REQUIRED_FIELDS - payload.keys())
    if missing:
        raise ValueError(f"missing fields: {', '.join(missing)}")

    raw_status = str(payload["raw_status"]).strip().upper()
    normalized_status = STATUS_MAP.get(raw_status)
    if normalized_status is None:
        raise ValueError(f"unknown raw_status: {raw_status}")

    try:
        price = Decimal(str(payload["price"])).quantize(Decimal("0.01"), ROUND_HALF_UP)
    except InvalidOperation as exc:
        raise ValueError("price is not a decimal amount") from exc
    if price < 0:
        raise ValueError("price cannot be negative")

    try:
        observed = datetime.fromisoformat(str(payload["observed_at"]).replace("Z", "+00:00"))
    except ValueError as exc:
        raise ValueError("observed_at must be an ISO-8601 timestamp") from exc
    if observed.tzinfo is None:
        raise ValueError("observed_at must include a timezone")

    return Observation(
        observed_at=observed.astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
        ingested_at=datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"),
        market=str(payload["market"]),
        store_id=str(payload["store_id"]),
        zip_code=str(payload.get("zip_code", "")),
        item_id=str(payload["item_id"]),
        seller=str(payload["seller"]),
        variant=str(payload["variant"]),
        fulfillment=str(payload["fulfillment"]),
        raw_status=raw_status,
        normalized_status=normalized_status,
        price_minor=int(price * 100),
        currency=str(payload["currency"]).upper(),
        source=str(payload["source"]),
        proxy_session_id=str(payload.get("proxy_session_id", "not-applicable")),
    )


def fetch_observation(url: str) -> Observation:
    response = requests.get(url, timeout=(3.05, 10))
    response.raise_for_status()
    payload = response.json()
    if not isinstance(payload, dict):
        raise ValueError("response must be a JSON object")
    return parse_observation(payload)


class Monitor:
    def __init__(self, output_dir: Path) -> None:
        self.last_valid: Dict[Tuple[str, ...], Observation] = {}
        self.observations_path = output_dir / "observations.jsonl"
        self.alerts_path = output_dir / "alerts.jsonl"

    @staticmethod
    def _append(path: Path, record: Dict[str, Any]) -> None:
        with path.open("a", encoding="utf-8") as stream:
            stream.write(json.dumps(record, sort_keys=True) + "\n")

    def process(self, current: Observation) -> str:
        key = current.comparison_key
        previous = self.last_valid.get(key)
        self._append(self.observations_path, asdict(current))
        self.last_valid[key] = current

        if previous is None:
            return f"BASELINE status={current.normalized_status}"
        if previous.normalized_status == current.normalized_status:
            return f"NO_CHANGE status={current.normalized_status}"

        event = {
            "comparison_key": key,
            "from": previous.normalized_status,
            "to": current.normalized_status,
            "observed_at": current.observed_at,
        }
        self._append(self.alerts_path, event)
        return f"ALERT {event['from']} -> {event['to']}"


def main() -> None:
    FixtureHandler.request_index = 0
    server = HTTPServer(("127.0.0.1", 0), FixtureHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    endpoint = f"http://127.0.0.1:{server.server_port}/authorized-inventory"

    print("Authorized local fixture: ready")
    print("Network boundary: loopback only; credentials in logs: none")

    try:
        with tempfile.TemporaryDirectory(prefix="inventory-demo-") as directory:
            monitor = Monitor(Path(directory))
            for check_number in range(1, len(FIXTURES) + 1):
                try:
                    current = fetch_observation(endpoint)
                    print(f"Check {check_number}: {monitor.process(current)}")
                except (ValueError, requests.RequestException) as exc:
                    print(
                        f"Check {check_number}: ERROR {type(exc).__name__}; "
                        "last valid state preserved"
                    )
                time.sleep(0.05)

            observation_count = len(monitor.observations_path.read_text(encoding="utf-8").splitlines())
            alert_count = len(monitor.alerts_path.read_text(encoding="utf-8").splitlines())
            print(f"Persisted valid observations: {observation_count}; alerts: {alert_count}")
    finally:
        server.shutdown()
        server.server_close()

    print(f"Stop condition: completed {len(FIXTURES)} bounded checks")


if __name__ == "__main__":
    main()

python3 -m pip install requests==2.32.5
python3 authorized_inventory_demo.py

Authorized local fixture: ready
Network boundary: loopback only; credentials in logs: none
Check 1: BASELINE status=OUT_OF_STOCK
Check 2: ERROR ValueError; last valid state preserved
Check 3: NO_CHANGE status=OUT_OF_STOCK
Check 4: ALERT OUT_OF_STOCK -> IN_STOCK
Persisted valid observations: 3; alerts: 1
Stop condition: completed 4 bounded checks

export AUTHORIZED_TARGET_URL="https://approved.example/inventory"
export ROLA_PROXY_URL="http://USERNAME:PASSWORD@HOST:PORT"

import os
import requests

target_url = os.environ["AUTHORIZED_TARGET_URL"]
proxy_url = os.environ["ROLA_PROXY_URL"]

try:
    response = requests.get(
        target_url,
        proxies={"http": proxy_url, "https": proxy_url},
        timeout=(5, 20),
        allow_redirects=True,
    )
except requests.RequestException:
    raise SystemExit("Stop: transport failure; review redacted operational logs")

if response.status_code in {401, 403, 407, 429}:
    raise SystemExit(f"Stop: source returned {response.status_code}")

response.raise_for_status()
if "application/json" not in response.headers.get("Content-Type", ""):
    raise SystemExit("Stop: authorized source did not return JSON")