e661d91c72
- Add translateCarName import from i18n.ts for proper multilingual support - Change default API language from 'ko' to 'en' for hero banners - Add checkbox column for Local Cars banner registration - Update Dockerfile with Playwright dependencies - Add PostgreSQL migration script - Add banner translation fix script 🤖 Generated with [Claude Code](https://claude.ai/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
298 lines
9.0 KiB
Python
298 lines
9.0 KiB
Python
"""
|
|
SQLite to PostgreSQL Migration Script
|
|
Handles boolean conversion and foreign key constraints
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import sqlite3
|
|
import psycopg2
|
|
from urllib.parse import quote_plus
|
|
|
|
SQLITE_PATH = os.path.join(os.path.dirname(__file__), "autonet.db")
|
|
PG_CONFIG = {
|
|
"host": "192.168.0.201",
|
|
"port": 5432,
|
|
"database": "autonet",
|
|
"user": "admin",
|
|
"password": "roskfl@1122"
|
|
}
|
|
|
|
# Tables in dependency order (parents first)
|
|
TABLES_ORDER = [
|
|
# Base tables (no FK dependencies)
|
|
"car_makers",
|
|
"car_models",
|
|
"translations",
|
|
"system_settings",
|
|
"cc_packages",
|
|
"exchange_rates",
|
|
"exchange_rate_history",
|
|
"hero_banner_settings",
|
|
# Users before user-dependent tables
|
|
"users",
|
|
# Car related
|
|
"cars",
|
|
"car_images",
|
|
"car_options",
|
|
"car_performance_checks",
|
|
"car_specifications",
|
|
"car_views",
|
|
"performance_check_views",
|
|
# Cache
|
|
"car_cache",
|
|
"car_detail_cache",
|
|
"cache_request_queue",
|
|
# Hero banners
|
|
"hero_banners",
|
|
# User activities
|
|
"charge_history",
|
|
"inquiries",
|
|
"inquiry_messages",
|
|
"vehicle_requests",
|
|
"request_vehicles",
|
|
"purchased_vehicles",
|
|
"dealer_applications",
|
|
"dealer_info",
|
|
"vehicle_shares",
|
|
"share_rewards",
|
|
"withdrawal_requests",
|
|
"referral_rewards",
|
|
"notifications",
|
|
"push_subscriptions",
|
|
"user_notification_preferences",
|
|
"verification_codes",
|
|
"visitor_logs",
|
|
"visitor_daily_stats",
|
|
"visitor_sessions",
|
|
]
|
|
|
|
def create_tables():
|
|
"""Create tables in PostgreSQL"""
|
|
print("\n[Step 1] Creating tables in PostgreSQL...")
|
|
|
|
os.environ["USE_SQLITE"] = "False"
|
|
os.environ["DB_HOST"] = PG_CONFIG["host"]
|
|
os.environ["DB_PORT"] = str(PG_CONFIG["port"])
|
|
os.environ["DB_NAME"] = PG_CONFIG["database"]
|
|
os.environ["DB_USER"] = PG_CONFIG["user"]
|
|
os.environ["DB_PASSWORD"] = PG_CONFIG["password"]
|
|
|
|
from sqlalchemy import create_engine
|
|
from app.database import Base
|
|
from app.models import (
|
|
CarMaker, CarModel, Car, CarImage, CarOption,
|
|
User, CarView, PerformanceCheckView, ChargeHistory, VerificationCode,
|
|
Inquiry, InquiryMessage, HeroBanner, HeroBannerSettings,
|
|
Translation, CarCache, CarDetailCache, CacheRequestQueue,
|
|
SystemSettings, VehicleRequest, RequestVehicle, PurchasedVehicle,
|
|
DealerApplication, DealerInfo, VehicleShare, ShareReward,
|
|
WithdrawalRequest, ReferralReward, Notification,
|
|
PushSubscription, UserNotificationPreference,
|
|
CarPerformanceCheck, CarSpecification,
|
|
ExchangeRate, ExchangeRateHistory, CCPackage,
|
|
VisitorLog, VisitorDailyStats, VisitorSession,
|
|
)
|
|
|
|
encoded_pw = quote_plus(PG_CONFIG['password'])
|
|
pg_url = f"postgresql://{PG_CONFIG['user']}:{encoded_pw}@{PG_CONFIG['host']}:{PG_CONFIG['port']}/{PG_CONFIG['database']}"
|
|
engine = create_engine(pg_url, echo=False)
|
|
Base.metadata.create_all(bind=engine)
|
|
print(" Tables created successfully!")
|
|
|
|
def get_boolean_columns(pg_cursor, table_name):
|
|
"""Get list of boolean columns for a table"""
|
|
pg_cursor.execute("""
|
|
SELECT column_name FROM information_schema.columns
|
|
WHERE table_schema = 'public' AND table_name = %s AND data_type = 'boolean'
|
|
""", (table_name,))
|
|
return [row[0] for row in pg_cursor.fetchall()]
|
|
|
|
def convert_row(row, columns, bool_cols):
|
|
"""Convert SQLite row values for PostgreSQL (handle booleans)"""
|
|
result = []
|
|
for i, val in enumerate(row):
|
|
col_name = columns[i]
|
|
if col_name in bool_cols and val is not None:
|
|
# Convert 0/1 to False/True
|
|
result.append(bool(val))
|
|
else:
|
|
result.append(val)
|
|
return tuple(result)
|
|
|
|
def migrate_table(sqlite_conn, pg_conn, table_name):
|
|
"""Migrate a single table"""
|
|
sqlite_cursor = sqlite_conn.cursor()
|
|
pg_cursor = pg_conn.cursor()
|
|
|
|
# Get SQLite columns
|
|
sqlite_cursor.execute(f"PRAGMA table_info({table_name})")
|
|
sqlite_cols = [col[1] for col in sqlite_cursor.fetchall()]
|
|
if not sqlite_cols:
|
|
return 0
|
|
|
|
# Get data
|
|
sqlite_cursor.execute(f"SELECT * FROM {table_name}")
|
|
rows = sqlite_cursor.fetchall()
|
|
if not rows:
|
|
print(f" {table_name}: 0 rows (empty)")
|
|
return 0
|
|
|
|
# Check PostgreSQL table exists
|
|
pg_cursor.execute("""
|
|
SELECT EXISTS (SELECT FROM information_schema.tables
|
|
WHERE table_schema = 'public' AND table_name = %s)
|
|
""", (table_name,))
|
|
if not pg_cursor.fetchone()[0]:
|
|
print(f" {table_name}: skipped (not in PostgreSQL)")
|
|
return 0
|
|
|
|
# Get PostgreSQL columns
|
|
pg_cursor.execute("""
|
|
SELECT column_name FROM information_schema.columns
|
|
WHERE table_schema = 'public' AND table_name = %s
|
|
ORDER BY ordinal_position
|
|
""", (table_name,))
|
|
pg_cols = [row[0] for row in pg_cursor.fetchall()]
|
|
|
|
# Find common columns
|
|
common_cols = [c for c in sqlite_cols if c in pg_cols]
|
|
if not common_cols:
|
|
print(f" {table_name}: skipped (no matching columns)")
|
|
return 0
|
|
|
|
col_indices = [sqlite_cols.index(c) for c in common_cols]
|
|
|
|
# Get boolean columns
|
|
bool_cols = set(get_boolean_columns(pg_cursor, table_name))
|
|
|
|
# Prepare query
|
|
cols_str = ", ".join(common_cols)
|
|
placeholders = ", ".join(["%s"] * len(common_cols))
|
|
insert_sql = f"INSERT INTO {table_name} ({cols_str}) VALUES ({placeholders})"
|
|
|
|
try:
|
|
# Truncate with CASCADE to handle FK
|
|
pg_cursor.execute(f"TRUNCATE TABLE {table_name} CASCADE")
|
|
|
|
success = 0
|
|
for row in rows:
|
|
try:
|
|
# Extract and convert row
|
|
filtered = tuple(row[i] for i in col_indices)
|
|
converted = convert_row(filtered, common_cols, bool_cols)
|
|
pg_cursor.execute(insert_sql, converted)
|
|
success += 1
|
|
except Exception as e:
|
|
pg_conn.rollback()
|
|
# Truncate again after rollback
|
|
pg_cursor.execute(f"TRUNCATE TABLE {table_name} CASCADE")
|
|
print(f" {table_name}: error - {str(e)[:80]}")
|
|
return 0
|
|
|
|
pg_conn.commit()
|
|
print(f" {table_name}: {success}/{len(rows)} rows migrated")
|
|
return success
|
|
except Exception as e:
|
|
pg_conn.rollback()
|
|
print(f" {table_name}: error - {str(e)[:80]}")
|
|
return 0
|
|
|
|
def reset_sequences(pg_conn):
|
|
"""Reset sequences to max(id) + 1"""
|
|
print("\n[Step 3] Resetting sequences...")
|
|
pg_cursor = pg_conn.cursor()
|
|
|
|
# Get all tables with id column
|
|
pg_cursor.execute("""
|
|
SELECT table_name FROM information_schema.columns
|
|
WHERE table_schema = 'public' AND column_name = 'id'
|
|
""")
|
|
tables = [row[0] for row in pg_cursor.fetchall()]
|
|
|
|
for table in tables:
|
|
try:
|
|
# Check if sequence exists
|
|
seq_name = f"{table}_id_seq"
|
|
pg_cursor.execute("""
|
|
SELECT EXISTS (SELECT FROM pg_sequences WHERE schemaname = 'public' AND sequencename = %s)
|
|
""", (seq_name,))
|
|
if pg_cursor.fetchone()[0]:
|
|
pg_cursor.execute(f"""
|
|
SELECT setval('{seq_name}', COALESCE((SELECT MAX(id) FROM {table}), 0) + 1, false)
|
|
""")
|
|
except:
|
|
pass
|
|
|
|
pg_conn.commit()
|
|
print(" Sequences reset completed")
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print("SQLite to PostgreSQL Migration")
|
|
print("=" * 60)
|
|
|
|
# Step 1: Create tables
|
|
try:
|
|
create_tables()
|
|
except Exception as e:
|
|
print(f" Failed: {e}")
|
|
sys.exit(1)
|
|
|
|
# Step 2: Connect and migrate
|
|
print("\n[Step 2] Migrating data...")
|
|
|
|
sqlite_conn = sqlite3.connect(SQLITE_PATH)
|
|
|
|
encoded_pw = quote_plus(PG_CONFIG['password'])
|
|
pg_conn = psycopg2.connect(
|
|
host=PG_CONFIG['host'],
|
|
port=PG_CONFIG['port'],
|
|
database=PG_CONFIG['database'],
|
|
user=PG_CONFIG['user'],
|
|
password=PG_CONFIG['password']
|
|
)
|
|
|
|
# Disable FK checks during migration
|
|
pg_cursor = pg_conn.cursor()
|
|
pg_cursor.execute("SET session_replication_role = 'replica';")
|
|
pg_conn.commit()
|
|
|
|
# Get SQLite tables
|
|
sqlite_cursor = sqlite_conn.cursor()
|
|
sqlite_cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name != 'sqlite_sequence'")
|
|
all_tables = set(t[0] for t in sqlite_cursor.fetchall())
|
|
|
|
# Migrate in order
|
|
total = 0
|
|
migrated_tables = set()
|
|
|
|
for table in TABLES_ORDER:
|
|
if table in all_tables:
|
|
total += migrate_table(sqlite_conn, pg_conn, table)
|
|
migrated_tables.add(table)
|
|
|
|
# Migrate remaining tables
|
|
remaining = all_tables - migrated_tables
|
|
for table in remaining:
|
|
total += migrate_table(sqlite_conn, pg_conn, table)
|
|
|
|
# Re-enable FK checks
|
|
pg_cursor.execute("SET session_replication_role = 'origin';")
|
|
pg_conn.commit()
|
|
|
|
# Step 3: Reset sequences
|
|
reset_sequences(pg_conn)
|
|
|
|
sqlite_conn.close()
|
|
pg_conn.close()
|
|
|
|
print("\n" + "=" * 60)
|
|
print(f"Migration completed! Total rows: {total}")
|
|
print("=" * 60)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|