News Emoji's Bug - Printable Version

Code
#!/usr/bin/env python3



import argparse

import logging

import os

import sys

import time

import pymysql



LIVE_DB = "NEWDB"

OLD_DB = "OLDDB"

TABLE = "mybb_posts"



MYSQL_SOCKET = "/run/mysqld/mysqld.sock"



LOG_FILE = "/extra/post-repair.log"

STATE_FILE = "/extra/post-repair.state"



DEFAULT_BATCH = 10000

DEFAULT_SLEEP = 0.25





def connect():

    db = pymysql.connect(

        unix_socket=MYSQL_SOCKET,

        user="root",

        charset="utf8mb4",

        autocommit=False,

        cursorclass=pymysql.cursors.DictCursor

    )



    with db.cursor() as cur:

        cur.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")

        cur.execute("SET SESSION innodb_lock_wait_timeout=5")



    db.commit()

    return db





def setup_logging():

    logging.basicConfig(

        level=logging.INFO,

        format="%(asctime)s %(message)s",

        handlers=[

            logging.FileHandler(LOG_FILE),

            logging.StreamHandler(sys.stdout)

        ]

    )





def get_state():

    try:

        with open(STATE_FILE, "r") as f:

            return int(f.read().strip())

    except (FileNotFoundError, ValueError):

        return 0





def save_state(pid):

    tmp = STATE_FILE + ".tmp"



    with open(tmp, "w") as f:

        f.write(str(pid))



    os.replace(tmp, STATE_FILE)





def single_pid(db, pid, dry_run=False):

    sql = f"""

        SELECT

            o.pid,

            o.edittime AS old_edittime,

            n.edittime AS new_edittime,

            OCTET_LENGTH(o.message) AS old_bytes,

            OCTET_LENGTH(n.message) AS new_bytes,

            SHA2(o.message, 256) AS old_hash,

            SHA2(n.message, 256) AS new_hash,

            BINARY o.message = BINARY n.message AS message_same

        FROM {OLD_DB}.{TABLE} o

        JOIN {LIVE_DB}.{TABLE} n ON n.pid = o.pid

        WHERE o.pid = %s

    """



    with db.cursor() as cur:

        cur.execute(sql, (pid,))

        row = cur.fetchone()



    if not row:

        print(f"PID {pid}: not present in both databases.")

        return



    print(f"PID:           {pid}")

    print(f"Edit time:     old={row['old_edittime']} new={row['new_edittime']}")

    print(f"Message bytes: old={row['old_bytes']} new={row['new_bytes']}")

    print(f"Old SHA256:    {row['old_hash']}")

    print(f"New SHA256:    {row['new_hash']}")

    print(f"Message same:  {'YES' if row['message_same'] else 'NO'}")



    if row["message_same"]:

        print("No repair needed.")

        return



    if row["old_edittime"] != row["new_edittime"]:

        print("NOT UPDATED: edittime differs.")

        return



    if dry_run:

        print("DRY RUN: would replace live message with old message.")

        return



    update = f"""

        UPDATE {LIVE_DB}.{TABLE} n

        JOIN {OLD_DB}.{TABLE} o ON o.pid = n.pid

        SET n.message = o.message

        WHERE n.pid = %s

          AND n.edittime = o.edittime

          AND BINARY n.message <> BINARY o.message

    """



    with db.cursor() as cur:

        cur.execute(update, (pid,))

        changed = cur.rowcount



    db.commit()



    if changed == 1:

        print("UPDATED successfully.")

        logging.info("REPAIRED PID %d", pid)

    else:

        print("NOT UPDATED: row no longer meets repair conditions.")





def get_pid_range(db, last_pid, batch_size):

    sql = f"""

        SELECT pid

        FROM {OLD_DB}.{TABLE}

        WHERE pid > %s

        ORDER BY pid

        LIMIT %s

    """



    with db.cursor() as cur:

        cur.execute(sql, (last_pid, batch_size))

        rows = cur.fetchall()



    if not rows:

        return None



    return rows[0]["pid"], rows[-1]["pid"], len(rows)





def find_candidates(db, start_pid, end_pid):

    sql = f"""

        SELECT o.pid

        FROM {OLD_DB}.{TABLE} o

        JOIN {LIVE_DB}.{TABLE} n ON n.pid = o.pid

        WHERE o.pid BETWEEN %s AND %s

          AND n.edittime = o.edittime

          AND BINARY n.message <> BINARY o.message

        ORDER BY o.pid

    """



    with db.cursor() as cur:

        cur.execute(sql, (start_pid, end_pid))

        return [row["pid"] for row in cur.fetchall()]





def repair_range(db, start_pid, end_pid):

    sql = f"""

        UPDATE {LIVE_DB}.{TABLE} n

        JOIN {OLD_DB}.{TABLE} o ON o.pid = n.pid

        SET n.message = o.message

        WHERE o.pid BETWEEN %s AND %s

          AND n.edittime = o.edittime

          AND BINARY n.message <> BINARY o.message

    """



    with db.cursor() as cur:

        cur.execute(sql, (start_pid, end_pid))

        return cur.rowcount





def full_run(db, batch_size, sleep_time, dry_run, resume, start_pid):

    if start_pid is not None:

        last_pid = start_pid

    elif resume:

        last_pid = get_state()

    else:

        last_pid = 0



    scanned = 0

    repaired = 0

    batches = 0



    logging.info(

        "START live=%s old=%s batch=%d sleep=%.2f start_pid=%d dry_run=%s",

        LIVE_DB, OLD_DB, batch_size, sleep_time, last_pid, dry_run

    )



    while True:

        pid_range = get_pid_range(db, last_pid, batch_size)



        if not pid_range:

            break



        start, end, count = pid_range

        candidates = find_candidates(db, start, end)



        scanned += count

        batches += 1



        if candidates:

            logging.info(

                "RANGE %d-%d scanned=%d candidates=%d",

                start, end, count, len(candidates)

            )



            if dry_run:

                for pid in candidates:

                    logging.info("DRY-RUN PID %d", pid)

            else:

                changed = repair_range(db, start, end)

                db.commit()



                repaired += changed



                logging.info(

                    "UPDATED range=%d-%d candidates=%d changed=%d",

                    start, end, len(candidates), changed

                )

        else:

            db.rollback()



        last_pid = end

        save_state(last_pid)



        if batches % 10 == 0:

            logging.info(

                "PROGRESS last_pid=%d scanned=%d repaired=%d",

                last_pid, scanned, repaired

            )



        if sleep_time > 0:

            time.sleep(sleep_time)



    logging.info(

        "FINISHED last_pid=%d scanned=%d repaired=%d",

        last_pid, scanned, repaired

    )





def main():

    parser = argparse.ArgumentParser(

        description="Repair HF post messages from old database."

    )



    parser.add_argument(

        "-p", "--pid",

        type=int,

        help="Check or repair a single PID, e.g. -p123456"

    )



    parser.add_argument(

        "--dry-run",

        action="store_true",

        help="Compare only; do not update."

    )



    parser.add_argument(

        "--batch",

        type=int,

        default=DEFAULT_BATCH,

        help=f"Rows per batch, default {DEFAULT_BATCH}"

    )



    parser.add_argument(

        "--sleep",

        type=float,

        default=DEFAULT_SLEEP,

        help=f"Sleep between batches, default {DEFAULT_SLEEP}"

    )



    parser.add_argument(

        "--resume",

        action="store_true",

        help="Resume from saved PID."

    )



    parser.add_argument(

        "--start-pid",

        type=int,

        help="Start scanning after this PID."

    )



    args = parser.parse_args()



    setup_logging()

    db = connect()



    try:

        if args.pid:

            single_pid(db, args.pid, args.dry_run)

        else:

            full_run(

                db,

                args.batch,

                args.sleep,

                args.dry_run,

                args.resume,

                args.start_pid

            )

    except KeyboardInterrupt:

        db.rollback()

        print("\nInterrupted. Progress preserved.")

    except Exception:

        db.rollback()

        logging.exception("FATAL ERROR")

        raise

    finally:

        db.close()





if __name__ == "__main__":

    main()