115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Import a full catalog.json into the admin SQLite DB (no HTTP / no size limit).
|
|
|
|
Usage:
|
|
python3 import_catalog.py /path/to/catalog.json
|
|
python3 import_catalog.py /path/to/catalog.json --export
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Reuse server helpers
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import server # noqa: E402
|
|
|
|
|
|
def map_export_category(raw: dict) -> dict:
|
|
"""Map export {id,parent_id,name,full_path} -> admin category row fields."""
|
|
name = str(raw.get("name") or raw.get("full_path") or "").strip()
|
|
if not name:
|
|
raise ValueError("category missing name")
|
|
slug = str(raw.get("slug") or server._slugify(name))
|
|
parent_slug = None
|
|
parent_id = raw.get("parent_id") or raw.get("parentSlug") or raw.get("parent_slug")
|
|
if parent_id:
|
|
parent_name = str(parent_id).split(" / ")[-1].strip()
|
|
parent_slug = server._slugify(parent_name)
|
|
return {
|
|
"slug": slug,
|
|
"name": name.title() if name == name.upper() else name,
|
|
"name_fa": str(raw.get("name_fa") or raw.get("nameFa") or ""),
|
|
"parent_slug": parent_slug,
|
|
"sort_order": int(raw.get("sort_order") or raw.get("sortOrder") or 0),
|
|
"active": raw.get("active", True),
|
|
}
|
|
|
|
|
|
def import_categories(payload: dict) -> int:
|
|
cats = payload.get("categories") or []
|
|
if not isinstance(cats, list):
|
|
return 0
|
|
# parents first (shorter paths)
|
|
cats = sorted(cats, key=lambda c: str(c.get("full_path") or c.get("id") or "").count("/"))
|
|
count = 0
|
|
with server.db() as conn:
|
|
for raw in cats:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
try:
|
|
server.upsert_category(conn, map_export_category(raw))
|
|
count += 1
|
|
except Exception as exc:
|
|
print(f" category skip: {exc}", file=sys.stderr)
|
|
return count
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Import catalog.json into admin DB")
|
|
parser.add_argument("catalog", type=Path, help="Path to catalog.json")
|
|
parser.add_argument(
|
|
"--export",
|
|
action="store_true",
|
|
help="Also export active products to storefront catalog.dedupe2.json",
|
|
)
|
|
parser.add_argument(
|
|
"--products-only",
|
|
action="store_true",
|
|
help="Skip categories",
|
|
)
|
|
args = parser.parse_args()
|
|
path = args.catalog.expanduser().resolve()
|
|
if not path.is_file():
|
|
print(f"File not found: {path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print(f"Loading {path} ({path.stat().st_size // 1_000_000} MB)…")
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(payload, (dict, list)):
|
|
print("Invalid JSON root", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
server.init_db()
|
|
|
|
if isinstance(payload, dict) and not args.products_only:
|
|
n = import_categories(payload)
|
|
print(f"Categories imported/updated: {n}")
|
|
|
|
print("Importing products…")
|
|
result = server.import_products_payload(payload)
|
|
print(
|
|
f"Products created={result['created']} updated={result['updated']} "
|
|
f"skipped={result['skipped']} errors={len(result['errors'])}"
|
|
)
|
|
if result["errors"]:
|
|
print("First errors:", file=sys.stderr)
|
|
for err in result["errors"][:10]:
|
|
print(f" {err}", file=sys.stderr)
|
|
|
|
if args.export:
|
|
out = server.export_catalog()
|
|
print(f"Exported {out['count']} products → {out['path']}")
|
|
|
|
with server.db() as conn:
|
|
pc = conn.execute("SELECT COUNT(*) AS c FROM products").fetchone()["c"]
|
|
cc = conn.execute("SELECT COUNT(*) AS c FROM categories").fetchone()["c"]
|
|
print(f"DB totals: products={pc} categories={cc}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|