""" LabWorks Archive Extractor ========================== Connects to a live SQL Server instance running LabWorks and exports customer, order, and line item data to a portable SQLite archive file. Requirements: pip install pyodbc pandas Usage: python labworks_extract.py The script will prompt for connection details, then write labworks_archive.sqlite in the same directory. """ import sqlite3 import sys import os def check_deps(): missing = [] try: import pyodbc except ImportError: missing.append('pyodbc') try: import pandas as pd except ImportError: missing.append('pandas') if missing: print(f"Missing packages: {', '.join(missing)}") print(f"Install with: pip install {' '.join(missing)}") sys.exit(1) check_deps() import pyodbc import pandas as pd # ─── CONNECTION ──────────────────────────────────────────────────────────── def get_connection(): print("\nLabWorks Archive Extractor") print("=" * 40) print("Enter your SQL Server connection details.") print("(Press Enter to accept defaults in [brackets])\n") server = input("Server name/IP [localhost]: ").strip() or "localhost" database = input("Database name [LabWorks]: ").strip() or "LabWorks" auth = input("Use Windows Authentication? [Y/n]: ").strip().lower() if auth in ('', 'y', 'yes'): conn_str = ( f"DRIVER={{ODBC Driver 17 for SQL Server}};" f"SERVER={server};" f"DATABASE={database};" f"Trusted_Connection=yes;" ) else: username = input("SQL Username: ").strip() password = input("SQL Password: ").strip() conn_str = ( f"DRIVER={{ODBC Driver 17 for SQL Server}};" f"SERVER={server};" f"DATABASE={database};" f"UID={username};" f"PWD={password};" ) try: conn = pyodbc.connect(conn_str, timeout=10) print(f"\n✓ Connected to {server}/{database}") return conn except Exception as e: print(f"\n✗ Connection failed: {e}") print("\nTroubleshooting:") print(" - Make sure SQL Server is running") print(" - Try 'localhost\\SQLEXPRESS' or '(local)' as server name") print(" - Make sure ODBC Driver 17 for SQL Server is installed") print(" Download: https://learn.microsoft.com/sql/connect/odbc/download-odbc-driver-for-sql-server") sys.exit(1) # ─── QUERIES ────────────────────────────────────────────────────────────── CUSTOMERS_SQL = """ SELECT c.CustomerID, c.AlternateCustomerID, c.CustomerName, c.CustomerSince, c.IsActive, c.CreditLimit, c.BillingNotes, c.PopupNotes, c.ProductionNotes, con.ContactName, con.PhoneNo1, con.MobileNo, con.EmailAddress, a.Address1, a.Address2, a.City, a.State, a.Zip FROM CustomerInfo c LEFT JOIN ContactInfo con ON con.ContactID = c.DefaultContactInfo LEFT JOIN AddressInfo a ON a.AddressID = c.DefaultBillingInfo """ ORDERS_SQL = """ SELECT o.OrderID, o.CustomerID, c.CustomerName, o.PONumber, o.Reference, o.OrderBy, o.OrderFor, o.ReceiveDateTime, o.DueDateTime, o.PickupDateTime, o.VoidedDateTime, o.Total, o.TaxAmount, o.Deposit, o.OrderStatusID, o.CustomerOriginalNotes, o.AdditionalNotes, o.ApprovalNotes, o.StoreID FROM OrderInfo o LEFT JOIN CustomerInfo c ON c.CustomerID = o.CustomerID """ ORDER_ITEMS_SQL = """ SELECT oi.OrderItemID, oi.OrderID, oi.Description, oi.Quantity, oi.Each, oi.Size1, oi.Size2, oi.SizeUnit, oi.UnitPrice, oi.BasePrice, oi.AdditionalPrice, oi.InstructionNote, oi.OrderStatusID, oi.DueDateTime, oi.ItemType FROM OrderItemInfo oi """ STATUS_SQL = """ SELECT StatusID, StatusName, StatusType FROM StatusInfo """ # ─── EXPORT ─────────────────────────────────────────────────────────────── def export_table(sql_conn, query, table_name, sqlite_conn, chunk_size=5000): print(f" Exporting {table_name}…", end='', flush=True) first = True total = 0 for chunk in pd.read_sql(query, sql_conn, chunksize=chunk_size): chunk.to_sql(table_name, sqlite_conn, if_exists='replace' if first else 'append', index=False) first = False total += len(chunk) print(f"\r Exporting {table_name}… {total:,} rows", end='', flush=True) print(f"\r ✓ {table_name}: {total:,} rows") return total def build_sqlite(sqlite_path, sql_conn): print(f"\nExporting to: {sqlite_path}") print("-" * 40) sqlite_conn = sqlite3.connect(sqlite_path) export_table(sql_conn, CUSTOMERS_SQL, 'customers', sqlite_conn) export_table(sql_conn, ORDERS_SQL, 'orders', sqlite_conn) export_table(sql_conn, ORDER_ITEMS_SQL,'order_items', sqlite_conn) export_table(sql_conn, STATUS_SQL, 'status', sqlite_conn) print("\n Building indexes and view…") c = sqlite_conn.cursor() # Enrich customers c.execute("ALTER TABLE customers ADD COLUMN IsAnonymous INTEGER DEFAULT 0") c.execute("UPDATE customers SET IsAnonymous = 1 WHERE CustomerName REGEXP '^\\([0-9]+\\)$'") # Add status names to orders try: c.execute("ALTER TABLE orders ADD COLUMN StatusName TEXT") except: pass c.execute(""" UPDATE orders SET StatusName = ( SELECT StatusName FROM status WHERE status.StatusID = orders.OrderStatusID ) """) # Indexes for idx_sql in [ "CREATE INDEX IF NOT EXISTS idx_orders_customer ON orders(CustomerID)", "CREATE INDEX IF NOT EXISTS idx_orders_receive ON orders(ReceiveDateTime)", "CREATE INDEX IF NOT EXISTS idx_items_order ON order_items(OrderID)", "CREATE INDEX IF NOT EXISTS idx_customer_name ON customers(CustomerName)", ]: c.execute(idx_sql) # Search view c.execute(""" CREATE VIEW IF NOT EXISTS order_search AS SELECT o.OrderID, o.CustomerID, c.CustomerName, c.EmailAddress, c.PhoneNo1, c.MobileNo, o.PONumber, o.Reference, o.OrderBy, o.OrderFor, o.ReceiveDateTime, o.DueDateTime, o.PickupDateTime, o.Total, o.Deposit, o.StatusName, o.CustomerOriginalNotes, o.AdditionalNotes FROM orders o LEFT JOIN customers c ON c.CustomerID = o.CustomerID """) sqlite_conn.commit() sqlite_conn.close() size_mb = os.path.getsize(sqlite_path) / 1024 / 1024 print(f" ✓ Done — {size_mb:.1f} MB") # ─── MAIN ───────────────────────────────────────────────────────────────── def main(): sql_conn = get_connection() out_path = input(f"\nOutput file [labworks_archive.sqlite]: ").strip() or "labworks_archive.sqlite" if not out_path.endswith('.sqlite'): out_path += '.sqlite' build_sqlite(out_path, sql_conn) sql_conn.close() print(f"\n✓ Archive created: {out_path}") print("\nNext step: Open https://labworks.gastgeber.us in your browser and load this file.") if __name__ == '__main__': main()