""" Migrate PartKeepr PartAttachment image records into parts_inventory.part_images. Default source files path matches your server layout: /var/www/html/Server/files/PartAttachment Example: python3 scripts/migrate_partkeepr_attachments.py \ --source-host 192.168.50.5 --source-port 3306 \ --source-db PartKeepr --source-user partkeepr --source-password partkeepr \ --target-host 192.168.50.5 --target-port 3306 \ --target-db parts_inventory --target-user bill --target-password 5074367 """ from __future__ import annotations import argparse import os import shutil import sys from pathlib import Path from typing import Optional import pymysql from pymysql.cursors import DictCursor # Make project-root imports work when run as a script. ROOT = Path(__file__).resolve().parent.parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from blueprints.partsinventory import _effective_upload_dir, _ensure_tables # noqa: E402 IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".tif", ".svg"} def connect(host: str, port: int, user: str, password: str, database: str): return pymysql.connect( host=host, port=port, user=user, password=password, database=database, cursorclass=DictCursor, autocommit=False, ) def parse_args(): parser = argparse.ArgumentParser(description="Migrate PartKeepr PartAttachment images to parts_inventory part_images.") parser.add_argument("--source-host", required=True) parser.add_argument("--source-port", type=int, default=3306) parser.add_argument("--source-db", required=True) parser.add_argument("--source-user", required=True) parser.add_argument("--source-password", required=True) parser.add_argument("--target-host", required=True) parser.add_argument("--target-port", type=int, default=3306) parser.add_argument("--target-db", required=True) parser.add_argument("--target-user", required=True) parser.add_argument("--target-password", required=True) parser.add_argument("--source-files-dir", default="/var/www/html/Server/files/PartAttachment") parser.add_argument("--dry-run", action="store_true", help="Show intended actions without writing files/DB.") return parser.parse_args() def candidate_source_file(base_dir: Path, filename: str, extension: Optional[str]) -> Optional[Path]: candidates = [base_dir / filename] if extension: candidates.append(base_dir / f"{filename}.{extension.lstrip('.')}") for c in candidates: if c.exists() and c.is_file(): return c return None def is_image_attachment(row: dict) -> bool: """ PartKeepr data is inconsistent: many image attachments have isImage=NULL. Detect images using multiple hints. """ is_image = row.get("isImage") if is_image == 1: return True mimetype = (row.get("mimetype") or "").lower() if mimetype.startswith("image/"): return True extension = (row.get("extension") or "").strip().lower() if extension and f".{extension.lstrip('.')}" in IMAGE_EXTS: return True original = (row.get("originalname") or "").lower() return any(original.endswith(ext) for ext in IMAGE_EXTS) def main(): args = parse_args() source_files_dir = Path(args.source_files_dir) if not source_files_dir.exists(): raise RuntimeError(f"Source files directory not found: {source_files_dir}") src_conn = connect(args.source_host, args.source_port, args.source_user, args.source_password, args.source_db) dst_conn = connect(args.target_host, args.target_port, args.target_user, args.target_password, args.target_db) copied = 0 inserted = 0 skipped_no_part = 0 skipped_missing_file = 0 skipped_duplicate = 0 try: _ensure_tables(dst_conn) with src_conn.cursor() as src, dst_conn.cursor() as dst: src.execute( """ SELECT id, part_id, filename, originalname, extension, isImage, mimetype FROM PartAttachment ORDER BY id """ ) attachments = [r for r in src.fetchall() if is_image_attachment(r)] upload_root = Path(_effective_upload_dir()) for row in attachments: src_part_id = row["part_id"] src_filename = row.get("filename") or "" src_extension = row.get("extension") original_name = row.get("originalname") or src_filename # Map source part to destination part via source_partkeepr_id. dst.execute("SELECT id FROM parts WHERE source_partkeepr_id = %s LIMIT 1", (src_part_id,)) part_row = dst.fetchone() if not part_row: skipped_no_part += 1 continue dst_part_id = int(part_row["id"]) src_file = candidate_source_file(source_files_dir, src_filename, src_extension) if src_file is None: skipped_missing_file += 1 continue # Avoid duplicate rows for same part + original filename. dst.execute( """ SELECT id FROM part_images WHERE part_id = %s AND original_filename = %s LIMIT 1 """, (dst_part_id, original_name), ) if dst.fetchone(): skipped_duplicate += 1 continue ext = src_file.suffix or ".jpg" relative_path = Path("parts") / str(dst_part_id) / f"{src_file.stem}{ext}" dest_file = upload_root / relative_path relative_path_str = str(relative_path).replace("\\", "/") if not args.dry_run: dest_file.parent.mkdir(parents=True, exist_ok=True) # Avoid accidental overwrite collisions. if dest_file.exists(): stem = src_file.stem idx = 1 while True: alt = dest_file.parent / f"{stem}_{idx}{ext}" if not alt.exists(): dest_file = alt relative_path_str = str(Path("parts") / str(dst_part_id) / alt.name).replace("\\", "/") break idx += 1 shutil.copy2(src_file, dest_file) copied += 1 # Preserve "first image is primary" behavior. dst.execute("SELECT COUNT(*) AS c FROM part_images WHERE part_id = %s", (dst_part_id,)) is_primary = 1 if int(dst.fetchone()["c"]) == 0 else 0 if not args.dry_run: dst.execute( """ INSERT INTO part_images (part_id, file_path, original_filename, source, is_primary) VALUES (%s, %s, %s, 'uploaded', %s) """, (dst_part_id, relative_path_str, original_name, is_primary), ) inserted += 1 if args.dry_run: dst_conn.rollback() else: dst_conn.commit() print( "PartAttachment migration complete:", f"copied={copied}", f"inserted={inserted}", f"skipped_no_part={skipped_no_part}", f"skipped_missing_file={skipped_missing_file}", f"skipped_duplicate={skipped_duplicate}", f"dry_run={args.dry_run}", ) except Exception: dst_conn.rollback() raise finally: src_conn.close() dst_conn.close() if __name__ == "__main__": main()