fix: fix Chinese mojibake in SQL files

[pigeon2049]

No description provided.

[pigeon2049]

你们这样vibe coding 是不行滴！
too young too simle ,sometimes naive!
编码错的看都不看的

fix_all_sql.py

import os
import sys
from pathlib import Path

# 确保控制台输出支持 UTF-8
if sys.platform == "win32":
    try:
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

# ==========================================
# 构造 Unicode -> 原始 Byte 反向映射 (MySQL latin1 / cp1252)
# ==========================================
byte_to_char = {}
for b in range(256):
    if b in (0x81, 0x8D, 0x8F, 0x90, 0x9D):  # cp1252 未定义但 MySQL latin1 支持的字节
        ch = chr(b)
    else:
        ch = bytes([b]).decode("cp1252")
    byte_to_char[b] = ch

char_to_byte = {ch: b for b, ch in byte_to_char.items()}


def reverse_mojibake(text: str):
    """
    尝试反向解析乱码：
    当前乱码字符 -> cp1252/latin1 原始字节 -> UTF-8 解码 -> 正确文字
    """
    if not isinstance(text, str) or not text:
        return None
    try:
        raw = bytes(char_to_byte[ch] for ch in text)
    except KeyError:
        # 包含正常中文或无法映射到 latin1 的字符，直接跳过
        return None
    try:
        fixed = raw.decode("utf-8")
    except UnicodeDecodeError:
        return None

    if fixed == text:
        return None
    return fixed


def is_cjk(ch: str) -> bool:
    """判断单个字符是否属于中文字符集"""
    cp = ord(ch)
    return (
        0x3400 <= cp <= 0x4DBF      # CJK Unified Ideographs Extension A
        or 0x4E00 <= cp <= 0x9FFF   # CJK Unified Ideographs
        or 0xF900 <= cp <= 0xFAFF   # CJK Compatibility Ideographs
        or 0x20000 <= cp <= 0x3134F # CJK Extensions B-I
    )


def cjk_count(text: str) -> int:
    return sum(1 for ch in text if is_cjk(ch))


def should_fix(old: str, new: str) -> bool:
    """
    安全策略：
    1. 原字符串中不含真正中文字符（全是一堆 æ/å/è 等西欧扩展字符或 ASCII）
    2. 反解后包含了合法的中文字符
    """
    if not new:
        return False
    return cjk_count(old) == 0 and cjk_count(new) > 0


def fix_string_content(text: str) -> str:
    """
    修复字符串内容（处理 SQL 内部可能存在的转义单引号与反斜杠）
    """
    if not text:
        return text

    # 先尝试整段直接逆向解码
    candidate = reverse_mojibake(text)
    if candidate and should_fix(text, candidate):
        return candidate

    return text


def process_sql(sql: str):
    """
    状态机解析 SQL，精准识别单引号字符串字面量、双横线注释、块注释
    只针对疑似乱码的字符串和注释进行无损修复
    """
    length = len(sql)
    i = 0
    result = []
    fix_count = 0

    while i < length:
        ch = sql[i]

        # 1. 检查 SQL 单行注释 -- 或 #
        if (ch == "-" and i + 1 < length and sql[i + 1] == "-") or ch == "#":
            comment_chars = []
            while i < length and sql[i] not in ("\r", "\n"):
                comment_chars.append(sql[i])
                i += 1
            comment_text = "".join(comment_chars)
            fixed_comment = reverse_mojibake(comment_text)
            if fixed_comment and should_fix(comment_text, fixed_comment):
                result.append(fixed_comment)
                fix_count += 1
            else:
                result.append(comment_text)
            continue

        # 2. 检查 SQL 块注释 /* ... */
        if ch == "/" and i + 1 < length and sql[i + 1] == "*":
            comment_chars = ["/*"]
            i += 2
            while i < length:
                if sql[i] == "*" and i + 1 < length and sql[i + 1] == "/":
                    comment_chars.append("*/")
                    i += 2
                    break
                comment_chars.append(sql[i])
                i += 1
            comment_text = "".join(comment_chars)
            fixed_comment = reverse_mojibake(comment_text)
            if fixed_comment and should_fix(comment_text, fixed_comment):
                result.append(fixed_comment)
                fix_count += 1
            else:
                result.append(comment_text)
            continue

        # 3. 检查单引号字符串字面量 '...'
        if ch == "'":
            result.append("'")
            i += 1
            content = []
            while i < length:
                c = sql[i]
                # MySQL 反斜杠转义，如 \' 或 \\
                if c == "\\" and i + 1 < length:
                    content.append(c)
                    content.append(sql[i + 1])
                    i += 2
                    continue
                # SQL 标准单引号转义 ''
                if c == "'":
                    if i + 1 < length and sql[i + 1] == "'":
                        content.append("''")
                        i += 2
                        continue
                    # 字符串结束
                    break
                content.append(c)
                i += 1

            orig_str = "".join(content)
            fixed_str = fix_string_content(orig_str)
            if fixed_str != orig_str:
                fix_count += 1
                result.append(fixed_str)
            else:
                result.append(orig_str)

            if i < length and sql[i] == "'":
                result.append("'")
                i += 1
            continue

        result.append(ch)
        i += 1

    return "".join(result), fix_count


def scan_and_fix_repo(root_dir: str):
    root = Path(root_dir)
    sql_files = []
    
    # 忽略 target, .git, repair_mojibake.sql 等
    ignore_dirs = {"target", ".git", ".idea", "node_modules", ".agents"}
    
    for path in root.rglob("*.sql"):
        # 排除忽略目录及自身生成的临时 sql
        if any(part in ignore_dirs for part in path.parts):
            continue
        if path.name in ("repair_mojibake.sql",):
            continue
        sql_files.append(path)

    print(f"[*] 共发现 {len(sql_files)} 个 .sql 文件需要检查...")

    total_fixed_files = 0
    total_fixed_items = 0

    for file_path in sql_files:
        try:
            content = file_path.read_text(encoding="utf-8", errors="replace")
        except Exception as e:
            print(f"[!] 无法读取文件 {file_path}: {e}")
            continue

        fixed_content, fix_count = process_sql(content)

        if fix_count > 0 and fixed_content != content:
            file_path.write_text(fixed_content, encoding="utf-8", newline="")
            print(f"[✓] 已修复文件 ({fix_count} 处乱码): {file_path.relative_to(root)}")
            total_fixed_files += 1
            total_fixed_items += fix_count

    print("\n" + "=" * 50)
    print(f"处理完成！")
    print(f"修复文件数: {total_fixed_files} / {len(sql_files)}")
    print(f"修复乱码项: {total_fixed_items} 处")
    print("=" * 50)


if __name__ == "__main__":
    scan_and_fix_repo(os.getcwd())

repair_mojibake.py

import argparse
import sys
from collections import defaultdict
import pymysql

# 确保 Windows 命令行输出 UTF-8 字符不会抛出 GBK 编码异常
if sys.platform == "win32":
    try:
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

# ==========================================
# 构造 Unicode -> 原始 Byte 反向映射 (MySQL latin1 / cp1252)
# ==========================================
byte_to_char = {}
for b in range(256):
    if b in (0x81, 0x8D, 0x8F, 0x90, 0x9D):  # cp1252 未定义但 MySQL latin1 支持的字节
        ch = chr(b)
    else:
        ch = bytes([b]).decode("cp1252")
    byte_to_char[b] = ch

char_to_byte = {ch: b for b, ch in byte_to_char.items()}


def reverse_mojibake(text: str):
    """
    尝试反向解析乱码：
    当前乱码字符 -> cp1252/latin1 原始字节 -> UTF-8 解码 -> 正确文字
    """
    if not isinstance(text, str) or not text:
        return None
    try:
        raw = bytes(char_to_byte[ch] for ch in text)
    except KeyError:
        # 包含正常中文或无法映射到 latin1 的字符，直接跳过
        return None
    try:
        fixed = raw.decode("utf-8")
    except UnicodeDecodeError:
        return None

    if fixed == text:
        return None
    return fixed


def is_cjk(ch: str) -> bool:
    """判断单个字符是否属于中文字符集"""
    cp = ord(ch)
    return (
        0x3400 <= cp <= 0x4DBF      # CJK Unified Ideographs Extension A
        or 0x4E00 <= cp <= 0x9FFF   # CJK Unified Ideographs
        or 0xF900 <= cp <= 0xFAFF   # CJK Compatibility Ideographs
        or 0x20000 <= cp <= 0x3134F # CJK Extensions B-I
    )


def cjk_count(text: str) -> int:
    return sum(1 for ch in text if is_cjk(ch))


def should_fix(old: str, new: str) -> bool:
    """
    安全策略：
    1. 原字符串中不含真正中文字符（全是一堆 æ/å/è 等西欧扩展字符）
    2. 反解后包含了合法的中文字符
    """
    if not new:
        return False
    return cjk_count(old) == 0 and cjk_count(new) > 0


def qi(identifier: str) -> str:
    """转义反引号标识符"""
    return "`" + identifier.replace("`", "``") + "`"


def main():
    parser = argparse.ArgumentParser(description="MySQL 乱码扫描与修复 SQL 生成工具")
    parser.add_argument("--host", default="100.66.245.17", help="MySQL 主机地址 (默认: 100.66.245.17)")
    parser.add_argument("--port", type=int, default=3306, help="MySQL 端口 (默认: 3306)")
    parser.add_argument("--user", default="app", help="MySQL 用户名 (默认: app)")
    parser.add_argument("--password", default="bZJcWQCtjH334GWS", help="MySQL 密码")
    parser.add_argument("--database", default="app", help="数据库名 (默认: app)")
    parser.add_argument("--output", default="repair_mojibake.sql", help="输出 SQL 文件名 (默认: repair_mojibake.sql)")
    parser.add_argument("--execute", action="store_true", help="是否直接执行更新（默认仅生成 SQL 文件）")

    args = parser.parse_args()

    db_config = {
        "host": args.host,
        "port": args.port,
        "user": args.user,
        "password": args.password,
        "database": args.database,
        "charset": "utf8mb4",
    }

    print(f"[*] 正在连接数据库: {db_config['host']}:{db_config['port']}/{db_config['database']} (用户: {db_config['user']})...")
    try:
        conn = pymysql.connect(
            **db_config,
            cursorclass=pymysql.cursors.DictCursor,
        )
    except Exception as e:
        print(f"[!] 数据库连接失败: {e}")
        sys.exit(1)

    db_name = db_config["database"]

    # 1. 获取所有数据表的文本字段
    columns_by_table = defaultdict(list)
    with conn.cursor() as cursor:
        cursor.execute(
            """
            SELECT c.TABLE_NAME, c.COLUMN_NAME, c.DATA_TYPE 
            FROM information_schema.COLUMNS c
            JOIN information_schema.TABLES t 
              ON t.TABLE_SCHEMA = c.TABLE_SCHEMA AND t.TABLE_NAME = c.TABLE_NAME
            WHERE c.TABLE_SCHEMA = %s 
              AND t.TABLE_TYPE = 'BASE TABLE'
              AND c.DATA_TYPE IN ('char', 'varchar', 'tinytext', 'text', 'mediumtext', 'longtext')
              AND c.EXTRA NOT LIKE '%%GENERATED%%'
            ORDER BY c.TABLE_NAME, c.ORDINAL_POSITION
            """,
            (db_name,),
        )
        for row in cursor.fetchall():
            columns_by_table[row["TABLE_NAME"]].append(row["COLUMN_NAME"])

    # 2. 获取每张表的主键
    pk_by_table = defaultdict(list)
    with conn.cursor() as cursor:
        cursor.execute(
            """
            SELECT TABLE_NAME, COLUMN_NAME 
            FROM information_schema.KEY_COLUMN_USAGE 
            WHERE TABLE_SCHEMA = %s 
              AND CONSTRAINT_NAME = 'PRIMARY'
            ORDER BY TABLE_NAME, ORDINAL_POSITION
            """,
            (db_name,),
        )
        for row in cursor.fetchall():
            pk_by_table[row["TABLE_NAME"]].append(row["COLUMN_NAME"])

    changes = []
    print(f"[*] 开始扫描数据库: {db_name}，共 {len(columns_by_table)} 张表包含文本字段...")

    # 3. 遍历扫描每张表
    for table, text_columns in columns_by_table.items():
        pk_columns = pk_by_table.get(table)
        if not pk_columns:
            continue

        valid_text_cols = [c for c in text_columns if c not in pk_columns]
        if not valid_text_cols:
            continue

        select_columns = pk_columns + valid_text_cols
        sql = "SELECT " + ", ".join(qi(c) for c in select_columns) + " FROM " + qi(table)

        stream_cursor = conn.cursor(pymysql.cursors.SSDictCursor)
        try:
            stream_cursor.execute(sql)
            for row in stream_cursor:
                for column in valid_text_cols:
                    old = row[column]
                    if not isinstance(old, str) or not old:
                        continue

                    new = reverse_mojibake(old)
                    if not should_fix(old, new):
                        continue

                    pk_values = {pk: row[pk] for pk in pk_columns}
                    changes.append({
                        "table": table,
                        "column": column,
                        "pk": pk_values,
                        "old": old,
                        "new": new,
                    })
                    print(f"[FOUND] {table}.{column} (PK: {pk_values}) -> 原值: '{old}' | 修复后: '{new}'")
        except Exception as ex:
            print(f"[WARN] 扫描表 {table} 出错: {ex}")
        finally:
            stream_cursor.close()

    # 4. 生成修复 SQL 文件
    if changes:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write("-- ==========================================\n")
            f.write("-- 自动生成的乱码修复脚本\n")
            f.write(f"-- 目标数据库: {db_name}\n")
            f.write(f"-- 发现乱码项: {len(changes)} 条\n")
            f.write("-- ==========================================\n\n")
            f.write("SET NAMES utf8mb4;\n")
            f.write("START TRANSACTION;\n\n")

            for item in changes:
                table = item["table"]
                column = item["column"]
                old = item["old"]
                new = item["new"]

                where_clauses = []
                for pk, value in item["pk"].items():
                    where_clauses.append(f"{qi(pk)} = {conn.escape(value)}")

                # 双重校验：原值必须完全匹配
                where_clauses.append(f"{qi(column)} <=> {conn.escape(old)}")

                sql = (
                    f"UPDATE {qi(table)} "
                    f"SET {qi(column)} = {conn.escape(new)} "
                    f"WHERE {' AND '.join(where_clauses)} "
                    f"LIMIT 1;"
                )
                f.write(sql + "\n")

            f.write("\nCOMMIT;\n")

    print("\n==============================")
    print("扫描完成！")
    print(f"发现疑似乱码: {len(changes)} 处")
    if changes:
        print(f"修复 SQL 文件已生成: {args.output}")
        if args.execute:
            print("[*] 正在执行修复 SQL 到数据库...")
            with conn.cursor() as cursor:
                for item in changes:
                    table = item["table"]
                    column = item["column"]
                    old = item["old"]
                    new = item["new"]
                    where_clauses = [f"{qi(pk)} = {conn.escape(v)}" for pk, v in item["pk"].items()]
                    where_clauses.append(f"{qi(column)} <=> {conn.escape(old)}")
                    update_sql = f"UPDATE {qi(table)} SET {qi(column)} = {conn.escape(new)} WHERE {' AND '.join(where_clauses)} LIMIT 1"
                    cursor.execute(update_sql)
                conn.commit()
            print("[✓] 数据库修复执行完毕并已提交事务！")
    else:
        print("未发现需要修复的乱码数据。")
    print("==============================")

    conn.close()


if __name__ == "__main__":
    main()