Tolan/user-star/backfill_intimacy_hearts.py

"""
一次性脚本：为历史用户补 3 颗心（在 user_intimacy 中插入 heart_count=3，仅当该用户尚无记录时）。
运行前请确保已执行 migrate_add_user_intimacy_table.py 创建 user_intimacy 表。
"""
import asyncio
import sys
from pathlib import Path

# 将 backend 加入 path，支持从项目根或 backend 目录运行
_backend = Path(__file__).resolve().parent.parent / "backend"
if _backend.exists() and str(_backend) not in sys.path:
    sys.path.insert(0, str(_backend))

import aiomysql

from config import settings


async def backfill_intimacy_hearts() -> bool:
    """为所有在 users 表中存在且尚未在 user_intimacy 中有记录的用户插入 heart_count=3"""
    try:
        conn: aiomysql.Connection = await aiomysql.connect(
            host=settings.MYSQL_HOST,
            port=settings.MYSQL_PORT,
            user=settings.MYSQL_USER,
            password=settings.MYSQL_PASSWORD,
            db=settings.MYSQL_DATABASE,
            charset="utf8mb4",
        )
        async with conn.cursor() as cursor:
            # 使用 COLLATE 统一排序规则，避免 users 与 user_intimacy 的 utf8mb4_unicode_ci / utf8mb4_general_ci 混用报错
            await cursor.execute(
                """
                INSERT INTO user_intimacy (user_id, heart_count)
                SELECT u.user_id, 3
                FROM users u
                LEFT JOIN user_intimacy i ON u.user_id = i.user_id COLLATE utf8mb4_unicode_ci
                WHERE i.user_id IS NULL
                """
            )
            affected = cursor.rowcount
            await conn.commit()
            print(f"✅ 已为 {affected} 个历史用户补 3 颗心")
        await conn.ensure_closed()
        return True
    except Exception as exc:
        print(f"❌ 脚本执行失败: {exc}")
        import traceback
        traceback.print_exc()
        return False


if __name__ == "__main__":
    ok = asyncio.run(backfill_intimacy_hearts())
    sys.exit(0 if ok else 1)