"""Read the supplied bouquet workbook and safely prepare browser-ready photos. Requires openpyxl, Pillow and pillow-heif. Source files are never modified. The resulting private manifest is consumed by import-catalog.php. """ from __future__ import annotations import argparse import hashlib import io import json from pathlib import Path, PurePosixPath import re import stat import textwrap import warnings import zipfile from decimal import Decimal, InvalidOperation from openpyxl import load_workbook from PIL import Image, ImageCms, ImageDraw, ImageFont, ImageOps from pillow_heif import register_heif_opener HEADERS = ["Имя файла", "Название букета", "Описание букета", "Что таит в себе этот букет", "Состав", "Уход за букетом", "Цена"] LETTERS = dict(zip("абвгдеёжзийклмнопрстуфхцчшщъыьэюя", ["a", "b", "v", "g", "d", "e", "yo", "zh", "z", "i", "y", "k", "l", "m", "n", "o", "p", "r", "s", "t", "u", "f", "kh", "ts", "ch", "sh", "sch", "", "y", "", "e", "yu", "ya"])) def sha256(path: Path) -> str: with path.open("rb") as source: return hashlib.file_digest(source, "sha256").hexdigest() def slugify(value: str) -> str: result = "".join(LETTERS.get(c, c) for c in value.lower()) return re.sub(r"[^a-z0-9]+", "-", result).strip("-") def shorten(value: str, limit: int = 155) -> str: value = " ".join(value.split()) if len(value) <= limit: return value return value[:limit - 1].rsplit(" ", 1)[0].rstrip(" ,.;:") + "…" def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--xlsx", type=Path, required=True) parser.add_argument("--photos", type=Path, required=True) parser.add_argument("--editorial", type=Path) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() register_heif_opener() Image.MAX_IMAGE_PIXELS = 45_000_000 warnings.simplefilter("error", Image.DecompressionBombWarning) editorial = json.loads(args.editorial.read_text(encoding="utf-8")) if args.editorial else {} categories = editorial.get("categories", [{"id": "bouquets", "name": "Авторские букеты", "sort_order": 10}]) workbook_hash, archive_hash = sha256(args.xlsx), sha256(args.photos) book = load_workbook(args.xlsx, read_only=True, data_only=False) try: if len(book.worksheets) != 1: raise ValueError("Expected one bouquet worksheet; choose the source explicitly before importing.") sheet = book.worksheets[0] rows = list(sheet.iter_rows()) if [cell.value for cell in rows[0]] != HEADERS: raise ValueError("Unexpected worksheet columns.") products = [] for row_number, cells in enumerate(rows[1:], start=2): if not any(cell.value is not None for cell in cells): continue if len(cells) != len(HEADERS) or any(cell.data_type == "f" for cell in cells): raise ValueError(f"Row {row_number}: formulas or unexpected columns are not import data.") values = [str(cell.value).strip().replace("\r\n", "\n") if cell.value is not None else "" for cell in cells] if any(not value for value in values): raise ValueError(f"Row {row_number}: required source value is missing.") filenames, name, description, story, composition, care, price = values photos = [part.strip() for part in filenames.split("|")] if not 1 <= len(photos) <= 12 or len(set(photos)) != len(photos): raise ValueError(f"Row {row_number}: invalid gallery.") try: minor = Decimal(price.replace(",", ".")) * 100 except InvalidOperation as error: raise ValueError(f"Row {row_number}: invalid price.") from error if not minor.is_finite() or minor != minor.to_integral_value() or not 1 <= minor <= 10_000_000: raise ValueError(f"Row {row_number}: price must be a positive amount with at most two decimals.") metadata = editorial.get("products", {}).get(str(row_number), {}) primary_photo = metadata.get("primary_photo", photos[0]) if primary_photo not in photos: raise ValueError(f"Row {row_number}: selected primary photo is not in its gallery.") product = { "source_row": row_number, "name": name, "slug": metadata.get("slug", slugify(name)), "category": metadata.get("category", "bouquets"), "description": description, "composition": composition, "care": care, "price_minor": int(minor), "stock": 0, "active": True, "is_demo": False, "badge": "", "sort_order": len(products) * 10, "story_title": metadata.get("story_title", "История букета"), "story_text": story, "seo_title": metadata.get("seo_title", f"{name} — букет в Солигорске | La Fleur & Café"), "seo_description": metadata.get("seo_description", shorten(description)), "occasions": metadata.get("occasions", []), "images": [], "_photos": photos, "_primary_photo": primary_photo, } products.append(product) sheet_name = sheet.title finally: book.close() if not products or len({p["slug"] for p in products}) != len(products): raise ValueError("Empty catalogue or duplicate product URLs.") if len({p["name"].casefold() for p in products}) != len(products): raise ValueError("Duplicate product names.") photo_names = [name for p in products for name in p["_photos"]] if len(photo_names) != len(set(photo_names)): raise ValueError("A source photo is assigned to more than one bouquet; review the mapping.") with zipfile.ZipFile(args.photos) as archive: entries = {} total_size = 0 for entry in archive.infolist(): raw = entry.filename.replace("\\", "/") relative = PurePosixPath(raw) if relative.is_absolute() or ".." in relative.parts or ":" in raw or stat.S_ISLNK(entry.external_attr >> 16): raise ValueError("Unsafe archive entry: " + raw) if entry.is_dir(): continue if entry.flag_bits & 1 or not 0 < entry.file_size <= 30_000_000: raise ValueError("Encrypted, empty or oversized photo: " + raw) if relative.suffix.lower() not in {".heic", ".heif", ".jpg", ".jpeg", ".png", ".webp"}: raise ValueError("Unexpected archive file: " + raw) if relative.name in entries: raise ValueError("Duplicate photo basename: " + relative.name) total_size += entry.file_size entries[relative.name] = entry if len(entries) > 1000 or total_size > 512_000_000: raise ValueError("The photo archive is too large.") if set(entries) != set(photo_names): raise ValueError(f"Unmatched photos: missing={sorted(set(photo_names)-set(entries))}, extra={sorted(set(entries)-set(photo_names))}") args.output.mkdir(parents=True, exist_ok=False) image_dir = args.output / "images" image_dir.mkdir() previews = [] srgb = ImageCms.createProfile("sRGB") for product in products: primary_photo = product.pop("_primary_photo") for index, name in enumerate(product.pop("_photos")): raw_bytes = archive.read(entries[name]) with Image.open(io.BytesIO(raw_bytes)) as source: if source.width * source.height > 45_000_000: raise ValueError("Photo exceeds 45 megapixels: " + name) image = ImageOps.exif_transpose(source) icc = image.info.get("icc_profile") if icc: image = ImageCms.profileToProfile(image, ImageCms.ImageCmsProfile(io.BytesIO(icc)), srgb, outputMode="RGB") else: image = image.convert("RGB") image.thumbnail((2400, 2400), Image.Resampling.LANCZOS) # Save a fresh image to remove source EXIF, GPS and embedded thumbnails. clean = Image.new("RGB", image.size) clean.paste(image) encoded = io.BytesIO() clean.save(encoded, format="WEBP", quality=88, method=6) data = encoded.getvalue() digest = hashlib.sha256(data).hexdigest() relative_path = "images/" + digest[:32] + ".webp" (args.output / relative_path).write_bytes(data) product["images"].append({"source_file": name, "path": relative_path, "sha256": digest, "width": clean.width, "height": clean.height, "is_primary": name == primary_photo}) previews.append((clean.copy(), product["name"], name, name == primary_photo)) print(f"Prepared row {product['source_row']}: {name} -> {clean.width}x{clean.height}", flush=True) manifest = {"schema_version": 1, "source": {"workbook": {"path": str(args.xlsx.resolve()), "sha256": workbook_hash, "sheet": sheet_name}, "photos": {"path": str(args.photos.resolve()), "sha256": archive_hash}}, "categories": categories, "products": products} (args.output / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") font = ImageFont.truetype("C:/Windows/Fonts/arial.ttf", 17) tile_w, tile_h, columns = 270, 390, 5 canvas = Image.new("RGB", (tile_w * columns, tile_h * ((len(previews) + columns - 1) // columns)), "#faf6f2") draw = ImageDraw.Draw(canvas) for index, (image, name, filename, primary) in enumerate(previews): x, y = (index % columns) * tile_w, (index // columns) * tile_h image.thumbnail((250, 310), Image.Resampling.LANCZOS) canvas.paste(image, (x + (tile_w - image.width) // 2, y + (310 - image.height) // 2)) title = "\n".join(textwrap.wrap(name, width=26)) draw.multiline_text((x + 10, y + 316), title, font=font, fill="#332b2b", spacing=2) draw.text((x + 10, y + 365), filename + (" *" if primary else ""), font=font, fill="#6c5c57") canvas.save(args.output / "contact-sheet.jpg", quality=90) print(json.dumps({"products": len(products), "images": len(photo_names), "manifest": str(args.output / "manifest.json")}, ensure_ascii=False)) if __name__ == "__main__": main()