提交

backend/main.py

@@ -81,6 +81,7 @@ from services.wechat_jsapi_service import WechatJSAPIService
 from services.energy_stone_service import energy_stone_service
 from services.scenario_topic_service import scenario_topic_service
 from services.llm_task_service import llm_task_service
+from services.ab_subtitle_service import ab_subtitle_service
 from scenario_config import is_special_ab_experiment_user
 from models_db import VerificationCode, UserAvatar, UserBackgroundImage, Feedback, BackgroundImage, User
 from session_store import SessionStore
@@ -391,6 +392,10 @@ class AdminUpdateLlmTaskRequest(AdminCreateLlmTaskRequest):
     pass
 
 
+class AdminUpdateAbSubtitleSettingRequest(BaseModel):
+    subtitle_on: int = Field(..., ge=0, le=1, description="字幕开关：0=关，1=开")
+
+
 @app.get("/")
 async def root():
     """健康检查"""
@@ -909,6 +914,17 @@ async def get_journal_latest_topic(user_id: str, _: None = Depends(require_user_
         raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})
 
 
+@app.get("/api/config/ab-subtitle")
+async def get_ab_subtitle_config():
+    """A/B 组字幕开关配置（A=0 关，B=1 开），供前端读取实验策略。"""
+    try:
+        settings_map = await ab_subtitle_service.get_all_settings_cached()
+        return {"success": True, "data": settings_map}
+    except Exception:
+        logger.exception("获取 A/B 字幕配置失败")
+        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})
+
+
 @app.get("/api/users/{user_id}/ab-llm-task")
 async def get_ab_llm_task(user_id: str, _: None = Depends(require_user_session)):
     """A/B 组实验用户：返回当前最新 LLM 任务名称（不含 prompt）。"""
@@ -4224,14 +4240,25 @@ async def get_user_ui_state(user_id: str, _: None = Depends(require_user_session
     """
     try:
         state = await user_service.get_user_ui_state(user_id)
+        user = await user_service.get_user(user_id)
+        special_group = user.get("special_group") if user else None
+
         if not state:
             # 没有记录时返回默认值
+            default_subtitle_on = await ab_subtitle_service.resolve_subtitle_on(special_group, False)
             default_state = UserUIStateResponse(
                 user_id=user_id,
                 current_page=None,
                 visited_routes=[],
+                subtitle_on=default_subtitle_on,
             )
             return {"success": True, "data": default_state.dict()}
+
+        if "subtitle_on" in state:
+            state["subtitle_on"] = await ab_subtitle_service.resolve_subtitle_on(
+                special_group,
+                bool(state.get("subtitle_on")),
+            )
         return {"success": True, "data": state}
     except Exception as e:  # noqa: BLE001
         logger.exception("获取用户UI状态失败")
@@ -4259,6 +4286,15 @@ async def update_user_ui_state(
     更新用户前端可视化状态（部分字段）
     """
     try:
+        user = await user_service.get_user(user_id)
+        special_group = user.get("special_group") if user else None
+        resolved_subtitle_on = request.subtitle_on
+        if request.subtitle_on is not None:
+            resolved_subtitle_on = await ab_subtitle_service.resolve_subtitle_on(
+                special_group,
+                bool(request.subtitle_on),
+            )
+
         updated = await user_service.upsert_user_ui_state(
             user_id=user_id,
             current_page=request.current_page,
@@ -4276,12 +4312,17 @@ async def update_user_ui_state(
             avatar_c_unlocked=request.avatar_c_unlocked,
             has_shared_avatar=request.has_shared_avatar,
             current_avatar_id=request.current_avatar_id,
-            subtitle_on=request.subtitle_on,
+            subtitle_on=resolved_subtitle_on,
             sfx_enabled=request.sfx_enabled,
             bgm_enabled=request.bgm_enabled,
         )
         if not updated:
             raise HTTPException(status_code=500, detail={"code": UPDATE_UI_STATE_FAILED})
+        if updated and "subtitle_on" in updated:
+            updated["subtitle_on"] = await ab_subtitle_service.resolve_subtitle_on(
+                special_group,
+                bool(updated.get("subtitle_on")),
+            )
         return {"success": True, "data": updated}
     except HTTPException:
         raise
@@ -5272,6 +5313,40 @@ async def admin_delete_llm_task(task_id: int):
         raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})
 
 
+@app.get("/api/admin/ab-subtitle-settings")
+async def admin_list_ab_subtitle_settings():
+    """A/B 实验：字幕开关配置列表（web-admin）。"""
+    try:
+        data = await ab_subtitle_service.list_settings()
+        return {"success": True, "data": data}
+    except Exception:
+        logger.exception("查询 A/B 字幕配置失败")
+        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})
+
+
+@app.put("/api/admin/ab-subtitle-settings/{group_key}")
+async def admin_update_ab_subtitle_setting(
+    group_key: str,
+    payload: AdminUpdateAbSubtitleSettingRequest,
+):
+    """A/B 实验：更新某组字幕开关（0=关，1=开）。"""
+    try:
+        row = await ab_subtitle_service.update_setting(
+            group_key=group_key.strip().upper(),
+            subtitle_on=payload.subtitle_on,
+        )
+        if not row:
+            raise HTTPException(status_code=404, detail={"code": "NOT_FOUND"})
+        return {"success": True, "data": row}
+    except HTTPException:
+        raise
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail={"code": "INVALID_PARAMS", "message": str(e)})
+    except Exception:
+        logger.exception("更新 A/B 字幕配置失败")
+        raise HTTPException(status_code=500, detail={"code": SERVER_ERROR})
+
+
 @app.get("/api/admin/users")
 async def admin_list_users(
     keyword: Optional[str] = Query(None, description="按 user_id/username/email/phone 模糊搜索"),

backend/migrations/add_ab_subtitle_settings_table.sql

@@ -0,0 +1,21 @@
+/*
+  A/B 组实验：字幕开关配置表
+  group_key: A / B
+  subtitle_on: 0=关, 1=开
+  执行前请备份。MySQL 5.7+ / 8.0+
+*/
+
+SET NAMES utf8mb4;
+
+CREATE TABLE IF NOT EXISTS `ab_subtitle_settings` (
+  `group_key` varchar(8) COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '实验分组：A / B',
+  `subtitle_on` tinyint(1) NOT NULL DEFAULT '0' COMMENT '字幕开关：0=关，1=开',
+  `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+  `updated_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+  PRIMARY KEY (`group_key`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='A/B组字幕开关配置表';
+
+INSERT INTO `ab_subtitle_settings` (`group_key`, `subtitle_on`) VALUES
+('A', 0),
+('B', 1)
+ON DUPLICATE KEY UPDATE `group_key` = VALUES(`group_key`);

backend/migrations/migrate_ab_subtitle_settings_init.py

@@ -0,0 +1,85 @@
+"""
+A/B 组字幕开关配置表：建表 + 种子数据（A=0 关，B=1 开）
+运行（在 backend 目录）: python migrations/migrate_ab_subtitle_settings_init.py
+"""
+import asyncio
+import sys
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import aiomysql
+
+from config import settings
+
+MIGRATIONS_DIR = Path(__file__).resolve().parent
+
+
+async def table_exists(cursor: Any) -> bool:
+    await cursor.execute(
+        """
+        SELECT COUNT(*) FROM information_schema.tables
+        WHERE table_schema = %s AND table_name = 'ab_subtitle_settings'
+        """,
+        (settings.MYSQL_DATABASE,),
+    )
+    row = await cursor.fetchone()
+    return bool(row and row[0] > 0)
+
+
+async def migrate() -> bool:
+    try:
+        conn: aiomysql.Connection = await aiomysql.connect(
+            host=settings.MYSQL_HOST,
+            port=settings.MYSQL_PORT,
+            user=settings.MYSQL_USER,
+            password=settings.MYSQL_PASSWORD,
+            db=settings.MYSQL_DATABASE,
+            charset="utf8mb4",
+        )
+
+        async with conn.cursor() as cursor:
+            if not await table_exists(cursor):
+                print("🔄 创建 ab_subtitle_settings 表...")
+                ddl = (MIGRATIONS_DIR / "add_ab_subtitle_settings_table.sql").read_text(encoding="utf-8")
+                for stmt in [s.strip() for s in ddl.split(";") if s.strip()]:
+                    if stmt.upper().startswith("SET NAMES"):
+                        continue
+                    await cursor.execute(stmt)
+                await conn.commit()
+                print("✅ ab_subtitle_settings 表创建完成")
+            else:
+                print("✅ ab_subtitle_settings 表已存在")
+                await cursor.execute(
+                    """
+                    INSERT INTO ab_subtitle_settings (group_key, subtitle_on) VALUES
+                    ('A', 0),
+                    ('B', 1)
+                    ON DUPLICATE KEY UPDATE group_key = VALUES(group_key)
+                    """
+                )
+                await conn.commit()
+                print("✅ A/B 组种子数据已确保存在")
+
+            await cursor.execute(
+                "SELECT group_key, subtitle_on, updated_at FROM ab_subtitle_settings ORDER BY group_key"
+            )
+            rows = await cursor.fetchall()
+            print("📊 当前配置:", rows)
+
+        conn.close()
+        return True
+    except Exception:
+        import traceback
+
+        traceback.print_exc()
+        return False
+
+
+if __name__ == "__main__":
+    confirm = input("\n是否继续执行迁移? (y/n): ") if len(sys.argv) <= 1 or sys.argv[1] != "--force" else "y"
+    if confirm.lower() != "y":
+        sys.exit(0)
+    ok: Any = asyncio.run(migrate())
+    sys.exit(0 if ok else 1)

backend/models_db.py

@@ -117,6 +117,31 @@ class User(Base):
         }
 
 
+class AbSubtitleSetting(Base):
+    """A/B 组字幕开关配置表"""
+
+    __tablename__ = "ab_subtitle_settings"
+
+    group_key = Column(String(8), primary_key=True, comment="实验分组：A / B")
+    subtitle_on = Column(Boolean, nullable=False, default=False, comment="字幕开关：0=关，1=开")
+    created_at = Column(TIMESTAMP, server_default=func.current_timestamp(), nullable=False, comment="创建时间")
+    updated_at = Column(
+        TIMESTAMP,
+        server_default=func.current_timestamp(),
+        onupdate=func.current_timestamp(),
+        nullable=False,
+        comment="更新时间",
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "group_key": self.group_key,
+            "subtitle_on": 1 if self.subtitle_on else 0,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
+        }
+
+
 class LlmTask(Base):
     """A/B 组 LLM 任务提示词表"""
 

backend/services/ab_subtitle_service.py

@@ -0,0 +1,116 @@
+"""
+A/B 组字幕开关配置服务
+"""
+import logging
+import time
+from typing import Any, Dict, Literal, Optional, TypedDict
+
+from sqlalchemy import select
+
+from database import db_manager
+from models_db import AbSubtitleSetting
+
+logger = logging.getLogger(__name__)
+
+SpecialGroupKey = Literal["A", "B"]
+_CACHE_TTL_SECONDS = 60.0
+_DEFAULT_SETTINGS: Dict[SpecialGroupKey, int] = {"A": 0, "B": 1}
+_cached_settings: Optional[Dict[SpecialGroupKey, int]] = None
+_cached_at: float = 0.0
+
+
+class AbSubtitleSettingsMap(TypedDict):
+    A: int
+    B: int
+
+
+class AbSubtitleService:
+    async def get_all_settings(self) -> AbSubtitleSettingsMap:
+        """读取 A/B 两组字幕开关配置。"""
+        try:
+            async with db_manager.get_session() as session:
+                result = await session.execute(
+                    select(AbSubtitleSetting).where(AbSubtitleSetting.group_key.in_(("A", "B")))
+                )
+                rows = result.scalars().all()
+
+            settings: Dict[SpecialGroupKey, int] = dict(_DEFAULT_SETTINGS)
+            for row in rows:
+                key = row.group_key
+                if key in ("A", "B"):
+                    settings[key] = 1 if row.subtitle_on else 0
+            return settings  # type: ignore[return-value]
+        except Exception as e:
+            logger.warning("[AbSubtitle] get_all_settings failed: %s", e)
+            return dict(_DEFAULT_SETTINGS)  # type: ignore[return-value]
+
+    async def get_all_settings_cached(
+        self, force_refresh: bool = False
+    ) -> AbSubtitleSettingsMap:
+        global _cached_settings, _cached_at
+        now = time.time()
+        if (
+            not force_refresh
+            and _cached_settings is not None
+            and (now - _cached_at) < _CACHE_TTL_SECONDS
+        ):
+            return _cached_settings  # type: ignore[return-value]
+
+        settings = await self.get_all_settings()
+        _cached_settings = settings
+        _cached_at = now
+        return settings
+
+    async def list_settings(self) -> list[Dict[str, Any]]:
+        rows = await self.get_all_settings_cached()
+        return [
+            {"group_key": "A", "subtitle_on": rows["A"]},
+            {"group_key": "B", "subtitle_on": rows["B"]},
+        ]
+
+    async def update_setting(self, group_key: str, subtitle_on: int) -> Optional[Dict[str, Any]]:
+        if group_key not in ("A", "B"):
+            raise ValueError("group_key must be A or B")
+        if subtitle_on not in (0, 1):
+            raise ValueError("subtitle_on must be 0 or 1")
+
+        try:
+            async with db_manager.get_session() as session:
+                result = await session.execute(
+                    select(AbSubtitleSetting).where(AbSubtitleSetting.group_key == group_key)
+                )
+                row = result.scalar_one_or_none()
+                if row is None:
+                    row = AbSubtitleSetting(group_key=group_key, subtitle_on=bool(subtitle_on))
+                    session.add(row)
+                else:
+                    row.subtitle_on = bool(subtitle_on)
+                await session.commit()
+                await session.refresh(row)
+
+            global _cached_settings, _cached_at
+            _cached_settings = None
+            _cached_at = 0.0
+            return row.to_dict()
+        except Exception as e:
+            logger.error("[AbSubtitle] update_setting failed: %s", e)
+            raise
+
+    async def resolve_subtitle_on(
+        self,
+        special_group: Optional[str],
+        stored_on: bool,
+    ) -> bool:
+        """A/B 组按数据库配置覆盖；普通用户保持原值。"""
+        if special_group not in ("A", "B"):
+            return stored_on
+        settings = await self.get_all_settings_cached()
+        return bool(settings.get(special_group, _DEFAULT_SETTINGS[special_group]))  # type: ignore[arg-type]
+
+    def invalidate_cache(self) -> None:
+        global _cached_settings, _cached_at
+        _cached_settings = None
+        _cached_at = 0.0
+
+
+ab_subtitle_service = AbSubtitleService()

frontend/src/components/UserGuard.tsx

@@ -11,7 +11,7 @@ import {
   syncUserDataFromDatabase,
   hasValidSession,
 } from '@/utils/storage';
-import { getUserInfo, getUserUIState, ensureSession, ApiError } from '@/utils/api';
+import { getUserInfo, getUserUIState, ensureSession, fetchAndCacheAbSubtitleSettings, ApiError } from '@/utils/api';
 import { preloadSfxAssets } from '@/utils/sfxPreload';
 
 /**
@@ -158,6 +158,10 @@ export const UserGuard = ({ children }: { children: React.ReactNode }) => {
             console.warn('[UserGuard] 获取用户UI状态失败:', error);
             return null;
           }),
+          fetchAndCacheAbSubtitleSettings().catch((error) => {
+            console.warn('[UserGuard] 获取 A/B 字幕配置失败:', error);
+            return null;
+          }),
         ]);
 
         const userInfoPayload = userInfoResult.ok ? userInfoResult.data : null;

frontend/src/pages/ChatBotPage.tsx

@@ -53,6 +53,7 @@ import {
   getUserInboxMessageDetail,
   claimUserInboxEnergyGift,
   getAbLlmTask,
+  fetchAndCacheAbSubtitleSettings,
   type InboxMessageListItem,
   type UserInboxMessageDetailResponse,
 } from '@/utils/api';
@@ -293,23 +294,36 @@ export const ChatBotPage = () => {
     greetingContextRef.current = greetingContext;
   }, [greetingContext]);
 
-  // 字幕相关状态（A 组强制关、B 组强制开；普通用户读 localStorage）
-  const subtitleAbPolicy = getSubtitleAbPolicy();
+  // 字幕相关状态（A/B 组按数据库配置强制开/关；普通用户读 localStorage）
+  const [subtitleAbPolicy, setSubtitleAbPolicy] = useState<ReturnType<typeof getSubtitleAbPolicy>>(
+    () => getSubtitleAbPolicy()
+  );
   const [subtitleOn, setSubtitleOn] = useState(() => {
     return resolveSubtitleOn(getStorage(STORAGE_KEYS.SUBTITLE_ON) === 'true');
   });
 
-  // A/B 组：进入 chatbot 时强制字幕策略并同步服务端
+  // A/B 组：进入 chatbot 时从数据库拉取配置，强制字幕策略并同步服务端
   useEffect(() => {
-    if (!subtitleAbPolicy) return;
-    const forcedOn = subtitleAbPolicy === 'force_on';
-    setStorage(STORAGE_KEYS.SUBTITLE_ON, forcedOn ? 'true' : 'false');
-    setSubtitleOn(forcedOn);
-    const uid = getUserId();
-    if (uid) {
-      updateUserUIState(uid, { subtitle_on: forcedOn }).catch(() => {});
-    }
-  }, [subtitleAbPolicy]);
+    let cancelled = false;
+    fetchAndCacheAbSubtitleSettings()
+      .then(() => {
+        if (cancelled) return;
+        const policy = getSubtitleAbPolicy();
+        setSubtitleAbPolicy(policy);
+        if (!policy) return;
+        const forcedOn = policy === 'force_on';
+        setStorage(STORAGE_KEYS.SUBTITLE_ON, forcedOn ? 'true' : 'false');
+        setSubtitleOn(forcedOn);
+        const uid = getUserId();
+        if (uid) {
+          updateUserUIState(uid, { subtitle_on: forcedOn }).catch(() => {});
+        }
+      })
+      .catch(() => {});
+    return () => {
+      cancelled = true;
+    };
+  }, []);
 
   const subtitleContainerRef = useRef<HTMLDivElement | null>(null);
   const [subtitleVisible, setSubtitleVisible] = useState(false);

frontend/src/utils/api.ts

@@ -421,6 +421,48 @@ export interface AbLlmTaskResponse {
   } | null;
 }
 
+export interface AbSubtitleSettingsResponse {
+  success: boolean;
+  data: {
+    A: 0 | 1;
+    B: 0 | 1;
+  };
+}
+
+/** A/B 组：获取字幕开关配置（A=0 关，B=1 开） */
+export const getAbSubtitleSettings = async (): Promise<AbSubtitleSettingsResponse> => {
+  const apiBaseUrl = getApiBaseUrl();
+  const response = await fetch(`${apiBaseUrl}/api/config/ab-subtitle`, {
+    method: 'GET',
+    headers: getApiHeaders(),
+  });
+
+  if (!response.ok) {
+    throw await parseApiError(response, 'SERVER_ERROR');
+  }
+
+  return response.json();
+};
+
+/** 拉取 A/B 字幕配置并写入 localStorage 缓存 */
+export const fetchAndCacheAbSubtitleSettings = async (): Promise<{ A: 0 | 1; B: 0 | 1 }> => {
+  const { setAbSubtitleSettingsCache, getAbSubtitleSettingsCache } = await import('./storage');
+  try {
+    const res = await getAbSubtitleSettings();
+    if (res.success && res.data) {
+      const settings = {
+        A: res.data.A === 1 ? 1 as const : 0 as const,
+        B: res.data.B === 1 ? 1 as const : 0 as const,
+      };
+      setAbSubtitleSettingsCache(settings);
+      return settings;
+    }
+  } catch (error) {
+    console.warn('[API] 获取 A/B 字幕配置失败，使用本地缓存', error);
+  }
+  return getAbSubtitleSettingsCache();
+};
+
 /** A/B 组：获取当前最新 LLM 任务名称 */
 export const getAbLlmTask = async (user_id: string): Promise<AbLlmTaskResponse> => {
   const apiBaseUrl = getApiBaseUrl();

frontend/src/utils/storage.ts

@@ -26,6 +26,7 @@ export const STORAGE_KEYS = {
   HAS_CLICKED_SHOP_IN_SETTINGS: 'Lyrria_has_clicked_shop_in_settings', // 是否已在设置弹窗中点击过商店
   AVATAR_C_UNLOCKED: 'Lyrria_avatar_c_unlocked', // 是否已永久展示 Avatar-C
   SUBTITLE_ON: 'Lyrria_subtitle_on', // 是否开启字幕显示
+  AB_SUBTITLE_SETTINGS: 'Lyrria_ab_subtitle_settings', // A/B 组字幕开关配置（来自数据库）
   IS_LOGGED_IN: 'Lyrria_is_logged_in', // 是否已登录
   USER_LEVEL: 'Lyrria_user_level', // 用户当前等级
   HAS_SHARED_AVATAR: 'Lyrria_has_shared_avatar', // 是否已分享头像奖励
@@ -51,14 +52,37 @@ export const isSpecialAbUser = (): boolean => {
   return getStorage(STORAGE_KEYS.IS_SPECIAL_USER) === 'true' && getSpecialGroup() !== null;
 };
 
-/** A/B 组字幕策略：A 强制关闭，B 强制开启，普通用户无策略 */
+/** A/B 组字幕开关配置：0=关，1=开 */
+export type AbSubtitleSettings = { A: 0 | 1; B: 0 | 1 };
+
+const DEFAULT_AB_SUBTITLE_SETTINGS: AbSubtitleSettings = { A: 0, B: 1 };
+
+export const setAbSubtitleSettingsCache = (settings: AbSubtitleSettings): void => {
+  setStorage(STORAGE_KEYS.AB_SUBTITLE_SETTINGS, JSON.stringify(settings));
+};
+
+export const getAbSubtitleSettingsCache = (): AbSubtitleSettings => {
+  try {
+    const raw = getStorage(STORAGE_KEYS.AB_SUBTITLE_SETTINGS);
+    if (!raw) return DEFAULT_AB_SUBTITLE_SETTINGS;
+    const parsed = JSON.parse(raw) as Partial<AbSubtitleSettings>;
+    return {
+      A: parsed.A === 1 ? 1 : 0,
+      B: parsed.B === 1 ? 1 : 0,
+    };
+  } catch {
+    return DEFAULT_AB_SUBTITLE_SETTINGS;
+  }
+};
+
+/** A/B 组字幕策略：按数据库配置决定强制开/关，普通用户无策略 */
 export type SubtitleAbPolicy = 'force_off' | 'force_on';
 
 export const getSubtitleAbPolicy = (): SubtitleAbPolicy | null => {
   const group = getSpecialGroup();
-  if (group === 'A') return 'force_off';
-  if (group === 'B') return 'force_on';
-  return null;
+  if (!group) return null;
+  const settings = getAbSubtitleSettingsCache();
+  return settings[group] === 1 ? 'force_on' : 'force_off';
 };
 
 export const resolveSubtitleOn = (storedOn: boolean): boolean => {

数据提取/export_user_data.py

@@ -0,0 +1,478 @@
+#!/usr/bin/env python3
+"""
+按手机号/邮箱导出用户对话记录与能量石消费记录。
+
+输入：用户数据.txt（每行一个手机号或邮箱）
+输出：output/ 目录下，每个用户一个 Excel（两个 sheet：对话记录、能量石消费记录）
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from dataclasses import dataclass
+from datetime import date, datetime
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+import pymysql
+from openpyxl import Workbook
+from openpyxl.styles import Font
+from openpyxl.utils import get_column_letter
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_INPUT = Path(__file__).resolve().parent / "用户数据.txt"
+DEFAULT_OUTPUT_DIR = Path(__file__).resolve().parent / "output"
+
+MESSAGE_HEADERS = [
+    "轮次",
+    "角色",
+    "内容",
+    "情感",
+    "Token数",
+    "创建时间",
+    "消息ID",
+    "扩展信息",
+]
+
+CONSUMPTION_HEADERS = [
+    "记录ID",
+    "订单ID",
+    "商品ID",
+    "商品名称",
+    "购买数量",
+    "能量石单价",
+    "水晶石单价",
+    "消耗能量石",
+    "消耗水晶石",
+    "图片名称",
+    "消费时间",
+    "数据来源",
+]
+
+SUMMARY_HEADERS = [
+    "序号",
+    "联系方式",
+    "类型",
+    "是否匹配",
+    "平台用户名",
+    "user_id",
+    "手机号",
+    "邮箱",
+    "对话条数",
+    "消费记录条数",
+    "导出文件名",
+    "备注",
+]
+
+
+@dataclass
+class Contact:
+    raw: str
+    is_email: bool
+
+    @property
+    def norm(self) -> str:
+        value = self.raw.strip()
+        return value.lower() if self.is_email else value
+
+
+def parse_env_file(env_path: Path) -> dict[str, str]:
+    env_map: dict[str, str] = {}
+    if not env_path.exists():
+        return env_map
+
+    for raw in env_path.read_text(encoding="utf-8").splitlines():
+        line = raw.strip()
+        if not line or line.startswith("#"):
+            continue
+        if line.startswith("export "):
+            line = line[len("export ") :].strip()
+        if "=" not in line:
+            continue
+        key, val = line.split("=", 1)
+        key = key.strip()
+        val = val.strip()
+        if not key:
+            continue
+        if len(val) >= 2 and ((val[0] == '"' and val[-1] == '"') or (val[0] == "'" and val[-1] == "'")):
+            val = val[1:-1]
+        env_map[key] = val
+    return env_map
+
+
+def load_db_env(project_root: Path, explicit_env_file: str | None = None) -> dict[str, str]:
+    candidates: list[Path] = []
+    if explicit_env_file:
+        env_path = Path(explicit_env_file)
+        if not env_path.is_absolute():
+            env_path = (project_root / env_path).resolve()
+        candidates.append(env_path)
+    else:
+        candidates.append((project_root / "backend/.env.lyrria").resolve())
+        candidates.append((project_root / "backend/.env").resolve())
+
+    merged: dict[str, str] = {}
+    for env_path in candidates:
+        merged.update(parse_env_file(env_path))
+
+    for key in ("MYSQL_HOST", "MYSQL_PORT", "MYSQL_USER", "MYSQL_PASSWORD", "MYSQL_DATABASE"):
+        if os.getenv(key) is not None:
+            merged[key] = os.getenv(key, "")
+    return merged
+
+
+def get_db_config(env_values: dict[str, str]) -> dict[str, Any]:
+    return {
+        "host": env_values.get("MYSQL_HOST", "127.0.0.1"),
+        "port": int(env_values.get("MYSQL_PORT", "3306")),
+        "user": env_values.get("MYSQL_USER", "root"),
+        "password": env_values.get("MYSQL_PASSWORD", ""),
+        "database": env_values.get("MYSQL_DATABASE", "Lyrria"),
+        "charset": "utf8mb4",
+        "cursorclass": pymysql.cursors.DictCursor,
+        "autocommit": True,
+        "connect_timeout": int(env_values.get("MYSQL_CONNECT_TIMEOUT", "10")),
+        "read_timeout": int(env_values.get("MYSQL_READ_TIMEOUT", "60")),
+    }
+
+
+def parse_contacts(input_path: Path) -> list[Contact]:
+    contacts: list[Contact] = []
+    seen: set[str] = set()
+
+    for raw in input_path.read_text(encoding="utf-8").splitlines():
+        line = raw.strip()
+        if not line:
+            continue
+        is_email = "@" in line
+        norm = line.lower() if is_email else line
+        if norm in seen:
+            continue
+        seen.add(norm)
+        contacts.append(Contact(raw=line, is_email=is_email))
+    return contacts
+
+
+def fetch_users_by_contacts(cur, contacts: list[Contact]) -> dict[str, dict[str, Any]]:
+    phones = sorted({c.norm for c in contacts if not c.is_email})
+    emails = sorted({c.norm for c in contacts if c.is_email})
+    matched: dict[str, dict[str, Any]] = {}
+
+    if phones:
+        placeholders = ", ".join(["%s"] * len(phones))
+        sql = (
+            "SELECT id, user_id, username, phone, email, area_code, user_type, "
+            "registered_at, total_messages, total_rounds, conversation_energy_stone_total, "
+            "crystal_stone_total, created_at "
+            f"FROM users WHERE phone IN ({placeholders})"
+        )
+        cur.execute(sql, phones)
+        for row in cur.fetchall():
+            phone = row.get("phone")
+            if phone:
+                matched[str(phone)] = row
+
+    if emails:
+        placeholders = ", ".join(["%s"] * len(emails))
+        sql = (
+            "SELECT id, user_id, username, phone, email, area_code, user_type, "
+            "registered_at, total_messages, total_rounds, conversation_energy_stone_total, "
+            "crystal_stone_total, created_at "
+            f"FROM users WHERE LOWER(email) IN ({placeholders})"
+        )
+        cur.execute(sql, emails)
+        for row in cur.fetchall():
+            email = row.get("email")
+            if email:
+                matched[email.strip().lower()] = row
+
+    return matched
+
+
+def fetch_messages(cur, user_id: str) -> list[dict[str, Any]]:
+    sql = (
+        "SELECT message_id, round_number, role, content, emotion, token_count, "
+        "message_metadata, created_at "
+        "FROM messages "
+        "WHERE user_id = %s "
+        "ORDER BY round_number ASC, id ASC"
+    )
+    cur.execute(sql, (user_id,))
+    return list(cur.fetchall())
+
+
+def _user_match_clause(user_uuid: str, user_pk: int) -> tuple[str, list[Any]]:
+    return (
+        "(CAST(apl.user_id AS CHAR) = %s OR apl.user_id = %s)",
+        [user_uuid, user_pk],
+    )
+
+
+def fetch_consumption_logs(cur, user_uuid: str, user_pk: int) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+
+    where_sql, params = _user_match_clause(user_uuid, user_pk)
+    try:
+        sql = (
+            "SELECT apl.id, NULL AS order_id, apl.item_id, apl.quantity, "
+            "apl.unit_price_energy, apl.unit_price_crystal, apl.cost_energy, apl.cost_crystal, "
+            "apl.item_name_snapshot, apl.image_name_snapshot, apl.created_at, "
+            "'asset_purchase_logs' AS source_table "
+            "FROM asset_purchase_logs apl "
+            f"WHERE {where_sql} "
+            "ORDER BY apl.created_at ASC, apl.id ASC"
+        )
+        cur.execute(sql, params)
+        rows.extend(cur.fetchall())
+    except Exception:
+        pass
+
+    if rows:
+        return rows
+
+    try:
+        sql = (
+            "SELECT poi.id, po.id AS order_id, poi.item_id, poi.quantity, "
+            "poi.unit_price_energy, poi.unit_price_crystal, poi.cost_energy, poi.cost_crystal, "
+            "poi.item_name_snapshot, poi.image_name_snapshot, "
+            "COALESCE(po.completed_at, poi.created_at) AS created_at, "
+            "'purchase_order_items' AS source_table "
+            "FROM purchase_order_items poi "
+            "JOIN purchase_orders po ON poi.order_id = po.id "
+            "WHERE po.status = 'completed' "
+            "AND (CAST(po.user_id AS CHAR) = %s OR po.user_id = %s) "
+            "ORDER BY created_at ASC, poi.id ASC"
+        )
+        cur.execute(sql, [user_uuid, user_pk])
+        rows.extend(cur.fetchall())
+    except Exception:
+        pass
+
+    return rows
+
+
+def normalize_cell(value: Any) -> Any:
+    if value is None:
+        return ""
+    if isinstance(value, datetime):
+        return value.strftime("%Y-%m-%d %H:%M:%S")
+    if isinstance(value, date):
+        return value.strftime("%Y-%m-%d")
+    if isinstance(value, Decimal):
+        return float(value)
+    if isinstance(value, (dict, list)):
+        return json.dumps(value, ensure_ascii=False)
+    return value
+
+
+def autosize_columns(ws, max_width: int = 60) -> None:
+    for col_idx, column_cells in enumerate(ws.columns, start=1):
+        max_len = 0
+        for cell in column_cells:
+            text = "" if cell.value is None else str(cell.value)
+            max_len = max(max_len, len(text))
+        width = min(max(max_len + 2, 10), max_width)
+        ws.column_dimensions[get_column_letter(col_idx)].width = width
+
+
+def write_sheet_header(ws, headers: list[str]) -> None:
+    ws.append(headers)
+    for cell in ws[1]:
+        cell.font = Font(bold=True)
+
+
+def sanitize_filename(name: str) -> str:
+    cleaned = re.sub(r'[\\/:*?"<>|]+', "_", name.strip())
+    cleaned = re.sub(r"\s+", "_", cleaned)
+    return cleaned[:120] or "unknown"
+
+
+def build_output_filename(contact: Contact, user_row: dict[str, Any]) -> str:
+    username = sanitize_filename(str(user_row.get("username") or "user"))
+    contact_part = sanitize_filename(contact.raw)
+    return f"{contact_part}_{username}.xlsx"
+
+
+def export_user_workbook(
+    output_path: Path,
+    user_row: dict[str, Any],
+    messages: list[dict[str, Any]],
+    consumptions: list[dict[str, Any]],
+) -> None:
+    wb = Workbook()
+
+    ws_messages = wb.active
+    ws_messages.title = "对话记录"
+    write_sheet_header(ws_messages, MESSAGE_HEADERS)
+    for row in messages:
+        ws_messages.append(
+            [
+                normalize_cell(row.get("round_number")),
+                normalize_cell(row.get("role")),
+                normalize_cell(row.get("content")),
+                normalize_cell(row.get("emotion")),
+                normalize_cell(row.get("token_count")),
+                normalize_cell(row.get("created_at")),
+                normalize_cell(row.get("message_id")),
+                normalize_cell(row.get("message_metadata")),
+            ]
+        )
+    autosize_columns(ws_messages)
+
+    ws_consume = wb.create_sheet("能量石消费记录")
+    write_sheet_header(ws_consume, CONSUMPTION_HEADERS)
+    for row in consumptions:
+        ws_consume.append(
+            [
+                normalize_cell(row.get("id")),
+                normalize_cell(row.get("order_id")),
+                normalize_cell(row.get("item_id")),
+                normalize_cell(row.get("item_name_snapshot")),
+                normalize_cell(row.get("quantity")),
+                normalize_cell(row.get("unit_price_energy")),
+                normalize_cell(row.get("unit_price_crystal")),
+                normalize_cell(row.get("cost_energy")),
+                normalize_cell(row.get("cost_crystal")),
+                normalize_cell(row.get("image_name_snapshot")),
+                normalize_cell(row.get("created_at")),
+                normalize_cell(row.get("source_table")),
+            ]
+        )
+    autosize_columns(ws_consume)
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    wb.save(output_path)
+
+
+def export_summary_workbook(output_dir: Path, summary_rows: list[list[Any]]) -> Path:
+    summary_path = output_dir / "_导出汇总.xlsx"
+    wb = Workbook()
+    ws = wb.active
+    ws.title = "导出汇总"
+    write_sheet_header(ws, SUMMARY_HEADERS)
+    for row in summary_rows:
+        ws.append(row)
+    autosize_columns(ws)
+    wb.save(summary_path)
+    return summary_path
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="按手机号/邮箱导出用户对话与能量石消费记录")
+    parser.add_argument(
+        "--input",
+        default=str(DEFAULT_INPUT),
+        help="用户名单文件路径（默认：数据提取/用户数据.txt）",
+    )
+    parser.add_argument(
+        "--output-dir",
+        default=str(DEFAULT_OUTPUT_DIR),
+        help="Excel 输出目录（默认：数据提取/output）",
+    )
+    parser.add_argument(
+        "--env-file",
+        default="",
+        help="可选：指定 env 文件路径（默认 backend/.env.lyrria）",
+    )
+    args = parser.parse_args()
+
+    input_path = Path(args.input).resolve()
+    output_dir = Path(args.output_dir).resolve()
+
+    if not input_path.exists():
+        print(f"输入文件不存在: {input_path}", file=sys.stderr)
+        return 1
+
+    contacts = parse_contacts(input_path)
+    if not contacts:
+        print("输入文件中没有有效的手机号或邮箱", file=sys.stderr)
+        return 1
+
+    env_values = load_db_env(PROJECT_ROOT, args.env_file or None)
+    db_conf = get_db_config(env_values)
+
+    try:
+        conn = pymysql.connect(**db_conf)
+    except Exception as exc:
+        print(f"数据库连接失败: {exc}", file=sys.stderr)
+        return 1
+
+    summary_rows: list[list[Any]] = []
+    exported_count = 0
+    matched_count = 0
+
+    try:
+        with conn.cursor() as cur:
+            user_map = fetch_users_by_contacts(cur, contacts)
+
+            for idx, contact in enumerate(contacts, start=1):
+                user_row = user_map.get(contact.norm)
+                if not user_row:
+                    summary_rows.append(
+                        [
+                            idx,
+                            contact.raw,
+                            "邮箱" if contact.is_email else "手机号",
+                            "否",
+                            "",
+                            "",
+                            "",
+                            "",
+                            0,
+                            0,
+                            "",
+                            "未找到对应用户",
+                        ]
+                    )
+                    continue
+
+                matched_count += 1
+                user_uuid = str(user_row["user_id"])
+                user_pk = int(user_row["id"])
+
+                messages = fetch_messages(cur, user_uuid)
+                consumptions = fetch_consumption_logs(cur, user_uuid, user_pk)
+
+                filename = build_output_filename(contact, user_row)
+                output_path = output_dir / filename
+                export_user_workbook(output_path, user_row, messages, consumptions)
+                exported_count += 1
+
+                summary_rows.append(
+                    [
+                        idx,
+                        contact.raw,
+                        "邮箱" if contact.is_email else "手机号",
+                        "是",
+                        user_row.get("username") or "",
+                        user_uuid,
+                        user_row.get("phone") or "",
+                        user_row.get("email") or "",
+                        len(messages),
+                        len(consumptions),
+                        filename,
+                        "",
+                    ]
+                )
+    finally:
+        conn.close()
+
+    summary_path = export_summary_workbook(output_dir, summary_rows)
+
+    print(f"输入联系方式（去重后）: {len(contacts)}")
+    print(f"匹配成功: {matched_count}")
+    print(f"未匹配: {len(contacts) - matched_count}")
+    print(f"已导出 Excel: {exported_count}")
+    print(f"输出目录: {output_dir}")
+    print(f"汇总文件: {summary_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())