445 lines
16 KiB
Python
445 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
aliexpress_to_partdb.py
|
|
=======================
|
|
|
|
Convert an AliExpress "orders.csv" export into a CSV ready to be imported
|
|
into Part-DB v2.15.x via the built-in CSV importer (Parts -> Import).
|
|
|
|
What it does
|
|
------------
|
|
* Maps each AliExpress order line to one Part-DB "part" row using the default
|
|
column set the Part-DB 2.15.x import wizard exposes:
|
|
|
|
name, description, category, footprint, manufacturer,
|
|
manufacturerpartnumber, quantity, minquantity, price, supplier,
|
|
supplierpartnumber, ordernumber, orderdetails, comment, keywords,
|
|
mass, lotnumber, storage_location, needs_review
|
|
|
|
* Downloads every "Item image url" to ./partdb_images/ (and tries to upgrade
|
|
the CDN's _220x220.jpg thumbnail to the full-size version).
|
|
* Writes ./partdb_images/attachments.json so you can either drag-drop the
|
|
images onto the imported parts in the Part-DB UI, or push them via the
|
|
REST API (POST /api/parts/{id}/attachments).
|
|
|
|
Usage
|
|
-----
|
|
# minimal
|
|
python3 aliexpress_to_partdb.py orders.csv
|
|
|
|
# with explicit output paths
|
|
python3 aliexpress_to_partdb.py orders.csv -o my_parts.csv -d ./images
|
|
|
|
# CSV only, no image download
|
|
python3 aliexpress_to_partdb.py orders.csv --no-images
|
|
|
|
Dependencies
|
|
------------
|
|
pip install requests
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
try:
|
|
import requests
|
|
except ImportError:
|
|
sys.stderr.write(
|
|
"ERROR: the 'requests' package is required.\n"
|
|
" install it with: pip install requests\n"
|
|
)
|
|
sys.exit(1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Part-DB target columns (verified against the official Part-DB 2.15.x
|
|
# "Tools -> Import parts" schema; see https://docs.part-db.de/usage/import_export.html)
|
|
#
|
|
# IMPORTANT: do NOT include `ordernumber`, `orderdetails` or `lotnumber` here.
|
|
# Those are properties of the related Orderdetail / PartLot entities, not of
|
|
# the Part entity itself, so flat CSV columns for them cause Doctrine to
|
|
# reject the import with: "must be one of Orderdetail[] (string given)".
|
|
# Order provenance is folded into the `notes` field instead.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Default category that new parts land in. Change this to any path that
|
|
# Part-DB should use (e.g. "Electronics/Sensors" or "Imported/Aliexpress").
|
|
# If the category doesn't exist yet, enable "Create unknown datastructures"
|
|
# in the Part-DB import wizard and Part-DB will create it automatically.
|
|
DEFAULT_CATEGORY = "New parts"
|
|
|
|
|
|
PARTDB_COLUMNS = [
|
|
"name",
|
|
"description",
|
|
"notes",
|
|
"category",
|
|
"footprint",
|
|
"manufacturer",
|
|
"manufacturer_product_number",
|
|
"manufacturer_product_url",
|
|
"amount",
|
|
"minamount",
|
|
"price",
|
|
"supplier",
|
|
"supplier_product_number",
|
|
"storage_location",
|
|
"tags",
|
|
"mass",
|
|
"needs_review",
|
|
]
|
|
|
|
# AliExpress header names (must match the export exactly)
|
|
AE = {
|
|
"order_id": "Order Id",
|
|
"order_date": "Order date",
|
|
"status": "Order Status",
|
|
"detail_url": "Order detail url",
|
|
"store": "Store Name",
|
|
"store_url": "Store url",
|
|
"currency": "Currency",
|
|
"tracking": "Tracking number",
|
|
"tracking_lnk": "Tracking link",
|
|
"title": "Item title",
|
|
"price": "Item price",
|
|
"quantity": "Item quantity",
|
|
"attrs": "Item attributes",
|
|
"image": "Item image url",
|
|
"product_url": "Item product link",
|
|
"net": "Total order net price",
|
|
"shipping": "Total Shipping",
|
|
"adjust": "Total price adjustments",
|
|
"discount": "Total discount",
|
|
"eu_tax": "Total EU Tax",
|
|
"vat": "Total VAT",
|
|
"total": "Total price",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
_CTRL = re.compile(r"[\x00-\x08\x0B-\x1F]")
|
|
|
|
|
|
def safe(text) -> str:
|
|
"""Strip control chars + trim, keep unicode."""
|
|
if text is None:
|
|
return ""
|
|
return _CTRL.sub(" ", str(text)).strip()
|
|
|
|
|
|
def larger_image_url(url: str) -> str:
|
|
"""Strip the trailing _220x220.jpg (or _50x50.jpg, _100x100.jpg …) so
|
|
the AliExpress CDN returns the full-size image.
|
|
|
|
AliExpress's CDN always serves paths like
|
|
kf/Se4663b49bf53454aa797e7a1b704c9a3V.jpg_220x220.jpg
|
|
where the original file already has its .jpg extension, and the
|
|
"_<W>x<H>.<ext>" suffix is appended only for the thumbnail. We strip
|
|
the entire size suffix; the file's real extension stays put.
|
|
"""
|
|
if not url:
|
|
return url
|
|
upgraded = re.sub(
|
|
r"_\d+x\d+\.(jpg|jpeg|png|webp)$",
|
|
"",
|
|
url,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
# Safety net: if we somehow stripped the only extension, return original
|
|
if not re.search(r"\.(jpg|jpeg|png|webp|gif)$", upgraded, re.IGNORECASE):
|
|
return url
|
|
return upgraded
|
|
|
|
|
|
def file_ext_from_url(url: str) -> str:
|
|
path = urlparse(url).path
|
|
ext = os.path.splitext(path)[1].lower()
|
|
return ext if ext in {".jpg", ".jpeg", ".png", ".webp", ".gif"} else ".jpg"
|
|
|
|
|
|
def make_manufacturer_part_number(attrs: str, title: str) -> str:
|
|
"""First sensible token of the attribute blob or title — AliExpress
|
|
stores e.g. 'ENS160 with AHT21' or 'DC5V, 5Pcs'."""
|
|
for source in (attrs, title):
|
|
if not source:
|
|
continue
|
|
first = re.split(r"[,;/]", source, maxsplit=1)[0].strip()
|
|
if 1 <= len(first) <= 64 and re.search(r"[A-Z0-9]", first, re.IGNORECASE):
|
|
return first
|
|
return ""
|
|
|
|
|
|
def build_keywords(attrs: str, store: str, order_id: str) -> str:
|
|
parts: list[str] = []
|
|
if attrs:
|
|
for piece in re.split(r"[,;/]+", attrs):
|
|
piece = piece.strip()
|
|
if piece and piece not in parts:
|
|
parts.append(piece)
|
|
if store:
|
|
parts.append(store)
|
|
if order_id:
|
|
parts.append(f"order:{order_id}")
|
|
return ", ".join(parts)
|
|
|
|
|
|
def normalize_url(url: str) -> str:
|
|
"""AliExpress sometimes uses protocol-relative URLs (//foo.com/...)."""
|
|
if url and url.startswith("//"):
|
|
return "https:" + url
|
|
return url
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Image downloader
|
|
# ---------------------------------------------------------------------------
|
|
def _try_download(session: requests.Session, url: str, dest: Path) -> bool:
|
|
"""Single download attempt. Returns True on success, False on any error."""
|
|
try:
|
|
headers = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
|
|
),
|
|
"Referer": "https://www.aliexpress.com/",
|
|
"Accept": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8",
|
|
}
|
|
with session.get(url, timeout=25, stream=True, headers=headers) as r:
|
|
r.raise_for_status()
|
|
with dest.open("wb") as f:
|
|
for chunk in r.iter_content(8192):
|
|
if chunk:
|
|
f.write(chunk)
|
|
return True
|
|
except Exception:
|
|
# remove a partially written file so retries are clean
|
|
try:
|
|
if dest.exists():
|
|
dest.unlink()
|
|
except OSError:
|
|
pass
|
|
return False
|
|
|
|
|
|
def download_image(
|
|
session: requests.Session,
|
|
primary_url: str,
|
|
fallback_url: str,
|
|
dest: Path,
|
|
) -> bool:
|
|
"""Try ``primary_url`` first; on failure try ``fallback_url`` (typically
|
|
the original thumbnail). Returns True if either succeeded."""
|
|
if not primary_url:
|
|
return False
|
|
|
|
if _try_download(session, primary_url, dest):
|
|
return True
|
|
|
|
if fallback_url and fallback_url != primary_url:
|
|
sys.stderr.write(
|
|
f" ! full-size 404, falling back to thumbnail: {fallback_url}\n"
|
|
)
|
|
if _try_download(session, fallback_url, dest):
|
|
return True
|
|
sys.stderr.write(f" ! image download failed: {fallback_url} (both attempts)\n")
|
|
else:
|
|
sys.stderr.write(f" ! image download failed: {primary_url}\n")
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Row transformation
|
|
# ---------------------------------------------------------------------------
|
|
def transform_row(row: dict) -> dict:
|
|
title = safe(row.get(AE["title"]))
|
|
attrs = safe(row.get(AE["attrs"]))
|
|
price = safe(row.get(AE["price"]))
|
|
qty_raw = safe(row.get(AE["quantity"]))
|
|
order_id = safe(row.get(AE["order_id"]))
|
|
order_date = safe(row.get(AE["order_date"]))
|
|
tracking = safe(row.get(AE["tracking"]))
|
|
store = safe(row.get(AE["store"]))
|
|
store_url = normalize_url(safe(row.get(AE["store_url"])))
|
|
detail_url = safe(row.get(AE["detail_url"]))
|
|
product_url = normalize_url(safe(row.get(AE["product_url"])))
|
|
|
|
mpn = make_manufacturer_part_number(attrs, title)
|
|
|
|
# description: title + variant (if variant not already in the title)
|
|
if attrs and attrs.lower() not in title.lower():
|
|
description = f"{title}\n\nVariant: {attrs}"
|
|
else:
|
|
description = title
|
|
|
|
# notes: full provenance trail (order id, date, tracking, URLs) goes
|
|
# here because ordernumber/orderdetails/lotnumber are NOT valid Part
|
|
# fields and would be rejected by the Part-DB CSV importer.
|
|
notes_lines: list[str] = []
|
|
if order_id:
|
|
notes_lines.append(f"AliExpress order ID: {order_id}")
|
|
if order_date:
|
|
notes_lines.append(f"AliExpress order date: {order_date}")
|
|
if tracking:
|
|
notes_lines.append(f"Tracking: {tracking}")
|
|
if product_url:
|
|
notes_lines.append(f"Product page: {product_url}")
|
|
if store_url:
|
|
notes_lines.append(f"Store: {store_url}")
|
|
if detail_url:
|
|
notes_lines.append(f"Order detail: {detail_url}")
|
|
|
|
# amount: drop ".00" if present, otherwise keep as-is
|
|
try:
|
|
qty_int = int(float(qty_raw))
|
|
amount_out = str(qty_int)
|
|
except (TypeError, ValueError):
|
|
amount_out = qty_raw
|
|
|
|
return {
|
|
"name": (title or f"AliExpress order {order_id}")[:120],
|
|
"description": description,
|
|
"notes": "\n".join(notes_lines),
|
|
"category": DEFAULT_CATEGORY,
|
|
"footprint": "",
|
|
"manufacturer": "",
|
|
"manufacturer_product_number": mpn,
|
|
"manufacturer_product_url": product_url,
|
|
"amount": amount_out,
|
|
"minamount": "",
|
|
"price": price,
|
|
"supplier": store,
|
|
"supplier_product_number": mpn,
|
|
"storage_location": "",
|
|
"tags": build_keywords(attrs, store, order_id),
|
|
"mass": "",
|
|
"needs_review": "1",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
def main(argv=None) -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Convert an AliExpress orders.csv into a Part-DB importable CSV.",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog=(
|
|
"Example:\n"
|
|
" python3 aliexpress_to_partdb.py orders.csv\n"
|
|
" python3 aliexpress_to_partdb.py orders.csv -o my_parts.csv -d ./images\n"
|
|
),
|
|
)
|
|
parser.add_argument("input", help="AliExpress orders.csv file")
|
|
parser.add_argument(
|
|
"-o", "--output", default="partdb_import.csv",
|
|
help="Output CSV path (default: partdb_import.csv)",
|
|
)
|
|
parser.add_argument(
|
|
"-d", "--images-dir", default="partdb_images",
|
|
help="Where to store downloaded item images (default: partdb_images)",
|
|
)
|
|
parser.add_argument(
|
|
"--no-images", action="store_true",
|
|
help="Skip image download (CSV only)",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
in_path = Path(args.input)
|
|
if not in_path.is_file():
|
|
sys.stderr.write(f"ERROR: input CSV not found: {in_path}\n")
|
|
return 1
|
|
|
|
out_path = Path(args.output)
|
|
images_dir = Path(args.images_dir)
|
|
images_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
with in_path.open(newline="", encoding="utf-8") as fh:
|
|
reader = csv.DictReader(fh)
|
|
if reader.fieldnames is None:
|
|
sys.stderr.write("ERROR: could not read CSV header.\n")
|
|
return 1
|
|
# warn about missing columns
|
|
missing = [v for v in AE.values() if v not in reader.fieldnames]
|
|
if missing:
|
|
sys.stderr.write("WARNING: these AliExpress columns were not found in the input CSV:\n")
|
|
for m in missing:
|
|
sys.stderr.write(f" - {m}\n")
|
|
rows = list(reader)
|
|
|
|
sys.stderr.write(f"Loaded {len(rows)} order line(s).\n")
|
|
|
|
session = requests.Session() if not args.no_images else None
|
|
attachments: list[dict] = []
|
|
image_attempts = 0
|
|
|
|
with out_path.open("w", newline="", encoding="utf-8") as fh:
|
|
writer = csv.DictWriter(fh, fieldnames=PARTDB_COLUMNS, quoting=csv.QUOTE_ALL)
|
|
writer.writeheader()
|
|
for idx, row in enumerate(rows, start=1):
|
|
mapped = transform_row(row)
|
|
writer.writerow(mapped)
|
|
|
|
if session is not None:
|
|
image_url = safe(row.get(AE["image"]))
|
|
if image_url:
|
|
hi = larger_image_url(image_url)
|
|
# Use the original URL's extension so the local file
|
|
# always has the right extension even if the upgrade
|
|
# collapses the suffix.
|
|
ext = file_ext_from_url(image_url)
|
|
# `ordernumber` is no longer in the mapped dict, so build
|
|
# the file id from the part name (truncated to keep
|
|
# Windows happy) and fall back to "row<idx>" if empty.
|
|
name_slug = re.sub(r"[^A-Za-z0-9_.-]+", "_", mapped["name"])[:60].strip("_")
|
|
safe_id = name_slug or f"row{idx}"
|
|
dest = images_dir / f"{safe_id}_{idx}{ext}"
|
|
image_attempts += 1
|
|
if download_image(session, hi, image_url, dest):
|
|
attachments.append({
|
|
"row": idx,
|
|
"part_name": mapped["name"],
|
|
"files": [str(dest.resolve())],
|
|
})
|
|
if idx % 25 == 0:
|
|
sys.stderr.write(f" processed {idx}/{len(rows)} rows...\n")
|
|
|
|
sidecar = images_dir / "attachments.json"
|
|
sidecar.write_text(
|
|
json.dumps(attachments, indent=2, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
sys.stderr.write("\nAll done.\n")
|
|
sys.stderr.write(f" CSV written to: {out_path.resolve()}\n")
|
|
sys.stderr.write(f" Images stored in: {images_dir.resolve()}\n")
|
|
sys.stderr.write(f" Attachments map: {sidecar.resolve()}\n")
|
|
if image_attempts:
|
|
sys.stderr.write(
|
|
f" Images downloaded: {len(attachments)}/{image_attempts} "
|
|
f"({image_attempts - len(attachments)} failed)\n"
|
|
)
|
|
sys.stderr.write(
|
|
"\nNext steps in Part-DB v2.15.x:\n"
|
|
" 1. Log in as admin -> Parts -> Import -> upload partdb_import.csv\n"
|
|
" (let the wizard auto-map columns; the headers already match\n"
|
|
" Part-DB's default import schema).\n"
|
|
" 2. Open each newly imported part and drag the matching image from\n"
|
|
" partdb_images/ onto the Attachments box, OR push them via the\n"
|
|
" REST API using the row -> file map in attachments.json.\n"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|