| 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: |
| |
| |
| |
| |
| 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) |
| |
| 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_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 |
| _LEADERBOARD_CACHE: Dict[str, Tuple[float, List[Dict[str, Any]]]] = {} |
| _CACHE_TTL_LEADERBOARD = 180 |
|
|
|
|
|
|
| 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)) |
|
|
| |
| 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 |
|
|
| |
| 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: |
| |
| 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)) |
| |
| 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 |
| |
| |
| 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)) |
| |
| 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))) |
|
|
| |
| 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, |
| } |
|
|
|
|
|
|
| |
| 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 |
|
|
