Migrating Kodi to MariaDB, with locked-down permissions and no import/export

#!/usr/bin/env python
"""
Script for migrating Kodi SQLite databases to MySQL / MariaDB.

Requires Python >= 3.7 and PyMySQL.

For usage, see:
https://blog.jasonantman.com/2021/03/migrating-kodi-to-mariadb-with-locked-down-permissions-and-no-importexport/

NOTE: This currently ignores triggers, as all triggers that Kodi currently uses
in these DBs are AFTER delete.
"""

import sys
import argparse
import logging
from typing import Any, List, Tuple

import pymysql
import pymysql.cursors
import sqlite3

FORMAT: str = "[%(asctime)s %(levelname)s] %(message)s"
logging.basicConfig(level=logging.WARNING, format=FORMAT)
logger: logging.Logger = logging.getLogger()


class KodiMigrator:

    def __init__(
        self, host: str, port: int, user: str, passwd: str,
        dry_run: bool = False
    ):
        self.dry_run: bool = dry_run
        logger.debug(
            'Connecting to MySQL at %s:%s as %s', host, port, user
        )
        self.mysql: pymysql.connections.Connection = pymysql.connect(
            host=host,
            user=user,
            port=port,
            password=passwd,
            charset='utf8mb4',
            cursorclass=pymysql.cursors.DictCursor
        )
        logger.info('Connected to MySQL')

    def run(self):
        for dbname in ['MyMusic72', 'MyVideos116']:
            self.handle_db(dbname)

    def handle_db(self, dbname: str):
        logger.info('Handling database: %s', dbname)
        logger.debug('MySQL select DB %s', dbname)
        self.mysql.select_db(dbname)
        mysql_tables = self._mysql_table_names()
        logger.debug('SQLite3 open %s.db', dbname)
        local: sqlite3.Connection = sqlite3.connect(f'{dbname}.db')
        sqlite_tables = self._sqlite_table_names(local)
        local.row_factory = sqlite3.Row
        if mysql_tables != sqlite_tables:
            raise RuntimeError(
                f'ERROR: MySQL table names in {dbname} ({mysql_tables}) do '
                f'not match SQLite table names ({sqlite_tables})!'
            )
        with self.mysql.cursor() as cursor:
            cursor.execute('set FOREIGN_KEY_CHECKS = 0;')
            self._fix_db(dbname, cursor)
            for tname in mysql_tables:
                self._handle_table(local, cursor, tname)
            cursor.execute('set FOREIGN_KEY_CHECKS = 1;')
            self.mysql.commit()

    def _fix_db(self, dbname: str, cur: pymysql.cursors.DictCursor):
        if dbname != 'MyVideos116':
            return
        # for data conversion, change tvshow.c06 from TEXT to MEDIUMTEXT
        self._mysql_execute(
            cur, 'ALTER TABLE tvshow MODIFY COLUMN c06 MEDIUMTEXT;'
        )

    def _handle_table(
        self, local: sqlite3.Connection, cur: pymysql.cursors.DictCursor,
        table_name: str
    ):
        # first empty the MySQL table
        self._mysql_execute(cur, f'DELETE FROM {table_name};')
        # now get the data from SQLite and copy it over
        sql = f'SELECT * FROM {table_name};'
        logger.debug('SQLite execute: %s', sql)
        local_cur: sqlite3.Cursor = local.cursor()
        local_cur.execute(sql)
        count: int = 0
        for row in local_cur:
            if table_name == 'tvshow' and row['idShow'] == 3:  # Doctor Who - c06 too long
                continue
            tmp = self._mysql_insert_for_row(table_name, row)
            self._mysql_execute(cur, *tmp)
            count += 1
        logger.info('Copied %d rows for table %s', count, table_name)

    def _mysql_insert_for_row(
        self, tname: str, row: sqlite3.Row
    ) -> Tuple[str, Tuple]:
        keys: List[str] = []
        values: List[str] = []
        for k, v in dict(row).items():
            keys.append(k)
            values.append(v)
        sql = f'INSERT INTO {tname} ({", ".join(keys)}) VALUES (' \
              f'{", ".join(["%s" for x in values])});'
        return sql, tuple(values)

    def _mysql_execute(
        self, cur: pymysql.cursors.Cursor, sql: str, data: tuple = ()
    ):
        if self.dry_run:
            logger.warning('DRY RUN: MySQL Execute: %s %s', sql, data)
        else:
            logger.debug('MySQL Execute: %s %s', sql, data)
            try:
                cur.execute(sql, data)
            except pymysql.err.MySQLError:
                logger.error('Error in query: %s %s', sql, data)
                raise

    def _sqlite_table_names(self, db: sqlite3.Connection) -> List[str]:
        cursor: sqlite3.Cursor = db.cursor()
        sql: str = "SELECT name FROM sqlite_master WHERE type='table';"
        logger.debug('SQLite execute: %s', sql)
        cursor.execute(sql)
        res: List[str] = sorted([x[0] for x in cursor.fetchall()])
        logger.debug('SQLite tables: %s', res)
        return res

    def _mysql_table_names(self) -> List[str]:
        sql: str = "show full tables where Table_Type != 'VIEW'"
        logger.debug('MySQL execute: %s', sql)
        with self.mysql.cursor(cursor=pymysql.cursors.Cursor) as cursor:
            cursor.execute(sql)
            res: List[str] = sorted([x[0] for x in cursor.fetchall()])
        logger.debug('MySQL tables: %s', res)
        return res


def parse_args(argv):
    """
    parse arguments/options

    this uses the new argparse module instead of optparse
    see: <https://docs.python.org/2/library/argparse.html>
    """
    p = argparse.ArgumentParser(description='Migrate Kodi SQLite to MySQL')
    p.add_argument('-d', '--dry-run', dest='dry_run', action='store_true',
                   default=False,
                   help="dry-run - print SQL that would be sent to MySQL but "
                        "don't actually run it")
    p.add_argument('-v', '--verbose', dest='verbose', action='store_true',
                   help='debug-level output.', default=False)
    p.add_argument('-H', '--host', type=str, dest='host',
                   help='MySQL host (default: 127.0.0.1)', default='127.0.0.1')
    p.add_argument('-P', '--port', type=int, dest='port', default=3306,
                   help='MySQL port (default: 3306)')
    p.add_argument('-u', '--user', type=str, dest='user', default='kodi',
                   help='MySQL user (default: kodi)')
    p.add_argument('-p', '--passwd', type=str, dest='passwd', default='kodi',
                   help='MySQL password (default: kodi)')
    args = p.parse_args(argv)
    return args


def set_log_info():
    """set logger level to INFO"""
    set_log_level_format(logging.INFO,
                         '%(asctime)s %(levelname)s:%(name)s:%(message)s')


def set_log_debug():
    """set logger level to DEBUG, and debug-level output format"""
    set_log_level_format(
        logging.DEBUG,
        "%(asctime)s [%(levelname)s %(filename)s:%(lineno)s - "
        "%(name)s.%(funcName)s() ] %(message)s"
    )


def set_log_level_format(level, format):
    """
    Set logger level and format.

    :param level: logging level; see the :py:mod:`logging` constants.
    :type level: int
    :param format: logging formatter format string
    :type format: str
    """
    formatter = logging.Formatter(fmt=format)
    logger.handlers[0].setFormatter(formatter)
    logger.setLevel(level)


if __name__ == "__main__":
    args = parse_args(sys.argv[1:])

    # set logging level
    if args.verbose:
        set_log_debug()
    else:
        set_log_info()
    KodiMigrator(
        args.host, args.port, args.user, args.passwd, dry_run=args.dry_run
    ).run()