Upload files to "/"
This commit is contained in:
@@ -0,0 +1,444 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user