Created
July 18, 2024 15:56
-
-
Save kuwapa/6610d718d68accebadc5cb27dab5a491 to your computer and use it in GitHub Desktop.
A script to take the sqlite db from postico1 and convert it to .sql files for postico2 query history
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
This script is to convert a SQLite DB file from postico1 which contains the SQL query | |
history to .sql files for each query for Postico2. | |
""" | |
import sqlite3 | |
import zlib | |
import os | |
from datetime import datetime, timedelta | |
import random | |
import string | |
import subprocess | |
# Path to your SQLite database file | |
db_path = '/path/to/dbfile' | |
# Directory to save the .sql files | |
output_dir = '/Users/yourname/Desktop/SQLfiles' | |
# Create the output directory if it doesn't exist | |
if not os.path.exists(output_dir): | |
os.makedirs(output_dir) | |
# Connect to the SQLite database | |
conn = sqlite3.connect(db_path) | |
cursor = conn.cursor() | |
# Query to select non-empty, non-null ZTEXT fields and compressed text fields | |
query = """ | |
SELECT ZTEXT, ZCOMPRESSEDTEXT, ZDATE | |
FROM ZPGEQUERYHISTORYELEMENT | |
WHERE (ZTEXT IS NOT NULL AND ZTEXT != '') OR (ZCOMPRESSEDTEXT IS NOT NULL) | |
""" | |
cursor.execute(query) | |
rows = cursor.fetchall() | |
# Function to generate a random 10-character alphanumeric string | |
def generate_random_id(length=10): | |
characters = string.ascii_uppercase + string.digits | |
return ''.join(random.choice(characters) for _ in range(length)) | |
# Function to convert ZDATE format to a readable timestamp and date folder | |
def convert_zdate(zdate): | |
# ZDATE is the number of seconds since 2001-01-01 | |
base_date = datetime(2001, 1, 1) | |
delta = timedelta(seconds=zdate) | |
date_time = base_date + delta | |
timestamp = date_time.strftime("%Y-%m-%d_%Hh%Mm%Ss%f")[:23] | |
date_folder = date_time.strftime("%Y-%m-%d") | |
return timestamp, date_folder | |
# Write each query to an individual file with the specified naming format | |
for i, row in enumerate(rows): | |
ztext = row[0] | |
zcompressedtext = row[1] | |
zdate = row[2] | |
if ztext: | |
query_text = ztext | |
elif zcompressedtext: | |
query_text = zlib.decompress(zcompressedtext).decode('utf-8') | |
else: | |
continue | |
timestamp, date_folder = convert_zdate(zdate) | |
random_id = generate_random_id() | |
file_name = f"{timestamp} {random_id}.sql" | |
# Create the date folder if it doesn't exist | |
folder_path = os.path.join(output_dir, date_folder) | |
if not os.path.exists(folder_path): | |
os.makedirs(folder_path) | |
file_path = os.path.join(folder_path, file_name) | |
with open(file_path, 'w') as file: | |
file.write(query_text) | |
# Hide the file extension using SetFile | |
try: | |
subprocess.run(['SetFile', '-a', 'E', file_path], check=True) | |
except subprocess.CalledProcessError as e: | |
print(f"Failed to hide extension for {file_path}: {e}") | |
# Close the database connection | |
conn.close() | |
print(f'Exported {len(rows)} queries to {output_dir}') |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment