- MRFZCCL // QNA STATS
- SYSTEM READY_
-
- """
- if is_retro
- else ""
- )
- inner_open = '' if is_retro else ""
- inner_close = "
" if is_retro else ""
- return f"""
-
-
-
-
-
-
-
-
- """
-
- def _html_to_image(self, html_str: str, filename: str, width: int, height: int) -> str:
- hti = Html2Image(output_path=str(self.output_dir))
- out = f"{filename}.png"
- hti.screenshot(html_str=html_str, save_as=out, size=(width, height))
- return str(self.output_dir / out)
-
- def render_to_image(self, body_html: str, filename: str, title: str, height: int) -> str:
- if self.theme in {"retro_win", "retro", "win95", "win"}:
- height = int(height) + self.RETRO_FRAME_EXTRA_HEIGHT
- html_str = self._build_html(body_html, title)
- return self._html_to_image(html_str, filename, self.CARD_WIDTH, height)
-
- # ======================= body builders (HTML) =======================
- def _build_leaderboard_body(
- self,
- users: List[UserQnAStats],
- title: str,
- sort_key: str,
- mode: str,
- avatar_map: Mapping[str, str],
- ) -> str:
- sorted_users = sorted(
- users,
- key=lambda u: self._safe_int(getattr(u, sort_key, 0), 0),
- reverse=True,
- )
-
- if mode == "correct":
- headers = ["排名", "用户", "正确", "错误", "提示", "准确率"]
- elif mode == "wrong":
- headers = ["排名", "用户", "错误", "正确", "提示", "准确率"]
- else: # hints
- headers = ["排名", "用户", "提示", "正确", "错误", "频率"]
-
- head_html = f"""
-
-
-
- {top_bar}
- {inner_open}
- {body_html}
-
- {inner_close}
-
-
-
-
- """
-
- th_html = "".join(f"
-
-
- Q&A STATS
- {self._esc(title)}
- {self._esc(self._pick_avatar_char(user_name_raw))}
'
-
- row_class = []
- if idx == 1:
- row_class.append("top1")
- row_class_str = f' class="{" ".join(row_class)}"' if row_class else ""
-
- rank_cell = f'{avatar_html}{self._esc(user_name_raw)}
-
-
-
- """
-
- th_html = "".join(f"
-
-
- MATCH
- {self._esc(title)}
- {self._esc(self._pick_avatar_char(user_name_raw))}
'
-
- row_class = []
- if idx == 1:
- row_class.append("top1")
- row_class_str = f' class="{" ".join(row_class)}"' if row_class else ""
-
- rank_cell = f'{avatar_html}{self._esc(user_name_raw)}
-
-
-
- {safe_pct:.1f}%
- {safe_pct/100.0:.2f}
-
-
-
-
-
-
-
- """
-
- def _build_user_profile_body_with_avatar(
- self,
- u: UserQnAStats,
- rank: Mapping[str, Any],
- avatar_data_url: Optional[str],
- honors: Optional[List[MatchHonor]] = None,
- ) -> str:
- user_name_raw = getattr(u, "user_name", "-")
- user_id_raw = getattr(u, "user_id", "-")
-
- correct = self._safe_int(getattr(u, "correct_count", 0))
- wrong = self._safe_int(getattr(u, "wrong_count", 0))
- tip = self._safe_int(getattr(u, "tip_count", 0))
- total = correct + wrong
- acc_pct = (correct / total * 100.0) if total else 0.0
- freq = (tip / total) if total else 0.0
-
- created_at = self._fmt_dt(getattr(u, "created_at", None))
- updated_at = self._fmt_dt(getattr(u, "updated_at", None))
-
- avatar_char = self._pick_avatar_char(user_name_raw)
-
- correct_rank = rank.get("correct_rank", "-")
- wrong_rank = rank.get("wrong_rank", "-")
- tip_rank = rank.get("tip_rank", "-")
-
- avatar_block = (
- f'比赛荣誉
- | 奖牌 | -比赛 | -名次 | -战绩 | -
|---|
{self._esc(avatar_char)}
'
- )
-
- honor_section = self._build_user_honor_section(list(honors or []))
-
- return f"""
-
- {avatar_block}
-
-
-
-
-
-
- USER PROFILE
- {self._esc(user_name_raw)}
- ID · {self._esc(user_id_raw)} · 频率 {freq:.2f}/题
-
-
- {honor_section}
- """
diff --git a/test_plugin/old/mrfzccl/src/QnAStatsRendererIndustrial.py b/test_plugin/old/mrfzccl/src/QnAStatsRendererIndustrial.py
deleted file mode 100644
index cb664556b..000000000
--- a/test_plugin/old/mrfzccl/src/QnAStatsRendererIndustrial.py
+++ /dev/null
@@ -1,13 +0,0 @@
-from .QnAStatsRenderer import QnAStatsRenderer
-
-class QnAStatsRendererIndustrial(QnAStatsRenderer):
- """
- 工业（深色）主题渲染器。
-
- 用法：
- - 白天（浅色）：QnAStatsRenderer(...)
- - 工业（深色）：QnAStatsRendererIndustrial(...)
- """
-
- def __init__(self, output_dir: str = "data/quiz_images"):
- super().__init__(output_dir=output_dir, theme="industrial")
diff --git a/test_plugin/old/mrfzccl/src/QnAStatsRendererRetroWin.py b/test_plugin/old/mrfzccl/src/QnAStatsRendererRetroWin.py
deleted file mode 100644
index 1a5ee3b52..000000000
--- a/test_plugin/old/mrfzccl/src/QnAStatsRendererRetroWin.py
+++ /dev/null
@@ -1,14 +0,0 @@
-from .QnAStatsRenderer import QnAStatsRenderer
-
-class QnAStatsRendererRetroWin(QnAStatsRenderer):
- """
- 复古 Win / 像素风主题渲染器。
-
- 用法：
- - 默认（浅色）：QnAStatsRenderer(...)
- - 工业（深色）：QnAStatsRendererIndustrial(...)
- - 复古（Win）：QnAStatsRendererRetroWin(...)
- """
-
- def __init__(self, output_dir: str = "data/quiz_images"):
- super().__init__(output_dir=output_dir, theme="retro_win")
diff --git a/test_plugin/old/mrfzccl/src/__init__.py b/test_plugin/old/mrfzccl/src/__init__.py
deleted file mode 100644
index e69de29bb..000000000
diff --git a/test_plugin/old/mrfzccl/src/db/__init__.py b/test_plugin/old/mrfzccl/src/db/__init__.py
deleted file mode 100644
index e69de29bb..000000000
diff --git a/test_plugin/old/mrfzccl/src/db/database.py b/test_plugin/old/mrfzccl/src/db/database.py
deleted file mode 100644
index edd472b3d..000000000
--- a/test_plugin/old/mrfzccl/src/db/database.py
+++ /dev/null
@@ -1,110 +0,0 @@
-from collections.abc import AsyncGenerator
-from contextlib import asynccontextmanager
-
-from sqlalchemy import text
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
-from sqlmodel import SQLModel
-from sqlalchemy.exc import OperationalError
-
-import os
-
-
-class DBManager:
- """数据库管理器，负责异步连接和会话管理"""
-
- def __init__(self, db_path: str):
- os.makedirs(os.path.dirname(os.path.abspath(db_path)), exist_ok=True)
-
- self.db_url = f"sqlite+aiosqlite:///{db_path}"
-
- # 创建异步引擎
- self.engine = create_async_engine(
- self.db_url,
- echo=False,
- pool_pre_ping=True,
- pool_recycle=3600,
- # pool_size=5,
- # max_overflow=5,
- )
-
- # 创建会话工厂
- self.async_session = async_sessionmaker(
- self.engine,
- class_=AsyncSession,
- expire_on_commit=False,
- )
- self.async_session_factory = self.async_session
-
- async def init_db(self):
- """初始化数据库，创建所有定义的表"""
-
- async with self.engine.begin() as conn:
- try:
- await conn.run_sync(SQLModel.metadata.create_all)
- except OperationalError as e:
- if "already exists" not in str(e):
- raise
-
- async with self.engine.begin() as conn:
- try:
- await conn.execute(text(
- "ALTER TABLE match ADD COLUMN question_limit INTEGER DEFAULT 0"
- ))
- except OperationalError:
- pass
- try:
- await conn.execute(text(
- "ALTER TABLE match ADD COLUMN time_limit INTEGER DEFAULT 0"
- ))
- except OperationalError:
- pass
- try:
- await conn.execute(text(
- "ALTER TABLE match ADD COLUMN started_at TIMESTAMP"
- ))
- except OperationalError:
- pass
- try:
- await conn.execute(text(
- "ALTER TABLE match_participant ADD COLUMN wrong_count INTEGER DEFAULT 0"
- ))
- except OperationalError:
- pass
- try:
- await conn.execute(text(
- "ALTER TABLE match_participant ADD COLUMN score REAL DEFAULT 0.0"
- ))
- except OperationalError:
- pass
- try:
- await conn.execute(text(
- "ALTER TABLE match_honor ADD COLUMN wrong_count INTEGER DEFAULT 0"
- ))
- except OperationalError:
- pass
- try:
- await conn.execute(text(
- "ALTER TABLE match_honor ADD COLUMN score REAL DEFAULT 0.0"
- ))
- except OperationalError:
- pass
-
- # SQLite 优化 PRAGMA
- async with self.engine.connect() as conn:
- await conn.execute(text("PRAGMA journal_mode=WAL"))
- await conn.execute(text("PRAGMA synchronous=NORMAL"))
- await conn.execute(text("PRAGMA cache_size=-20000"))
- await conn.execute(text("PRAGMA temp_store=MEMORY"))
- await conn.execute(text("PRAGMA mmap_size=134217728"))
- await conn.execute(text("PRAGMA optimize"))
- await conn.commit()
-
- @asynccontextmanager
- async def get_session(self) -> AsyncGenerator[AsyncSession, None]:
- """异步获取数据库会话的上下文管理器"""
- session = self.async_session_factory()
- try:
- async with session.begin():
- yield session
- finally:
- await session.close()
diff --git a/test_plugin/old/mrfzccl/src/db/repo.py b/test_plugin/old/mrfzccl/src/db/repo.py
deleted file mode 100644
index a5e53c1dc..000000000
--- a/test_plugin/old/mrfzccl/src/db/repo.py
+++ /dev/null
@@ -1,1140 +0,0 @@
-from datetime import datetime
-from typing import List, Optional, Tuple
-from sqlalchemy import desc, func, select, update, and_, delete
-from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.engine import CursorResult
-
-from .tables import UserQnAStats, Match, MatchParticipant, MatchHonor
-from .database import DBManager
-
-class UserQnARepo:
- """用户问答统计仓库，封装所有的数据库交互逻辑"""
-
- def __init__(self, db_manager: DBManager):
- self.db = db_manager
-
- # 获取或者创建用户数据
- async def get_or_create_user_stats(self, session: AsyncSession, user_id: str, user_name: str = "") -> UserQnAStats:
- """并发安全的 get_or_create 用户统计记录"""
- stmt = select(UserQnAStats).where(
- UserQnAStats.user_id == user_id
- )
-
- result = await session.execute(stmt)
- record = result.scalar_one_or_none()
-
- if record:
- # 如果用户名称有变化，更新用户名称
- if user_name and record.user_name != user_name:
- record.user_name = user_name
- record.updated_at = datetime.now()
- return record
-
- # 创建新记录（包含 tip_count）
- record = UserQnAStats(
- user_id=user_id,
- user_name=user_name or f"用户_{user_id}",
- correct_count=0,
- wrong_count=0,
- tip_count=0,
- created_at=datetime.now(),
- updated_at=datetime.now()
- )
- session.add(record)
-
- try:
- await session.flush()
- return record
- except IntegrityError:
- # 并发下被其他事务插入，重新读取
- result = await session.execute(stmt)
- return result.scalar_one()
-
- # 获取用户数据
- async def get_user_stats_only(self, session: AsyncSession, user_id: str) -> Optional[UserQnAStats]:
- """
- 只获取用户数据，如果用户不存在则返回None
-
- 参数:
- session: 数据库会话
- user_id: 用户ID
-
- 返回:
- UserQnAStats 或 None
- """
- stmt = select(UserQnAStats).where(
- UserQnAStats.user_id == user_id
- )
-
- result = await session.execute(stmt)
- return result.scalar_one_or_none()
-
- # ========== 增加操作 ==========
-
- # 增加答题正确数量
- async def increment_correct_count(self, user_id: str, user_name: str = "", increment: int = 1) -> bool:
- """增加用户答对次数（原子操作）"""
- async with self.db.get_session() as session:
- # 先尝试更新现有记录
- stmt = (
- update(UserQnAStats)
- .where(UserQnAStats.user_id == user_id)
- .values(
- correct_count=UserQnAStats.correct_count + increment,
- updated_at=datetime.now()
- )
- )
-
- result: CursorResult = await session.execute(stmt)
-
- if result.rowcount == 0:
- # 记录不存在，创建新记录
- record = await self.get_or_create_user_stats(session, user_id, user_name)
- record.correct_count += increment
- record.updated_at = datetime.now()
- await session.commit()
- return True
-
- # 如果提供了user_name，检查是否需要更新
- if user_name:
- user_stmt = select(UserQnAStats).where(UserQnAStats.user_id == user_id)
- user_result = await session.execute(user_stmt)
- user_record = user_result.scalar_one()
- if user_record.user_name != user_name:
- user_record.user_name = user_name
- user_record.updated_at = datetime.now()
-
- return True
-
- # 增加答题错误数量
- async def increment_wrong_count(self, user_id: str, user_name: str = "", increment: int = 1) -> bool:
- """增加用户答错次数（原子操作）"""
- async with self.db.get_session() as session:
- stmt = (
- update(UserQnAStats)
- .where(UserQnAStats.user_id == user_id)
- .values(
- wrong_count=UserQnAStats.wrong_count + increment,
- updated_at=datetime.now()
- )
- )
-
- result: CursorResult = await session.execute(stmt)
-
- if result.rowcount == 0:
- # 记录不存在，创建新记录
- record = await self.get_or_create_user_stats(session, user_id, user_name)
- record.wrong_count += increment
- record.updated_at = datetime.now()
- await session.commit()
- return True
-
- # 如果提供了user_name，检查是否需要更新
- if user_name:
- user_stmt = select(UserQnAStats).where(UserQnAStats.user_id == user_id)
- user_result = await session.execute(user_stmt)
- user_record = user_result.scalar_one()
- if user_record.user_name != user_name:
- user_record.user_name = user_name
- user_record.updated_at = datetime.now()
-
- return True
-
- # 增加提示次数
- async def increment_tip_count(self, user_id: str, user_name: str = "", increment: int = 1) -> bool:
- """增加用户提示次数（原子操作）"""
- async with self.db.get_session() as session:
- stmt = (
- update(UserQnAStats)
- .where(UserQnAStats.user_id == user_id)
- .values(
- tip_count=UserQnAStats.tip_count + increment,
- updated_at=datetime.now()
- )
- )
-
- result: CursorResult = await session.execute(stmt)
-
- if result.rowcount == 0:
- # 记录不存在，创建新记录
- record = await self.get_or_create_user_stats(session, user_id, user_name)
- record.tip_count += increment
- record.updated_at = datetime.now()
- await session.commit()
- return True
-
- # 如果提供了user_name，检查是否需要更新
- if user_name:
- user_stmt = select(UserQnAStats).where(UserQnAStats.user_id == user_id)
- user_result = await session.execute(user_stmt)
- user_record = user_result.scalar_one()
- if user_record.user_name != user_name:
- user_record.user_name = user_name
- user_record.updated_at = datetime.now()
-
- return True
-
- # 增加答题正确和答题错误数量
- async def increment_both_counts(self, user_id: str, user_name: str = "", correct_increment: int = 1, wrong_increment: int = 1) -> bool:
- """同时增加用户答对和答错次数（原子操作）"""
- async with self.db.get_session() as session:
- # 先尝试更新现有记录
- stmt = (
- update(UserQnAStats)
- .where(UserQnAStats.user_id == user_id)
- .values(
- correct_count=UserQnAStats.correct_count + correct_increment,
- wrong_count=UserQnAStats.wrong_count + wrong_increment,
- updated_at=datetime.now()
- )
- )
-
- result: CursorResult = await session.execute(stmt)
-
- if result.rowcount == 0:
- # 记录不存在，创建新记录
- record = await self.get_or_create_user_stats(session, user_id, user_name)
- record.correct_count += correct_increment
- record.wrong_count += wrong_increment
- record.updated_at = datetime.now()
- await session.commit()
- return True
-
- # 如果提供了user_name，检查是否需要更新
- if user_name:
- user_stmt = select(UserQnAStats).where(UserQnAStats.user_id == user_id)
- user_result = await session.execute(user_stmt)
- user_record = user_result.scalar_one()
- if user_record.user_name != user_name:
- user_record.user_name = user_name
- user_record.updated_at = datetime.now()
-
- return True
-
- # 增加所有计数器（正确、错误、提示）
- async def increment_all_counts(self, user_id: str, user_name: str = "", correct_increment: int = 0, wrong_increment: int = 0, tip_increment: int = 0) -> bool:
- """同时增加用户所有计数器（原子操作）"""
- async with self.db.get_session() as session:
- # 构建更新值字典
- update_values = {"updated_at": datetime.now()}
-
- if correct_increment != 0:
- update_values["correct_count"] = UserQnAStats.correct_count + correct_increment
- if wrong_increment != 0:
- update_values["wrong_count"] = UserQnAStats.wrong_count + wrong_increment
- if tip_increment != 0:
- update_values["tip_count"] = UserQnAStats.tip_count + tip_increment
-
- # 如果没有实际更新，直接返回
- if len(update_values) == 1: # 只有 updated_at
- return True
-
- # 先尝试更新现有记录
- stmt = (
- update(UserQnAStats)
- .where(UserQnAStats.user_id == user_id)
- .values(**update_values)
- )
-
- result: CursorResult = await session.execute(stmt)
-
- if result.rowcount == 0:
- # 记录不存在，创建新记录
- record = await self.get_or_create_user_stats(session, user_id, user_name)
-
- if correct_increment != 0:
- record.correct_count += correct_increment
- if wrong_increment != 0:
- record.wrong_count += wrong_increment
- if tip_increment != 0:
- record.tip_count += tip_increment
-
- record.updated_at = datetime.now()
- await session.commit()
- return True
-
- # 如果提供了user_name，检查是否需要更新
- if user_name:
- user_stmt = select(UserQnAStats).where(UserQnAStats.user_id == user_id)
- user_result = await session.execute(user_stmt)
- user_record = user_result.scalar_one()
- if user_record.user_name != user_name:
- user_record.user_name = user_name
- user_record.updated_at = datetime.now()
-
- return True
-
- # ========== 查询操作 ==========
-
- # 按ID查找用户并获取当前排名
- async def get_user_stats_with_rank(self, user_id: str) -> Tuple[Optional[UserQnAStats], Optional[int], int]:
- """
- 按ID查找用户并获取当前排名
-
- 返回:
- (用户统计记录, 排名(从1开始), 总用户数)
- 如果用户不存在，返回(None, None, 总用户数)
- """
- async with self.db.get_session() as session:
- # 获取总用户数
- total_stmt = select(func.count()).select_from(UserQnAStats)
- total_result = await session.execute(total_stmt)
- total_users = total_result.scalar_one() or 0
-
- if total_users == 0:
- return None, None, 0
-
- # 获取用户记录（包含 tip_count）
- user_stmt = select(UserQnAStats).where(UserQnAStats.user_id == user_id)
- user_result = await session.execute(user_stmt)
- user_record = user_result.scalar_one_or_none()
-
- if not user_record:
- return None, None, total_users
-
- # 计算排名: correct_count 越高排名越高
- # 使用窗口函数或子查询计算排名
- rank_stmt = select(
- func.count().label('rank')
- ).where(
- and_(
- UserQnAStats.correct_count > user_record.correct_count,
- UserQnAStats.user_id != user_id
- )
- )
- rank_result = await session.execute(rank_stmt)
- # 排名从1开始
- rank = (rank_result.scalar_one() or 0) + 1
-
- return user_record, rank, total_users
-
- # 获取用户统计数据（简单版本）
- async def get_user_stats(self, user_id: str) -> Optional[UserQnAStats]:
- """获取用户统计数据（包含 tip_count）"""
- async with self.db.get_session() as session:
- stmt = select(UserQnAStats).where(UserQnAStats.user_id == user_id)
- result = await session.execute(stmt)
- return result.scalar_one_or_none()
-
- # 按照正确的题目数量从大到小排序，返回前N个用户
- async def get_top_users(self, limit: int = 10) -> List[UserQnAStats]:
- """
- 按照正确的题目数量从大到小排序，返回前N个用户
-
- 参数:
- limit: 返回的用户数量
-
- 返回:
- 排名前N的用户列表（包含 tip_count）
- """
- async with self.db.get_session() as session:
- stmt = (
- select(UserQnAStats)
- .order_by(desc(UserQnAStats.correct_count), UserQnAStats.updated_at)
- .limit(limit)
- )
-
- result = await session.execute(stmt)
- return list(result.scalars().all())
-
- # 按照提示次数从多到少排序，返回前N个用户
- async def get_top_tip_users(self, limit: int = 10) -> List[UserQnAStats]:
- """
- 按照提示次数从多到少排序，返回前N个用户
-
- 参数:
- limit: 返回的用户数量
-
- 返回:
- 提示次数最多的用户列表
- """
- async with self.db.get_session() as session:
- stmt = (
- select(UserQnAStats)
- .order_by(desc(UserQnAStats.tip_count), desc(UserQnAStats.correct_count))
- .limit(limit)
- )
-
- result = await session.execute(stmt)
- return list(result.scalars().all())
-
- # 获取用户提示次数统计
- async def get_user_tip_stats(self, user_id: str) -> Tuple[int, int, int]:
- """
- 获取用户提示次数及相关统计
-
- 返回:
- (tip_count, correct_count, wrong_count)
- """
- async with self.db.get_session() as session:
- stmt = select(
- UserQnAStats.tip_count,
- UserQnAStats.correct_count,
- UserQnAStats.wrong_count
- ).where(UserQnAStats.user_id == user_id)
-
- result = await session.execute(stmt)
- row = result.one_or_none()
-
- if row:
- return row.tip_count, row.correct_count, row.wrong_count
- return 0, 0, 0
-
- # ========== 创建/更新操作 ==========
-
- # 创建或更新用户统计记录（完整记录）
- async def create_or_update_user(self, user_id: str, user_name: str, correct_count: int = 0, wrong_count: int = 0, tip_count: int = 0) -> UserQnAStats:
- """创建或更新用户统计记录（完整记录，包含 tip_count）"""
- async with self.db.get_session() as session:
- record = await self.get_or_create_user_stats(session, user_id, user_name)
-
- # 更新数据
- record.correct_count = correct_count
- record.wrong_count = wrong_count
- record.tip_count = tip_count
- record.updated_at = datetime.now()
-
- return record
-
- # 批量创建或更新用户统计记录
- async def batch_create_or_update_users(self, users_data: List[dict]) -> int:
- """
- 批量创建或更新用户统计记录
-
- 参数:
- users_data: 用户数据列表，每个元素为字典，包含:
- - user_id: 用户ID
- - user_name: 用户名
- - correct_count: 答对数量
- - wrong_count: 答错数量
- - tip_count: 提示数量（可选）
-
- 返回:
- 成功处理的数量
- """
- if not users_data:
- return 0
-
- processed_count = 0
- async with self.db.get_session() as session:
- for user_data in users_data:
- try:
- record = await self.get_or_create_user_stats(
- session,
- user_data['user_id'],
- user_data.get('user_name', '')
- )
-
- # 更新数据（可以设置为增量或覆盖，这里用增量）
- if 'correct_count' in user_data:
- record.correct_count += user_data['correct_count']
- if 'wrong_count' in user_data:
- record.wrong_count += user_data['wrong_count']
- if 'tip_count' in user_data:
- record.tip_count += user_data['tip_count']
-
- record.updated_at = datetime.now()
- processed_count += 1
-
- except Exception as e:
- # 记录错误但继续处理其他用户
- print(f"处理用户 {user_data.get('user_id')} 时出错: {e}")
- continue
-
- await session.commit()
-
- return processed_count
-
- # 更新用户提示次数（直接设置）
- async def update_tip_count(self, user_id: str, tip_count: int, user_name: str = "") -> bool:
- """更新用户提示次数（直接设置值）"""
- async with self.db.get_session() as session:
- stmt = (
- update(UserQnAStats)
- .where(UserQnAStats.user_id == user_id)
- .values(
- tip_count=tip_count,
- updated_at=datetime.now()
- )
- )
-
- result: CursorResult = await session.execute(stmt)
-
- if result.rowcount == 0:
- # 记录不存在，创建新记录
- record = await self.get_or_create_user_stats(session, user_id, user_name)
- record.tip_count = tip_count
- record.updated_at = datetime.now()
- await session.commit()
- return True
-
- # 如果提供了user_name，检查是否需要更新
- if user_name:
- user_stmt = select(UserQnAStats).where(UserQnAStats.user_id == user_id)
- user_result = await session.execute(user_stmt)
- user_record = user_result.scalar_one()
- if user_record.user_name != user_name:
- user_record.user_name = user_name
- user_record.updated_at = datetime.now()
-
- return True
-
- # ========== 其他操作 ==========
-
- # 分页获取用户排名
- async def get_user_rankings_page(self, page: int = 1, page_size: int = 20) -> Tuple[List[UserQnAStats], int]:
- """
- 分页获取用户排名
-
- 参数:
- page: 页码（从1开始）
- page_size: 每页数量
-
- 返回:
- (当前页的用户列表, 总用户数)
- """
- async with self.db.get_session() as session:
- # 获取总用户数
- total_stmt = select(func.count()).select_from(UserQnAStats)
- total_result = await session.execute(total_stmt)
- total_users = total_result.scalar_one() or 0
-
- # 计算偏移量
- offset = (page - 1) * page_size
-
- # 获取分页数据
- stmt = (
- select(UserQnAStats)
- .order_by(desc(UserQnAStats.correct_count), UserQnAStats.updated_at)
- .offset(offset)
- .limit(page_size)
- )
-
- result = await session.execute(stmt)
- users = list(result.scalars().all())
-
- return users, total_users
-
- # 根据用户名关键词搜索用户
- async def search_users_by_name(self, name_keyword: str, limit: int = 10) -> List[UserQnAStats]:
- """
- 根据用户名关键词搜索用户
-
- 参数:
- name_keyword: 用户名关键词
- limit: 返回的最大数量
-
- 返回:
- 匹配的用户列表，按正确数量排序
- """
- async with self.db.get_session() as session:
- stmt = (
- select(UserQnAStats)
- .where(UserQnAStats.user_name.contains(name_keyword))
- .order_by(desc(UserQnAStats.correct_count), UserQnAStats.updated_at)
- .limit(limit)
- )
-
- result = await session.execute(stmt)
- return list(result.scalars().all())
-
- # 获取总用户数量
- async def get_user_total_count(self) -> int:
- """获取总用户数量"""
- async with self.db.get_session() as session:
- stmt = select(func.count()).select_from(UserQnAStats)
- result = await session.execute(stmt)
- return result.scalar_one() or 0
-
- # 获取提示次数统计
- async def get_tip_stats_summary(self) -> Tuple[int, float]:
- """
- 获取提示次数统计摘要
-
- 返回:
- (总提示次数, 平均每用户提示次数)
- """
- async with self.db.get_session() as session:
- # 总提示次数
- total_tips_stmt = select(func.sum(UserQnAStats.tip_count))
- total_tips_result = await session.execute(total_tips_stmt)
- total_tips = total_tips_result.scalar_one() or 0
-
- # 总用户数
- total_users_stmt = select(func.count()).select_from(UserQnAStats)
- total_users_result = await session.execute(total_users_stmt)
- total_users = total_users_result.scalar_one() or 0
-
- # 平均提示次数
- avg_tips = total_tips / total_users if total_users > 0 else 0
-
- return total_tips, avg_tips
-
- # 删除用户统计记录
- async def delete_user_stats(self, user_id: str) -> bool:
- """删除用户统计记录"""
- async with self.db.get_session() as session:
- stmt = select(UserQnAStats).where(UserQnAStats.user_id == user_id)
- result = await session.execute(stmt)
- record = result.scalar_one_or_none()
-
- if record:
- await session.delete(record)
- return True
- return False
-
- # 批量获取多个用户的统计信息
- async def get_user_stats_by_ids(self, user_ids: List[str]) -> List[UserQnAStats]:
- """批量获取多个用户的统计信息"""
- if not user_ids:
- return []
-
- async with self.db.get_session() as session:
- stmt = select(UserQnAStats).where(UserQnAStats.user_id.in_(user_ids))
- result = await session.execute(stmt)
- return list(result.scalars().all())
-
- # ========== 信息获取操作 ==========
-
- # 获取正确个数排行榜
- async def get_correct_answers_leaderboard(self, limit: int = 10, offset: int = 0) -> List[UserQnAStats]:
- """
- 获取正确个数排行榜
-
- 参数:
- limit: 返回的用户数量
- offset: 偏移量（用于分页）
-
- 返回:
- 按正确个数降序排列的用户列表
- """
- async with self.db.get_session() as session:
- stmt = (
- select(UserQnAStats)
- .order_by(desc(UserQnAStats.correct_count), UserQnAStats.updated_at)
- .offset(offset)
- .limit(limit)
- )
-
- result = await session.execute(stmt)
- return list(result.scalars().all())
-
- # 重置用户答题数据
- async def reset_user_stats(self, user_id: str):
- """重置用户答题数据"""
- async with self.db.get_session() as session:
- stmt = (
- update(UserQnAStats)
- .where(UserQnAStats.user_id == user_id)
- .values(
- correct_count=0,
- wrong_count=0,
- tip_count=0,
- updated_at=datetime.now()
- )
- )
- await session.execute(stmt)
- await session.commit()
-
- # 重置所有用户的答题数据
- async def reset_all_stats(self):
- """重置所有用户的答题数据"""
- async with self.db.get_session() as session:
- stmt = (
- update(UserQnAStats)
- .values(
- correct_count=0,
- wrong_count=0,
- tip_count=0,
- updated_at=datetime.now()
- )
- )
- await session.execute(stmt)
- await session.commit()
-
- # 获取错误个数排行榜
- async def get_wrong_answers_leaderboard(self, limit: int = 10, offset: int = 0) -> List[UserQnAStats]:
- """
- 获取错误个数排行榜
-
- 参数:
- limit: 返回的用户数量
- offset: 偏移量（用于分页）
-
- 返回:
- 按错误个数降序排列的用户列表
- """
- async with self.db.get_session() as session:
- stmt = (
- select(UserQnAStats)
- .order_by(desc(UserQnAStats.wrong_count), UserQnAStats.updated_at)
- .offset(offset)
- .limit(limit)
- )
-
- result = await session.execute(stmt)
- return list(result.scalars().all())
-
- # 获取提示次数排行榜
- async def get_hints_usage_leaderboard(self, limit: int = 10, offset: int = 0) -> List[UserQnAStats]:
- """
- 获取提示次数排行榜
-
- 参数:
- limit: 返回的用户数量
- offset: 偏移量（用于分页）
-
- 返回:
- 按提示次数降序排列的用户列表
- """
- async with self.db.get_session() as session:
- stmt = (
- select(UserQnAStats)
- .order_by(desc(UserQnAStats.tip_count), UserQnAStats.updated_at)
- .offset(offset)
- .limit(limit)
- )
-
- result = await session.execute(stmt)
- return list(result.scalars().all())
-
- # 获取用户个人信息及在各种排行榜中的排名
- async def get_user_profile_with_rank(self, user_id: str) -> Tuple[Optional[UserQnAStats], dict]:
- """
- 获取用户个人信息及在各种排行榜中的排名
-
- 参数:
- user_id: 用户ID
-
- 返回:
- (用户数据, 排名信息字典)
- 如果用户不存在: (None, {})
- """
- async with self.db.get_session() as session:
- # 获取用户数据
- user_stats = await self.get_user_stats_only(session, user_id)
- if not user_stats:
- return None, {}
-
- # 计算正确个数排名
- correct_rank_stmt = select(
- func.count().label('rank')
- ).where(
- and_(
- UserQnAStats.correct_count > user_stats.correct_count,
- UserQnAStats.user_id != user_id
- )
- )
- correct_rank_result = await session.execute(correct_rank_stmt)
- correct_rank = (correct_rank_result.scalar_one() or 0) + 1
-
- # 计算错误个数排名
- wrong_rank_stmt = select(
- func.count().label('rank')
- ).where(
- and_(
- UserQnAStats.wrong_count > user_stats.wrong_count,
- UserQnAStats.user_id != user_id
- )
- )
- wrong_rank_result = await session.execute(wrong_rank_stmt)
- wrong_rank = (wrong_rank_result.scalar_one() or 0) + 1
-
- # 计算提示次数排名
- tip_rank_stmt = select(
- func.count().label('rank')
- ).where(
- and_(
- UserQnAStats.tip_count > user_stats.tip_count,
- UserQnAStats.user_id != user_id
- )
- )
- tip_rank_result = await session.execute(tip_rank_stmt)
- tip_rank = (tip_rank_result.scalar_one() or 0) + 1
-
- # 获取总用户数
- total_stmt = select(func.count()).select_from(UserQnAStats)
- total_result = await session.execute(total_stmt)
- total_users = total_result.scalar_one() or 0
-
- # 计算准确率（如果答过题）
- total_answers = user_stats.correct_count + user_stats.wrong_count
- accuracy = (user_stats.correct_count / total_answers * 100) if total_answers > 0 else 0
-
- rank_info = {
- 'correct_rank': correct_rank,
- 'wrong_rank': wrong_rank,
- 'tip_rank': tip_rank,
- 'total_users': total_users,
- 'accuracy': accuracy,
- 'total_answers': total_answers
- }
-
- return user_stats, rank_info
-
- # 获取排行榜概要信息
- async def get_leaderboard_summary(self) -> dict:
- """
- 获取排行榜概要信息
-
- 返回:
- 包含排行榜统计信息的字典
- """
- async with self.db.get_session() as session:
- # 获取总用户数
- total_users_stmt = select(func.count()).select_from(UserQnAStats)
- total_users_result = await session.execute(total_users_stmt)
- total_users = total_users_result.scalar_one() or 0
-
- # 获取总正确数
- total_correct_stmt = select(func.sum(UserQnAStats.correct_count))
- total_correct_result = await session.execute(total_correct_stmt)
- total_correct = total_correct_result.scalar_one() or 0
-
- # 获取总错误数
- total_wrong_stmt = select(func.sum(UserQnAStats.wrong_count))
- total_wrong_result = await session.execute(total_wrong_stmt)
- total_wrong = total_wrong_result.scalar_one() or 0
-
- # 获取总提示次数
- total_tips_stmt = select(func.sum(UserQnAStats.tip_count))
- total_tips_result = await session.execute(total_tips_stmt)
- total_tips = total_tips_result.scalar_one() or 0
-
- # 获取平均正确数
- avg_correct = total_correct / total_users if total_users > 0 else 0
-
- return {
- 'total_users': total_users,
- 'total_correct': total_correct,
- 'total_wrong': total_wrong,
- 'total_tips': total_tips,
- 'avg_correct': avg_correct,
- 'total_questions': total_correct + total_wrong
- }
-
-class MatchRepo:
- """比赛数据仓库"""
-
- def __init__(self, db_manager: DBManager):
- self.db = db_manager
-
- # 创建新比赛
- async def create_match(self, group_id: str, match_name: str, question_limit: int = 0, time_limit: int = 0) -> Match:
- """
- 创建新比赛
-
- 参数:
- group_id: 群组ID
- match_name: 比赛名称
- question_limit: 题目数量限制（0表示无限制）
- time_limit: 时间限制（分钟，0表示无限制）
-
- 返回:
- 创建的比赛对象
- """
- async with self.db.get_session() as session:
- match = Match(
- group_id=group_id,
- match_name=match_name,
- is_active=True,
- question_limit=question_limit,
- time_limit=time_limit
- )
- session.add(match)
- await session.commit()
- return match
-
- # 获取活跃比赛
- async def get_active_match(self, group_id: str) -> Optional[Match]:
- """
- 获取指定群组的活跃比赛
-
- 参数:
- group_id: 群组ID
-
- 返回:
- 活跃的比赛对象，如果不存在则返回None
- """
- async with self.db.get_session() as session:
- stmt = select(Match).where(
- and_(Match.group_id == group_id, Match.is_active)
- )
- result = await session.execute(stmt)
- return result.scalar_one_or_none()
-
- # 开始比赛
- async def start_match(self, match_id: int):
- """
- 开始指定ID的比赛
-
- 参数:
- match_id: 比赛ID
- """
- async with self.db.get_session() as session:
- stmt = select(Match).where(
- Match.match_id == match_id
- )
- result = await session.execute(stmt)
- match = result.scalar_one_or_none()
- if match:
- match.is_active = True
- match.started_at = datetime.now()
- await session.commit()
-
- # 结束比赛
- async def end_match(self, match_id: int):
- """
- 结束指定ID的比赛
-
- 参数:
- match_id: 比赛ID
- """
- async with self.db.get_session() as session:
- stmt = select(Match).where(
- Match.match_id == match_id
- )
- result = await session.execute(stmt)
- match = result.scalar_one_or_none()
- if match:
- match.is_active = False
- match.ended_at = datetime.now()
- await session.commit()
-
- # 添加参赛者
- async def add_participant(self, match_id: int, user_id: str, user_name: str) -> MatchParticipant:
- """
- 向比赛添加参赛者
-
- 参数:
- match_id: 比赛ID
- user_id: 用户ID
- user_name: 用户名称
-
- 返回:
- 参赛者对象（如果已存在则返回现有对象）
- """
- async with self.db.get_session() as session:
- stmt = select(MatchParticipant).where(
- and_(
- MatchParticipant.match_id == match_id,
- MatchParticipant.user_id == user_id
- )
- )
- result = await session.execute(stmt)
- existing = result.scalar_one_or_none()
- if existing:
- # 同步最新昵称，避免排行榜/名片长期显示旧昵称
- if existing.user_name != user_name:
- existing.user_name = user_name
- await session.commit()
- return existing
- participant = MatchParticipant(
- match_id=match_id,
- user_id=user_id,
- user_name=user_name,
- )
- session.add(participant)
- await session.commit()
- return participant
-
- # 获取参赛者
- async def get_participant(self, match_id: int, user_id: str) -> Optional[MatchParticipant]:
- """
- 获取指定比赛的指定参赛者
-
- 参数:
- match_id: 比赛ID
- user_id: 用户ID
-
- 返回:
- 参赛者对象，如果不存在则返回None
- """
- async with self.db.get_session() as session:
- stmt = select(MatchParticipant).where(
- and_(
- MatchParticipant.match_id == match_id,
- MatchParticipant.user_id == user_id
- )
- )
- result = await session.execute(stmt)
- return result.scalar_one_or_none()
-
- # 获取所有参赛者
- async def get_participants(self, match_id: int) -> List[MatchParticipant]:
- """
- 获取指定比赛的所有参赛者
-
- 参数:
- match_id: 比赛ID
-
- 返回:
- 参赛者对象列表
- """
- async with self.db.get_session() as session:
- stmt = select(MatchParticipant).where(MatchParticipant.match_id == match_id)
- result = await session.execute(stmt)
- return list(result.scalars().all())
-
- # 增加参赛者得分
- async def increment_participant_score(self, match_id: int, user_id: str):
- """
- 增加参赛者的正确答题数和分数
-
- 参数:
- match_id: 比赛ID
- user_id: 用户ID
- """
- async with self.db.get_session() as session:
- stmt = select(MatchParticipant).where(
- and_(
- MatchParticipant.match_id == match_id,
- MatchParticipant.user_id == user_id
- )
- )
- result = await session.execute(stmt)
- participant = result.scalar_one_or_none()
- if participant:
- participant.correct_count += 1
- participant.score = participant.correct_count - participant.wrong_count / 3.0
- await session.commit()
-
- # 增加参赛者错误数
- async def increment_participant_wrong(self, match_id: int, user_id: str):
- """
- 增加参赛者的错误答题数并更新分数
-
- 参数:
- match_id: 比赛ID
- user_id: 用户ID
- """
- async with self.db.get_session() as session:
- stmt = select(MatchParticipant).where(
- and_(
- MatchParticipant.match_id == match_id,
- MatchParticipant.user_id == user_id
- )
- )
- result = await session.execute(stmt)
- participant = result.scalar_one_or_none()
- if participant:
- participant.wrong_count += 1
- participant.score = participant.correct_count - participant.wrong_count / 3.0
- await session.commit()
-
- # 保存荣誉记录
- async def save_honor(self, user_id: str, match_id: int, match_name: str, rank: int, correct_count: int, wrong_count: int = 0, score: float = 0.0):
- """
- 保存用户的比赛荣誉记录
-
- 参数:
- user_id: 用户ID
- match_id: 比赛ID
- match_name: 比赛名称
- rank: 排名
- correct_count: 正确答题数
- wrong_count: 错误答题数（默认为0）
- score: 最终得分（默认为0.0）
- """
- medals = {1: "🥇", 2: "🥈", 3: "🥉"}
- medal = medals.get(rank, f"{rank}名")
- async with self.db.get_session() as session:
- # match_id=0 为“虚拟比赛ID”（手动授予荣誉），不适用按 match_id 去重
- if match_id != 0:
- existing_stmt = select(MatchHonor).where(
- and_(MatchHonor.user_id == user_id, MatchHonor.match_id == match_id)
- )
- result = await session.execute(existing_stmt)
- existing_list = list(result.scalars().all())
-
- if existing_list:
- keep = existing_list[0]
- keep.match_name = match_name
- keep.rank = rank
- keep.correct_count = correct_count
- keep.wrong_count = wrong_count
- keep.score = score
- keep.medal = medal
-
- # 兼容历史重复数据：清理多余的重复荣誉记录
- for extra in existing_list[1:]:
- try:
- session.delete(extra)
- except Exception:
- pass
-
- await session.commit()
- return
-
- honor = MatchHonor(
- user_id=user_id, match_id=match_id, match_name=match_name,
- rank=rank, correct_count=correct_count, wrong_count=wrong_count,
- score=score, medal=medal
- )
- session.add(honor)
- await session.commit()
-
- # 获取用户荣誉
- async def get_user_honors(self, user_id: str) -> List[MatchHonor]:
- """
- 获取用户的荣誉记录
-
- 参数:
- user_id: 用户ID
-
- 返回:
- 荣誉记录列表，按排名降序排列
- """
- async with self.db.get_session() as session:
- stmt = select(MatchHonor).where(MatchHonor.user_id == user_id).order_by(desc(MatchHonor.rank))
- result = await session.execute(stmt)
- honors = list(result.scalars().all())
-
- # 兼容历史数据：同一场比赛可能被重复写入荣誉，名片展示时去重
- deduped: list[MatchHonor] = []
- seen_match_ids: set[int] = set()
- seen_virtual: set[tuple] = set()
- for h in honors:
- mid = getattr(h, "match_id", 0) or 0
- if mid != 0:
- if int(mid) in seen_match_ids:
- continue
- seen_match_ids.add(int(mid))
- else:
- key = (
- str(getattr(h, "match_name", "") or ""),
- int(getattr(h, "rank", 0) or 0),
- int(getattr(h, "correct_count", 0) or 0),
- int(getattr(h, "wrong_count", 0) or 0),
- float(getattr(h, "score", 0.0) or 0.0),
- )
- if key in seen_virtual:
- continue
- seen_virtual.add(key)
- deduped.append(h)
-
- return deduped
-
- # 重置用户荣誉
- async def reset_user_honors(self, user_id: str):
- """
- 重置指定用户的所有荣誉数据
-
- 参数:
- user_id: 用户ID
- """
- async with self.db.get_session() as session:
- stmt = delete(MatchHonor).where(MatchHonor.user_id == user_id)
- await session.execute(stmt)
- await session.commit()
-
- # 重置所有荣誉
- async def reset_all_honors(self):
- """
- 重置所有用户的荣誉数据
- """
- async with self.db.get_session() as session:
- stmt = delete(MatchHonor)
- await session.execute(stmt)
- await session.commit()
diff --git a/test_plugin/old/mrfzccl/src/db/tables.py b/test_plugin/old/mrfzccl/src/db/tables.py
deleted file mode 100644
index 8fea8ac18..000000000
--- a/test_plugin/old/mrfzccl/src/db/tables.py
+++ /dev/null
@@ -1,67 +0,0 @@
-from datetime import datetime
-from typing import Optional
-from sqlmodel import Field, SQLModel
-
-
-class UserQnAStats(SQLModel, table=True):
- """用户问答统计表 - 记录用户的答题统计数据"""
-
- __tablename__ = "user_qna_stats"
- __table_args__ = {"extend_existing": True}
-
- id: Optional[int] = Field(default=None, primary_key=True, description="主键ID，自增唯一标识")
- user_id: str = Field(index=True, description="用户ID")
- user_name: str = Field(index=True, description="用户名称")
- correct_count: int = Field(default=0, description="答对次数")
- wrong_count: int = Field(default=0, description="答错次数")
- tip_count: int = Field(default=0, description="提示次数")
- created_at: datetime = Field(default_factory=datetime.now, description="创建时间")
- updated_at: datetime = Field(default_factory=datetime.now, description="更新时间")
-
-
-class Match(SQLModel, table=True):
- """比赛表"""
- __tablename__ = "match"
- __table_args__ = {"extend_existing": True}
-
- match_id: Optional[int] = Field(default=None, primary_key=True)
- group_id: str = Field(index=True, description="群ID")
- match_name: str = Field(description="比赛名称")
- is_active: bool = Field(default=True, description="是否进行中")
- question_limit: int = Field(default=0, description="答题数量限制(0不限制)")
- time_limit: int = Field(default=0, description="时间限制分钟(0不限制)")
- created_at: datetime = Field(default_factory=datetime.now)
- started_at: Optional[datetime] = Field(default=None, description="开始时间")
- ended_at: Optional[datetime] = Field(default=None, description="结束时间")
-
-
-class MatchParticipant(SQLModel, table=True):
- """比赛参与者表"""
- __tablename__ = "match_participant"
- __table_args__ = {"extend_existing": True}
-
- id: Optional[int] = Field(default=None, primary_key=True)
- match_id: int = Field(index=True, description="比赛ID")
- user_id: str = Field(index=True, description="用户ID")
- user_name: str = Field(description="用户名称")
- correct_count: int = Field(default=0, description="答对数")
- wrong_count: int = Field(default=0, description="答错数")
- score: float = Field(default=0.0, description="得分(正确数-错误数*1/3)")
- joined_at: datetime = Field(default_factory=datetime.now)
-
-
-class MatchHonor(SQLModel, table=True):
- """比赛荣誉表"""
- __tablename__ = "match_honor"
- __table_args__ = {"extend_existing": True}
-
- id: Optional[int] = Field(default=None, primary_key=True)
- user_id: str = Field(index=True, description="用户ID")
- match_id: int = Field(description="比赛ID")
- match_name: str = Field(description="比赛名称")
- rank: int = Field(description="名次")
- correct_count: int = Field(default=0, description="答对数")
- wrong_count: int = Field(default=0, description="答错数")
- score: float = Field(default=0.0, description="得分(正确数-错误数*1/3)")
- medal: str = Field(description="奖牌")
- created_at: datetime = Field(default_factory=datetime.now)
\ No newline at end of file
diff --git a/test_plugin/old/mrfzccl/src/handlers/__init__.py b/test_plugin/old/mrfzccl/src/handlers/__init__.py
deleted file mode 100644
index 49a6f893b..000000000
--- a/test_plugin/old/mrfzccl/src/handlers/__init__.py
+++ /dev/null
@@ -1,2 +0,0 @@
-"""Command handler modules for Mrfzccl plugin (stage-1 split)."""
-
diff --git a/test_plugin/old/mrfzccl/src/handlers/ccl_admin.py b/test_plugin/old/mrfzccl/src/handlers/ccl_admin.py
deleted file mode 100644
index 44559a3c5..000000000
--- a/test_plugin/old/mrfzccl/src/handlers/ccl_admin.py
+++ /dev/null
@@ -1,123 +0,0 @@
-from __future__ import annotations
-
-from typing import Any, AsyncIterator
-
-from astrbot.api.event import AstrMessageEvent
-
-
-async def handle_reset_user_data(
- self,
- event: AstrMessageEvent,
- target_user_id: str = "",
-) -> AsyncIterator[Any]:
- """清除用户答题数据（仅管理员）/ccl 清除数据 [user_id]"""
- sender_id = str(event.get_sender_id())
-
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- yield event.plain_result("❌ 只有管理员可以清除数据")
- return
-
- if target_user_id:
- await self.user_qna_repo.reset_user_stats(target_user_id)
- yield event.plain_result(f"✅ 用户 {target_user_id} 的答题数据已清除")
- else:
- await self.user_qna_repo.reset_user_stats(sender_id)
- yield event.plain_result("✅ 您的答题数据已清除")
-
-
-async def handle_reset_user_honors_cmd(
- self,
- event: AstrMessageEvent,
- target_user_id: str = "",
-) -> AsyncIterator[Any]:
- """清除用户荣誉数据（仅管理员）/ccl 清除荣誉 [user_id]"""
- sender_id = str(event.get_sender_id())
-
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- yield event.plain_result("❌ 只有管理员可以清除荣誉")
- return
-
- if target_user_id:
- await self.match_repo.reset_user_honors(target_user_id)
- yield event.plain_result(f"✅ 用户 {target_user_id} 的荣誉数据已清除")
- else:
- await self.match_repo.reset_user_honors(sender_id)
- yield event.plain_result("✅ 您的荣誉数据已清除")
-
-
-async def handle_reset_all_data_cmd(self, event: AstrMessageEvent) -> AsyncIterator[Any]:
- """清除所有用户的答题数据（仅管理员）/ccl 清除所有数据"""
- sender_id = str(event.get_sender_id())
-
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- yield event.plain_result("❌ 只有管理员可以清除所有数据")
- return
-
- await self.user_qna_repo.reset_all_stats()
- yield event.plain_result("✅ 所有用户的答题数据已清除")
-
-
-async def handle_reset_all_honors_cmd(self, event: AstrMessageEvent) -> AsyncIterator[Any]:
- """清除所有用户的荣誉数据（仅管理员）/ccl 清除所有荣誉"""
- sender_id = str(event.get_sender_id())
-
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- yield event.plain_result("❌ 只有管理员可以清除所有荣誉")
- return
-
- await self.match_repo.reset_all_honors()
- yield event.plain_result("✅ 所有用户的荣誉数据已清除")
-
-
-async def handle_grant_honor_cmd(
- self,
- event: AstrMessageEvent,
- target_user_id: str = "",
- rank: int = 1,
- match_name: str = "",
- correct_count: int = 0,
-) -> AsyncIterator[Any]:
- """授予用户特定荣誉（仅管理员）/ccl 授予荣誉 [user_id] [名次] [比赛名称] [答对数量]"""
- sender_id = str(event.get_sender_id())
-
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- yield event.plain_result("❌ 只有管理员可以授予荣誉")
- return
-
- # 检查参数完整性
- if not target_user_id or not match_name:
- yield event.plain_result("❌ 请提供完整参数: /ccl 授予荣誉 [user_id] [名次] [比赛名称] [答对数量]")
- return
-
- # 根据名次生成奖牌表情
- if rank == 1:
- medal = "🥇"
- elif rank == 2:
- medal = "🥈"
- elif rank == 3:
- medal = "🥉"
- else:
- medal = f"{rank}"
-
- # 计算得分（错误数默认为0）
- score = correct_count - 0
-
- # 保存荣誉
- await self.match_repo.save_honor(
- user_id=target_user_id,
- match_id=0, # 虚拟比赛ID
- match_name=match_name,
- rank=rank,
- correct_count=correct_count,
- wrong_count=0,
- score=score,
- )
-
- yield event.plain_result(
- f"✅ 已授予用户 {target_user_id} 荣誉: {medal} {match_name} 第{rank}名, 答对{correct_count}题"
- )
diff --git a/test_plugin/old/mrfzccl/src/handlers/ccl_leaderboard.py b/test_plugin/old/mrfzccl/src/handlers/ccl_leaderboard.py
deleted file mode 100644
index d189a72dd..000000000
--- a/test_plugin/old/mrfzccl/src/handlers/ccl_leaderboard.py
+++ /dev/null
@@ -1,137 +0,0 @@
-from __future__ import annotations
-
-from typing import Any, AsyncIterator
-
-from astrbot.api.event import AstrMessageEvent
-
-from ..tool import (
- generate_correct_leaderboard_text,
- generate_hints_leaderboard_text,
- generate_image_or_fallback,
- generate_user_profile_text,
- generate_wrong_leaderboard_text,
-)
-
-async def handle_correct_answers_leaderboard(self, event: AstrMessageEvent) -> AsyncIterator[Any]:
- """获取正确个数的排行榜命令 /ccl 排行榜"""
- if self.require_admin:
- sender_id = str(event.get_sender_id())
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- yield event.plain_result("❌ 只有管理员可以查看排行榜")
- return
-
- try:
- # 获取排行榜数据（前10名）
- users = await self.user_qna_repo.get_correct_answers_leaderboard(limit=10)
-
- if not users:
- yield event.plain_result("📊 当前还没有用户的答题记录哦~")
- return
-
- # 获取统计信息
- summary = await self.user_qna_repo.get_leaderboard_summary()
-
- # 使用统一的图片/文本生成函数
- async for result in generate_image_or_fallback(
- event=event,
- generate_image_func=lambda: self.renderer.generate_correct_leaderboard_image(users),
- generate_text_func=lambda: generate_correct_leaderboard_text(users, summary),
- ):
- yield result
-
- except Exception as e:
- yield event.plain_result(f"获取排行榜时出现错误: {str(e)}")
-
-async def handle_wrong_answers_leaderboard(self, event: AstrMessageEvent) -> AsyncIterator[Any]:
- """获取错误个数的排行榜命令 /ccl 错误排行榜"""
- if self.require_admin:
- sender_id = str(event.get_sender_id())
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- yield event.plain_result("❌ 只有管理员可以查看排行榜")
- return
-
- try:
- # 获取排行榜数据（前10名）
- users = await self.user_qna_repo.get_wrong_answers_leaderboard(limit=10)
-
- if not users:
- yield event.plain_result("📊 当前还没有用户的答题记录哦~")
- return
-
- # 使用统一的图片/文本生成函数
- async for result in generate_image_or_fallback(
- event=event,
- generate_image_func=lambda: self.renderer.generate_wrong_leaderboard_image(users),
- generate_text_func=lambda: generate_wrong_leaderboard_text(users),
- ):
- yield result
-
- except Exception as e:
- yield event.plain_result(f"获取排行榜时出现错误: {str(e)}")
-
-async def handle_hints_usage_leaderboard(self, event: AstrMessageEvent) -> AsyncIterator[Any]:
- """获取使用提示次数的排行榜命令 /ccl 提示排行榜"""
- if self.require_admin:
- sender_id = str(event.get_sender_id())
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- yield event.plain_result("❌ 只有管理员可以查看排行榜")
- return
-
- try:
- # 获取排行榜数据（前10名）
- users = await self.user_qna_repo.get_hints_usage_leaderboard(limit=10)
-
- if not users:
- yield event.plain_result("📊 当前还没有用户的答题记录哦~")
- return
-
- # 使用统一的图片/文本生成函数
- async for result in generate_image_or_fallback(
- event=event,
- generate_image_func=lambda: self.renderer.generate_hints_leaderboard_image(users),
- generate_text_func=lambda: generate_hints_leaderboard_text(users),
- ):
- yield result
-
- except Exception as e:
- yield event.plain_result(f"获取排行榜时出现错误: {str(e)}")
-
-async def handle_user_profile_retrieval(
- self,
- event: AstrMessageEvent,
- user_id: str | None = None,
-) -> AsyncIterator[Any]:
- """获取个人信息获取 /ccl 名片 [user_id] (如果user_id为空默认为发送人)"""
- try:
- # 确定用户ID
- target_user_id = user_id or event.get_sender_id()
-
- # 获取用户信息及排名
- user_stats, rank_info = await self.user_qna_repo.get_user_profile_with_rank(target_user_id)
-
- # 获取用户荣誉
- honors = await self.match_repo.get_user_honors(str(target_user_id))
-
- # 没有任何记录时直接返回
- if not user_stats and not honors:
- yield event.plain_result("❌ 未找到该用户的答题记录")
- return
-
- # 没有答题记录但有荣誉：直接用文本输出（名片图片依赖 user_stats）
- if not user_stats:
- yield event.plain_result(generate_user_profile_text(user_stats, rank_info, honors, str(target_user_id)))
- return
-
- # 使用统一的图片/文本生成函数
- async for result in generate_image_or_fallback(
- event=event,
- generate_image_func=lambda: self.renderer.generate_user_profile_image(user_stats, rank_info, honors),
- generate_text_func=lambda: generate_user_profile_text(user_stats, rank_info, honors, str(target_user_id)),
- ):
- yield result
-
- except Exception as e:
- yield event.plain_result(f"获取用户信息时出现错误: {str(e)}")
diff --git a/test_plugin/old/mrfzccl/src/handlers/ccl_match.py b/test_plugin/old/mrfzccl/src/handlers/ccl_match.py
deleted file mode 100644
index f1975e946..000000000
--- a/test_plugin/old/mrfzccl/src/handlers/ccl_match.py
+++ /dev/null
@@ -1,199 +0,0 @@
-from __future__ import annotations
-
-import time
-from typing import Any, AsyncIterator
-
-import astrbot.api.message_components as Comp
-from astrbot.api.event import AstrMessageEvent
-
-from ..tool import generate_image_or_fallback, generate_match_leaderboard_text
-
-async def handle_match_help(self, event: AstrMessageEvent) -> AsyncIterator[Any]:
- """比赛模式帮助"""
- if event.get_group_id() is None:
- yield event.plain_result("请在群聊使用")
- return
- yield event.plain_result(
- """📋 比赛模式指令帮助
-━━━━━━━━━━━━━━
-/ccl 比赛创建 [名称] [题目限制] [时间限制(分钟)] - 创建比赛(仅管理员)
-/ccl 比赛开始 - 开始比赛(仅管理员)
-/ccl 比赛结束/结束比赛 - 结束比赛(仅管理员)
-/ccl 比赛排行/排行 - 查看比赛排行榜
-━━━━━━━━━━━━━━"""
- )
-
-async def handle_match_create(
- self,
- event: AstrMessageEvent,
- name: str = "",
- question_limit: int = 0,
- time_limit: int = 0,
-) -> AsyncIterator[Any]:
- """创建比赛（仅管理员）用法: /ccl 比赛创建 [名称] [题目限制] [时间限制(分钟)]"""
- group_id_raw = event.get_group_id()
- if group_id_raw is None:
- yield event.plain_result("请在群聊使用")
- return
- group_id = str(group_id_raw)
- sender_id = str(event.get_sender_id())
-
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- yield event.plain_result("❌ 只有管理员可以创建比赛")
- return
-
- # 检查是否已有进行中的比赛
- existing = await self.match_repo.get_active_match(group_id)
- if existing:
- yield event.plain_result("❌ 当前群已有进行中的比赛")
- return
-
- # 设置题目限制和时间限制
- q_limit = question_limit if question_limit >= 0 else self.match_question_limit
- t_limit = time_limit if time_limit >= 0 else self.match_time_limit
-
- if q_limit < 0 or t_limit < 0:
- yield event.plain_result(f"参数未通过检验,q_limit:{q_limit},t_limit:{t_limit}")
- return
-
- # 创建比赛名称
- match_name = name if name else f"比赛_{int(time.time())}"
- # 创建比赛
- await self.match_repo.create_match(group_id, match_name, q_limit, t_limit)
-
- # 构建响应信息
- info = f"✅ 比赛「{match_name}」已创建！"
- if q_limit > 0:
- info += f"\n📝 题目限制: {q_limit}题"
- if t_limit > 0:
- info += f"\n⏱️ 时间限制: {t_limit}分钟"
- info += "\n进行答题即可参与比赛"
- yield event.plain_result(info)
-
-async def match_start_precheck(self, event: AstrMessageEvent) -> tuple[bool, str | None, Any | None]:
- """`/ccl 比赛开始` 的锁外校验与 DB 状态更新。"""
- group_id_raw = event.get_group_id()
- if group_id_raw is None:
- return False, None, event.plain_result("请在群聊使用")
-
- sender_id = str(event.get_sender_id())
- group_id = str(group_id_raw)
- self.match_sessions[group_id] = event.unified_msg_origin
-
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- return False, group_id, event.plain_result("❌ 只有管理员可以开始比赛")
-
- # 获取活跃比赛
- match = await self.match_repo.get_active_match(group_id)
- if not match:
- return False, group_id, event.plain_result("❌ 当前没有进行中的比赛")
-
- # 开始比赛
- await self.match_repo.start_match(match.match_id)
-
- return True, group_id, None
-
-async def match_start_inlock(self, group_id: str) -> bytes | str | None:
- """`/ccl 比赛开始` 的锁内状态清理 + 出题逻辑。"""
- # 防止上次比赛残留题目导致 fc_init 返回 already_exists
- self.end_game(group_id)
-
- # 取消旧的比赛循环/提示任务（防止重复启动）
- if group_id in self.match_next_task:
- self._safe_cancel_task(self.match_next_task.pop(group_id, None))
- if group_id in self.match_loop_task:
- self._safe_cancel_task(self.match_loop_task.pop(group_id, None))
- self.match_question_state.pop(group_id, None)
-
- # 初始化第一题
- result = await self.fc_init(group_id)
- if result and result != "already_exists":
- self.match_question_state[group_id] = time.time()
- self._schedule_match_hint(group_id)
-
- return result
-
-def build_match_start_response(event: AstrMessageEvent, result: bytes | str | None) -> Any:
- if result and result != "already_exists":
- return event.chain_result(
- [
- Comp.Plain("🏁 比赛已开始！答题即为参与比赛\n干员立绘,请使用/fcc [干员名称] 进行猜测"),
- Comp.Image.fromBytes(result),
- ]
- )
- return event.plain_result("🏁 比赛已开始！第一题获取失败，请重试")
-
-async def match_end_precheck(self, event: AstrMessageEvent) -> tuple[bool, str | None, Any | None]:
- """`/ccl 比赛结束` 的锁外校验。"""
- group_id_raw = event.get_group_id()
- if group_id_raw is None:
- return False, None, event.plain_result("请在群聊使用")
-
- sender_id = str(event.get_sender_id())
- group_id = str(group_id_raw)
-
- # 检查管理员权限
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- return False, group_id, event.plain_result("❌ 只有管理员可以结束比赛")
-
- return True, group_id, None
-
-async def match_end_inlock(self, group_id: str) -> tuple[bool, str, list]:
- """`/ccl 比赛结束` 的锁内结算逻辑。"""
- # 重新获取活跃比赛，避免与自动结束并发导致重复荣誉
- match_now = await self.match_repo.get_active_match(group_id)
- if not match_now or not match_now.is_active:
- return False, "", []
-
- match_name, _, top_participants = await self._end_match_and_collect_top(group_id, match_now)
- return True, match_name, top_participants
-
-async def iter_match_end_results(
- self,
- event: AstrMessageEvent,
- match_name: str,
- top_participants: list,
-) -> AsyncIterator[Any]:
- """`/ccl 比赛结束` 的锁外输出（图片优先，失败回退文本）。"""
- async for result in generate_image_or_fallback(
- event=event,
- generate_image_func=lambda: self.renderer.generate_match_leaderboard_image(
- match_name,
- top_participants,
- title=f"比赛「{match_name}」已结束排行榜",
- ),
- generate_text_func=lambda: generate_match_leaderboard_text(match_name, top_participants, ended=True),
- ):
- yield result
-
-async def handle_match_leaderboard(self, event: AstrMessageEvent) -> AsyncIterator[Any]:
- """使用`/ccl 比赛排行`获取比赛排行榜"""
- group_id_raw = event.get_group_id()
- if group_id_raw is None:
- yield event.plain_result("请在群聊使用")
- return
- group_id = str(group_id_raw)
-
- # 获取活跃比赛
- match = await self.match_repo.get_active_match(group_id)
- if not match:
- yield event.plain_result("❌ 无进行中比赛")
- return
-
- # 获取参赛者列表并排序
- participants = await self.match_repo.get_participants(match.match_id)
- participants.sort(key=lambda p: p.score, reverse=True)
-
- top_participants = participants[:10]
-
- async for result in generate_image_or_fallback(
- event=event,
- generate_image_func=lambda: self.renderer.generate_match_leaderboard_image(
- match.match_name,
- top_participants,
- ),
- generate_text_func=lambda: generate_match_leaderboard_text(match.match_name, top_participants),
- ):
- yield result
diff --git a/test_plugin/old/mrfzccl/src/handlers/fc_handlers.py b/test_plugin/old/mrfzccl/src/handlers/fc_handlers.py
deleted file mode 100644
index b27aecebf..000000000
--- a/test_plugin/old/mrfzccl/src/handlers/fc_handlers.py
+++ /dev/null
@@ -1,351 +0,0 @@
-from __future__ import annotations
-
-import time
-import traceback
-from difflib import SequenceMatcher
-from typing import Any, AsyncIterator
-
-import astrbot.api.message_components as Comp
-from astrbot.api import logger
-from astrbot.api.event import AstrMessageEvent
-
-from ..tool import (
- calculate_char_coverage_set,
- check_daily_limit,
- check_homophone,
- generate_image_or_fallback,
- generate_match_leaderboard_text,
- has_active_game,
- resolve_alias,
-)
-
-async def handle_fc(
- self,
- event: AstrMessageEvent,
- *,
- user_id: str,
- sender_id: str,
- is_group: bool,
- group_id: str | None,
-) -> Any | None:
- """Core logic for `/fc` (expects room lock is held by caller)."""
- response = None
-
- # 确保数据库初始化
- try:
- await self.db.init_db()
- logger.info("[Mrfzccl] 数据库初始化完成")
- except Exception as e:
- logger.error(f"[Mrfzccl] 数据库初始化失败: {e}")
- response = event.chain_result(
- [
- Comp.At(qq=sender_id),
- Comp.Plain(" 数据库初始化失败，请联系管理员"),
- ]
- )
-
- if response is None:
- # 检查是否在比赛模式和是否限制（仅群聊）
- match = await self.match_repo.get_active_match(group_id) if is_group else None
- # 非管理员进行次数检测
- if self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- # 非比赛模式下检查每日限制
- if not match and not check_daily_limit(sender_id, self.daily_counter, self.daily_limit):
- response = event.plain_result(f"今日游戏次数已达上限({self.daily_limit}次)，请明天再来！")
-
- if response is None:
- try:
- # 调用初始化游戏方法
- result = await self.fc_init(user_id)
- if result == "already_exists":
- response = event.plain_result("已经初始化,请不要重复操作")
- elif result is None:
- response = event.plain_result("图片获取失败,请重试")
- else:
- # 发送游戏图片
- response = event.chain_result(
- [
- Comp.Plain("干员立绘,请使用/fcc [干员名称] 进行猜测"),
- Comp.Image.fromBytes(result),
- ]
- )
- except Exception as e:
- logger.error(f"[fc] 命令执行失败: {e}")
- logger.error(traceback.format_exc())
- response = event.plain_result("游戏初始化失败，请稍后重试")
-
- return response
-
-async def handle_fcc(
- self,
- event: AstrMessageEvent,
- *,
- user_id: str,
- sender_id: str,
- is_group: bool,
- group_id: str | None,
-) -> tuple[list[Any], tuple[str, list] | None]:
- """Core logic for `/fcc` (expects room lock is held by caller)."""
- responses: list[Any] = []
- match_end_payload: tuple[str, list] | None = None # (ended_match_name, ended_top_participants)
-
- logger.debug(
- f"[fcc] user_id={user_id}, player_keys={list(self.player.keys())}, has_active={has_active_game(self.player, user_id)}"
- )
-
- # 检查是否有活跃比赛
- match = await self.match_repo.get_active_match(group_id) if is_group else None
-
- # 检查用户是否有活跃游戏
- if not has_active_game(self.player, user_id):
- if match:
- responses.append(event.plain_result("比赛期间请等待管理员发送题目"))
- else:
- responses.append(event.plain_result("没有初始化房间,请使用/fc"))
- return responses, match_end_payload
-
- # 提取并清理用户输入的猜测内容
- guess_text = self.extract_and_sanitize_input(event.message_str, "fcc")
- if not guess_text:
- responses.append(
- event.chain_result(
- [
- Comp.At(qq=sender_id),
- Comp.Plain(" 请输入要猜测的干员名称"),
- ]
- )
- )
- return responses, match_end_payload
-
- correct_name = self.player[user_id]["name"] # 获取正确答案
-
- # 解析别名（将用户输入的别名转换为正式名称）
- resolved_guess = resolve_alias(guess_text, self.alias_map)
-
- # 计算相似度
- similarity = SequenceMatcher(None, correct_name, resolved_guess).ratio()
- # 计算字符覆盖率
- calculate = calculate_char_coverage_set(correct_name, resolved_guess)
- # 检查是否为同音字
- homophone_match = check_homophone(correct_name, resolved_guess, enable_homophone=self.enable_homophone)
- # 综合判断是否正确
- is_correct = (similarity > self.similarity_threshold) or (calculate > self.calculate_threshold) or homophone_match
-
- logger.debug(
- f"[答题判断] 正确答案: {correct_name}, 用户回答: {resolved_guess}, 相似度: {similarity:.2f}, "
- f"字匹配率: {calculate:.2f}, 同音匹配: {homophone_match}, 阈值: {self.similarity_threshold}/{self.calculate_threshold}, "
- f"结果: {is_correct}"
- )
-
- sender_name = event.get_sender_name()
-
- # 如果是比赛模式，更新比赛数据
- if is_group and match and group_id is not None:
- self.match_sessions[group_id] = event.unified_msg_origin
- await self.match_repo.add_participant(match.match_id, str(sender_id), sender_name)
- if is_correct:
- await self.match_repo.increment_participant_score(match.match_id, str(sender_id))
- # 取消当前题目的自动提示任务
- if group_id in self.match_next_task:
- self._safe_cancel_task(self.match_next_task.pop(group_id, None))
- else:
- await self.match_repo.increment_participant_wrong(match.match_id, str(sender_id))
-
- # 处理回答结果
- if is_correct:
- chain = [
- Comp.At(qq=sender_id),
- Comp.Plain(f" 回答正确! 答案为: {correct_name}"),
- ]
- responses.append(event.chain_result(chain))
- responses.append(await self.send_original_image(user_id, event)) # 发送原图
-
- # 更新用户正确计数
- await self.user_qna_repo.increment_correct_count(
- user_id=sender_id,
- user_name=sender_name,
- )
-
- # 比赛模式：自动出下一题 / 自动结束
- if is_group and match and group_id is not None:
- # 重新获取活跃比赛，避免已自动结束后再次结算导致重复荣誉
- match_now = await self.match_repo.get_active_match(group_id)
- if match_now and match_now.is_active:
- # 若已经有人推进到下一题，当前协程无需重复出题
- existing = self.player.get(group_id)
- if not (existing and existing.get("status") in {"active", "loading"}):
- end_reason = await self._get_match_end_reason(match_now)
- if end_reason:
- ended_match_name, _, ended_top_participants = await self._end_match_and_collect_top(
- group_id,
- match_now,
- )
- if end_reason == "time_limit":
- responses.append(
- event.plain_result(f"⏱️ 已达到时间限制，比赛「{ended_match_name}」自动结束！")
- )
- else:
- responses.append(
- event.plain_result(f"📝 已达到题目上限，比赛「{ended_match_name}」自动结束！")
- )
- match_end_payload = (ended_match_name, ended_top_participants)
- else:
- next_bytes = await self.fc_init(group_id)
- if next_bytes and next_bytes != "already_exists":
- self.match_question_state[group_id] = time.time()
- self._schedule_match_hint(group_id)
- responses.append(
- event.chain_result(
- [
- Comp.Plain("下一题来啦！\n干员立绘,请使用/fcc [干员名称] 进行猜测"),
- Comp.Image.fromBytes(next_bytes),
- ]
- )
- )
- else:
- responses.append(event.plain_result("下一题获取失败，请管理员使用 /ccl 比赛开始 重试"))
- else:
- chain = [
- Comp.At(qq=sender_id),
- Comp.Plain(" 回答错误!"),
- ]
- responses.append(event.chain_result(chain))
- # 更新用户错误计数
- await self.user_qna_repo.increment_wrong_count(
- user_id=sender_id,
- user_name=sender_name,
- )
-
- return responses, match_end_payload
-
-async def iter_match_end_leaderboard(
- self,
- event: AstrMessageEvent,
- match_end_payload: tuple[str, list],
-) -> AsyncIterator[Any]:
- """Post-lock output for `/fcc` when a match ends (image preferred, text fallback)."""
- ended_match_name, ended_top_participants = match_end_payload
- async for result in generate_image_or_fallback(
- event=event,
- generate_image_func=lambda: self.renderer.generate_match_leaderboard_image(
- ended_match_name,
- ended_top_participants,
- title=f"比赛「{ended_match_name}」已结束排行榜",
- ),
- generate_text_func=lambda: generate_match_leaderboard_text(
- ended_match_name,
- ended_top_participants,
- ended=True,
- ),
- ):
- yield result
-
-async def handle_fce(
- self,
- event: AstrMessageEvent,
- *,
- user_id: str,
- sender_id: str,
- is_group: bool,
- group_id: str | None,
-) -> list[Any]:
- """Core logic for `/fce` (expects room lock is held by caller)."""
- responses: list[Any] = []
-
- # 检查比赛模式下是否有权限
- match = await self.match_repo.get_active_match(group_id) if is_group else None
- if match and self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- responses.append(event.plain_result("❌ 比赛期间只有管理员可以强制结束"))
- elif not has_active_game(self.player, user_id):
- responses.append(event.plain_result("没有初始化房间,请使用/fc"))
- else:
- answer = self.player[user_id]["name"] # 获取答案
- chain = [
- Comp.At(qq=sender_id),
- Comp.Plain(f" 游戏已结束,答案为: {answer}"),
- ]
- responses.append(event.chain_result(chain))
- responses.append(await self.send_original_image(user_id, event)) # 发送原图
-
- return responses
-
-async def handle_fct(
- self,
- event: AstrMessageEvent,
- *,
- user_id: str,
- sender_id: str,
- is_group: bool,
- group_id: str | None,
-) -> Any | None:
- """Core logic for `/fct` (expects room lock is held by caller)."""
- response = None
-
- # 检查比赛模式下是否有权限（仅群聊）
- match = await self.match_repo.get_active_match(group_id) if is_group else None
- if match and self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- response = event.plain_result("❌ 比赛期间只有管理员可以使用提示")
- elif not has_active_game(self.player, user_id):
- response = event.plain_result("没有初始化房间,请使用/fc")
- else:
- hint_text, _ = self._next_hint_text_and_advance(user_id)
- response = event.plain_result(hint_text)
-
- # 更新用户提示使用次数
- await self.user_qna_repo.increment_tip_count(
- user_id=event.get_sender_id(),
- user_name=event.get_sender_name(),
- )
-
- return response
-
-async def handle_fcw(
- self,
- event: AstrMessageEvent,
- *,
- user_id: str,
- sender_id: str,
- is_group: bool,
- group_id: str | None,
-) -> Any | None:
- """Core logic for `/fcw` (expects room lock is held by caller)."""
- response = None
-
- # 检查比赛模式下是否有权限（仅群聊）
- match = await self.match_repo.get_active_match(group_id) if is_group else None
- if match and self.admin_ids and sender_id not in [str(x) for x in self.admin_ids]:
- response = event.plain_result("❌ 比赛期间只有管理员可以使用提示")
- elif not has_active_game(self.player, user_id):
- response = event.plain_result("没有初始化房间,请使用/fc")
- else:
- char_data = self.data.get(self.player[user_id]["name"], {})
-
- logger.info(f"[fcw] player={self.player[user_id]}, char_data keys={list(char_data.keys()) if char_data else 'None'}")
-
- # 获取职业及分支
- profession = char_data.get("职业及分支", char_data.get("职业分支", "该干员没有该属性"))
- # 星级转换为中文
- star = char_data.get("星级", "")
- star_map = {"1": "一星", "2": "二星", "3": "三星", "4": "四星", "5": "五星", "6": "六星"}
- star_cn = star_map.get(str(star), star)
- # 阵营
- camp = char_data.get("阵营", char_data.get("所属阵营", "该干员没有该属性"))
-
- # 构建提示消息
- msg = "💡 一次性提示:\n"
- msg += f"职业: {profession}\n"
- msg += f"星级: {star_cn}\n"
- msg += f"阵营: {camp}"
-
- response = event.plain_result(msg)
-
- # 设置提示计数为4（跳过属性提示阶段）
- self.player[user_id]["fctn"] = 4
- # 更新用户提示使用次数
- await self.user_qna_repo.increment_tip_count(
- user_id=sender_id,
- user_name=event.get_sender_name(),
- increment=3,
- )
-
- return response
diff --git a/test_plugin/old/mrfzccl/src/tool.py b/test_plugin/old/mrfzccl/src/tool.py
deleted file mode 100644
index 65fe43a63..000000000
--- a/test_plugin/old/mrfzccl/src/tool.py
+++ /dev/null
@@ -1,306 +0,0 @@
-from collections import Counter
-import os
-from typing import Any, Callable, Iterable, Mapping, Optional
-
-from astrbot.api.event import AstrMessageEvent
-import astrbot.api.message_components as Comp
-from pypinyin import lazy_pinyin, Style
-from datetime import datetime
-
-def calculate_char_coverage_set(correct_name: str, guess_text: str) -> float:
- """
- 计算guess_text包含correct_name中字符的比例（去重版本）
-
- Args:
- correct_name: 正确答案
- guess_text: 用户猜测的答案
-
- Returns:
- float: 字符覆盖率 (0-1之间)
- """
- if not correct_name:
- return 0.0
-
- # 转换为集合去重
- correct_chars = set(correct_name)
- guess_chars = set(guess_text)
-
- # 计算匹配的字符数比例
- matched_chars = correct_chars & guess_chars
- coverage = len(matched_chars) / len(correct_chars)
-
- return coverage
-
-def calculate_char_coverage_counter(correct_name: str, guess_text: str) -> float:
- """
- 计算guess_text包含correct_name中字符的比例（不去重版本）
-
- Args:
- correct_name: 正确答案
- guess_text: 用户猜测的答案
-
- Returns:
- float: 字符覆盖率 (0-1之间)
- """
- if not correct_name:
- return 0.0
-
- # 使用Counter统计字符出现次数
- correct_counter = Counter(correct_name)
- guess_counter = Counter(guess_text)
-
- # 计算总字符数和匹配的字符数
- total_chars = sum(correct_counter.values())
- matched_chars = 0
-
- for char, count in correct_counter.items():
- matched_chars += min(count, guess_counter.get(char, 0))
-
- coverage = matched_chars / total_chars if total_chars > 0 else 0.0
-
- return coverage
-
-def generate_correct_leaderboard_text(users: Iterable[Any], summary: Optional[Mapping[str, Any]] = None) -> str:
- """生成正确量排行榜文本"""
- users = list(users or [])
- if not users:
- return "📊 当前还没有用户的答题记录哦~"
-
- message = "🏆 **正确量排行榜** 🏆\n\n"
-
- for i, user in enumerate(users, 1):
- if i == 1:
- medal = "🥇"
- elif i == 2:
- medal = "🥈"
- elif i == 3:
- medal = "🥉"
- else:
- medal = f"{i}."
-
- total_answers = getattr(user, "correct_count", 0) + getattr(user, "wrong_count", 0)
- accuracy = (getattr(user, "correct_count", 0) / total_answers * 100) if total_answers > 0 else 0
-
- message += f"{medal} {getattr(user, 'user_name', '-')}\n"
- message += (
- f" ✅ 正确: {getattr(user, 'correct_count', 0)} | ❌ 错误: {getattr(user, 'wrong_count', 0)} | 💡 提示: {getattr(user, 'tip_count', 0)}\n"
- )
- updated_at = getattr(user, "updated_at", None)
- updated_str = updated_at.strftime("%Y-%m-%d") if hasattr(updated_at, "strftime") else "-"
- message += f" 📈 准确率: {accuracy:.1f}% | 📅 最后更新: {updated_str}\n\n"
-
- if summary:
- message += "📊 **统计信息**\n"
- message += f"总用户数: {summary.get('total_users', '-')} | 总答题数: {summary.get('total_questions', '-')}\n"
- message += f"总正确数: {summary.get('total_correct', '-')} | 总错误数: {summary.get('total_wrong', '-')}\n"
- message += f"平均正确数: {summary.get('avg_correct', 0):.1f}"
-
- return message
-
-def generate_wrong_leaderboard_text(users: Iterable[Any]) -> str:
- """生成错误个数排行榜文本"""
- users = list(users or [])
- if not users:
- return "📊 当前还没有用户的答题记录哦~"
-
- message = "💥 **错误个数排行榜** 💥\n\n"
-
- for i, user in enumerate(users, 1):
- if i == 1:
- medal = "💣"
- elif i == 2:
- medal = "🧨"
- elif i == 3:
- medal = "🎆"
- else:
- medal = f"{i}."
-
- total_answers = getattr(user, "correct_count", 0) + getattr(user, "wrong_count", 0)
- error_rate = (getattr(user, "wrong_count", 0) / total_answers * 100) if total_answers > 0 else 0
-
- message += f"{medal} {getattr(user, 'user_name', '-')}\n"
- message += (
- f" ❌ 错误: {getattr(user, 'wrong_count', 0)} | ✅ 正确: {getattr(user, 'correct_count', 0)} | 💡 提示: {getattr(user, 'tip_count', 0)}\n"
- )
- updated_at = getattr(user, "updated_at", None)
- updated_str = updated_at.strftime("%Y-%m-%d") if hasattr(updated_at, "strftime") else "-"
- message += f" 📉 错误率: {error_rate:.1f}% | 📅 最后更新: {updated_str}\n\n"
-
- return message
-
-def generate_hints_leaderboard_text(users: Iterable[Any]) -> str:
- """生成提示次数排行榜文本"""
- users = list(users or [])
- if not users:
- return "📊 当前还没有用户的答题记录哦~"
-
- message = "💡 **提示次数排行榜** 💡\n\n"
-
- for i, user in enumerate(users, 1):
- if i == 1:
- medal = "🎯"
- elif i == 2:
- medal = "🔍"
- elif i == 3:
- medal = "🧩"
- else:
- medal = f"{i}."
-
- total_answers = getattr(user, "correct_count", 0) + getattr(user, "wrong_count", 0)
- tips_per_question = (getattr(user, "tip_count", 0) / total_answers) if total_answers > 0 else 0
-
- message += f"{medal} {getattr(user, 'user_name', '-')}\n"
- message += (
- f" 💡 提示: {getattr(user, 'tip_count', 0)} | ✅ 正确: {getattr(user, 'correct_count', 0)} | ❌ 错误: {getattr(user, 'wrong_count', 0)}\n"
- )
- updated_at = getattr(user, "updated_at", None)
- updated_str = updated_at.strftime("%Y-%m-%d") if hasattr(updated_at, "strftime") else "-"
- message += f" 📊 提示频率: {tips_per_question:.2f}/题 | 📅 最后更新: {updated_str}\n\n"
-
- return message
-
-def generate_match_leaderboard_text(match_name: str, participants: Iterable[Any], ended: bool = False) -> str:
- """生成比赛排行榜文本（图片生成失败时的回退）"""
- participants = list(participants or [])
- if not participants:
- status = "已结束" if ended else "排行榜"
- return f"比赛「{match_name}」{status}\n\n暂无参赛记录"
-
- try:
- participants.sort(key=lambda p: float(getattr(p, "score", 0.0) or 0.0), reverse=True)
- except Exception:
- pass
-
- title = f"比赛「{match_name}」已结束\n排行榜" if ended else f"比赛「{match_name}」排行榜"
- message = f"{title}\n----------------\n"
- for i, p in enumerate(participants[:10], 1):
- user_name = getattr(p, "user_name", "-")
- correct = getattr(p, "correct_count", 0)
- wrong = getattr(p, "wrong_count", 0)
- try:
- score_str = f"{float(getattr(p, 'score', 0.0) or 0.0):.2f}"
- except Exception:
- score_str = "-"
- message += f"{i}. {user_name}: {correct}对 {wrong}错 {score_str}分\n"
- return message
-
-def generate_user_profile_text(user_stats: Any, rank_info: Mapping[str, Any], honors=None, user_id: str | None = None) -> str:
- """生成用户个人信息文本"""
- honors = list(honors or [])
-
- title = getattr(user_stats, "user_name", None) if user_stats else None
- title = title or (user_id or "未知用户")
- message = f"👤 **用户信息 - {title}**\n\n"
-
- if not user_stats:
- message += "📊 **基础统计**\n"
- message += "暂无答题记录\n"
- else:
- total_answers = getattr(user_stats, "correct_count", 0) + getattr(user_stats, "wrong_count", 0)
- accuracy = (getattr(user_stats, "correct_count", 0) / total_answers * 100) if total_answers > 0 else 0
-
- message += "📊 **基础统计**\n"
- message += f"✅ 正确: {getattr(user_stats, 'correct_count', 0)}\n"
- message += f"❌ 错误: {getattr(user_stats, 'wrong_count', 0)}\n"
- message += f"💡 提示: {getattr(user_stats, 'tip_count', 0)}\n"
- message += f"🎯 准确率: {accuracy:.1f}%\n"
- message += f"📝 总答题数: {total_answers}\n\n"
-
- if rank_info:
- message += f"🏆 **排名信息** (共{rank_info.get('total_users', '?')}人)\n"
- message += f"✅ 正确排名: 第{rank_info.get('correct_rank', '?')}名\n"
- message += f"❌ 错误排名: 第{rank_info.get('wrong_rank', '?')}名\n"
- message += f"💡 提示排名: 第{rank_info.get('tip_rank', '?')}名\n\n"
-
- created_at = getattr(user_stats, "created_at", None)
- updated_at = getattr(user_stats, "updated_at", None)
- created_str = created_at.strftime("%Y-%m-%d %H:%M") if hasattr(created_at, "strftime") else "-"
- updated_str = updated_at.strftime("%Y-%m-%d %H:%M") if hasattr(updated_at, "strftime") else "-"
-
- message += "📅 **时间信息**\n"
- message += f"⏰ 注册时间: {created_str}\n"
- message += f"🔄 最后更新: {updated_str}\n"
-
- if honors:
- message += "\n🏅 **比赛荣誉**\n"
- for h in honors[:5]:
- try:
- score_str = f"{float(getattr(h, 'score', 0.0)):.1f}"
- except Exception:
- score_str = "-"
- message += (
- f"{getattr(h, 'medal', '')} {getattr(h, 'match_name', '-')}: "
- f"第{getattr(h, 'rank', '?')}名（✅{getattr(h, 'correct_count', 0)}/"
- f"❌{getattr(h, 'wrong_count', 0)}，S{score_str}）\n"
- )
- else:
- message += "\n暂无荣誉记录\n"
-
- return message
-
-async def generate_image_or_fallback(
- event: AstrMessageEvent,
- generate_image_func: Callable[..., Any],
- generate_text_func: Callable[..., str],
- *args,
- **kwargs,
-):
- """统一的图片生成和回退处理"""
- try:
- image_path = await generate_image_func(*args, **kwargs)
-
- if image_path and os.path.exists(image_path):
- yield event.chain_result([Comp.Image.fromFileSystem(image_path)])
- return
-
- text_message = generate_text_func(*args, **kwargs)
- yield event.plain_result(f"图片生成失败，使用文本模式显示\n\n{text_message}")
-
- except Exception as render_error:
- text_message = generate_text_func(*args, **kwargs)
- yield event.plain_result(f"图片生成失败，使用文本模式显示\n错误: {str(render_error)}\n\n{text_message}")
-
-def parse_aliases(alias_str: str) -> dict[str, str]:
- """解析别名配置字符串为映射表：别名:正名,别名:正名"""
- alias_map: dict[str, str] = {}
- if not alias_str:
- return alias_map
- for pair in str(alias_str).split(","):
- if ":" not in pair:
- continue
- alias, name = pair.split(":", 1)
- alias = alias.strip()
- name = name.strip()
- if not alias or not name:
- continue
- alias_map[alias] = name
- return alias_map
-
-def resolve_alias(name: str, alias_map: Mapping[str, str]) -> str:
- """将别名解析为正名（若不存在则返回原值）"""
- return (alias_map or {}).get(name, name)
-
-def get_pinyin(text: str) -> str:
- """获取汉字的拼音（不带声调）"""
- return "".join(lazy_pinyin(text, style=Style.NORMAL))
-
-def check_homophone(correct: str, guess: str, enable_homophone: bool = False) -> bool:
- """检查两个字符串是否同音（基于拼音）"""
- if not enable_homophone:
- return False
- return get_pinyin(correct) == get_pinyin(guess)
-
-def check_daily_limit(user_id: str, daily_counter: dict, daily_limit: int) -> bool:
- """检查并更新每日计数器，返回是否允许继续游戏"""
- today = datetime.now().date()
- key = f"{user_id}_{today}"
- count = daily_counter.get(key, 0)
- if count >= daily_limit:
- return False
- daily_counter[key] = count + 1
- return True
-
-def has_active_game(player: Mapping[str, Any], user_id: str) -> bool:
- """检查用户是否有活跃游戏"""
- data = (player or {}).get(user_id)
- return bool(data and data.get("status") == "active")
diff --git a/tests_v4/test_legacy_plugin_integration.py b/tests_v4/test_legacy_plugin_integration.py
index f9ee172d8..77e25ff37 100644
--- a/tests_v4/test_legacy_plugin_integration.py
+++ b/tests_v4/test_legacy_plugin_integration.py
@@ -8,7 +8,6 @@ from __future__ import annotations
import sys
import tempfile
import textwrap
-import types
from pathlib import Path
import pytest
@@ -21,34 +20,6 @@ from astrbot_sdk.runtime.loader import (
)
-def _install_mrfzccl_optional_dependency_stubs(monkeypatch: pytest.MonkeyPatch) -> None:
- """为真实旧插件夹具补齐仓库测试环境里缺失的可选依赖。"""
- pypinyin = types.ModuleType("pypinyin")
-
- class _Style:
- NORMAL = "normal"
-
- def _lazy_pinyin(text, style=None):
- return list(str(text or ""))
-
- pypinyin.Style = _Style
- pypinyin.lazy_pinyin = _lazy_pinyin
- monkeypatch.setitem(sys.modules, "pypinyin", pypinyin)
-
- html2image = types.ModuleType("html2image")
-
- class _Html2Image:
- def __init__(self, *args, **kwargs):
- self.args = args
- self.kwargs = kwargs
-
- def screenshot(self, *args, **kwargs):
- return []
-
- html2image.Html2Image = _Html2Image
- monkeypatch.setitem(sys.modules, "html2image", html2image)
-
-
class TestLegacyPluginImports:
"""测试旧版 API 导入路径是否可用。"""
@@ -565,68 +536,6 @@ class TestRealTestPlugin:
if p in sys.path:
sys.path.remove(p)
- def test_load_real_external_legacy_plugin_fixture(
- self, monkeypatch: pytest.MonkeyPatch
- ):
- """测试仓库中的真实外部旧插件夹具。"""
- project_root = Path(__file__).parent.parent
- plugin_dir = project_root / "test_plugin" / "old" / "mrfzccl"
-
- if not plugin_dir.exists():
- pytest.skip("external legacy plugin fixture not found")
-
- _install_mrfzccl_optional_dependency_stubs(monkeypatch)
- spec = load_plugin_spec(plugin_dir)
-
- paths_to_add = []
- if str(plugin_dir) not in sys.path:
- sys.path.insert(0, str(plugin_dir))
- paths_to_add.append(str(plugin_dir))
-
- src_new = project_root / "src-new"
- if str(src_new) not in sys.path:
- sys.path.insert(0, str(src_new))
- paths_to_add.append(str(src_new))
-
- try:
- loaded = load_plugin(spec)
-
- assert loaded.plugin.name == "astrbot_plugin_mrfzccl"
- assert len(loaded.instances) == 1
- assert len(loaded.handlers) >= 10
-
- command_names = {
- handler.descriptor.trigger.command
- for handler in loaded.handlers
- if isinstance(handler.descriptor.trigger, CommandTrigger)
- }
- assert {
- "fc",
- "fcc",
- "fce",
- "fct",
- "fcw",
- "ccl 排行榜",
- "ccl 错误排行榜",
- "ccl 提示排行榜",
- } <= command_names
-
- instance = loaded.instances[0]
- assert Path(instance.storage_dir) == plugin_dir / "data"
- assert instance.db_path == str((plugin_dir / "data" / "mrfzccl.db"))
- assert instance.renderer.theme == "light"
- finally:
- synthetic_modules = [
- name
- for name in list(sys.modules)
- if name.startswith("_astrbot_legacy_pkg_mrfzccl")
- ]
- for module_name in synthetic_modules:
- sys.modules.pop(module_name, None)
- for p in paths_to_add:
- if p in sys.path:
- sys.path.remove(p)
-
if __name__ == "__main__":
pytest.main([__file__, "-v"])
-
-
- 统计
-
-
-
-
-
- 正确
{self._fmt_int(correct)}
错误
{self._fmt_int(wrong)}
提示
{self._fmt_int(tip)}
-
-
-
-
-
- 排名
-
-
- 正确#{self._esc(correct_rank)}
- 错误#{self._esc(wrong_rank)}
- 提示#{self._esc(tip_rank)}
-
- 创建 {self._esc(created_at)}
- 更新 {self._esc(updated_at)} -
- - 更新 {self._esc(updated_at)} -