"""Convert this project's PostgreSQL backup without executing it or dropping data. Strings are parsed, not regex-rewritten. Output is for an EMPTY MySQL 8.0.13+ or MariaDB 10.6+ database. The source backup is never modified. """ import argparse import datetime as dt import hashlib import json import re from pathlib import Path def statements(text): start = i = 0 quote = None while i < len(text): c = text[i] if quote: if c == quote: if i + 1 < len(text) and text[i + 1] == quote: i += 2 continue quote = None elif text.startswith('--', i): end = text.find('\n', i) if end < 0: break if not text[start:i].strip(): start = end + 1 i = end elif c in "'\"": quote = c elif c == ';': statement = text[start:i].strip() if statement: yield statement start = i + 1 i += 1 if quote or text[start:].strip().strip('-').strip(): tail = text[start:].strip() if tail and not tail.startswith('--'): raise ValueError('Unterminated SQL statement') def values(text): result = [] i = 0 while i < len(text): while i < len(text) and text[i] in ' \r\n\t,': i += 1 if i == len(text): break if text[i] == "'": i += 1 value = [] while i < len(text): if text[i] == "'": if i + 1 < len(text) and text[i + 1] == "'": value.append("'") i += 2 continue i += 1 break value.append(text[i]) i += 1 else: raise ValueError('Unterminated value') result.append(('string', ''.join(value))) else: end = text.find(',', i) end = len(text) if end < 0 else end raw = text[i:end].strip() if raw.upper() == 'NULL': result.append(('null', None)) elif raw in ('true', 'false'): result.append(('number', '1' if raw == 'true' else '0')) elif re.fullmatch(r'[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?', raw): result.append(('number', raw)) else: raise ValueError(f'Unsupported SQL value syntax: {raw[:30]}') i = end return result UNIQUE = { 'users': [('email',), ('google_id',), ('account_id',)], 'wallets': [('user_id',)], 'api_keys': [('key_hash',)], 'permissions': [('name',)], 'sms_providers': [('name',)], 'sms_numbers': [('number',)], 'telegram_sessions': [('telegram_id',)], 'boost_services': [('provider', 'provider_service_id')], 'products': [('seed_key',)], } KEYS = {'app_settings': ('key',), 'schema_patches': ('name',), 'role_permissions': ('role', 'permission_id')} def convert(source, destination): raw = source.read_bytes() tables = {} for statement in statements(raw.decode('utf-8-sig')): if statement.startswith('DROP TABLE IF EXISTS'): continue # Never emit the source's destructive DROP commands. create = re.fullmatch(r'CREATE TABLE "(\w+)"\s*\((.*)\)', statement, re.S) if create: name, body = create.groups() columns = [] for line in body.strip().splitlines(): column = re.fullmatch(r'\s*"(\w+)"\s+(.+?),?\s*', line) if not column: raise ValueError(f'Unrecognized column in {name}') col, definition = column.groups() definition = definition.rstrip(',') columns.append({'name': col, 'definition': definition}) tables[name] = {'columns': columns, 'rows': []} continue insert = re.fullmatch(r'INSERT INTO "(\w+)" \((.*?)\) VALUES \((.*)\)', statement, re.S) if not insert: raise ValueError(f'Unsupported statement: {statement[:60]}') name, cols, data = insert.groups() names = re.findall(r'"(\w+)"', cols) parsed = values(data) if len(parsed) != len(names) or name not in tables: raise ValueError(f'Invalid row in {name}') if names != [c['name'] for c in tables[name]['columns']]: raise ValueError(f'Unexpected column order in {name}') tables[name]['rows'].append(parsed) output = [ '-- Converted Dublogs backup. Import into an EMPTY database only.', '-- Source SHA256: ' + hashlib.sha256(raw).hexdigest(), 'SET NAMES utf8mb4;', "SET time_zone = '+00:00';", "SET SESSION sql_mode = 'STRICT_ALL_TABLES,NO_AUTO_VALUE_ON_ZERO,NO_ENGINE_SUBSTITUTION';", ] schema = [] manifest = {'source_sha256': hashlib.sha256(raw).hexdigest(), 'tables': {}} for name, table in tables.items(): columns = table['columns'] primary = KEYS.get(name, ('id',)) uniques = UNIQUE.get(name, []) indexed = set(primary) | {c for key in uniques for c in key} definitions = [] for col in columns: definition = col['definition'] colname = col['name'] pgtype = re.split(r' NOT NULL| DEFAULT ', definition)[0] col['type'] = pgtype sqltype = {'integer': 'INT', 'bigint': 'BIGINT', 'smallint': 'SMALLINT', 'double precision': 'DOUBLE', 'real': 'FLOAT', 'timestamp with time zone': 'DATETIME(6)', 'jsonb': 'LONGTEXT', 'text': 'VARCHAR(255)' if colname in indexed else 'LONGTEXT'}.get(pgtype) if not sqltype: raise ValueError(f'Unknown type {pgtype}') if sqltype == 'VARCHAR(255)': index = [c['name'] for c in columns].index(colname) if any(v[index][0] == 'string' and len(v[index][1]) > 255 for v in table['rows']): raise ValueError(f'Indexed value exceeds 255 characters: {name}.{colname}') nullable = ' NOT NULL' if 'NOT NULL' in definition else ' NULL' default = '' if 'nextval(' in definition: default = ' AUTO_INCREMENT' elif ' DEFAULT ' in definition: expression = definition.split(' DEFAULT ', 1)[1] expression = re.sub(r'::(?:text|jsonb)$', '', expression) if expression == 'now()': default = ' DEFAULT CURRENT_TIMESTAMP(6)' else: value = values(expression)[0] literal = sql_value(value, pgtype) default = f' DEFAULT ({literal})' if sqltype == 'LONGTEXT' else f' DEFAULT {literal}' definitions.append(f' `{colname}` {sqltype}{nullable}{default}') if pgtype == 'jsonb': definitions.append(f' CHECK (JSON_VALID(`{colname}`))') definitions.append(' PRIMARY KEY (' + ','.join(f'`{c}`' for c in primary) + ')') for key in uniques: definitions.append(' UNIQUE KEY `uq_' + '_'.join(key) + '` (' + ','.join(f'`{c}`' for c in key) + ')') for col in columns: if col['name'].endswith('_id') and col['name'] not in indexed: definitions.append(f" KEY `idx_{col['name']}` (`{col['name']}`)") ddl = f'CREATE TABLE `{name}` (\n' + ',\n'.join(definitions) + '\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;' output.append(ddl) schema.append(ddl) # Detect duplicates before writing an artifact that would fail half way. names = [c['name'] for c in columns] for key in [primary, *uniques]: seen = set() for row in table['rows']: val = tuple(row[names.index(c)][1] for c in key) if None in val: continue if val in seen: raise ValueError(f'Duplicate key in {name}: {key}; no output written') seen.add(val) digest = hashlib.sha256() output.append('START TRANSACTION;') for row in table['rows']: converted = [sql_value(v, c['type']) for v, c in zip(row, columns)] output.append(f'INSERT INTO `{name}` (' + ','.join(f'`{c}`' for c in names) + ') VALUES (' + ','.join(converted) + ');') digest.update(json.dumps(row, ensure_ascii=False, separators=(',', ':')).encode()) output.append('COMMIT;') manifest['tables'][name] = {'rows': len(table['rows']), 'columns': names, 'data_sha256': digest.hexdigest()} destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text('\n'.join(output) + '\n', encoding='utf-8') destination.with_suffix('.schema.sql').write_text('SET NAMES utf8mb4;\n' + '\n'.join(schema) + '\n', encoding='utf-8') manifest['converted_sha256'] = hashlib.sha256(destination.read_bytes()).hexdigest() destination.with_suffix('.manifest.json').write_text(json.dumps(manifest, indent=2) + '\n', encoding='utf-8') print(f'Converted {len(tables)} tables, {sum(len(t["rows"]) for t in tables.values())} rows. No source data changed.') def sql_value(value, pgtype): kind, raw = value if kind == 'null': return 'NULL' if kind == 'number': return raw if pgtype == 'timestamp with time zone': parsed = dt.datetime.fromisoformat(raw.replace('Z', '+00:00')) if parsed.tzinfo is None: raise ValueError('Timestamp lacks timezone') raw = parsed.astimezone(dt.timezone.utc).strftime('%Y-%m-%d %H:%M:%S.%f') return "'" + raw + "'" if pgtype == 'jsonb': json.loads(raw) return "CONVERT(X'" + raw.encode('utf-8').hex() + "' USING utf8mb4)" if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('source', type=Path) parser.add_argument('destination', type=Path) args = parser.parse_args() convert(args.source, args.destination)