""" One-way sync from PartKeepr schema into parts_inventory schema. Usage example: python scripts/sync_partkeepr.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 root --target-password secret """ from __future__ import annotations import argparse from typing import Dict, Tuple import pymysql from pymysql.cursors import DictCursor from blueprints.partsinventory import _ensure_tables 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 upsert_categories(src_conn, dst_conn) -> Dict[int, int]: mapping: Dict[int, int] = {} with src_conn.cursor() as src, dst_conn.cursor() as dst: src.execute("SELECT id, parent_id, name, description FROM PartCategory ORDER BY id") rows = src.fetchall() # Phase 1: ensure all nodes exist without parent links. for row in rows: src_id = row["id"] dst.execute("SELECT id FROM categories WHERE source_partkeepr_id = %s LIMIT 1", (src_id,)) hit = dst.fetchone() if hit: mapping[src_id] = hit["id"] dst.execute( """ UPDATE categories SET name = %s, description = %s WHERE id = %s """, (row.get("name") or "", row.get("description"), hit["id"]), ) continue dst.execute( """ INSERT INTO categories (parent_id, source_partkeepr_id, name, description, sort_order) VALUES (NULL, %s, %s, %s, 0) """, (src_id, row.get("name") or "", row.get("description")), ) mapping[src_id] = dst.lastrowid # Phase 2: parent links. for row in rows: src_id = row["id"] parent_src = row.get("parent_id") dst_id = mapping[src_id] parent_dst = mapping.get(parent_src) dst.execute("UPDATE categories SET parent_id = %s WHERE id = %s", (parent_dst, dst_id)) return mapping def upsert_storage_locations(src_conn, dst_conn) -> Dict[int, int]: mapping: Dict[int, int] = {} with src_conn.cursor() as src, dst_conn.cursor() as dst: src.execute("SELECT id, category_id, name FROM StorageLocation ORDER BY id") rows = src.fetchall() for row in rows: src_id = row["id"] dst.execute("SELECT id FROM storage_locations WHERE source_partkeepr_id = %s LIMIT 1", (src_id,)) hit = dst.fetchone() if hit: mapping[src_id] = hit["id"] dst.execute( "UPDATE storage_locations SET name = %s, description = %s WHERE id = %s", (row.get("name") or "", f"PartKeepr category_id={row.get('category_id')}", hit["id"]), ) continue dst.execute( """ INSERT INTO storage_locations (parent_id, source_partkeepr_id, name, description, sort_order) VALUES (NULL, %s, %s, %s, 0) """, (src_id, row.get("name") or "", f"PartKeepr category_id={row.get('category_id')}"), ) mapping[src_id] = dst.lastrowid return mapping def upsert_parts(src_conn, dst_conn, cat_map: Dict[int, int], loc_map: Dict[int, int]) -> Tuple[int, int, int]: inserted = 0 updated = 0 matched_existing = 0 with src_conn.cursor() as src, dst_conn.cursor() as dst: src.execute( """ SELECT p.id, p.name, p.description, p.comment, p.internalPartNumber, p.stockLevel, p.storageLocation_id, p.category_id, p.partUnit_id, u.symbol AS unit_symbol FROM Part p LEFT JOIN Unit u ON p.partUnit_id = u.id ORDER BY p.id """ ) rows = src.fetchall() for row in rows: src_id = row["id"] category_id = cat_map.get(row.get("category_id")) storage_location_id = loc_map.get(row.get("storageLocation_id")) description_parts = [] if row.get("description"): description_parts.append(str(row["description"])) if row.get("comment"): description_parts.append(str(row["comment"])) description = "\n\n".join(description_parts).strip() unit = row.get("unit_symbol") or "" quantity = int(row.get("stockLevel") or 0) part_number = (row.get("internalPartNumber") or "").strip() dst.execute("SELECT id FROM parts WHERE source_partkeepr_id = %s LIMIT 1", (src_id,)) hit = dst.fetchone() if hit: dst.execute( """ UPDATE parts SET category_id = %s, storage_location_id = %s, name = %s, description = %s, part_number = %s, quantity = %s, unit = %s, manufacturer = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s """, ( category_id, storage_location_id, row.get("name") or "", description, part_number, quantity, unit, None, hit["id"], ), ) updated += 1 continue # If a manual/imported record already exists by name (often with blank part_number), # adopt it instead of creating a PK-* duplicate record. dst.execute( """ SELECT id, part_number, quantity, unit, description FROM parts WHERE source_partkeepr_id IS NULL AND name = %s ORDER BY id LIMIT 1 """, (row.get("name") or "",), ) existing_name_match = dst.fetchone() if existing_name_match: merged_part_number = existing_name_match["part_number"] if existing_name_match.get("part_number") else part_number merged_quantity = existing_name_match["quantity"] if existing_name_match.get("quantity") not in (None, 0) else quantity merged_unit = existing_name_match["unit"] if existing_name_match.get("unit") else unit merged_description = existing_name_match["description"] if existing_name_match.get("description") else description dst.execute( """ UPDATE parts SET category_id = %s, storage_location_id = %s, source_partkeepr_id = %s, description = %s, part_number = %s, quantity = %s, unit = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s """, ( category_id, storage_location_id, src_id, merged_description, merged_part_number or "", merged_quantity, merged_unit or "", existing_name_match["id"], ), ) matched_existing += 1 continue dst.execute( """ INSERT INTO parts ( category_id, storage_location_id, source_partkeepr_id, name, description, part_number, quantity, unit, barcode, datasheet_url, manufacturer, footprint ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NULL, NULL, %s, NULL) """, ( category_id, storage_location_id, src_id, row.get("name") or "", description, part_number, quantity, unit, None, ), ) inserted += 1 return inserted, updated, matched_existing def parse_args(): parser = argparse.ArgumentParser(description="Sync PartKeepr data into parts_inventory schema.") 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) return parser.parse_args() def main(): args = parse_args() 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) try: _ensure_tables(dst_conn) cat_map = upsert_categories(src_conn, dst_conn) loc_map = upsert_storage_locations(src_conn, dst_conn) inserted, updated, matched_existing = upsert_parts(src_conn, dst_conn, cat_map, loc_map) dst_conn.commit() print( "Sync complete:", f"categories={len(cat_map)}", f"storage_locations={len(loc_map)}", f"parts_inserted={inserted}", f"parts_updated={updated}", f"parts_matched_existing={matched_existing}", ) except Exception: dst_conn.rollback() raise finally: src_conn.close() dst_conn.close() if __name__ == "__main__": main()