122 lines
3.5 KiB
Python
122 lines
3.5 KiB
Python
"""
|
|
migration_photo_verification_toggle.py
|
|
=======================================
|
|
Adds `photo_verification_enabled` column to the `qr_codes` table.
|
|
|
|
Uses pymysql directly to avoid SQLAlchemy ORM loading the model
|
|
(which would fail if the column doesn't exist yet).
|
|
|
|
Default: 1 (True) for all existing rows — preserves current behaviour.
|
|
|
|
Run once on each server (LT and GOV):
|
|
python3 tools/migration_photo_verification_toggle.py
|
|
|
|
Safe to re-run — skips if column already exists.
|
|
"""
|
|
|
|
import os, sys, re
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
|
|
import pymysql
|
|
|
|
TABLE = 'qr_codes'
|
|
COLUMN = 'photo_verification_enabled'
|
|
|
|
|
|
def parse_db_url(url):
|
|
"""
|
|
Parse DATABASE_URL robustly using regex to handle special characters
|
|
(including @ or : ) in the password.
|
|
|
|
Supports:
|
|
mysql+pymysql://user:pass@host:port/dbname
|
|
mysql+pymysql://user:pass@host/dbname
|
|
"""
|
|
# Strip driver prefix
|
|
url = re.sub(r'^mysql\+pymysql://', '', url)
|
|
url = re.sub(r'^mysql://', '', url)
|
|
|
|
# Split credentials from host/db on the LAST @ before the host
|
|
# Pattern: user:password@host[:port]/dbname[?...]
|
|
m = re.match(
|
|
r'^(?P<user>[^:]+):(?P<password>.+)@(?P<host>[^@:/]+)(?::(?P<port>\d+))?/(?P<db>[^?]+)',
|
|
url
|
|
)
|
|
if not m:
|
|
print(f"[ERROR] Could not parse DATABASE_URL. Raw (redacted): {url[:30]}...")
|
|
sys.exit(1)
|
|
|
|
return {
|
|
'host': m.group('host'),
|
|
'port': int(m.group('port')) if m.group('port') else 3306,
|
|
'user': m.group('user'),
|
|
'password': m.group('password'),
|
|
'database': m.group('db'),
|
|
}
|
|
|
|
|
|
def get_connection():
|
|
db_url = os.environ.get('DATABASE_URL', '')
|
|
if not db_url:
|
|
print("[ERROR] DATABASE_URL not set in .env")
|
|
sys.exit(1)
|
|
|
|
params = parse_db_url(db_url)
|
|
return pymysql.connect(
|
|
host=params['host'],
|
|
port=params['port'],
|
|
user=params['user'],
|
|
password=params['password'],
|
|
database=params['database'],
|
|
charset='utf8mb4',
|
|
autocommit=False,
|
|
)
|
|
|
|
|
|
def run():
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
# Check if column already exists
|
|
cur.execute(
|
|
"SELECT COUNT(*) FROM information_schema.COLUMNS "
|
|
"WHERE TABLE_SCHEMA = DATABASE() "
|
|
f"AND TABLE_NAME = '{TABLE}' "
|
|
f"AND COLUMN_NAME = '{COLUMN}'"
|
|
)
|
|
exists = cur.fetchone()[0] > 0
|
|
|
|
if exists:
|
|
print(f"[SKIP] Column '{COLUMN}' already exists on '{TABLE}'. Nothing to do.")
|
|
return
|
|
|
|
print(f"[ADD] Adding column '{COLUMN}' to '{TABLE}' ...")
|
|
cur.execute(
|
|
f"ALTER TABLE `{TABLE}` "
|
|
f"ADD COLUMN `{COLUMN}` TINYINT(1) NOT NULL DEFAULT 1"
|
|
)
|
|
conn.commit()
|
|
|
|
# Verify
|
|
cur.execute(
|
|
"SELECT COUNT(*) FROM information_schema.COLUMNS "
|
|
"WHERE TABLE_SCHEMA = DATABASE() "
|
|
f"AND TABLE_NAME = '{TABLE}' "
|
|
f"AND COLUMN_NAME = '{COLUMN}'"
|
|
)
|
|
if cur.fetchone()[0] > 0:
|
|
print(f"[OK] Column '{COLUMN}' added. All existing rows default to 1 (enabled).")
|
|
else:
|
|
print(f"[FAIL] Column was not created — check DB permissions.")
|
|
sys.exit(1)
|
|
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run() |