85 lines
2.4 KiB
Python
85 lines
2.4 KiB
Python
"""Example fetcher: read products / stock from a local or remote JSON file.
|
|
|
|
Config (JSON in admin UI):
|
|
{
|
|
"url": "https://example.com/products.json", // or omit for local path
|
|
"path": "/absolute/or/relative/products.json",
|
|
"stock_only": false
|
|
}
|
|
|
|
Expected JSON shapes:
|
|
{ "products": [ {...}, ... ] }
|
|
or a bare array of products.
|
|
|
|
For stock sync, each item needs at least: id|externalId|sku|slug and stock.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
NAME = "Example JSON source"
|
|
DESCRIPTION = (
|
|
"Import products or sync stock from a JSON URL/path. "
|
|
"Replace this module with a real site-specific scraper when ready."
|
|
)
|
|
|
|
|
|
def _load_payload(config: dict):
|
|
url = (config or {}).get("url")
|
|
path = (config or {}).get("path")
|
|
if url:
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={"User-Agent": "KikomikoAdmin/1.0"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
if path:
|
|
p = Path(path).expanduser()
|
|
if not p.is_file():
|
|
raise FileNotFoundError(f"JSON path not found: {p}")
|
|
return json.loads(p.read_text(encoding="utf-8"))
|
|
raise ValueError("config needs url or path")
|
|
|
|
|
|
def _as_list(payload):
|
|
if isinstance(payload, list):
|
|
return payload
|
|
if isinstance(payload, dict):
|
|
for key in ("products", "items", "data"):
|
|
if isinstance(payload.get(key), list):
|
|
return payload[key]
|
|
raise ValueError("JSON must be a product array or {products:[...]}")
|
|
|
|
|
|
def fetch_products(config: dict):
|
|
rows = _as_list(_load_payload(config))
|
|
for row in rows:
|
|
row.setdefault("source", "example_json")
|
|
return rows
|
|
|
|
|
|
def sync_stocks(config: dict, products: list):
|
|
"""Return [{id|external_id|sku|slug, stock}, ...] for matching DB rows."""
|
|
rows = _as_list(_load_payload(config))
|
|
out = []
|
|
for row in rows:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
stock = row.get("stock")
|
|
if stock is None:
|
|
continue
|
|
out.append(
|
|
{
|
|
"id": row.get("id"),
|
|
"external_id": row.get("externalId") or row.get("external_id"),
|
|
"sku": row.get("sku"),
|
|
"slug": row.get("slug"),
|
|
"stock": int(stock),
|
|
}
|
|
)
|
|
return out
|