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