79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
"""CSV feed adapter.
|
|
|
|
Reads a supplier CSV with columns:
|
|
sku,title,cost,inventory,shipping_cost,shipping_time,category,image_url
|
|
(optional: description, brand)
|
|
`data_or_url` is either raw CSV text or an http(s) URL to fetch.
|
|
"""
|
|
import csv
|
|
import io
|
|
import urllib.request
|
|
|
|
from app.engines.suppliers.base import SupplierAdapter, register_adapter
|
|
|
|
|
|
@register_adapter
|
|
class CSVAdapter(SupplierAdapter):
|
|
adapter_type = "csv"
|
|
|
|
def __init__(self, supplier, data_or_url=""):
|
|
super().__init__(supplier, data_or_url=data_or_url)
|
|
self._rows = None
|
|
|
|
def _load(self):
|
|
if self._rows is not None:
|
|
return
|
|
src = (self.data_or_url or "").strip()
|
|
if src.startswith(("http://", "https://")):
|
|
with urllib.request.urlopen(src, timeout=30) as resp:
|
|
src = resp.read().decode("utf-8", "replace")
|
|
self._rows = list(csv.DictReader(io.StringIO(src)))
|
|
|
|
def get_products(self):
|
|
self._load()
|
|
products = []
|
|
for row in self._rows:
|
|
sku = (row.get("sku") or "").strip()
|
|
title = (row.get("title") or "").strip()
|
|
if not sku or not title:
|
|
continue
|
|
products.append(
|
|
{
|
|
"sku": sku,
|
|
"title": title,
|
|
"description": (row.get("description") or "").strip() or None,
|
|
"category": (row.get("category") or "").strip() or None,
|
|
"brand": (row.get("brand") or "").strip() or None,
|
|
"cost": float(row.get("cost") or 0),
|
|
"inventory": int(float(row.get("inventory") or 0)),
|
|
"shipping_cost": float(row.get("shipping_cost") or 0),
|
|
"shipping_time": (row.get("shipping_time") or "").strip() or "3-5 days",
|
|
"image_url": (row.get("image_url") or "").strip() or None,
|
|
}
|
|
)
|
|
return products
|
|
|
|
def get_inventory(self):
|
|
return [
|
|
{"supplier_sku": p["sku"], "inventory": p["inventory"]}
|
|
for p in self.get_products()
|
|
]
|
|
|
|
def get_price(self, sku):
|
|
for p in self.get_products():
|
|
if p["sku"] == sku:
|
|
return p["cost"]
|
|
return None
|
|
|
|
def create_order(self, items):
|
|
return {"adapter": "csv", "ref": f"csv:{len(items)}:items", "items": items}
|
|
|
|
def get_order_status(self, ref):
|
|
return "confirmed"
|
|
|
|
def get_tracking(self, ref):
|
|
return None
|
|
|
|
def cancel_order(self, ref):
|
|
return {"ref": ref, "status": "cancelled"}
|