#!/usr/bin/env python3
"""
sync_mirror.py - one-way full-reload mirror of a SQL Server database into MariaDB.

Run on the cPanel server (cron). Pulls from SQL Server with a read-only login and
rebuilds every table in the target MariaDB database. Each table is loaded into a
"<name>__new" table and swapped in with an atomic RENAME only after the row count
matches, so readers never see a half-loaded table and a failed table keeps its
previous copy.

Requires: python3, pyodbc, pymysql, Microsoft ODBC Driver 18 for SQL Server.

Usage:
  python3 sync_mirror.py --config /root/sync/sync_mirror.ini
  python3 sync_mirror.py --config ... --tables Customers,Orders
  python3 sync_mirror.py --config ... --exclude AuditLog,SessionData
  python3 sync_mirror.py --config ... --schema-only      # create empty tables, no data

Notes:
  * Tables are created with lowercase names, utf8mb4 / utf8mb4_unicode_ci
    (case-insensitive, like SQL Server's default collation).
  * Only the primary key is recreated. Add secondary indexes later if you query
    the mirror heavily; the mirror is reference data for the Stage 2 ETL.
  * Tables in a schema other than dbo are named <schema>__<table>.
"""
import argparse
import configparser
import fcntl
import logging
import sys
import time

import pymysql
import pyodbc

BATCH = 1000
MAX_ROW_BYTES = 60000          # MariaDB row limit is 65535; stay under it
VARCHAR_MAX_CHARS = 16000      # above this, use longtext
log = logging.getLogger("sync_mirror")


# ----------------------------------------------------------------- type mapping
def map_column(c):
    """Return (maria_type, select_expr_template, est_row_bytes) for a column.

    c: dict with name, type, max_length (bytes), precision, scale
    select_expr_template uses {col} as the quoted column placeholder.
    """
    t = c["type"].lower()
    ml, p, s = c["max_length"], c["precision"], c["scale"]
    ident = "{col}"

    if t in ("char", "varchar", "nchar", "nvarchar"):
        chars = ml if ml == -1 else (ml // 2 if t.startswith("n") else ml)
        if chars == -1 or chars > VARCHAR_MAX_CHARS:
            return "longtext", ident, 0
        if t in ("char", "nchar") and chars <= 255:
            return f"char({chars})", ident, chars * 4
        return f"varchar({chars})", ident, chars * 4 + 2
    if t in ("text", "ntext"):
        return "longtext", ident, 0
    if t == "int":
        return "int", ident, 4
    if t == "bigint":
        return "bigint", ident, 8
    if t == "smallint":
        return "smallint", ident, 2
    if t == "tinyint":
        return "tinyint unsigned", ident, 1
    if t == "bit":
        return "tinyint(1)", ident, 1
    if t in ("decimal", "numeric"):
        return f"decimal({p},{s})", ident, 16
    if t == "money":
        return "decimal(19,4)", ident, 16
    if t == "smallmoney":
        return "decimal(10,4)", ident, 16
    if t == "float":
        return "double", ident, 8
    if t == "real":
        return "float", ident, 4
    if t in ("datetime", "smalldatetime"):
        return "datetime(3)" if t == "datetime" else "datetime", ident, 8
    if t == "datetime2":
        return f"datetime({min(s, 6)})", ident, 8
    if t == "date":
        return "date", ident, 3
    if t == "time":
        return f"time({min(s, 6)})", ident, 6
    if t == "datetimeoffset":
        return "varchar(40)", "CONVERT(varchar(34), {col}, 127)", 162
    if t == "uniqueidentifier":
        return "char(36)", "CONVERT(char(36), {col})", 144
    if t in ("binary", "varbinary"):
        if ml == -1 or ml > 8000:
            return "longblob", ident, 0
        return (f"binary({ml})" if t == "binary" else f"varbinary({ml})"), ident, ml + 2
    if t in ("image",):
        return "longblob", ident, 0
    if t in ("timestamp", "rowversion"):
        return "binary(8)", ident, 8
    # xml, sql_variant, hierarchyid, geography, geometry, anything unknown
    return "longtext", "CONVERT(nvarchar(max), {col})", 0


def shrink_to_fit(cols):
    """Convert the widest varchar columns to longtext until the row fits."""
    total = sum(c["bytes"] for c in cols)
    while total > MAX_ROW_BYTES:
        cand = [c for c in cols if c["maria"].startswith(("varchar", "char")) and c["bytes"] > 0]
        if not cand:
            break
        w = max(cand, key=lambda c: c["bytes"])
        total -= w["bytes"]
        w["maria"], w["bytes"] = "longtext", 0
        log.info("  row too wide; %s converted to longtext", w["name"])
    return cols


# ----------------------------------------------------------------- helpers
def q(name):
    return "`" + name.replace("`", "``") + "`"


def mssql_connect(cfg):
    s = cfg["mssql"]
    conn_str = (
        f"DRIVER={{{s.get('driver', 'ODBC Driver 18 for SQL Server')}}};"
        f"SERVER={s['server']};DATABASE={s['database']};"
        f"UID={s['user']};PWD={s['password']};"
        f"Encrypt={s.get('encrypt', 'yes')};"
        f"TrustServerCertificate={s.get('trust_server_certificate', 'yes')};"
        "ApplicationIntent=ReadOnly;"
    )
    conn = pyodbc.connect(conn_str, timeout=30)
    conn.setdecoding(pyodbc.SQL_CHAR, encoding=s.get("char_encoding", "utf-8"))
    conn.setdecoding(pyodbc.SQL_WCHAR, encoding="utf-16le")
    conn.setencoding(encoding="utf-8")
    if s.getboolean("snapshot", fallback=False):
        # requires: ALTER DATABASE <db> SET ALLOW_SNAPSHOT_ISOLATION ON
        conn.execute("SET TRANSACTION ISOLATION LEVEL SNAPSHOT")
    return conn


def maria_connect(cfg):
    m = cfg["mariadb"]
    return pymysql.connect(
        host=m.get("host", "localhost"), port=m.getint("port", 3306),
        user=m["user"], password=m["password"], database=m["database"],
        charset="utf8mb4", autocommit=True, local_infile=False,
    )


def list_tables(ms):
    rows = ms.cursor().execute(
        "SELECT s.name, t.name FROM sys.tables t JOIN sys.schemas s ON s.schema_id=t.schema_id "
        "WHERE t.is_ms_shipped = 0 ORDER BY s.name, t.name"
    ).fetchall()
    return [(r[0], r[1]) for r in rows]


def table_columns(ms, schema, table):
    full = f"[{schema}].[{table}]"
    rows = ms.cursor().execute(
        "SELECT c.name, ty.name, c.max_length, c.precision, c.scale, c.is_nullable "
        "FROM sys.columns c "
        "JOIN sys.types ty ON ty.system_type_id = c.system_type_id AND ty.user_type_id = ty.system_type_id "
        "WHERE c.object_id = OBJECT_ID(?) ORDER BY c.column_id", full).fetchall()
    cols = []
    for name, tname, ml, p, s, nullable in rows:
        mtype, expr, nbytes = map_column(
            {"name": name, "type": tname, "max_length": ml, "precision": p, "scale": s})
        cols.append({"name": name, "maria": mtype, "expr": expr, "bytes": nbytes,
                     "nullable": bool(nullable), "type": tname})
    return shrink_to_fit(cols)


def primary_key(ms, schema, table):
    full = f"[{schema}].[{table}]"
    rows = ms.cursor().execute(
        "SELECT c.name FROM sys.indexes i "
        "JOIN sys.index_columns ic ON ic.object_id=i.object_id AND ic.index_id=i.index_id "
        "JOIN sys.columns c ON c.object_id=ic.object_id AND c.column_id=ic.column_id "
        "WHERE i.is_primary_key=1 AND i.object_id=OBJECT_ID(?) ORDER BY ic.key_ordinal", full).fetchall()
    return [r[0] for r in rows]


def maria_name(schema, table):
    return (table if schema.lower() == "dbo" else f"{schema}__{table}").lower()


def table_exists(my, name):
    cur = my.cursor()
    cur.execute("SELECT 1 FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name=%s", (name,))
    return cur.fetchone() is not None


def create_table(my, name, cols, pk):
    defs = [f"{q(c['name'].lower())} {c['maria']} {'NULL' if c['nullable'] else 'NOT NULL'}" for c in cols]
    base = f"CREATE TABLE {q(name)} (\n  " + ",\n  ".join(defs)
    tail = ") ENGINE=InnoDB ROW_FORMAT=DYNAMIC DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci"
    my.cursor().execute(f"DROP TABLE IF EXISTS {q(name)}")
    if pk:
        pk_sql = ",\n  PRIMARY KEY (" + ",".join(q(p.lower()) for p in pk) + ")"
        try:
            my.cursor().execute(base + pk_sql + "\n" + tail)
            return
        except pymysql.MySQLError as e:
            log.warning("  PK not created (%s); creating without it", e)
            my.cursor().execute(f"DROP TABLE IF EXISTS {q(name)}")
    my.cursor().execute(base + "\n" + tail)


# ----------------------------------------------------------------- per-table sync
def sync_table(ms, my, schema, table, schema_only=False):
    name = maria_name(schema, table)
    tmp = name + "__new"
    cols = table_columns(ms, schema, table)
    pk = primary_key(ms, schema, table)
    create_table(my, tmp, cols, pk)

    read_count = 0
    if not schema_only:
        select_list = ", ".join(c["expr"].format(col=f"[{c['name']}]") + f" AS [{c['name']}]" for c in cols)
        src = ms.cursor()
        src.execute(f"SELECT {select_list} FROM [{schema}].[{table}]")
        insert = (f"INSERT INTO {q(tmp)} (" + ",".join(q(c["name"].lower()) for c in cols) +
                  ") VALUES (" + ",".join(["%s"] * len(cols)) + ")")
        dst = my.cursor()
        while True:
            rows = src.fetchmany(BATCH)
            if not rows:
                break
            dst.executemany(insert, [tuple(r) for r in rows])
            read_count += len(rows)

        cur = my.cursor()
        cur.execute(f"SELECT COUNT(*) FROM {q(tmp)}")
        loaded = cur.fetchone()[0]
        if loaded != read_count:
            raise RuntimeError(f"row count mismatch: read {read_count}, loaded {loaded}")

    old = name + "__old"
    cur = my.cursor()
    cur.execute(f"DROP TABLE IF EXISTS {q(old)}")
    if table_exists(my, name):
        cur.execute(f"RENAME TABLE {q(name)} TO {q(old)}, {q(tmp)} TO {q(name)}")
        cur.execute(f"DROP TABLE IF EXISTS {q(old)}")
    else:
        cur.execute(f"RENAME TABLE {q(tmp)} TO {q(name)}")
    return read_count


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", required=True)
    ap.add_argument("--tables", help="comma-separated table names to sync (default: all)")
    ap.add_argument("--exclude", help="comma-separated table names to skip")
    ap.add_argument("--schema-only", action="store_true")
    ap.add_argument("--lock", default="/tmp/sync_mirror.lock")
    a = ap.parse_args()

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    cfg = configparser.ConfigParser()
    cfg.read(a.config)

    lock = open(a.lock, "w")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        log.error("another sync is already running; exiting")
        return 2

    only = {t.strip().lower() for t in a.tables.split(",")} if a.tables else None
    skip = {t.strip().lower() for t in a.exclude.split(",")} if a.exclude else set()

    ms, my = mssql_connect(cfg), maria_connect(cfg)
    my.cursor().execute("SET SESSION foreign_key_checks=0, unique_checks=0, sql_mode='NO_AUTO_VALUE_ON_ZERO'")

    tables = [(s, t) for s, t in list_tables(ms)
              if (only is None or t.lower() in only) and t.lower() not in skip]
    log.info("syncing %d tables", len(tables))
    failures, total_rows, start = [], 0, time.time()

    for schema, table in tables:
        t0 = time.time()
        try:
            n = sync_table(ms, my, schema, table, a.schema_only)
            total_rows += n
            log.info("OK   %s.%s  %d rows  %.1fs", schema, table, n, time.time() - t0)
        except Exception as e:  # keep going; the previous copy of this table stays in place
            failures.append((schema, table, str(e)))
            log.error("FAIL %s.%s  %s", schema, table, e)
            try:
                my.cursor().execute(f"DROP TABLE IF EXISTS {q(maria_name(schema, table) + '__new')}")
            except Exception:
                pass

    log.info("done: %d tables, %d rows, %d failures, %.1fs",
             len(tables) - len(failures), total_rows, len(failures), time.time() - start)
    for s, t, e in failures:
        log.error("  failed: %s.%s: %s", s, t, e)
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())