Blame for database.py

import asyncio
import json
import os
import time
from contextlib import asynccontextmanager
from datetime import datetime
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

from client_source import current_client_source, validate_client_source

import aiomysql
from aiomysql.cursors import DictCursor

from config import (
    IMAGES_DIR,
    logger,
    MYSQL_HOST,
    MYSQL_PORT,
    MYSQL_DB,
    MYSQL_USER,
    MYSQL_PASSWORD,
    MYSQL_POOL_MIN_SIZE,
    MYSQL_POOL_MAX_SIZE,
    LEADERBOARD_DAYS_LIMIT,
)


_pool: Optional[aiomysql.Pool] = None
_pool_lock = asyncio.Lock()


async def init_mysql_pool() -> aiomysql.Pool:
  """初始化 MySQL 连接池"""
  global _pool
  if _pool is not None:
    return _pool

  async with _pool_lock:
    if _pool is not None:
      return _pool
    try:
      _pool = await aiomysql.create_pool(
          host=MYSQL_HOST,
          port=MYSQL_PORT,
          user=MYSQL_USER,
          password=MYSQL_PASSWORD,
          db=MYSQL_DB,
          minsize=MYSQL_POOL_MIN_SIZE,
          maxsize=MYSQL_POOL_MAX_SIZE,
          autocommit=True,
          charset="utf8mb4",
          cursorclass=DictCursor,
      )
      logger.info(
          "MySQL 连接池初始化成功，host=%s db=%s",
          MYSQL_HOST,
          MYSQL_DB,
      )
    except Exception as exc:
      logger.error(f"初始化 MySQL 连接池失败: {exc}")
      raise
  return _pool


async def close_mysql_pool() -> None:
  """关闭 MySQL 连接池"""
  global _pool
  if _pool is None:
    return

  async with _pool_lock:
    if _pool is None:
      return
    _pool.close()
    await _pool.wait_closed()
    _pool = None
    logger.info("MySQL 连接池已关闭")


@asynccontextmanager
async def get_connection():
  """获取连接池中的连接"""
  if _pool is None:
    await init_mysql_pool()
  assert _pool is not None
  conn = await _pool.acquire()
  try:
    yield conn
  finally:
    _pool.release(conn)


async def execute(
    query: str, params: Sequence[Any] | Dict[str, Any] | None = None
) -> int:
  """执行写入类 SQL"""
  start_time = time.perf_counter()
  async with get_connection() as conn:
    async with conn.cursor() as cursor:
      try:
        await cursor.execute(query, params or ())
        affected = cursor.rowcount
        duration = (time.perf_counter() - start_time) * 1000
        formatted_query = " ".join(query.split())
        logger.info(
            "SQL Execute: %s | Params: %s | Affected: %d | Cost: %.2fms",
            formatted_query,
            params,
            affected,
            duration,
        )
        return affected
      except Exception as exc:
        duration = (time.perf_counter() - start_time) * 1000
        formatted_query = " ".join(query.split())
        logger.error(
            "SQL Execute Failed: %s | Params: %s | Cost: %.2fms | Error: %s",
            formatted_query,
            params,
            duration,
            exc,
        )
        raise


async def fetch_all(
    query: str, params: Sequence[Any] | Dict[str, Any] | None = None
) -> List[Dict[str, Any]]:
  """执行查询并返回全部结果"""
  start_time = time.perf_counter()
  async with get_connection() as conn:
    async with conn.cursor() as cursor:
      try:
        await cursor.execute(query, params or ())
        rows = await cursor.fetchall()
        duration = (time.perf_counter() - start_time) * 1000
        formatted_query = " ".join(query.split())
        logger.info(
            "SQL Query: %s | Params: %s | Rows: %d | Cost: %.2fms",
            formatted_query,
            params,
            len(rows),
            duration,
        )
        return list(rows)
      except Exception as exc:
        duration = (time.perf_counter() - start_time) * 1000
        formatted_query = " ".join(query.split())
        logger.error(
            "SQL Query Failed: %s | Params: %s | Cost: %.2fms | Error: %s",
            formatted_query,
            params,
            duration,
            exc,
        )
        raise


async def fetch_one(
    query: str, params: Sequence[Any] | Dict[str, Any] | None = None
) -> Optional[Dict[str, Any]]:
  """执行查询并返回单条结果"""
  rows = await fetch_all(query, params)
  return rows[0] if rows else None


def _serialize_extra(extra: Optional[Dict[str, Any]]) -> Optional[str]:
  if extra is None:
    return None
  try:
    return json.dumps(extra, ensure_ascii=False)
  except Exception:
    logger.warning("无法序列化 extra_metadata，已忽略")
    return None


async def upsert_image_record(
    *,
    file_path: str,
    category: str,
    nickname: Optional[str],
    score: float,
    is_cropped_face: bool,
    size_bytes: int,
    last_modified: datetime,
    bos_uploaded: bool,
    hostname: Optional[str] = None,
    region: Optional[str] = None,
    extra_metadata: Optional[Dict[str, Any]] = None,
    client_source: Optional[str] = None,
) -> None:
  """写入或更新图片记录"""
  source = validate_client_source(client_source if client_source is not None else current_client_source.get())
  query = """
        INSERT INTO tpl_app_processed_images (
            file_path,
            category,
            nickname,
            score,
            is_cropped_face,
            size_bytes,
            last_modified,
            bos_uploaded,
            hostname,
            region,
            extra_metadata,
            client_source
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
            client_source = IF(client_source = 'unknown', VALUES(client_source), client_source),
            category = VALUES(category),
            nickname = VALUES(nickname),
            score = VALUES(score),
            is_cropped_face = VALUES(is_cropped_face),
            size_bytes = VALUES(size_bytes),
            last_modified = VALUES(last_modified),
            bos_uploaded = VALUES(bos_uploaded),
            hostname = VALUES(hostname),
            region = VALUES(region),
            extra_metadata = VALUES(extra_metadata),
            updated_at = CURRENT_TIMESTAMP
    """
  extra_value = _serialize_extra(extra_metadata)
  await execute(
      query,
      (
        file_path,
        category,
        nickname,
        score,
        1 if is_cropped_face else 0,
        size_bytes,
        last_modified,
        1 if bos_uploaded else 0,
        hostname,
        region,
        extra_value,
        source,
      ),
  )


async def fetch_paged_image_records(
    *,
    category: Optional[str],
    nickname: Optional[str],
    region: Optional[str] = None,
    platform: Optional[str] = None,
    offset: int,
    limit: int,
    is_cropped_face: Optional[int] = None,
) -> List[Dict[str, Any]]:
  """按条件分页查询图片记录"""
  where_clauses: List[str] = []
  params: List[Any] = []
  if category and category != "all":
    where_clauses.append("category = %s")
    params.append(category)
  if nickname:
    where_clauses.append("nickname LIKE %s")
    params.append(f"{nickname}%")
  if platform in {"android", "ios", "web", "wechat", "miniprogram"}:
    where_clauses.append("client_source = %s")
    params.append("wechat" if platform == "miniprogram" else platform)
  if region:
    where_clauses.append("region = %s")
    params.append(region)
  if is_cropped_face is not None:
    where_clauses.append("is_cropped_face = %s")
    params.append(is_cropped_face)
  where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
  query = f"""
        SELECT
            file_path,
            category,
            nickname,
            score,
            is_cropped_face,
            size_bytes,
            last_modified,
            bos_uploaded,
            hostname,
            region
        FROM tpl_app_processed_images
        {where_sql}
        ORDER BY last_modified DESC, id DESC
        LIMIT %s OFFSET %s
    """
  params.extend([limit, offset])
  return await fetch_all(query, params)


async def count_image_records(
    *,
    category: Optional[str],
    nickname: Optional[str],
    region: Optional[str] = None,
    platform: Optional[str] = None,
    is_cropped_face: Optional[int] = None,
) -> int:
  """按条件统计图片记录数量"""
  where_clauses: List[str] = []
  params: List[Any] = []
  if category and category != "all":
    where_clauses.append("category = %s")
    params.append(category)
  if nickname:
    where_clauses.append("nickname LIKE %s")
    params.append(f"{nickname}%")
  if platform in {"android", "ios", "web", "wechat", "miniprogram"}:
    where_clauses.append("client_source = %s")
    params.append("wechat" if platform == "miniprogram" else platform)
  if region:
    where_clauses.append("region = %s")
    params.append(region)
  if is_cropped_face is not None:
    where_clauses.append("is_cropped_face = %s")
    params.append(is_cropped_face)
  where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
  query = f"SELECT COUNT(*) AS total FROM tpl_app_processed_images {where_sql}"
  rows = await fetch_all(query, params)
  if not rows:
    return 0
  return int(rows[0].get("total", 0) or 0)


async def fetch_today_category_counts() -> List[Dict[str, Any]]:
  """统计当天按类别分组的数量"""
  query = """
      SELECT
          COALESCE(category, 'unknown') AS category,
          COUNT(*) AS count
      FROM tpl_app_processed_images
      WHERE last_modified >= CURDATE()
        AND last_modified < DATE_ADD(CURDATE(), INTERVAL 1 DAY)
      GROUP BY COALESCE(category, 'unknown')
    """
  rows = await fetch_all(query)
  return [
    {
      "category": str(row.get("category") or "unknown"),
      "count": int(row.get("count") or 0),
    }
    for row in rows
  ]


async def fetch_records_by_paths(file_paths: Iterable[str]) -> Dict[
  str, Dict[str, Any]]:
  """根据文件名批量查询图片记录"""
  paths = list({path for path in file_paths if path})
  if not paths:
    return {}

  placeholders = ", ".join(["%s"] * len(paths))
  query = f"""
        SELECT
            file_path,
            category,
            nickname,
            score,
            is_cropped_face,
            size_bytes,
            last_modified,
            bos_uploaded,
            hostname,
            region
        FROM tpl_app_processed_images
        WHERE file_path IN ({placeholders})
    """
  rows = await fetch_all(query, paths)
  return {row["file_path"]: row for row in rows}


_IMAGES_DIR_ABS = os.path.abspath(os.path.expanduser(IMAGES_DIR))


def _normalize_file_path(file_path: str) -> Optional[str]:
  """将绝对路径转换为相对 IMAGES_DIR 的文件名"""
  try:
    abs_path = os.path.abspath(os.path.expanduser(file_path))
    if os.path.isdir(abs_path):
      return None
    if os.path.commonpath([_IMAGES_DIR_ABS, abs_path]) != _IMAGES_DIR_ABS:
      return os.path.basename(abs_path)
    rel_path = os.path.relpath(abs_path, _IMAGES_DIR_ABS)
    return rel_path.replace("\\", "/")
  except Exception:
    return None


def infer_category_from_filename(filename: str, default: str = "other") -> str:
  """根据文件名推断类别"""
  lower_name = filename.lower()
  if "_face" in lower_name:
    return "face"
  if lower_name.endswith("_original.webp") or "_original" in lower_name:
    return "original"
  if "_restore" in lower_name:
    return "restore"
  if "_upcolor" in lower_name:
    return "upcolor"
  if "_compress" in lower_name:
    return "compress"
  if "_upscale" in lower_name:
    return "upscale"
  if "_anime_style_" in lower_name:
    return "anime_style"
  if "_grayscale" in lower_name:
    return "grayscale"
  if "_id_photo" in lower_name or "_save_id_photo" in lower_name:
    return "id_photo"
  if "_grid_" in lower_name:
    return "grid"
  if "_rvm_id_photo" in lower_name:
    return "rvm"
  if "_celebrity_" in lower_name or "_celebrity" in lower_name:
    return "celebrity"
  return default


from config import HOSTNAME

async def upsert_device_record(
    *,
    device_id: str,
    device_type: Optional[str] = None,
    device_model: Optional[str] = None,
    os_version: Optional[str] = None,
    app_version: Optional[str] = None,
    timezone: Optional[str] = None,
    region: Optional[str] = None,
    language: Optional[str] = None,
    client_source: str = "android",
) -> None:
    """写入或更新设备记录"""
    source = validate_client_source(client_source)
    query = """
        INSERT INTO tpl_app_user_devices (
            device_id,
            device_type,
            device_model,
            os_version,
            app_version,
            timezone,
            region,
            language,
            client_source
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
            client_source = IF(client_source = 'unknown', VALUES(client_source), client_source),
            device_type = VALUES(device_type),
            device_model = VALUES(device_model),
            os_version = VALUES(os_version),
            app_version = VALUES(app_version),
            timezone = VALUES(timezone),
            region = VALUES(region),
            language = VALUES(language),
            updated_at = CURRENT_TIMESTAMP
    """
    import time
    start_time = time.perf_counter()
    try:
        await execute(
            query,
            (
                device_id,
                device_type,
                device_model,
                os_version,
                app_version,
                timezone,
                region,
                language,
                source,
            ),
        )
        duration = (time.perf_counter() - start_time) * 1000
        logger.info(f"Device record updated for {device_id}, cost: {duration:.2f}ms")
    except Exception as exc:
        logger.warning(f"写入设备记录失败: {exc}")


async def get_user_display_name(device_id: str) -> str:
    return (await get_user_profile(device_id))["display_name"]


async def get_user_profile(device_id: str) -> Dict[str, Any]:
    versioned = os.getenv("ENABLE_PROFILE_REVISIONS", "false").lower() == "true"
    fields = "display_name, display_name_initialized" + (", profile_revision" if versioned else "")
    row = await fetch_one(
        f"SELECT {fields} FROM tpl_app_user_devices WHERE device_id = %s",
        (device_id,),
    )
    name = str(row.get("display_name") or "") if row else ""
    result = {"display_name": name,
              "nickname_initialized": bool(name or (row and row.get("display_name_initialized")))}
    if versioned:
        result["client_revision"] = int(row.get("profile_revision") or 0) if row else 0
    return result


async def set_user_display_name(
    device_id: str,
    display_name: str,
    *,
    initialize_only: bool = False,
    client_revision: Optional[int] = None,
    device_type: Optional[str] = None,
    device_model: Optional[str] = None,
    os_version: Optional[str] = None,
    app_version: Optional[str] = None,
    timezone: Optional[str] = None,
    region: Optional[str] = None,
    language: Optional[str] = None,
) -> None:
    # 独立更新资料，不触碰评分记录的 updated_at、分数或原 nickname。
    # 首次引导同时写入设备资料，单条语句保证昵称与设备信息一起保存。
    # 旧客户端未传或传空的设备字段保留原值；空昵称仍表示主动清空。
    # 条件判断与写入在同一条 SQL 内，避免初始化重试覆盖并发手动改名或清空。
    name_update = ("IF(display_name_initialized = 0 AND display_name = '', VALUES(display_name), display_name)"
                   if initialize_only else "VALUES(display_name)")
    versioned = client_revision is not None and os.getenv("ENABLE_PROFILE_REVISIONS", "false").lower() == "true"
    revision_column = ", profile_revision" if versioned else ""
    revision_value = ", %s" if versioned else ""
    revision_update = ", profile_revision = GREATEST(profile_revision, VALUES(profile_revision))" if versioned else ""
    if versioned and not initialize_only:
        name_update = "IF(VALUES(profile_revision) > profile_revision, VALUES(display_name), display_name)"
    await execute(
        f"""INSERT INTO tpl_app_user_devices (
               device_id, display_name, display_name_initialized, client_source, device_type, device_model,
               os_version, app_version, timezone, region, language{revision_column}
           ) VALUES (%s, %s, 1, 'android', %s, %s, %s, %s, %s, %s, %s{revision_value})
           ON DUPLICATE KEY UPDATE display_name = {name_update},
               display_name_initialized = 1,
               client_source = IF(client_source = 'unknown', VALUES(client_source), client_source),
               device_type = COALESCE(NULLIF(VALUES(device_type), ''), device_type),
               device_model = COALESCE(NULLIF(VALUES(device_model), ''), device_model),
               os_version = COALESCE(NULLIF(VALUES(os_version), ''), os_version),
               app_version = COALESCE(NULLIF(VALUES(app_version), ''), app_version),
               timezone = COALESCE(NULLIF(VALUES(timezone), ''), timezone),
               region = COALESCE(NULLIF(VALUES(region), ''), region),
               language = COALESCE(NULLIF(VALUES(language), ''), language){revision_update}""",
        (device_id, display_name, device_type, device_model, os_version,
         app_version, timezone, region, language) + ((client_revision,) if versioned else ()) ,
    )


async def fetch_user_devices(
    *,
    device_id: Optional[str] = None,
    region: Optional[str] = None,
    offset: int,
    limit: int,
) -> List[Dict[str, Any]]:
    """查询用户设备列表，支持按 device_id 和 region 筛选，按 created_at 倒序"""
    where_clauses: List[str] = []
    params: List[Any] = []

    if device_id:
        where_clauses.append("d.device_id LIKE %s")
        params.append(f"{device_id}%")

    if region:
        where_clauses.append("d.region = %s")
        params.append(region)

    where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""

    query = f"""
        SELECT
            d.device_id,
            d.device_type,
            d.device_model,
            d.os_version,
            d.app_version,
            d.timezone,
            d.region,
            d.language,
            d.created_at,
            d.updated_at,
            COALESCE((
                SELECT COUNT(*) 
                FROM tpl_app_processed_images p 
                WHERE p.nickname LIKE CONCAT('android_release_', d.device_id, '%%')
                    AND p.category = 'original'
            ), 0) AS detection_count
        FROM tpl_app_user_devices d
        {where_sql}
        ORDER BY d.updated_at DESC
        LIMIT %s OFFSET %s
    """
    params.extend([limit, offset])
    return await fetch_all(query, params)


async def count_user_devices(
    *,
    device_id: Optional[str] = None,
    region: Optional[str] = None,
) -> int:
    """统计用户设备数量"""
    where_clauses: List[str] = []
    params: List[Any] = []

    if device_id:
        where_clauses.append("d.device_id LIKE %s")
        params.append(f"{device_id}%")

    if region:
        where_clauses.append("d.region = %s")
        params.append(region)

    where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""

    query = f"SELECT COUNT(*) AS total FROM tpl_app_user_devices d {where_sql}"
    rows = await fetch_all(query, params)
    if not rows:
        return 0
    return int(rows[0].get("total", 0) or 0)


async def fetch_device_region_stats() -> List[Dict[str, Any]]:
    """统计用户设备按地区分组的数量"""
    query = """
        SELECT
            COALESCE(region, '') AS region,
            COUNT(*) AS count
        FROM tpl_app_user_devices
        GROUP BY COALESCE(region, '')
        ORDER BY count DESC
    """
    return await fetch_all(query)


async def fetch_image_region_stats(
    *,
    category: Optional[str],
    nickname: Optional[str],
    region: Optional[str] = None,
    platform: Optional[str] = None,
    is_cropped_face: Optional[int] = None,
) -> List[Dict[str, Any]]:
    """统计图片记录按地区分组的数量"""
    where_clauses: List[str] = []
    params: List[Any] = []
    if category and category != "all":
        where_clauses.append("category = %s")
        params.append(category)
    if nickname:
        where_clauses.append("nickname LIKE %s")
        params.append(f"{nickname}%")
    if platform in {"android", "ios", "web", "wechat", "miniprogram"}:
        where_clauses.append("client_source = %s")
        params.append("wechat" if platform == "miniprogram" else platform)
    if region:
        where_clauses.append("region = %s")
        params.append(region)
    if is_cropped_face is not None:
        where_clauses.append("is_cropped_face = %s")
        params.append(is_cropped_face)
    where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
    query = f"""
        SELECT
            COALESCE(region, '') AS region,
            COUNT(*) AS count
        FROM tpl_app_processed_images
        {where_sql}
        GROUP BY COALESCE(region, '')
        ORDER BY count DESC
    """
    return await fetch_all(query, params)


async def record_image_creation(
    *,
    file_path: str,
    nickname: Optional[str],
    score: float = 0.0,
    category: Optional[str] = None,
    bos_uploaded: bool = False,
    region: Optional[str] = None,
    extra_metadata: Optional[Dict[str, Any]] = None,
) -> None:
  """
  记录图片元数据到数据库，如果数据库不可用则静默忽略。
  :param file_path: 绝对或相对文件路径
  :param nickname: 用户昵称
  :param score: 关联得分
  :param category: 文件类别，未提供时自动根据文件名推断
  :param bos_uploaded: 是否已上传至 BOS
  :param extra_metadata: 额外信息
  """
  normalized = _normalize_file_path(file_path)
  if normalized is None:
    logger.info("record_image_creation: 无法计算文件名，路径=%s", file_path)
    return

  abs_path = os.path.join(_IMAGES_DIR_ABS, normalized)
  if not os.path.isfile(abs_path):
    logger.info("record_image_creation: 文件不存在，跳过记录 file=%s", abs_path)
    return

  try:
    stat = os.stat(abs_path)
    category_name = category or infer_category_from_filename(normalized)
    # 逻辑调整：支持 ugc_xxxx_face1.webp 这种格式
    is_cropped_face = "_face" in normalized and normalized.count("_") >= 2
    last_modified = datetime.fromtimestamp(stat.st_mtime)

    nickname_value = nickname.strip() if isinstance(nickname,
                                                    str) and nickname.strip() else None

    # 如果 region 为空且为 Android 客户端请求，尝试通过 device_id 回填历史记录的 region
    region_val = region.strip() if isinstance(region, str) and region.strip() else None
    if not region_val and nickname_value and nickname_value.startswith("android_"):
        parts = nickname_value.split("_")
        device_id = parts[-1] if len(parts) >= 3 else None
        if device_id:
            try:
                row = await fetch_one(
                    "SELECT region FROM tpl_app_user_devices WHERE device_id = %s AND region IS NOT NULL AND region != '' LIMIT 1",
                    (device_id,),
                )
                if row and row.get("region"):
                    region_val = row.get("region")
            except Exception:
                pass

    await upsert_image_record(
        file_path=normalized,
        category=category_name,
        nickname=nickname_value,
        score=score,
        is_cropped_face=is_cropped_face,
        size_bytes=stat.st_size,
        last_modified=last_modified,
        bos_uploaded=bos_uploaded,
        hostname=HOSTNAME,
        region=region_val,
        extra_metadata=extra_metadata,
    )
  except Exception as exc:
    logger.warning(f"写入图片记录失败: {exc}")


# 排名与排行榜内存缓存
_RANK_CACHE: Dict[str, Tuple[float, Dict[str, Any]]] = {}
_CACHE_TTL_RANK = 60  # 60秒
_LEADERBOARD_CACHE: Dict[str, Tuple[float, List[Dict[str, Any]]]] = {}
_CACHE_TTL_LEADERBOARD = 180  # 3分钟



def _ranking_scope(source: str, alias: str = "") -> str:
    validate_client_source(source)
    prefix = alias + "." if alias else ""
    condition = f"{prefix}client_source = %s"
    if source == "android":
        condition += f" AND ({prefix}nickname IS NULL OR LEFT({prefix}nickname, 14) <> 'android_debug_')"
    return condition


async def get_face_score_rank(
    score: float,
    region: Optional[str] = None,
    gender: Optional[str] = None,
    age: Optional[str] = None,
    client_source: str = "android",
) -> Dict[str, Any]:
    """
    计算颜值评分在全球和当前地区的排名及击败百分比
    :param score: 颜值分数 (0.0 ~ 100.0)
    :param region: 用户所在地区代码 (如 "CN", "US")
    :param gender: 预留性别
    :param age: 预留年龄
    :return: 包含地区和全球排名统计的字典
    """
    from region_code_utils import get_country_name_by_region_code
    scope = _ranking_scope(client_source)
    score_val = max(0.0, min(100.0, float(score)))
    region_code = region.strip().upper() if region and region.strip() and region.strip().upper() != "GLOBAL" else ""
    region_name = get_country_name_by_region_code(region_code) if region_code else "全球"

    days_limit = max(1, int(LEADERBOARD_DAYS_LIMIT))

    # 排名对小数分数敏感，不能按 0.1 分缓存，否则相邻分数会误用同一名次。
    cache_key = f"{client_source}:{score_val:.4f}:{region_code}:{days_limit}"
    now = time.time()
    if cache_key in _RANK_CACHE:
        cached_time, cached_data = _RANK_CACHE[cache_key]
        if now - cached_time < _CACHE_TTL_RANK:
            return cached_data

    # 统一分数比较表达式：兼容历史库内可能存在的 10 分制 (0~10) 与 100 分制 (0~100)
    score_expr = "(CASE WHEN score <= 10.0 THEN score * 10.0 ELSE score END)"

    success = True
    global_rank = 0
    global_total = 0
    global_percentile = 0.0

    region_rank = 0
    region_total = 0
    region_percentile = 0.0

    try:
        # 排行榜按用户最高分去重；/rank 必须使用完全相同的用户最高分去重口径。
        user_key_expr = "(CASE WHEN nickname IS NOT NULL AND nickname != '' THEN nickname ELSE CONCAT('id_', id) END)"
        g_stats_row = await fetch_one(
            f"""
            SELECT COUNT(*) AS t,
                   COALESCE(SUM(CASE WHEN max_score > %s THEN 1 ELSE 0 END), 0) + 1 AS r
            FROM (
                SELECT {user_key_expr} AS user_key,
                       MAX({score_expr}) AS max_score
                FROM tpl_app_processed_images
                WHERE category = 'face'
                  AND {scope}
                  AND updated_at >= DATE_SUB(NOW(), INTERVAL {days_limit} DAY)
                  AND score > 0
                GROUP BY user_key
            ) AS ranked_users
            """,
            (score_val, client_source)
        )
        if g_stats_row is not None:
            db_g_total = int(g_stats_row.get("t") or 0)
            db_g_rank = int(g_stats_row.get("r") or 1)
            if db_g_total > 0:
                global_total = db_g_total
                global_rank = min(global_total, max(1, db_g_rank))
                # 击败百分比 = (total - rank + 1) / total * 100
                global_percentile = round((global_total - global_rank + 1) / global_total * 100.0, 2)
                global_percentile = max(0.01, min(100.0, global_percentile))

        # 地区排名计算
        if region_code:
            r_stats_row = await fetch_one(
                f"""
                SELECT COUNT(*) AS t,
                       COALESCE(SUM(CASE WHEN max_score > %s THEN 1 ELSE 0 END), 0) + 1 AS r
                FROM (
                    SELECT {user_key_expr} AS user_key,
                           MAX({score_expr}) AS max_score
                    FROM tpl_app_processed_images
                    WHERE category = 'face' AND region = %s
                      AND {scope}
                      AND updated_at >= DATE_SUB(NOW(), INTERVAL {days_limit} DAY)
                      AND score > 0
                    GROUP BY user_key
                ) AS ranked_users
                """,
                (score_val, region_code, client_source)
            )
            if r_stats_row is not None:
                db_r_total = int(r_stats_row.get("t") or 0)
                db_r_rank = int(r_stats_row.get("r") or 1)
                if db_r_total > 0:
                    region_total = db_r_total
                    region_rank = min(region_total, max(1, db_r_rank))
                    region_percentile = round((region_total - region_rank + 1) / region_total * 100.0, 2)
                    region_percentile = max(0.01, min(100.0, region_percentile))
        else:
            region_total = global_total
            region_rank = global_rank
            region_percentile = global_percentile

    except Exception as exc:
        logger.warning(f"获取颜值排名数据库查询异常: {exc}")
        success = False

    result = {
        "success": success,
        "score": score_val,
        "region_code": region_code or "GLOBAL",
        "region_name": region_name,
        "region_rank": region_rank,
        "region_total": region_total,
        "region_percentile": region_percentile,
        "global_rank": global_rank,
        "global_total": global_total,
        "global_percentile": global_percentile,
        "days_limit": days_limit,
    }
    if success:
        _RANK_CACHE[cache_key] = (now, result)
    return result


async def get_face_score_leaderboard(
    region: Optional[str] = None,
    limit: int = 50,
    current_device_id: Optional[str] = None,
    client_source: str = "android",
) -> Dict[str, Any]:
    result = await _get_cached_face_score_leaderboard(region, limit, current_device_id, client_source)
    if client_source != "android" or not result["success"] or not result["items"]:
        return result
    # 排名缓存只保存评分记录；昵称每次批量读取，跨 worker 改名/清空也立即生效。
    # Android 上传格式为 prefix_androidId，历史 android_* 前缀同样兼容。
    device_ids = sorted({item["nickname"].rsplit("_", 1)[-1]
                         for item in result["items"] if item["nickname"].startswith("android_")})
    try:
        names: Dict[str, str] = {}
        if device_ids:
            placeholders = ", ".join(["%s"] * len(device_ids))
            rows = await fetch_all(
                f"SELECT device_id, display_name FROM tpl_app_user_devices WHERE client_source = 'android' AND device_id IN ({placeholders})",
                device_ids,
            )
            names = {row["device_id"]: str(row.get("display_name") or "") for row in rows}
        for item in result["items"]:
            item["display_name"] = names.get(item["nickname"].rsplit("_", 1)[-1], "") if item["nickname"].startswith("android_") else ""
        return result
    except Exception as exc:
        logger.warning("读取排行榜昵称失败: %s", exc)
        return {**result, "success": False, "items": [], "total_records": 0}


async def _get_cached_face_score_leaderboard(
    region: Optional[str] = None,
    limit: int = 50,
    current_device_id: Optional[str] = None,
    client_source: str = "android",
) -> Dict[str, Any]:
    """
    获取颜值评分记录排行榜 Top N (支持地区榜与全球榜)。
    按用户（nickname）去重，每个用户仅保留最高分记录；无昵称的匿名记录独立保留。
    """
    from region_code_utils import get_country_name_by_region_code
    scope = _ranking_scope(client_source, "candidate")
    region_code = region.strip().upper() if region and region.strip() and region.strip().upper() != "GLOBAL" else None
    days_limit = max(1, int(LEADERBOARD_DAYS_LIMIT))
    # limit 会改变缓存内容，必须纳入 key，避免一次小 limit 请求污染后续 Top 50。
    safe_limit = max(1, min(100, int(limit)))
    cache_key = f"{client_source}:{region_code or 'GLOBAL'}:{days_limit}:{safe_limit}"
    now = time.time()

    def response_items(cached_or_fresh_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        device_suffix = f"_{current_device_id}" if current_device_id and client_source == "android" else None
        result_items: List[Dict[str, Any]] = []
        for cached_item in cached_or_fresh_items[:limit]:
            nickname = str(cached_item.get("nickname") or "")
            public_item = dict(cached_item)
            public_item["is_current_user"] = bool(device_suffix and nickname.endswith(device_suffix))
            result_items.append(public_item)
        return result_items

    if cache_key in _LEADERBOARD_CACHE:
        cached_time, cached_items = _LEADERBOARD_CACHE[cache_key]
        if now - cached_time < _CACHE_TTL_LEADERBOARD:
            return {
                "success": True,
                "region_code": region_code or "GLOBAL",
                "region_name": get_country_name_by_region_code(region_code) if region_code else "全球",
                "total_records": len(cached_items),
                "items": response_items(cached_items),
                "days_limit": days_limit,
            }

    success = True
    items = []
    candidate_score_expr = "(CASE WHEN candidate.score <= 10.0 THEN candidate.score * 10.0 ELSE candidate.score END)"
    better_score_expr = "(CASE WHEN better.score <= 10.0 THEN better.score * 10.0 ELSE better.score END)"
    better_scope = _ranking_scope(client_source, "better")
    try:
        if region_code:
            query = f"""
                SELECT candidate.id, {candidate_score_expr} AS score, candidate.region,
                       candidate.nickname, candidate.file_path, candidate.extra_metadata, candidate.updated_at, candidate.created_at
                FROM tpl_app_processed_images AS candidate
                WHERE candidate.category = 'face' AND candidate.score > 0 AND candidate.region = %s
                  AND {scope}
                  AND candidate.updated_at >= DATE_SUB(NOW(), INTERVAL {days_limit} DAY)
                  AND (
                      candidate.nickname IS NULL
                      OR candidate.nickname = ''
                      OR NOT EXISTS (
                          SELECT 1
                          FROM tpl_app_processed_images AS better
                          WHERE better.nickname = candidate.nickname
                            AND better.category = 'face' AND better.score > 0 AND better.region = %s
                            AND {better_scope}
                            AND better.updated_at >= DATE_SUB(NOW(), INTERVAL {days_limit} DAY)
                            AND (
                                {better_score_expr} > {candidate_score_expr}
                                OR ({better_score_expr} = {candidate_score_expr} AND better.id < candidate.id)
                            )
                      )
                  )
                ORDER BY score DESC, candidate.id ASC
                LIMIT {safe_limit}
            """
            rows = await fetch_all(query, (region_code, client_source, region_code, client_source))
        else:
            query = f"""
                SELECT candidate.id, {candidate_score_expr} AS score, candidate.region,
                       candidate.nickname, candidate.file_path, candidate.extra_metadata, candidate.updated_at, candidate.created_at
                FROM tpl_app_processed_images AS candidate
                WHERE candidate.category = 'face' AND candidate.score > 0
                  AND {scope}
                  AND candidate.updated_at >= DATE_SUB(NOW(), INTERVAL {days_limit} DAY)
                  AND (
                      candidate.nickname IS NULL
                      OR candidate.nickname = ''
                      OR NOT EXISTS (
                          SELECT 1
                          FROM tpl_app_processed_images AS better
                          WHERE better.nickname = candidate.nickname
                            AND better.category = 'face' AND better.score > 0
                            AND {better_scope}
                            AND better.updated_at >= DATE_SUB(NOW(), INTERVAL {days_limit} DAY)
                            AND (
                                {better_score_expr} > {candidate_score_expr}
                                OR ({better_score_expr} = {candidate_score_expr} AND better.id < candidate.id)
                            )
                      )
                  )
                ORDER BY score DESC, candidate.id ASC
                LIMIT {safe_limit}
            """
            rows = await fetch_all(query, (client_source, client_source))

        seen_users = set()
        for row in rows:
            raw_nick = str(row.get("nickname") or "").strip()
            user_key = raw_nick if raw_nick else f"id_{row.get('id')}"
            if user_key in seen_users:
                continue
            seen_users.add(user_key)

            row_id = int(row.get("id") or 0)
            score_val = round(float(row.get("score") or 0.0), 2)
            row_region = row.get("region") or region_code or "US"
            row_country_name = get_country_name_by_region_code(row_region)

            meta = row.get("extra_metadata")
            if isinstance(meta, str):
                try:
                    meta = json.loads(meta)
                except Exception:
                    meta = {}
            elif not isinstance(meta, dict):
                meta = {}

            raw_gender = str(meta.get("gender") or "").strip().lower()
            if any(w in raw_gender for w in ["female", "woman", "women", "女", "f"]):
                gender = "Female"
            elif any(w in raw_gender for w in ["male", "man", "men", "男", "m"]):
                gender = "Male"
            else:
                gender = "Female" if (row_id % 2 == 0) else "Male"
            age = str(meta.get("age") or (20 + (row_id % 12)))

            # 确定性假头像 ID (1..16)，女 1..8，男 9..16
            if gender == "Female":
                avatar_id = (row_id % 8) + 1
            else:
                avatar_id = (row_id % 8) + 9

            updated_at_val = row.get("updated_at") or row.get("created_at")
            updated_at_str = str(updated_at_val or "")

            items.append({
                "record_id": row_id,
                "file_path": str(row.get("file_path") or ""),
                "rank": len(items) + 1,
                "score": score_val,
                "nickname": str(row.get("nickname") or ""),
                "region_code": row_region,
                "region_name": row_country_name,
                "gender": gender,
                "age": age,
                "avatar_id": avatar_id,
                "created_at": updated_at_str,
                "updated_at": updated_at_str,
            })
            if len(items) >= limit:
                break

    except Exception as exc:
        logger.warning(f"获取颜值排行榜数据库异常: {exc}")
        success = False

    if success:
        _LEADERBOARD_CACHE[cache_key] = (now, items)
    return {
        "success": success,
        "region_code": region_code or "GLOBAL",
        "region_name": get_country_name_by_region_code(region_code) if region_code else "全球",
        "total_records": len(items),
        "items": response_items(items),
        "days_limit": days_limit,
    }



# 清理调试数据 SQL 列表，可在此处增删 SQL
CLEAN_DEBUG_SQLS = [
    "DELETE FROM tpl_app_user_devices WHERE device_model='Xiaomi 23127PN0CC' AND region='CN'",
    "DELETE FROM tpl_app_user_devices WHERE device_id='cfcae0ec66c8b429'",
    "DELETE FROM tpl_app_user_devices WHERE device_id='7bfefa2d7f8c0be3'",
    "DELETE FROM tpl_app_user_devices WHERE device_id='782831daa5e25b35'",
    "DELETE FROM tpl_app_processed_images WHERE nickname LIKE 'android_debug%%'",
    "DELETE FROM tpl_app_processed_images WHERE nickname='string'",
    "DELETE FROM tpl_app_processed_images WHERE nickname='android_release_1b708ef79ac4f484'",
    "DELETE FROM tpl_app_processed_images WHERE nickname='android_release_3912d8c0e7f9c153'",
    "DELETE FROM tpl_app_processed_images WHERE nickname='android_release_782831daa5e25b35'",
    "DELETE FROM tpl_app_processed_images WHERE nickname='9527'",
]


async def clean_debug_data() -> Dict[str, int]:
    """
    清理调试数据
    :return: 返回每个 SQL 影响的行数
    """
    results = {}
    for idx, sql in enumerate(CLEAN_DEBUG_SQLS):
        try:
            affected = await execute(sql)
            results[f"sql_{idx + 1}"] = affected
            logger.info(f"Clean debug SQL #{idx + 1} executed, affected: {affected}")
        except Exception as exc:
            logger.warning(f"Clean debug SQL #{idx + 1} failed: {exc}")
            results[f"sql_{idx + 1}"] = -1
    return results