一、功能介绍
1. 探险模式
探险模式采用游戏画面识别和模拟键盘、鼠标操作的方式完成挑战流程，支持以下两种运行方式：
- 自动开火模式：根据游戏倒计时、击败提示和奖励弹窗自动执行开火、停止开火、补充弹药及确认奖励等操作。
- 自动召唤模式：循环检测召唤入口，自动按 E 进入选卡页面，根据设定的卡片星级和卡槽位置完成选卡并召唤 Boss。
- 持续开火模式：持续开枪，手动接管鼠标后自动停止运行。
2. 刷水晶箱子模式
刷水晶箱子模式会自动识别游戏大厅、房间、准备界面、游戏内画面和结算界面，并按照流程完成刷水晶箱子操作。
- 进入游戏后首次右键操作会随机等待 8～12 秒。
- 首次右键后，每隔随机 10～20 秒点按一次鼠标左键和鼠标右键。
- 分数达到目标值后，会连续确认分数，确认成功后退出当前局。
- 当前局结束后会继续下一轮，运行轮数和运行时间不设上限，直到手动停止。
二、操作教程(必看)
1. 设置游戏窗口
- 启动穿越火线。
- 将游戏设置为 1024×768 窗口模式(和助手设置的分辨率对应即可)。注意设置完成后请大退游戏进程重新进入才会生效！！！！！
- 确认游戏画面没有被其他窗口、弹窗或桌面内容遮挡。
- 将换弹按键(R)设置为键盘右侧的方向键(上下左右箭头)的右箭头->
- 注意电脑分辨率缩放设置，一定要设置100%，其他会无法识别游戏窗口。
- 游戏画质建议开启为性能优先，低画质下识别率更精准
![图片[1]-穿越火线探险挑战模式挂机-速易博客](https://gamehook.top/wp-content/uploads/2026/10/image-1024x793.png)
![图片[2]-穿越火线探险挑战模式挂机-速易博客](https://gamehook.top/wp-content/uploads/2026/10/image-2-1024x793.png)
2. 安装罗技鼠标驱动
- 1.下载驱动LGS_9.02.65_x64_Logitech.exe后进行安装，安装后电脑会自动重启。win11可能出现无法在此设备上加载驱动程序弹窗警告，点击弹窗底部的了解详情按钮后会跳转网页，参考网页内解决方案即可。
- 2.电脑重启后，在开始菜单中打开罗技游戏软件。
![图片[3]-穿越火线探险挑战模式挂机-速易博客](https://gamehook.top/wp-content/uploads/2026/10/image-6.png)
![图片[4]-穿越火线探险挑战模式挂机-速易博客](https://gamehook.top/wp-content/uploads/2026/10/image-8.png)
软件使用罗技鼠标驱动中的控制功能发送鼠标左右键操作。不需要实体罗技鼠标设备，对你手里拿的鼠标没要求。
因游戏检测非法外设输入，故罗技鼠标驱动安装是运行脚本的必要条件。未安装驱动时，两个模式都不会启动。
3. 安装使用挑战助手
3.1 下载
从网盘下载最新cf挑战助手.zip后，解压，打开cf挑战助手.exe运行即可(需给予管理员权限启动)。
![图片[5]-穿越火线探险挑战模式挂机-速易博客](https://gamehook.top/wp-content/uploads/2026/10/image-7.png)
3.2 检查顶部驱动状态
打开 CF挑战助手后，软件顶部会显示罗技鼠标驱动状态：
- 绿色“已安装并运行”：已检测到罗技驱动安装文件，并且 LCore.exe 正在运行。
- 红色“已安装，未运行”：已检测到驱动已经安装，但当前没有检测到 LCore.exe 运行。
- 红色“未安装”：未检测到有效的罗技驱动安装信息，启动脚本时会被拦截。
3.3 启动和停止脚本
- 进入“探险模式”或“刷水晶箱子”页签。
- 点击页面上的“启动”按钮，或按下 F9。
- 需要停止脚本时，点击“停止”按钮，或按下 F10。
- 首次运行脚本时，大概率触发ACE非法外设信号输入异常弹窗，忽略关闭即可，不会封号。
5. 视频教程演示
三、刷水晶箱子使用步骤
1. 设置目标分数
软件会使用数字图片模板识别游戏内分数。分数位数增加时，会按照右侧对齐、左侧延长的方式识别。
2. 启动刷水晶箱子
- 进入“刷水晶箱子”页签。
- 确认地图和目标分数设置正确。
- 点击“启动”或按 F9。
- 脚本会自动执行大厅、房间、频道和局内流程。
- 建议使用KAC-圣炎切割者III
3. 局内操作说明
- 进入局内后，首次右键操作随机等待 8～12 秒。
- 之后每隔随机 10～20 秒进行一次鼠标左键点按和鼠标右键点按。
- 不会长时间持续按住鼠标左键。
- 如果游戏画面暂时无法识别，脚本会等待并继续检测，避免直接执行不确定操作。
4. 停止刷水晶箱子
需要停止时按 F10 或点击“停止”。如果脚本正在局内操作，建议等待鼠标按键释放后再关闭游戏窗口。
四、常见问题
1. 提示未检测到罗技鼠标驱动怎么办？
请先通过教程安装 Logitech Gaming Software。安装完成后，至少手动打开一次 Logitech Gaming Software，等待驱动完成初始化，然后重新启动 CF挑战助手。
如果仍然显示未安装，请确认没有删除 LCore.exe，并检查 Windows 的程序安装信息是否完整。安装路径不要求固定为 C:\Program Files\Logitech Gaming Software，软件会自动检查常见安装路径和注册表安装登记。
2. 已安装但没有运行 LCore，还能启动脚本吗？
可以。只要软件能够确认 Logitech Gaming Software 已经安装，并且此前至少成功打开过一次，当前 LCore.exe 未运行通常不会阻止脚本启动。
如果鼠标左键或右键操作没有生效，请手动打开 Logitech Gaming Software 后重新运行脚本。
3. 提示未找到 CF 游戏窗口怎么办？
- 确认穿越火线已经启动。
- 确认游戏处于 1024×768 窗口模式。
- 确认游戏窗口没有最小化和未遮挡。
- 将游戏切换到前台后，再重新启动脚本。
4. 游戏窗口移动后会不会识别错误？
软件会动态读取游戏窗口的位置，并重新计算截图区域和点击坐标。运行中拖动窗口后，松开窗口即可继续检测。拖动过程中的短暂识别失败属于正常现象。
5. 刷水晶箱子分数显示为“–”怎么办？
- 确认已经进入游戏内，而不是停留在大厅或加载界面。
- 确认游戏画面没有被其他窗口遮挡。
- 确认分辨率和助手对应、画质调整为最低(性能优先)。
6. 按 F9 或 F10 没有反应怎么办？
请确认软件已经正常打开，并且当前没有其他模式正在运行。F9 和 F10 根据当前页签判断操作对象；如果当前位于“关于/必看教程”页签，F9 默认启动探险模式。
如果软件未完成启动检查、未找到 CF 窗口或未安装罗技鼠标驱动，脚本不会启动。
五、QQ 交流群
QQ 交流群号：1107740569
欢迎一起组队、交流使用经验，也欢迎反馈问题、提交建议和讨论功能改进。
六、核心实现代码
main.py
from __future__ import annotations
import ctypes
import hashlib
import hmac
import json
import logging
import os
import queue
import re
import threading
import tkinter as tk
import urllib.parse
import urllib.request
import uuid
import webbrowser
from dataclasses import replace
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from tkinter import messagebox, ttk
from typing import Any, Optional
from PIL import Image, ImageTk
try:
import psutil # type: ignore
except ImportError: # pragma: no cover - requirements.txt includes psutil
psutil = None # type: ignore[assignment]
import challenge_bot
import expedition_bot
ROOT = Path(__file__).resolve().parent
APP_VERSION = "1.0.4"
VERSION_API = ""
USAGE_API = ""
USAGE_INTERVAL_MS = 30_000
TUTORIAL_URL = "https://gamehook.top/430.html"
DONATION_IMAGE = ROOT / "resource" / "wx_zanshang.jpg"
CHALLENGE_CONFIG = ROOT / "challenge_config.json"
EXPEDITION_CONFIG = ROOT / "expedition_config.json"
VOICE_FILES = {
"started": ROOT / "resource" / "脚本已运行.mp3",
"stopped": ROOT / "resource" / "脚本已停止.mp3",
"boss_10": ROOT / "resource" / "10分钟提醒.mp3",
"boss_20": ROOT / "resource" / "20分钟提醒.mp3",
"boss_35": ROOT / "resource" / "35分钟提醒.mp3",
}
ACTIVATION_FILE = Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming")) / "cf挑战助手" / "activation.json"
USAGE_ID_FILE = ACTIVATION_FILE.with_name("usage.json")
LOGITECH_LCORE_PATHS = (
Path(r"C:\Program Files\Logitech Gaming Software\LCore.exe"),
Path(r"C:\Program Files (x86)\Logitech Gaming Software\LCore.exe"),
)
LOGITECH_TUTORIAL_MESSAGE = (
"未检测到已安装的罗技鼠标驱动。请先通过“关于/必看教程”中的教程完成安装，"
"并至少打开 Logitech Gaming Software 一次，确认驱动初始化后再运行脚本。"
)
@dataclass(frozen=True)
class VersionInfo:
version: str
notes: list[str]
url: str
password: str
password_tips: str
@dataclass(frozen=True)
class UsageCounts:
online: int
today: int
def load_usage_id(path: Path = USAGE_ID_FILE) -> str:
try:
value = json.loads(path.read_text(encoding="utf-8"))["client_id"]
if isinstance(value, str):
return str(uuid.UUID(value))
raise ValueError("无效的匿名客户端 ID")
except (OSError, ValueError, KeyError, TypeError):
client_id = str(uuid.uuid4())
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_suffix(".tmp")
temporary.write_text(json.dumps({"client_id": client_id}), encoding="utf-8")
os.replace(temporary, path)
return client_id
def report_usage(client_id: str) -> UsageCounts:
body = json.dumps({"client_id": client_id}).encode("utf-8")
request = urllib.request.Request(
USAGE_API,
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=5) as response:
payload = json.load(response)
if not isinstance(payload, dict) or payload.get("code") != 0 or not isinstance(payload.get("data"), dict):
raise ValueError("使用人数接口返回无效")
data = payload["data"]
online, today = data.get("online"), data.get("today")
if any(type(value) is not int or value < 0 for value in (online, today)):
raise ValueError("使用人数接口统计值无效")
return UsageCounts(online, today)
def version_is_newer(remote: str, current: str) -> bool:
"""Return whether a remote dotted numeric version is newer than local."""
pattern = re.compile(r"^\s*v?(\d+(?:\.\d+)*)\s*$", re.IGNORECASE)
remote_match = pattern.fullmatch(remote)
current_match = pattern.fullmatch(current)
if remote_match is None or current_match is None:
return False
remote_parts = tuple(int(part) for part in remote_match.group(1).split("."))
current_parts = tuple(int(part) for part in current_match.group(1).split("."))
width = max(len(remote_parts), len(current_parts))
return remote_parts + (0,) * (width - len(remote_parts)) > current_parts + (0,) * (width - len(current_parts))
def activation_digest(password: str) -> str:
return hashlib.sha256(password.encode("utf-8")).hexdigest()
def activation_matches(entered: str, expected: str) -> bool:
return hmac.compare_digest(entered.encode("utf-8"), expected.encode("utf-8"))
def is_administrator() -> bool:
if os.name != "nt":
return False
try:
return bool(ctypes.windll.shell32.IsUserAnAdmin())
except (AttributeError, OSError):
return False
def saved_activation_matches(password: str, path: Path = ACTIVATION_FILE) -> bool:
try:
data = json.loads(path.read_text(encoding="utf-8"))
saved = data.get("sha256") if isinstance(data, dict) else None
return isinstance(saved, str) and hmac.compare_digest(saved, activation_digest(password))
except (OSError, ValueError):
return False
def save_activation(password: str, path: Path = ACTIVATION_FILE) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_suffix(".tmp")
temporary.write_text(json.dumps({"sha256": activation_digest(password)}), encoding="utf-8")
os.replace(temporary, path)
def _path_from_registry_value(value: Any) -> Optional[Path]:
if not isinstance(value, str):
return None
match = re.match(r'^\s*"([^"]+\.exe)"|^\s*([^,\s]+\.exe)', value, re.IGNORECASE)
raw = (match.group(1) or match.group(2)) if match else value.split(",", 1)[0].strip().strip('"')
path = Path(raw)
return path if path.name.casefold() == "lcore.exe" else None
def _registry_logitech_paths() -> set[Path]:
if os.name != "nt":
return set()
try:
import winreg
except ImportError:
return set()
paths: set[Path] = set()
def add_values(key: Any) -> None:
try:
names = {"", "displayicon", "installlocation", "path", "installpath"}
for index in range(winreg.QueryInfoKey(key)[1]):
name, value, _ = winreg.EnumValue(key, index)
if name.casefold() not in names or not isinstance(value, str):
continue
candidate = _path_from_registry_value(value)
if candidate is not None:
paths.add(candidate)
else:
folder = Path(value.strip().strip('"'))
if folder.name.casefold() != "lcore.exe":
paths.add(folder / "LCore.exe")
except (OSError, ValueError):
return
def scan_root(hive: Any, root: str, uninstall: bool) -> None:
try:
with winreg.OpenKey(hive, root) as key:
if not uninstall:
add_values(key)
for index in range(winreg.QueryInfoKey(key)[0]):
try:
name = winreg.EnumKey(key, index)
with winreg.OpenKey(key, name) as child:
if uninstall:
try:
display_name = str(winreg.QueryValueEx(child, "DisplayName")[0]).casefold()
except OSError:
continue
if not (
("logitech" in display_name or "罗技" in display_name)
and ("gaming software" in display_name or "游戏软件" in display_name)
):
continue
add_values(child)
except OSError:
continue
except OSError:
return
hives = (winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER)
uninstall_roots = (
r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall",
r"SOFTWARE\WOW6432Node\Microsoft\Windows\CurrentVersion\Uninstall",
)
app_paths = (
r"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\LCore.exe",
r"SOFTWARE\WOW6432Node\Microsoft\Windows\CurrentVersion\App Paths\LCore.exe",
)
for hive in hives:
for root in uninstall_roots:
scan_root(hive, root, True)
for root in app_paths:
scan_root(hive, root, False)
return paths
def _logitech_candidate_paths(paths: tuple[Path, ...]) -> set[Path]:
candidates = set(paths)
if paths == LOGITECH_LCORE_PATHS:
for variable in ("ProgramFiles", "ProgramFiles(x86)", "LOCALAPPDATA"):
base = os.environ.get(variable)
if base:
candidates.add(Path(base) / "Logitech Gaming Software" / "LCore.exe")
candidates.update(_registry_logitech_paths())
return candidates
def detect_logitech_driver(paths: tuple[Path, ...] = LOGITECH_LCORE_PATHS) -> tuple[bool, bool]:
"""Return (installed, running), even when Logitech uses a custom path."""
candidates = _logitech_candidate_paths(paths)
normalized_paths = {os.path.normcase(os.path.normpath(str(path))) for path in candidates}
installed = any(path.is_file() for path in candidates)
running = False
if psutil is not None:
try:
for process in psutil.process_iter(["name", "exe"]):
info = process.info
name = str(info.get("name") or "").casefold()
executable = info.get("exe")
normalized_executable = (
os.path.normcase(os.path.normpath(str(executable))) if executable else ""
)
if name != "lcore.exe":
continue
if executable and Path(str(executable)).name.casefold() == "lcore.exe" and Path(str(executable)).is_file():
installed = True
running = True
break
if installed and (not normalized_executable or normalized_executable in normalized_paths):
running = True
break
except Exception:
logging.debug("检测罗技驱动进程失败", exc_info=True)
return installed, running
def latest_version_info() -> VersionInfo:
body = urllib.parse.urlencode({"key": "cf_helper_version"}).encode("ascii")
request = urllib.request.Request(
VERSION_API,
data=body,
headers={"Content-Type": "application/x-www-form-urlencoded"},
method="POST",
)
with urllib.request.urlopen(request, timeout=8) as response:
payload = json.load(response)
if payload.get("code") != 0 or not isinstance(payload.get("data"), dict):
raise ValueError("版本接口返回无效")
data = payload["data"]
version = data.get("new_version")
if not isinstance(version, str) or not version.strip():
raise ValueError("版本接口缺少版本号")
content = data.get("content", [])
if not isinstance(content, list):
content = []
notes = [item for item in content if isinstance(item, str)]
raw_url = data.get("download_url", "")
# Some API values contain a pasted Markdown link; take its URL target.
url = raw_url.rsplit("](", 1)[-1].rstrip(")") if isinstance(raw_url, str) else ""
parsed = urllib.parse.urlparse(url)
if parsed.scheme != "https" or not parsed.netloc:
url = TUTORIAL_URL
password = data.get("password", "")
password_tips = data.get("password_tips", "")
if not isinstance(password, str) or not isinstance(password_tips, str):
raise ValueError("版本接口的激活信息格式无效")
return VersionInfo(version.strip(), notes, url, password, password_tips)
class QueueLogHandler(logging.Handler):
def __init__(self, messages: queue.Queue[str]):
super().__init__()
self.messages = messages
def emit(self, record: logging.LogRecord) -> None:
try:
self.messages.put_nowait(self.format(record))
except Exception:
self.handleError(record)
class VoicePlayer:
"""Play the bundled MP3 prompts through Windows MCI without dependencies."""
ALIAS = "cf_challenge_voice"
def __init__(self) -> None:
self._lock = threading.Lock()
self.enabled = True
self.boss_enabled = True
def play(self, event: str) -> None:
if (event.startswith("boss_") and not self.boss_enabled) or (not event.startswith("boss_") and not self.enabled):
return
path = VOICE_FILES.get(event)
if os.name != "nt" or path is None or not path.is_file():
logging.warning("语音资源不可用：%s", path)
return
threading.Thread(target=self._play, args=(path,), daemon=True, name="voice-prompt").start()
def _play(self, path: Path) -> None:
with self._lock:
mci = ctypes.windll.winmm.mciSendStringW
mci(f"close {self.ALIAS}", None, 0, None)
result = mci(f'open "{path}" type mpegvideo alias {self.ALIAS}', None, 0, None)
if result:
logging.warning("无法加载语音资源：%s（MCI %s）", path.name, result)
return
result = mci(f"play {self.ALIAS} from 0", None, 0, None)
if result:
logging.warning("无法播放语音资源：%s（MCI %s）", path.name, result)
def close(self) -> None:
if os.name == "nt":
with self._lock:
ctypes.windll.winmm.mciSendStringW(f"close {self.ALIAS}", None, 0, None)
class ChallengePage(ttk.Frame):
def __init__(self, app: "ChallengeAssistant") -> None:
super().__init__(app.notebook, padding=12)
self.app = app
self.score_target = tk.StringVar(value="700000")
self.status = tk.StringVar(value="未启动")
self.scene = tk.StringVar(value="场景：未启动")
self.score = tk.StringVar(value="当前分数：--")
self.rounds = tk.StringVar(value="已完成：0 轮")
self.bot: Optional[challenge_bot.ChallengeBot] = None
self.worker: Optional[threading.Thread] = None
self._auto_save_job: Optional[str] = None
self._loading = True
self._build()
self._load()
self._loading = False
self.score_target.trace_add("write", self._schedule_auto_save)
self.after(500, self._refresh)
def _build(self) -> None:
self.columnconfigure(0, weight=1)
settings = ttk.LabelFrame(self, text="刷水晶箱子")
settings.grid(row=0, column=0, sticky="ew")
settings.columnconfigure(1, weight=1)
fixed = (
("地图", "新巨人成废墟，建议使用枪械KAC-圣炎切割者III"),
("首次右键", "进入局内后随机 8～12 秒"),
("周期点按", "每隔随机 10～20 秒左键+右键"),
)
for row, (label, value) in enumerate(fixed):
ttk.Label(settings, text=label, style="Field.TLabel").grid(row=row, column=0, padx=12, pady=6, sticky="w")
ttk.Label(settings, text=value, wraplength=260, justify="left").grid(
row=row, column=1, padx=12, pady=6, sticky="w"
)
ttk.Label(settings, text="目标分数", style="Field.TLabel").grid(row=len(fixed), column=0, padx=12, pady=6, sticky="w")
ttk.Entry(settings, textvariable=self.score_target, width=18).grid(
row=len(fixed), column=1, padx=12, pady=(6, 12), sticky="ew"
)
actions = ttk.Frame(self)
actions.grid(row=1, column=0, pady=10, sticky="ew")
actions.columnconfigure((0, 1), weight=1)
self.start_button = ttk.Button(actions, text="F9 启动", style="Primary.TButton", command=self.app.start_selected)
self.start_button.grid(row=0, column=0, padx=(0, 5), sticky="ew")
self.stop_button = ttk.Button(actions, text="F10 停止", command=self.app.stop_active, state="disabled")
self.stop_button.grid(row=0, column=1, padx=(5, 0), sticky="ew")
runtime = ttk.LabelFrame(self, text="运行状态")
runtime.grid(row=2, column=0, sticky="ew")
runtime.columnconfigure((0, 1), weight=1, uniform="challenge-status")
for index, variable in enumerate((self.status, self.rounds, self.scene, self.score)):
ttk.Label(runtime, textvariable=variable, wraplength=165, anchor="w").grid(
row=index // 2, column=index % 2, padx=(10, 4), pady=2, sticky="ew"
)
def _read(self) -> tuple[dict[str, Any], challenge_bot.Settings, dict[str, challenge_bot.TemplateSpec], Path]:
raw = dict(challenge_bot.load_json(CHALLENGE_CONFIG))
settings = challenge_bot.build_settings(raw)
templates = challenge_bot.read_templates(raw)
templates_dir = (ROOT / str(raw.get("templates_dir", "challenge_templates"))).resolve()
return raw, settings, templates, templates_dir
def _load(self) -> None:
try:
_, settings, _, _ = self._read()
self.score_target.set(str(settings.score_target))
self.status.set("配置已加载，启动后将直接控制游戏")
except Exception as exc:
self.status.set("配置加载失败")
messagebox.showerror("挑战配置错误", str(exc))
def _settings(self) -> tuple[challenge_bot.Settings, dict[str, challenge_bot.TemplateSpec], Path]:
_, settings, templates, templates_dir = self._read()
try:
target = max(1, int(self.score_target.get()))
except ValueError as exc:
raise challenge_bot.BotError("目标分数必须是有效整数") from exc
settings = replace(
settings,
poll_interval=challenge_bot.POLL_INTERVAL_SECONDS,
score_target=target,
)
return settings, templates, templates_dir
def save(self, quiet: bool = False) -> None:
try:
settings, _, _ = self._settings()
raw = dict(challenge_bot.load_json(CHALLENGE_CONFIG))
raw["score_target"] = settings.score_target
for removed in (
"poll_interval",
"max_rounds",
"max_runtime_minutes",
"right_click_delay",
"right_click_delay_min",
"right_click_delay_max",
"score_confirmations",
"tesseract_cmd",
"left_click_wait",
"left_click_hold",
):
raw.pop(removed, None)
CHALLENGE_CONFIG.write_text(json.dumps(raw, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
if not quiet:
self.status.set("参数已保存")
except Exception as exc:
if quiet:
raise
messagebox.showerror("保存失败", str(exc))
def _schedule_auto_save(self, *_args: Any) -> None:
if self._loading:
return
if self._auto_save_job is not None:
self.after_cancel(self._auto_save_job)
self._auto_save_job = self.after(500, self._auto_save)
def _auto_save(self) -> None:
self._auto_save_job = None
try:
self.save(quiet=True)
self.status.set("目标分数已自动保存")
except Exception:
self.status.set("目标分数输入无效，尚未保存")
def start(self) -> bool:
if self.worker and self.worker.is_alive():
return False
if not self.app.claim_runner(self):
return False
try:
self.save(quiet=True)
settings, templates, templates_dir = self._settings()
missing = [spec.file for spec in templates.values() if not (templates_dir / spec.file).is_file()]
if missing:
raise challenge_bot.BotError("缺少模板：" + "、".join(missing))
challenge_bot.WindowGuard(settings).check()
ui = challenge_bot.UI(
templates_dir,
live=True,
jitter=settings.click_jitter,
window_titles=settings.window_titles,
process_names=settings.process_names,
hid_dll=(ROOT / settings.hid_dll).resolve() if settings.hid_dll else None,
)
self.bot = challenge_bot.ChallengeBot(settings, templates, ui)
except Exception as exc:
self.app.release_runner(self)
messagebox.showerror("无法启动挑战模式", str(exc))
return False
self.status.set("运行中（直接控制游戏）")
self.start_button.configure(state="disabled")
self.stop_button.configure(state="normal")
self.worker = threading.Thread(target=self._run, daemon=True, name="challenge-mode")
self.worker.start()
self.app.set_running_title(self)
return True
def _run(self) -> None:
result = 1
try:
assert self.bot is not None
result = self.bot.run()
except Exception:
logging.exception("挑战模式异常退出")
self.after(0, lambda: self._finished(result))
def _finished(self, result: int) -> None:
self.start_button.configure(state="normal")
self.stop_button.configure(state="disabled")
self.status.set("已停止" if result == 0 else "运行失败，已停止")
self.app.release_runner(self)
def stop(self) -> bool:
if self.worker and self.worker.is_alive() and self.bot is not None:
self.bot.stop()
self.status.set("正在停止...")
return True
return False
def _refresh(self) -> None:
if self.bot is not None:
self.scene.set(f"场景：{self.bot.stage}")
self.rounds.set(f"已完成：{self.bot.completed_rounds} 轮")
score = self.bot.current_score
self.score.set(
"分数：--"
if score is None
else f"分数：{score:,}（{self.bot.score_confirmation_count}/{challenge_bot.SCORE_CONFIRMATIONS}）"
)
self.after(500, self._refresh)
class ExpeditionPage(ttk.Frame):
MODE_TO_CODE = {"自动开火模式": "smart_fire", "自动召唤模式": "smart_card"}
CODE_TO_MODE = {value: key for key, value in MODE_TO_CODE.items()}
def __init__(self, app: "ChallengeAssistant") -> None:
super().__init__(app.notebook, padding=12)
self.app = app
self.operation_mode = tk.StringVar(value="自动开火模式")
self.card_star = tk.StringVar(value="1")
self.card_slot = tk.StringVar(value="1")
self.hotkey_sound = tk.BooleanVar(value=True)
self.boss_sound = tk.BooleanVar(value=True)
self.status = tk.StringVar(value="未启动")
self.stage = tk.StringVar(value="等待启动")
self.game_elapsed = tk.StringVar(value="游戏开局时间 --:--")
self.bot: Optional[expedition_bot.CFHelperCloneBot] = None
self.worker: Optional[threading.Thread] = None
self._loading = True
self._auto_save_job: Optional[str] = None
self._build()
self._load()
self._loading = False
for variable in (self.operation_mode, self.card_star, self.card_slot, self.hotkey_sound, self.boss_sound):
variable.trace_add("write", self._schedule_auto_save)
self.hotkey_sound.trace_add("write", self._sync_hotkey_sound)
self.boss_sound.trace_add("write", self._sync_boss_sound)
self.after(300, self._refresh)
def _build(self) -> None:
self.columnconfigure(0, weight=1)
settings = ttk.LabelFrame(self, text="探险模式")
settings.grid(row=0, column=0, sticky="ew")
settings.columnconfigure((0, 1), weight=1, uniform="expedition-settings")
mode_field = ttk.Frame(settings)
mode_field.grid(row=0, column=0, padx=(12, 6), pady=(8, 4), sticky="ew")
mode_field.columnconfigure(0, weight=1)
ttk.Label(mode_field, text="运行模式", style="Field.TLabel").grid(row=0, column=0, sticky="w")
ttk.Combobox(
mode_field,
textvariable=self.operation_mode,
values=tuple(self.MODE_TO_CODE),
state="readonly",
width=12,
).grid(row=1, column=0, pady=(4, 0), sticky="ew")
sound_field = ttk.Frame(settings)
sound_field.grid(row=0, column=1, padx=(6, 12), pady=(8, 4), sticky="ew")
ttk.Label(sound_field, text="声音设置", style="Field.TLabel").grid(row=0, column=0, columnspan=2, sticky="w")
ttk.Checkbutton(sound_field, text="热键声音", variable=self.hotkey_sound).grid(row=1, column=0, pady=(4, 0), sticky="w")
ttk.Checkbutton(sound_field, text="Boss提醒", variable=self.boss_sound).grid(row=1, column=1, pady=(4, 0), sticky="w")
star_field = ttk.Frame(settings)
star_field.grid(row=1, column=0, padx=(12, 6), pady=(4, 12), sticky="ew")
star_field.columnconfigure(0, weight=1)
ttk.Label(star_field, text="卡片星级", style="Field.TLabel").grid(row=0, column=0, sticky="w")
ttk.Combobox(
star_field, textvariable=self.card_star, values=("1", "2", "3", "4", "5", "6"), state="readonly", width=12
).grid(row=1, column=0, pady=(4, 0), sticky="ew")
slot_field = ttk.Frame(settings)
slot_field.grid(row=1, column=1, padx=(6, 12), pady=(4, 12), sticky="ew")
slot_field.columnconfigure(0, weight=1)
ttk.Label(slot_field, text="卡槽位置", style="Field.TLabel").grid(row=0, column=0, sticky="w")
ttk.Combobox(
slot_field, textvariable=self.card_slot, values=("1", "2", "3", "4", "5"), state="readonly", width=12
).grid(row=1, column=0, pady=(4, 0), sticky="ew")
ttk.Label(settings, text="游戏分辨率 固定 1024x768", style="Field.TLabel").grid(
row=2, column=0, padx=(12, 6), pady=(0, 10), sticky="w"
)
ttk.Label(settings, textvariable=self.game_elapsed, style="Timer.TLabel").grid(
row=2, column=1, padx=(6, 12), pady=(0, 10), sticky="e"
)
actions = ttk.Frame(self)
actions.grid(row=1, column=0, pady=10, sticky="ew")
actions.columnconfigure((0, 1), weight=1)
self.start_button = ttk.Button(actions, text="F9 启动", style="Primary.TButton", command=self.app.start_selected)
self.start_button.grid(row=0, column=0, padx=(0, 5), sticky="ew")
self.stop_button = ttk.Button(actions, text="F10 停止", command=self.app.stop_active, state="disabled")
self.stop_button.grid(row=0, column=1, padx=(5, 0), sticky="ew")
runtime = ttk.LabelFrame(self, text="运行状态")
runtime.grid(row=2, column=0, sticky="ew")
ttk.Label(runtime, textvariable=self.status).pack(anchor="w", padx=10, pady=4)
ttk.Label(runtime, textvariable=self.stage).pack(anchor="w", padx=10, pady=4)
def _load(self) -> None:
try:
raw = expedition_bot.load_json(EXPEDITION_CONFIG)
self.operation_mode.set(self.CODE_TO_MODE.get(str(raw.get("operation_mode")), "自动开火模式"))
self.card_star.set(str(raw.get("card_star", 1)))
self.card_slot.set(str(raw.get("card_slot", 1)))
self.hotkey_sound.set(raw.get("hotkey_sound", True) is not False)
self.boss_sound.set(raw.get("boss_sound", True) is not False)
self.app.voice.enabled = self.hotkey_sound.get()
self.app.voice.boss_enabled = self.boss_sound.get()
self.status.set("配置已加载，启动后将直接控制游戏")
except Exception as exc:
self.status.set("配置加载失败")
messagebox.showerror("探险配置错误", str(exc))
def _raw_from_form(self) -> dict[str, Any]:
raw = dict(expedition_bot.load_json(EXPEDITION_CONFIG))
try:
for removed in (
"poll_interval", "confidence", "countdown_delay", "post_defeat_delay", "repeat_count",
"max_runtime_minutes", "right_after_defeat", "summon_timeout",
):
raw.pop(removed, None)
raw.update(
{
"resolution": [1024, 768],
"capture_region": [0, 0, 1024, 768],
"live": True,
"input_backend": "cfhelper",
"operation_mode": self.MODE_TO_CODE[self.operation_mode.get()],
"card_star": max(1, min(6, int(self.card_star.get()))),
"card_slot": max(1, min(5, int(self.card_slot.get()))),
"hotkey_sound": self.hotkey_sound.get(),
"boss_sound": self.boss_sound.get(),
}
)
except (ValueError, KeyError) as exc:
raise expedition_bot.BotError("探险模式参数必须是有效数字或有效选项") from exc
return raw
def _settings(self) -> expedition_bot.Settings:
return expedition_bot.build_settings(self._raw_from_form())
def save(self) -> None:
raw = self._raw_from_form()
expedition_bot.build_settings(raw)
EXPEDITION_CONFIG.write_text(json.dumps(raw, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def _schedule_auto_save(self, *_args: Any) -> None:
if self._loading:
return
if self._auto_save_job is not None:
self.after_cancel(self._auto_save_job)
self._auto_save_job = self.after(300, self._auto_save)
def _sync_hotkey_sound(self, *_args: Any) -> None:
self.app.voice.enabled = self.hotkey_sound.get()
def _sync_boss_sound(self, *_args: Any) -> None:
self.app.voice.boss_enabled = self.boss_sound.get()
def _auto_save(self) -> None:
self._auto_save_job = None
try:
self.save()
self.status.set("设置已自动保存")
except Exception:
logging.exception("探险设置自动保存失败")
self.status.set("设置保存失败")
def start(self) -> bool:
if self.worker and self.worker.is_alive():
return False
if not self.app.claim_runner(self):
return False
try:
self.save()
settings = self._settings()
expedition_bot.WindowGuard(settings).check()
self.bot = expedition_bot.CFHelperCloneBot(settings, live=True, on_reminder=self.app.voice.play)
except Exception as exc:
self.app.release_runner(self)
messagebox.showerror("无法启动探险模式", str(exc))
return False
self.status.set("运行中（直接控制游戏）")
self.game_elapsed.set("游戏开局时间 --:--")
self.start_button.configure(state="disabled")
self.stop_button.configure(state="normal")
self.worker = threading.Thread(target=self._run, daemon=True, name="expedition-mode")
self.worker.start()
self.app.set_running_title(self)
return True
def _run(self) -> None:
result = 1
try:
assert self.bot is not None
result = self.bot.run()
except Exception:
logging.exception("探险模式异常退出")
self.after(0, lambda: self._finished(result))
def _finished(self, result: int) -> None:
self.start_button.configure(state="normal")
self.stop_button.configure(state="disabled")
reason = self.bot.stop_reason if self.bot is not None else None
if reason:
self.status.set(f"{reason}，已停止")
self.app.voice.play("stopped")
else:
self.status.set("已停止" if result == 0 else "运行失败，已停止")
self.game_elapsed.set("游戏开局时间 --:--")
self.app.release_runner(self)
def stop(self) -> bool:
if self.worker and self.worker.is_alive() and self.bot is not None:
self.bot.stop()
self.status.set("正在停止...")
return True
return False
def _refresh(self) -> None:
if self.bot is not None:
self.stage.set(self.bot.stage)
if self.worker and self.worker.is_alive():
elapsed = getattr(self.bot.detector, "game_elapsed", None) or "--:--"
self.game_elapsed.set(f"游戏开局时间 {elapsed}")
self.after(300, self._refresh)
class AboutPage(ttk.Frame):
def __init__(self, app: "ChallengeAssistant") -> None:
super().__init__(app.notebook)
canvas = tk.Canvas(self, background="#f3f5f7", highlightthickness=0)
scrollbar = ttk.Scrollbar(self, orient="vertical", command=canvas.yview)
canvas.configure(yscrollcommand=scrollbar.set)
canvas.pack(side="left", fill="both", expand=True)
scrollbar.pack(side="right", fill="y")
content = ttk.Frame(canvas, padding=(20, 16))
window = canvas.create_window((0, 0), window=content, anchor="nw")
canvas.bind("<Configure>", lambda event: canvas.itemconfigure(window, width=event.width))
ttk.Label(content, text="赞赏支持", style="Section.TLabel").pack(anchor="w")
ttk.Label(content, text="自愿赞赏，感谢支持。", style="Field.TLabel").pack(anchor="w", pady=(3, 0))
try:
with Image.open(DONATION_IMAGE) as source:
picture = source.resize((280, 280), Image.Resampling.LANCZOS)
self.donation_photo = ImageTk.PhotoImage(picture)
ttk.Label(content, image=self.donation_photo).pack(anchor="center", pady=(8, 14))
except (OSError, tk.TclError):
ttk.Label(content, text="赞赏码暂时无法显示").pack(anchor="w", pady=(8, 14))
ttk.Separator(content).pack(fill="x", pady=(0, 12))
tutorial = ttk.Label(content, text="使用教程 gamehook.top/430.html ↗", style="Link.TLabel", cursor="hand2")
tutorial.pack(anchor="w", pady=(0, 12))
tutorial.bind("<Button-1>", lambda _event: webbrowser.open(TUTORIAL_URL))
ttk.Separator(content).pack(fill="x", pady=(0, 12))
ttk.Label(content, text="QQ 交流群 1107740569", style="Section.TLabel").pack(anchor="w")
ttk.Label(content, text="一起组队、交流经验，也欢迎反馈问题和改进建议。", wraplength=340).pack(
anchor="w", fill="x", pady=(5, 8)
)
ttk.Button(content, text="复制群号", command=self._copy_qq).pack(anchor="w", pady=(0, 12))
ttk.Separator(content).pack(fill="x", pady=(0, 12))
ttk.Label(content, text="免责声明", style="Section.TLabel").pack(anchor="w")
disclaimer = (
"本工具仅依据屏幕画面进行识别，并模拟普通键盘、鼠标操作；"
"不读取、注入或修改游戏进程与内存，不拦截网络通信，也不篡改游戏文件或数据。\n\n"
"本软件完全免费，请勿二次售卖、捆绑收费或以本软件名义收取费用。\n\n"
"画面识别和自动操作可能受分辨率、网络状态及游戏更新影响，无法保证始终准确。"
"使用前请了解游戏的相关规则，并自行决定是否使用；因误操作、游戏规则变化或使用本工具"
"产生的账号及其他风险，由使用者自行承担。"
)
disclaimer_label = ttk.Label(content, text=disclaimer, justify="left", wraplength=320)
disclaimer_label.pack(anchor="w", fill="x", pady=(7, 16))
def resize_content(event: tk.Event) -> None:
canvas.configure(scrollregion=canvas.bbox("all"))
wraplength = max(200, event.width - 40)
if int(disclaimer_label.cget("wraplength")) != wraplength:
disclaimer_label.configure(wraplength=wraplength)
content.bind("<Configure>", resize_content)
def scroll_content(event: tk.Event) -> str:
if event.delta:
canvas.yview_scroll(-1 if event.delta > 0 else 1, "units")
return "break"
for widget in (canvas, content, *content.winfo_children()):
widget.bind("<MouseWheel>", scroll_content)
def _copy_qq(self) -> None:
self.clipboard_clear()
self.clipboard_append("1107740569")
class ChallengeAssistant(tk.Tk):
def __init__(self) -> None:
super().__init__()
self.withdraw()
self.title("cf挑战助手")
self.geometry("420x760")
self.minsize(420, 680)
self.configure(background="#f3f5f7")
self.columnconfigure(0, weight=1)
self.rowconfigure(1, weight=3)
self.rowconfigure(2, weight=2)
self.messages: queue.Queue[str] = queue.Queue()
self.version_results: queue.Queue[VersionInfo | Exception] = queue.Queue()
self.usage_results: queue.Queue[UsageCounts | Exception] = queue.Queue()
self.usage_stop = threading.Event()
self.active_runner: Optional[Any] = None
self.hotkey_stop = threading.Event()
self._activated = False
self._activation_dialog: Optional[tk.Toplevel] = None
self.voice = VoicePlayer()
self._configure_styles()
self._configure_logging()
header = ttk.Frame(self, style="Header.TFrame", padding=(18, 12))
header.grid(row=0, column=0, sticky="ew")
header.columnconfigure(0, weight=1)
ttk.Label(header, text=f"版本 {APP_VERSION}", style="HeaderVersion.TLabel").grid(
row=0, column=0, sticky="w"
)
ttk.Label(header, text="F9 启动 F10 停止", style="HeaderHint.TLabel").grid(
row=1, column=0, pady=(4, 0), sticky="w"
)
self.clock_time = tk.StringVar()
self.clock_date = tk.StringVar()
ttk.Label(header, textvariable=self.clock_time, style="HeaderClock.TLabel").grid(row=0, column=1, sticky="e")
ttk.Label(header, textvariable=self.clock_date, style="HeaderDate.TLabel").grid(
row=1, column=1, pady=(4, 0), sticky="e"
)
ttk.Separator(header, orient="horizontal").grid(row=2, column=0, columnspan=2, pady=(10, 7), sticky="ew")
self.logitech_status = tk.StringVar(value="罗技鼠标驱动：检测中")
self.logitech_label = ttk.Label(
header, textvariable=self.logitech_status, style="DriverError.TLabel"
)
self.logitech_label.grid(row=3, column=0, sticky="w")
ttk.Label(header, text="QQ群 1107740569", style="HeaderQQ.TLabel").grid(
row=3, column=1, sticky="e"
)
self.online_count = tk.StringVar(value="当前在线人数 --")
self.today_count = tk.StringVar(value="今日总使用人数 --")
ttk.Label(header, textvariable=self.online_count, style="HeaderCount.TLabel").grid(
row=4, column=0, pady=(8, 0), sticky="w"
)
ttk.Label(header, textvariable=self.today_count, style="HeaderCount.TLabel").grid(
row=4, column=1, pady=(8, 0), sticky="e"
)
self._refresh_clock()
self._refresh_logitech_status()
self.notebook = ttk.Notebook(self)
self.notebook.grid(row=1, column=0, padx=14, pady=(14, 7), sticky="nsew")
self.expedition_page = ExpeditionPage(self)
self.challenge_page = ChallengePage(self)
self.about_page = AboutPage(self)
self.notebook.add(self.expedition_page, text="探险模式")
self.notebook.add(self.challenge_page, text="刷水晶箱子")
self.notebook.add(self.about_page, text="关于/必看教程")
self.notebook.bind("<<NotebookTabChanged>>", self._on_tab_changed)
logs = ttk.LabelFrame(self, text="运行日志")
logs.grid(row=2, column=0, padx=14, pady=(7, 14), sticky="nsew")
logs.columnconfigure(0, weight=1)
logs.rowconfigure(0, weight=1)
self.log_text = tk.Text(logs, height=10, state="disabled", wrap="word")
self.log_text.grid(row=0, column=0, sticky="nsew")
scrollbar = ttk.Scrollbar(logs, orient="vertical", command=self.log_text.yview)
scrollbar.grid(row=0, column=1, sticky="ns")
self.log_text.configure(yscrollcommand=scrollbar.set)
self.logs = logs
# Keep the controls and hotkeys unavailable until the version endpoint
# has been reached and any required activation code has been verified.
self.after(120, self._drain_logs)
self.after(200, self._drain_version_results)
self.after(200, self._drain_usage_results)
threading.Thread(target=self._check_version, daemon=True, name="version-check").start()
self.protocol("WM_DELETE_WINDOW", self._close)
def _configure_styles(self) -> None:
style = ttk.Style(self)
if "vista" in style.theme_names():
style.theme_use("vista")
style.configure("TFrame", background="#f3f5f7")
style.configure("Header.TFrame", background="#18353b")
style.configure("HeaderVersion.TLabel", background="#18353b", foreground="#7ce2bd", font=("Microsoft YaHei UI", 11, "bold"))
style.configure("HeaderHint.TLabel", background="#18353b", foreground="#d0e3e5", font=("Microsoft YaHei UI", 9))
style.configure("HeaderQQ.TLabel", background="#18353b", foreground="#ffd166", font=("Microsoft YaHei UI", 10, "bold"))
style.configure("HeaderClock.TLabel", background="#18353b", foreground="#ffffff", font=("Microsoft YaHei UI", 16, "bold"))
style.configure("HeaderDate.TLabel", background="#18353b", foreground="#b0c8c9", font=("Microsoft YaHei UI", 9))
style.configure("DriverReady.TLabel", background="#18353b", foreground="#7ce2bd", font=("Microsoft YaHei UI", 9, "bold"))
style.configure("DriverError.TLabel", background="#18353b", foreground="#ff9b91", font=("Microsoft YaHei UI", 9, "bold"))
style.configure("HeaderCount.TLabel", background="#18353b", foreground="#ffffff", font=("Microsoft YaHei UI", 10, "bold"))
style.configure("TLabel", background="#f3f5f7", foreground="#1f2933", font=("Microsoft YaHei UI", 9))
style.configure("Field.TLabel", foreground="#52606d")
style.configure("Timer.TLabel", foreground="#087a97", font=("Microsoft YaHei UI", 9, "bold"))
style.configure("Hint.TLabel", foreground="#66788a", anchor="center")
style.configure("Section.TLabel", foreground="#1f2933", font=("Microsoft YaHei UI", 10, "bold"))
style.configure("Link.TLabel", foreground="#087a97", font=("Microsoft YaHei UI", 9, "underline"))
style.configure("TLabelframe", background="#f3f5f7", borderwidth=1, relief="solid")
style.configure("TLabelframe.Label", background="#f3f5f7", foreground="#1f2933", font=("Microsoft YaHei UI", 10, "bold"))
style.configure("TNotebook", background="#f3f5f7", borderwidth=0)
style.configure("TNotebook.Tab", font=("Microsoft YaHei UI", 10, "bold"), padding=(18, 9))
style.configure("TButton", font=("Microsoft YaHei UI", 10), padding=(10, 8))
style.configure("Primary.TButton", font=("Microsoft YaHei UI", 10, "bold"))
style.configure("TEntry", padding=6)
style.configure("TCombobox", padding=5)
def _configure_logging(self) -> None:
formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s", "%H:%M:%S")
gui_handler = QueueLogHandler(self.messages)
gui_handler.setFormatter(formatter)
log_dir = ROOT / "logs"
log_dir.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(log_dir / "assistant.log", encoding="utf-8")
file_handler.setFormatter(formatter)
logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(gui_handler)
logger.addHandler(file_handler)
self.gui_handler = gui_handler
self.file_handler = file_handler
def _on_tab_changed(self, _event: tk.Event) -> None:
if self.nametowidget(self.notebook.select()) is self.about_page:
self.logs.grid_remove()
self.rowconfigure(1, weight=1)
self.rowconfigure(2, weight=0)
else:
self.logs.grid()
self.rowconfigure(1, weight=3)
self.rowconfigure(2, weight=2)
def _refresh_clock(self) -> None:
now = datetime.now()
self.clock_time.set(now.strftime("%H:%M:%S"))
self.clock_date.set(now.strftime("%Y-%m-%d"))
self.after(1000, self._refresh_clock)
def _refresh_logitech_status(self) -> None:
installed, running = detect_logitech_driver()
self._show_logitech_status(installed, running)
self.after(2000, self._refresh_logitech_status)
def _show_logitech_status(self, installed: bool, running: bool) -> None:
if installed and running:
self.logitech_status.set("罗技鼠标驱动：已安装并运行")
self.logitech_label.configure(style="DriverReady.TLabel")
elif not installed:
self.logitech_status.set("罗技鼠标驱动：未安装")
self.logitech_label.configure(style="DriverError.TLabel")
else:
self.logitech_status.set("罗技鼠标驱动：已安装，未运行")
self.logitech_label.configure(style="DriverError.TLabel")
def claim_runner(self, runner: Any) -> bool:
if not self._activated:
return False
installed, running = detect_logitech_driver()
self._show_logitech_status(installed, running)
if not installed:
messagebox.showwarning("请先安装罗技鼠标驱动", LOGITECH_TUTORIAL_MESSAGE, parent=self)
return False
if self.active_runner is not None and self.active_runner is not runner:
messagebox.showwarning("已有模式运行", "请先停止当前模式，再启动另一个模式。")
return False
self.active_runner = runner
return True
def release_runner(self, runner: Any) -> None:
if self.active_runner is runner:
self.active_runner = None
self.title("cf挑战助手")
def set_running_title(self, runner: Any) -> None:
if runner is self.active_runner:
mode = "探险模式" if runner is self.expedition_page else "刷水晶箱子"
self.title(f"cf挑战助手-{mode}-运行中")
def start_selected(self) -> bool:
if not self._activated:
return False
selected = self.nametowidget(self.notebook.select())
if selected is self.about_page:
selected = self.expedition_page
started = bool(selected.start())
if started:
self.voice.play("started")
return started
def stop_active(self) -> bool:
selected = self.nametowidget(self.notebook.select())
if selected is self.about_page:
selected = self.expedition_page
if self.active_runner is None or self.active_runner is not selected:
logging.warning("当前页签没有正在运行的脚本")
return False
stopped = bool(selected.stop())
if stopped:
self.voice.play("stopped")
return stopped
def _check_version(self) -> None:
try:
info = latest_version_info()
except Exception as exc:
self.version_results.put(exc)
return
self.version_results.put(info)
def _drain_version_results(self) -> None:
try:
result = self.version_results.get_nowait()
except queue.Empty:
pass
else:
if isinstance(result, Exception):
self._show_version_error(result)
else:
self._authorize(result)
if self.winfo_exists():
self.after(200, self._drain_version_results)
def _show_version_error(self, error: Exception) -> None:
if not self.winfo_exists() or self._activated:
return
retry = messagebox.askretrycancel(
"无法验证软件版本",
f"未能完成版本和激活校验：\n{error}\n\n未验证前不能使用软件。",
parent=self,
)
if retry:
threading.Thread(target=self._check_version, daemon=True, name="version-check-retry").start()
else:
self._close()
def _authorize(self, info: VersionInfo) -> None:
if self._activated or not self.winfo_exists():
return
if info.password and not saved_activation_matches(info.password):
self._show_activation_dialog(info)
return
self._complete_authorization(info)
def _show_activation_dialog(self, info: VersionInfo) -> None:
if self._activation_dialog is not None:
return
dialog = tk.Toplevel(self)
self._activation_dialog = dialog
dialog.title("cf挑战助手 - 激活")
dialog.resizable(False, False)
dialog.configure(background="#f3f5f7")
dialog.protocol("WM_DELETE_WINDOW", self._close)
content = ttk.Frame(dialog, padding=20)
content.pack(fill="both", expand=True)
ttk.Label(content, text="请输入激活码", style="Section.TLabel").pack(anchor="w")
tips = info.password_tips.strip() or "输入激活码后即可使用。"
ttk.Label(content, text=tips, wraplength=360, justify="left").pack(anchor="w", fill="x", pady=(8, 12))
code = tk.StringVar()
entry = ttk.Entry(content, textvariable=code, show="*", width=38)
entry.pack(fill="x")
error = ttk.Label(content, text="", foreground="#b42318")
error.pack(anchor="w", pady=(6, 0))
buttons = ttk.Frame(content)
buttons.pack(fill="x", pady=(12, 0))
def submit() -> None:
entered = code.get().strip()
if not activation_matches(entered, info.password):
error.configure(text="激活码不匹配，请重新输入。")
code.set("")
entry.focus_set()
return
try:
save_activation(entered)
except OSError as exc:
logging.warning("无法保存激活码：%s", exc)
messagebox.showwarning("激活码保存失败", "验证成功，但无法保存激活码；下次启动需重新输入。", parent=dialog)
dialog.grab_release()
dialog.destroy()
self._activation_dialog = None
self._complete_authorization(info)
ttk.Button(buttons, text="退出", command=self._close).pack(side="right")
ttk.Button(buttons, text="激活", style="Primary.TButton", command=submit).pack(side="right", padx=(0, 8))
dialog.bind("<Return>", lambda _event: submit())
dialog.bind("<Escape>", lambda _event: self._close())
dialog.update_idletasks()
x = (dialog.winfo_screenwidth() - dialog.winfo_width()) // 2
y = (dialog.winfo_screenheight() - dialog.winfo_height()) // 2
dialog.geometry(f"+{x}+{y}")
dialog.grab_set()
dialog.lift()
entry.focus_set()
def _complete_authorization(self, info: VersionInfo) -> None:
self._activated = True
self.deiconify()
threading.Thread(target=self._usage_heartbeat, daemon=True, name="usage-heartbeat").start()
if os.name == "nt":
threading.Thread(target=self._poll_global_hotkeys, daemon=True, name="global-hotkeys").start()
if version_is_newer(info.version, APP_VERSION):
self._show_update(info.version, info.notes, info.url)
def _usage_heartbeat(self) -> None:
try:
client_id = load_usage_id()
except OSError as exc:
self.usage_results.put(exc)
return
while not self.usage_stop.is_set():
try:
self.usage_results.put(report_usage(client_id))
except (OSError, ValueError) as exc:
self.usage_results.put(exc)
if self.usage_stop.wait(USAGE_INTERVAL_MS / 1000):
return
def _drain_usage_results(self) -> None:
try:
while True:
result = self.usage_results.get_nowait()
if isinstance(result, UsageCounts):
self.online_count.set(f"当前在线人数 {result.online}")
self.today_count.set(f"今日总使用人数 {result.today}")
else:
self.online_count.set("当前在线人数 --")
self.today_count.set("今日总使用人数 --")
logging.debug("使用人数暂不可用：%s", result)
except queue.Empty:
pass
if self.winfo_exists():
self.after(200, self._drain_usage_results)
def _show_update(self, version: str, notes: list[str], url: str) -> None:
if not self.winfo_exists():
return
details = "\n".join(notes[:8])
message = f"发现新版本 v{version}（当前 v{APP_VERSION}）。"
if details:
message += f"\n\n更新内容：\n{details}"
message += "\n\n现在前往下载页面吗？"
if messagebox.askyesno("cf挑战助手更新", message, parent=self):
webbrowser.open(url)
def _poll_global_hotkeys(self) -> None:
user32 = ctypes.windll.user32
callbacks = {
0x78: self.start_selected,
0x79: self.stop_active,
}
previous = {key: False for key in callbacks}
while not self.hotkey_stop.wait(0.035):
for key, callback in callbacks.items():
pressed = bool(user32.GetAsyncKeyState(key) & 0x8000)
if pressed and not previous[key]:
self.after(0, callback)
previous[key] = pressed
def _drain_logs(self) -> None:
while True:
try:
message = self.messages.get_nowait()
except queue.Empty:
break
self.log_text.configure(state="normal")
self.log_text.insert("end", message + "\n")
self.log_text.see("end")
self.log_text.configure(state="disabled")
self.after(120, self._drain_logs)
def _close(self) -> None:
self.hotkey_stop.set()
self.usage_stop.set()
if self.active_runner is not None:
self.active_runner.stop()
self.voice.close()
self.destroy()
def run_app() -> int:
if not is_administrator():
messagebox.showerror("需要管理员权限", "cf挑战助手需要管理员权限运行。请右键程序，选择“以管理员身份运行”。")
return 1
ChallengeAssistant().mainloop()
return 0
if __name__ == "__main__":
raise SystemExit(run_app())
expedition_bot.py
"""CFHelper 风格的封神台自动开火助手（独立第二版）。
本文件不依赖第一版的 BOSS 血条/OCR 逻辑，使用灰度模板匹配模拟 CFHelper
的流程：倒计时 ``1`` -> 延迟 -> HID 鼠标左键持续开火 -> ``击败了`` -> 松开，
随后固定使用右方向键补充弹药并完成确认。模板和配置均位于本文件旁边的独立目录。
"""
from __future__ import annotations
import argparse
import ctypes
import json
import logging
import os
import signal
import sys
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Mapping, Optional
import cf_window
from game_timer_reader import GameTimerReader, GameTimerTracker
ROOT = Path(__file__).resolve().parent
DEFAULT_CONFIG = ROOT / "expedition_config.json"
TEMPLATE_ROOT = ROOT / "expedition_templates"
REFERENCE_RESOLUTION = (1024, 768)
POLL_INTERVAL_SECONDS = 0.025
SUMMON_POLL_INTERVAL_SECONDS = 0.015
MATCH_CONFIDENCE = 0.77
SUMMON_BUTTON_CONFIDENCE = 0.58
COUNTDOWN_DELAY_SECONDS = 0.4
POST_DEFEAT_DELAY_SECONDS = 0.35
SUMMON_SELECT_DELAY_SECONDS = 0.02
SUMMON_SUBMIT_DELAY_SECONDS = 0.02
SUMMON_E_HOLD_SECONDS = 0.02
SUMMON_E_RETRY_SECONDS = 0.35
SUMMON_MENU_DELAY_SECONDS = 0.12
SUMMON_POST_DEFEAT_E_SECONDS = 5.0
SUMMON_POST_DEFEAT_E_INTERVAL_SECONDS = 0.12
REWARD_PRIORITY_SECONDS = 8.0
REWARD_CLICK_RETRY_SECONDS = 0.25
SUMMON_CARD_EXHAUSTED_SECONDS = 3.0
SUMMON_CARD_EXHAUSTED_ATTEMPTS = 6
SUMMON_ATTEMPT_IDLE_RESET_SECONDS = 2.0
GAME_TIME_REMINDERS = ((9 * 60 + 40, "boss_10"), (19 * 60 + 40, "boss_20"), (34 * 60 + 40, "boss_35"))
SUMMON_GAME_LIMIT_SECONDS = 35 * 60
class BotError(RuntimeError):
pass
class WindowGuard:
"""Use the same startup window check as the crystal-box mode."""
def __init__(self, settings: "Settings"):
self.settings = settings
def check(self) -> None:
try:
cf_window.require_window(self.settings.window_titles, ("crossfire.exe",))
except RuntimeError as exc:
raise BotError(str(exc)) from exc
@dataclass(frozen=True)
class Settings:
resolution: tuple[int, int]
window_titles: tuple[str, ...]
capture_region: tuple[int, int, int, int]
arrow_hold: float
live: bool
input_backend: str
hid_dll: Optional[str]
operation_mode: str
card_star: int
card_slot: int
star_point: tuple[int, int]
card_point: tuple[int, int]
summon_button_point: tuple[int, int]
countdown_template: str
defeat_template: str
summon_template: str
confirm_template: str
def load_json(path: Path) -> Mapping[str, Any]:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise BotError(f"找不到配置文件：{path}") from exc
except json.JSONDecodeError as exc:
raise BotError(f"配置文件不是有效 JSON：{path}: {exc}") from exc
if not isinstance(value, dict):
raise BotError("配置文件根节点必须是对象")
return value
def _region(raw: Mapping[str, Any], name: str, default: tuple[int, int, int, int]) -> tuple[int, int, int, int]:
value = raw.get(name, list(default))
if not isinstance(value, list) or len(value) != 4:
raise BotError(f"{name} 必须是 [x,y,width,height]")
return tuple(max(0, int(item)) for item in value) # type: ignore[return-value]
def _point(raw: Mapping[str, Any], name: str, default: tuple[int, int]) -> tuple[int, int]:
value = raw.get(name, list(default))
if not isinstance(value, list) or len(value) != 2:
raise BotError(f"{name} 必须是 [x,y]")
return int(value[0]), int(value[1])
def build_settings(raw: Mapping[str, Any]) -> Settings:
resolution = raw.get("resolution", [1024, 768])
titles = raw.get("window_titles", ["穿越火线", "CrossFire"])
if not isinstance(resolution, list) or len(resolution) != 2:
raise BotError("resolution 必须是 [宽, 高]")
if not isinstance(titles, list) or not all(isinstance(item, str) for item in titles):
raise BotError("window_titles 必须是字符串数组")
backend = str(raw.get("input_backend", "cfhelper")).strip().lower()
if backend not in {"cfhelper", "dry"}:
raise BotError("input_backend 只能是 cfhelper 或 dry")
width, height = 1024, 768
# 1024x768 召唤窗口：6 个星级页签；卡片槽从 x=278 开始，间距约 135。
default_stars = [(255, 286), (363, 286), (471, 286), (579, 286), (687, 286), (795, 286)]
default_slots = [(278, 420), (413, 420), (548, 420), (683, 420), (818, 420)]
star_index = max(0, min(5, int(raw.get("card_star", 1)) - 1))
slot_index = max(0, min(4, int(raw.get("card_slot", 1)) - 1))
scale_x, scale_y = width / REFERENCE_RESOLUTION[0], height / REFERENCE_RESOLUTION[1]
operation_mode = str(raw.get("operation_mode", "smart_fire")).strip().lower()
if operation_mode in {"smart_card", "card", "智能开卡模式"}:
operation_mode = "smart_card"
elif operation_mode not in {"smart_fire", "fire", "自动开火模式"}:
operation_mode = "smart_fire"
card_star = max(1, min(6, int(raw.get("card_star", 1))))
card_slot = max(1, min(5, int(raw.get("card_slot", 1))))
return Settings(
resolution=(1024, 768),
window_titles=tuple(titles),
capture_region=(0, 0, width, height),
# These values are part of the second-version workflow and are no
# longer user-configurable.
arrow_hold=max(0.01, float(raw.get("arrow_hold", 0.05))),
live=True,
input_backend=backend,
hid_dll=str(raw.get("hid_dll", "dll/HIDController.dll")),
operation_mode=operation_mode,
card_star=card_star,
card_slot=card_slot,
star_point=(round(default_stars[star_index][0] * scale_x), round(default_stars[star_index][1] * scale_y)),
card_point=(round(default_slots[slot_index][0] * scale_x), round(default_slots[slot_index][1] * scale_y)),
summon_button_point=_point(raw, "summon_button_point", (788, 598)),
countdown_template=str(raw.get("countdown_template", "daojishi.png")),
defeat_template=str(raw.get("defeat_template", "jb.png")),
summon_template=str(raw.get("summon_template", "zhbs.png")),
confirm_template=str(raw.get("confirm_template", "js.png")),
)
class HIDMouse:
MOUSE_LEFT_DOWN = 0x0002
MOUSE_LEFT_UP = 0x0004
def __init__(self, path: Path):
if not path.is_file():
raise BotError(f"HID 驱动 DLL 不存在：{path}")
self.dll = ctypes.WinDLL(str(path))
self.down = self._bind("MouseDown", [ctypes.c_uint16])
self.up = self._bind("MouseUp", [ctypes.c_uint16])
self.release = self._bind("release_all_keys", [])
def _bind(self, name: str, args: list[Any]) -> Any:
fn = getattr(self.dll, name)
fn.argtypes = args
fn.restype = None if name == "release_all_keys" else ctypes.c_bool
return fn
def left_down(self) -> bool:
return bool(self.down(self.MOUSE_LEFT_DOWN))
def left_up(self) -> bool:
return bool(self.up(self.MOUSE_LEFT_UP))
def release_all(self) -> None:
self.release()
class ArrowInput:
"""CFHelper-compatible keyboard input through pynput/Windows SendInput."""
def __init__(self, settings: Settings, live: bool):
self.live = live
self._left_down = False
self._keyboard = None
self._key = None
self._user32 = None
self._hid: Optional[HIDMouse] = None
if not live:
return
if os.name != "nt":
raise BotError("实际操作模式只能在 Windows 上运行")
try:
from pynput import keyboard # type: ignore
except ImportError as exc:
raise BotError("控制游戏需要 pynput，请先安装 requirements.txt") from exc
self._keyboard = keyboard.Controller()
self._key = keyboard.Key
self._user32 = ctypes.windll.user32
if settings.hid_dll:
try:
self._hid = HIDMouse((ROOT / settings.hid_dll).resolve())
except Exception as exc:
raise BotError(f"无法加载 HID 鼠标驱动：{exc}") from exc
def activate_game(self, titles: tuple[str, ...]) -> bool:
if not self.live:
return True
try:
user32 = ctypes.windll.user32
user32.GetForegroundWindow.restype = ctypes.c_void_p
hwnd = cf_window.find_window(titles, ("crossfire.exe",))
if hwnd:
if user32.GetForegroundWindow() == hwnd:
return True
user32.ShowWindow(hwnd, 9)
user32.SetForegroundWindow(hwnd)
time.sleep(0.08)
return user32.GetForegroundWindow() == hwnd
except Exception as exc:
logging.debug("激活游戏窗口失败：%s", exc)
return False
def _press(self, key: Any) -> None:
if self.live and self._keyboard:
self._keyboard.press(key)
def _release(self, key: Any) -> None:
if self.live and self._keyboard:
self._keyboard.release(key)
def press_e(self, titles: tuple[str, ...], hold: float) -> None:
self.activate_game(titles)
logging.info("CFHelper：发送 E 召唤%s", "（dry-run）" if not self.live else "")
self._press("e")
try:
if self.live:
time.sleep(hold)
finally:
self._release("e")
def fire_down(self, titles: tuple[str, ...]) -> None:
if self._left_down:
return
self.activate_game(titles)
logging.info("HID：按住鼠标左键开火%s", "（dry-run）" if not self.live else "")
if self.live and self._hid:
if not self._hid.left_down():
raise BotError("HID 鼠标左键按下失败")
self._left_down = True
def user_left_button_down(self) -> bool:
"""Return the physical left-button state before the bot owns firing."""
return bool(self.live and self._user32 and self._user32.GetAsyncKeyState(0x01) & 0x8000)
def release_user_left_button(self, titles: tuple[str, ...]) -> bool:
"""Release a manually held left button so it cannot bypass bot state."""
if not self.user_left_button_down():
return False
self.activate_game(titles)
logging.warning("检测到用户手动按住鼠标左键，已自动松开")
if self._hid:
self._hid.left_up()
try:
import pyautogui # type: ignore
pyautogui.mouseUp(button="left")
except Exception:
pass
self._left_down = False
return True
def fire_up(self) -> None:
if self._left_down:
logging.info("HID：松开鼠标左键，停止开火%s", "（dry-run）" if not self.live else "")
if self.live and self._hid:
self._hid.left_up()
self._left_down = False
def reload_action(self, titles: tuple[str, ...], hold: float) -> None:
self.activate_game(titles)
logging.info("击败后短按键盘右方向键补充弹药%s", "（dry-run）" if not self.live else "")
key = self._key.right if self._key else "<right>"
self._press(key)
if self.live:
time.sleep(hold)
self._release(key)
def click(self, point: tuple[int, int], titles: tuple[str, ...], label: str, fast: bool = False) -> None:
if self.live:
if not self.activate_game(titles):
raise BotError("未能激活 CF 游戏窗口，取消点击")
rect = cf_window.window_rect(titles, ("crossfire.exe",))
if rect is None:
raise BotError("未找到 CF 游戏窗口，取消点击")
point = (point[0] + rect[0], point[1] + rect[1])
logging.info("点击%s：%s%s", label, point, "（dry-run）" if not self.live else "")
if self.live:
try:
import pyautogui # type: ignore
if fast:
pyautogui.click(*point, _pause=False)
else:
pyautogui.click(*point)
except Exception as exc:
raise BotError(f"点击{label}失败：{exc}") from exc
def release_all(self) -> None:
self.fire_up()
if self._hid:
self._hid.release_all()
if self.live and self._keyboard and self._key:
for key in (self._key.right, "e"):
try:
self._keyboard.release(key)
except Exception:
pass
class TemplateDetector:
def __init__(self, settings: Settings):
try:
import cv2 # type: ignore
import numpy as np # type: ignore
import pyautogui # type: ignore
except ImportError as exc:
raise BotError("缺少 pyautogui、opencv-python 或 numpy 依赖") from exc
self.cv2, self.np, self.pag = cv2, np, pyautogui
self.settings = settings
self.templates: dict[str, Any] = {}
folder = self._template_folder()
self.timer_reader = GameTimerReader(folder / "timer_digits")
self.timer_tracker = GameTimerTracker()
self.game_elapsed: Optional[str] = None
self.game_elapsed_seconds: Optional[int] = None
self._next_timer_read_at = 0.0
defeat_path = folder / "defeat_text.png"
if not defeat_path.is_file():
raise BotError(f"缺少1024x768击败文字模板：{defeat_path}")
defeat_image = self._read_image(defeat_path)
if defeat_image is None:
raise BotError(f"无法读取击败文字模板：{defeat_path}")
self.defeat_mask = (defeat_image > 175).astype(np.uint8) * 255
self.last_defeat_score = 0.0
self.last_defeat_position: Optional[tuple[int, int]] = None
for key, name in (("countdown", settings.countdown_template), ("defeat", settings.defeat_template), ("summon_prompt", "e.png"), ("summon", settings.summon_template), ("confirm", settings.confirm_template)):
if not name:
continue
path = folder / name
if path.is_file():
image = self._read_image(path)
if image is not None:
self.templates[key] = image
else:
logging.warning("缺少模板：%s", path)
def _read_image(self, path: Path) -> Any:
"""Read an image even when the Windows project path contains Chinese."""
encoded = self.np.frombuffer(path.read_bytes(), dtype=self.np.uint8)
return self.cv2.imdecode(encoded, self.cv2.IMREAD_GRAYSCALE)
def _template_folder(self) -> Path:
folder = TEMPLATE_ROOT / "1024x768"
if not folder.is_dir():
raise BotError(f"缺少 1024x768 模板目录：{folder}")
return folder
def screenshot(self) -> Any:
rect = cf_window.window_rect(self.settings.window_titles, ("crossfire.exe",))
if rect is None:
raise BotError("未找到 CF 游戏窗口，暂不截屏")
image = self.pag.screenshot(region=(rect[0], rect[1], *self.settings.resolution))
frame = self.cv2.cvtColor(self.np.array(image), self.cv2.COLOR_RGB2BGR)
if hasattr(self, "timer_reader") and time.monotonic() >= self._next_timer_read_at:
now = time.monotonic()
self.game_elapsed_seconds = self.timer_tracker.observe(self.timer_reader.read(frame), now)
self.game_elapsed = self.timer_tracker.display
self._next_timer_read_at = now + 0.3
return frame
def _crop(self, frame: Any) -> Any:
x, y, width, height = self.settings.capture_region
return frame[y:y + height, x:x + width]
def locate(self, frame: Any, key: str) -> Optional[tuple[int, int, int, int, float]]:
template = self.templates.get(key)
crop = self._crop(frame)
if template is None or crop.size == 0:
return None
if key == "summon":
offset_x, offset_y = 650, 545
crop = crop[545:650, 650:920]
elif key == "summon_prompt":
offset_x, offset_y = 350, 240
crop = crop[240:430, 350:700]
elif key == "countdown":
offset_x, offset_y = 380, 220
crop = crop[220:440, 380:650]
else:
offset_x, offset_y = 0, 0
if crop.size == 0:
return None
gray = self.cv2.cvtColor(crop, self.cv2.COLOR_BGR2GRAY)
best: Optional[tuple[int, int, int, int, float]] = None
scale_x = self.settings.resolution[0] / REFERENCE_RESOLUTION[0]
scale_y = self.settings.resolution[1] / REFERENCE_RESOLUTION[1]
width = max(8, int(template.shape[1] * scale_x))
height = max(8, int(template.shape[0] * scale_y))
for extra_scale in (0.90, 0.96, 1.0, 1.04, 1.10):
match_width = max(8, int(width * extra_scale))
match_height = max(8, int(height * extra_scale))
if match_width >= gray.shape[1] or match_height >= gray.shape[0]:
continue
resized = self.cv2.resize(template, (match_width, match_height), interpolation=self.cv2.INTER_AREA)
result = self.cv2.matchTemplate(gray, resized, self.cv2.TM_CCOEFF_NORMED)
_, score, _, point = self.cv2.minMaxLoc(result)
candidate = (point[0] + self.settings.capture_region[0] + offset_x, point[1] + self.settings.capture_region[1] + offset_y, match_width, match_height, float(score))
if best is None or candidate[4] > best[4]:
best = candidate
confidence = SUMMON_BUTTON_CONFIDENCE if key == "summon" else MATCH_CONFIDENCE
return best if best and best[4] >= confidence else None
def present(self, frame: Any, key: str) -> bool:
return self.locate(frame, key) is not None
def defeat_present(self, frame: Any) -> bool:
"""Match the complete '击败了' glyphs at their native video size."""
x, y, width, height = (280, 140, 450, 60)
crop = frame[y:y + height, x:x + width]
self.last_defeat_score = 0.0
self.last_defeat_position = None
if crop.size == 0:
return False
hsv = self.cv2.cvtColor(crop, self.cv2.COLOR_BGR2HSV)
mask = ((hsv[:, :, 1] < 100) & (hsv[:, :, 2] > 175)).astype(self.np.uint8) * 255
th, tw = self.defeat_mask.shape
if mask.shape[0] < th or mask.shape[1] < tw:
return False
result = self.cv2.matchTemplate(mask, self.defeat_mask, self.cv2.TM_CCOEFF_NORMED)
_, score, _, point = self.cv2.minMaxLoc(result)
self.last_defeat_score = float(score)
self.last_defeat_position = (x + point[0], y + point[1])
if score < 0.86:
return False
matched = mask[point[1]:point[1] + th, point[0]:point[0] + tw] > 0
expected = self.defeat_mask > 0
# Each character must match, so partial words cannot confirm a kill.
for left, right in ((0, 18), (18, 34), (34, tw)):
observed = matched[:, left:right]
reference = expected[:, left:right]
overlap = int((observed & reference).sum())
if overlap / max(1, int(reference.sum())) < 0.80:
return False
if overlap / max(1, int(observed.sum())) < 0.80:
return False
return True
def reward_present(self, frame: Any) -> bool:
"""辅助检测奖励面板；CFHelper真正使用的是确认按钮模板。"""
crop = frame[180:640, 150:880]
if crop.size == 0:
return False
gray = self.cv2.cvtColor(crop, self.cv2.COLOR_BGR2GRAY)
hsv = self.cv2.cvtColor(crop, self.cv2.COLOR_BGR2HSV)
dark_ratio = float((gray < 90).mean())
# 奖励窗是居中的大面积深色面板，且下方有蓝色确认按钮。
blue = ((hsv[:, :, 0] >= 90) & (hsv[:, :, 0] <= 135) & (hsv[:, :, 1] >= 45) & (hsv[:, :, 2] >= 70)).astype(self.np.uint8) * 255
components, _, stats, _ = self.cv2.connectedComponentsWithStats(blue)
button = False
for index in range(1, len(stats)):
x, y, w, h, area = [int(v) for v in stats[index]]
if 120 <= w <= 230 and 18 <= h <= 55 and area >= 500 and y > 350:
button = True
break
return dark_ratio > 0.16 and button
def confirm_button(self, frame: Any) -> Optional[tuple[int, int, int, int, float]]:
"""Match the actual confirmation-button template, then verify blue fill."""
template = self.templates.get("confirm")
if template is None:
return None
x0, y0 = 360, 510
crop = frame[y0:650, x0:700]
if crop.size == 0:
return None
hsv = self.cv2.cvtColor(crop, self.cv2.COLOR_BGR2HSV)
blue = (
(hsv[:, :, 0] >= 90) & (hsv[:, :, 0] <= 135)
& (hsv[:, :, 1] >= 35) & (hsv[:, :, 2] >= 60)
)
if int(blue.sum()) < 100:
return None
gray = self.cv2.cvtColor(crop, self.cv2.COLOR_BGR2GRAY)
best: Optional[tuple[int, int, int, int, float]] = None
for scale in (1.0, 0.95, 1.05, 0.90, 1.10):
width = max(20, int(template.shape[1] * scale))
height = max(10, int(template.shape[0] * scale))
if width >= gray.shape[1] or height >= gray.shape[0]:
continue
resized = self.cv2.resize(template, (width, height), interpolation=self.cv2.INTER_AREA)
result = self.cv2.matchTemplate(gray, resized, self.cv2.TM_CCOEFF_NORMED)
_, score, _, point = self.cv2.minMaxLoc(result)
candidate = (point[0] + x0, point[1] + y0, width, height, float(score))
if best is None or candidate[4] > best[4]:
best = candidate
if scale == 1.0 and score >= 0.98:
break
if best is None or best[4] < 0.90:
return None
x, y, width, height, score = best
hsv = self.cv2.cvtColor(frame[y:y + height, x:x + width], self.cv2.COLOR_BGR2HSV)
mask = ((hsv[:, :, 0] >= 90) & (hsv[:, :, 0] <= 135) & (hsv[:, :, 1] >= 35) & (hsv[:, :, 2] >= 60)).astype(self.np.uint8) * 255
if float((mask > 0).mean()) < 0.20:
return None
return best
class DefeatConfirmation:
"""Require stable glyph matches across distinct frames before reloading."""
def __init__(self) -> None:
self.hits = 0
self.first_at = 0.0
self.last_at = 0.0
self.position: Optional[tuple[int, int]] = None
def update(self, present: bool, position: Optional[tuple[int, int]], now: float) -> bool:
if not present or position is None:
self.hits = 0
self.position = None
return False
shifted = self.position is not None and max(abs(position[i] - self.position[i]) for i in (0, 1)) > 6
if not self.hits or shifted or now - self.last_at > 0.40:
self.hits = 0
self.first_at = now
self.hits += 1
self.last_at = now
self.position = position
return self.hits >= 3 and now - self.first_at >= 0.10
class RewardConfirmation:
"""Require three stable frames of the reward panel and its button."""
def __init__(self) -> None:
self.hits = 0
self.position: Optional[tuple[int, int]] = None
def update(self, match: Optional[tuple[int, int, int, int, float]], panel: bool) -> Optional[tuple[int, int]]:
if match is None or not panel:
self.hits = 0
self.position = None
return None
point = (match[0] + match[2] // 2, match[1] + match[3] // 2)
stable = self.position is None or max(
abs(point[0] - self.position[0]), abs(point[1] - self.position[1])
) <= 4
self.hits = self.hits + 1 if stable else 1
self.position = point
return point if self.hits >= 3 else None
class CFHelperCloneBot:
def __init__(self, settings: Settings, live: Optional[bool] = None, on_reminder: Optional[Callable[[str], None]] = None):
self.settings = settings
self.live = settings.live if live is None else live
self.stop_event = threading.Event()
self.detector = TemplateDetector(settings)
self.input = ArrowInput(settings, self.live and settings.input_backend == "cfhelper")
self.stage = "未启动"
self.stop_reason: Optional[str] = None
self.started_at = 0.0
self._rapid_e_until = 0.0
self._reward_priority_until = 0.0
self.on_reminder = on_reminder
self._timer_session_id = 0
self._timer_announced: set[str] = set()
def _capture_frame(self) -> Any:
frame = self.detector.screenshot()
elapsed = getattr(self.detector, "game_elapsed_seconds", None)
if not isinstance(elapsed, int):
return frame
session_id = getattr(self.detector.timer_tracker, "session_id", 0)
if session_id != getattr(self, "_timer_session_id", 0):
self._timer_session_id = session_id
self._timer_announced = set()
for threshold, event in GAME_TIME_REMINDERS:
if threshold <= elapsed <= threshold + 2 and event not in self._timer_announced:
self._timer_announced.add(event)
logging.info("游戏开局时间 %s，播报时间提醒", self.detector.game_elapsed)
if self.on_reminder is not None:
self.on_reminder(event)
if self.settings.operation_mode == "smart_card" and elapsed >= SUMMON_GAME_LIMIT_SECONDS:
self.stop_reason = "游戏开局已满35分钟"
self.stage = "游戏开局已满35分钟，脚本已停止"
logging.info("自动召唤模式达到35分钟，停止脚本")
self.stop_event.set()
self.input.release_all()
return frame
def stop(self) -> None:
self.stop_event.set()
self.input.release_all()
self.stage = "已停止"
def timed_out(self) -> bool:
return False
def _check_reward(self, frame: Any, tracker: RewardConfirmation, label: str, firing: bool = False) -> bool:
match = self.detector.confirm_button(frame)
panel = bool(match) and self.detector.reward_present(frame)
point = tracker.update(match, panel)
if point is None:
return False
logging.info("奖励面板与确认按钮已连续确认 3 帧，点击关闭：%s（模板分数 %.3f）", point, match[4])
if firing:
self.input.fire_up()
self.input.click(point, self.settings.window_titles, label, fast=True)
self.stage = "奖励弹窗已确认"
return True
def _submit_summon_selection(self) -> bool:
"""Select the configured card and submit it without the default click pause."""
self.input.click(self.settings.star_point, self.settings.window_titles, f"{self.settings.card_star}星卡页签", fast=True)
if self.stop_event.wait(SUMMON_SELECT_DELAY_SECONDS if self.live else 0.01):
return False
self.input.click(
self.settings.card_point,
self.settings.window_titles,
f"{self.settings.card_star}星卡槽{self.settings.card_slot}",
fast=True,
)
if self.stop_event.wait(SUMMON_SELECT_DELAY_SECONDS if self.live else 0.01):
return False
if self.stop_event.wait(SUMMON_SUBMIT_DELAY_SECONDS if self.live else 0.01):
return False
if self.stop_event.is_set():
return False
self.input.click(self.settings.summon_button_point, self.settings.window_titles, "召唤Boss", fast=True)
logging.info("已点击召唤Boss按钮：%s，返回循环继续检测", self.settings.summon_button_point)
return True
def _active_summon_once(self, cycle: int) -> bool:
"""Open the summon panel on its prompt and retry its button until play."""
self.stage = f"第 {cycle} 次：循环检测 E 召唤提示与召唤按钮"
last_retry_log = time.monotonic()
last_e_at = float("-inf")
reward_confirmation = RewardConfirmation()
reward_dismissed = False
reward_last_click_at = 0.0
summon_attempt_started_at: Optional[float] = None
summon_last_attempt_at = 0.0
summon_attempts = 0
while not self.stop_event.is_set():
frame = self._capture_frame()
if self.stop_event.is_set():
return False
now = time.monotonic()
rapid_e = now < getattr(self, "_rapid_e_until", 0.0)
if not rapid_e:
self._rapid_e_until = 0.0
reward_wait = False
if reward_dismissed:
match = self.detector.confirm_button(frame)
if match and self.detector.reward_present(frame):
reward_wait = True
if now - reward_last_click_at >= REWARD_CLICK_RETRY_SECONDS:
point = (match[0] + match[2] // 2, match[1] + match[3] // 2)
logging.info("奖励面板仍在，重试点击确认：%s", point)
self.input.click(point, self.settings.window_titles, "奖励确认", fast=True)
reward_last_click_at = time.monotonic()
else:
reward_dismissed = False
else:
if self._check_reward(frame, reward_confirmation, "奖励确认"):
reward_dismissed = True
reward_wait = True
reward_last_click_at = time.monotonic()
elif reward_confirmation.hits:
reward_wait = True
if reward_wait:
self.stop_event.wait(SUMMON_POLL_INTERVAL_SECONDS)
continue
summon_popup = not reward_wait and self.detector.locate(frame, "summon")
if summon_popup:
if summon_attempt_started_at is None or now - summon_last_attempt_at > SUMMON_ATTEMPT_IDLE_RESET_SECONDS:
summon_attempt_started_at = now
summon_attempts = 0
logging.info("检测到召唤选卡弹窗，开始本轮选卡尝试")
if (summon_attempts >= SUMMON_CARD_EXHAUSTED_ATTEMPTS
and now - summon_attempt_started_at >= SUMMON_CARD_EXHAUSTED_SECONDS):
if self.detector.locate(frame, "countdown"):
self._rapid_e_until = 0.0
self._summon_countdown_seen = True
logging.info("选卡重试期间检测到倒计时，召唤已成功，进入开火流程")
return True
self.stop_reason = "Boss 卡片不足"
self.stage = "Boss 卡片不足，脚本已停止"
logging.warning("选卡 %.1f 秒仍未进入倒计时，已尝试 %s 次，判定 Boss 卡片不足并停止脚本",
now - summon_attempt_started_at, summon_attempts)
self.stop_event.set()
return False
self._rapid_e_until = 0.0
self.stage = f"第 {cycle} 次：检测到召唤Boss按钮，选择卡片"
if not self._submit_summon_selection():
return False
summon_attempts += 1
summon_last_attempt_at = time.monotonic()
else:
prompt = None if rapid_e or reward_wait else self.detector.locate(frame, "summon_prompt")
interval = SUMMON_POST_DEFEAT_E_INTERVAL_SECONDS if rapid_e else SUMMON_E_RETRY_SECONDS
if (rapid_e or prompt) and now - last_e_at >= interval:
if not rapid_e:
logging.info("检测到 E 召唤Boss 提示，按 E 打开选卡面板")
self.input.press_e(self.settings.window_titles, SUMMON_E_HOLD_SECONDS)
last_e_at = time.monotonic()
delay = SUMMON_POLL_INTERVAL_SECONDS if rapid_e else (SUMMON_MENU_DELAY_SECONDS if self.live else 0.01)
if self.stop_event.wait(delay):
return False
continue
if not reward_wait and self._check_countdown(frame):
self._rapid_e_until = 0.0
self._summon_countdown_seen = True
logging.info("已检测到召唤后的倒计时，进入开火流程")
return True
if time.monotonic() - last_retry_log >= 30.0:
logging.info("仍在循环检测 E 召唤提示与召唤Boss按钮")
last_retry_log = time.monotonic()
# The summon area is small and checked before the full-screen
# countdown match, so use a tighter retry interval here.
self.stop_event.wait(SUMMON_POLL_INTERVAL_SECONDS)
return False
def _check_countdown(self, frame: Any) -> bool:
"""Search the small center region on every frame so '1' is not missed."""
return bool(self.detector.locate(frame, "countdown"))
def _active_summon(self, cycle: int) -> bool:
return self._active_summon_once(cycle)
def _run_cycle(self, cycle: int) -> bool:
self._summon_countdown_seen = False
if self.settings.operation_mode == "smart_card":
if not self._active_summon(cycle):
return False
self.stage = f"第 {cycle} 次：等待倒计时"
countdown_seen = False
firing = False
defeat_seen = False
defeat_confirmation = DefeatConfirmation()
reward_confirmation = RewardConfirmation()
try:
while not self.stop_event.is_set() and not self.timed_out():
if not firing:
self.input.release_user_left_button(self.settings.window_titles)
frame = self._capture_frame()
if self.stop_event.is_set():
return False
if self._check_reward(frame, reward_confirmation, "奖励确认", firing=firing):
return True
if reward_confirmation.hits:
continue
countdown = (self._summon_countdown_seen or self._check_countdown(frame)) if not firing else False
defeat_match = self.detector.defeat_present(frame) if firing else False
defeat = defeat_confirmation.update(
defeat_match, self.detector.last_defeat_position if firing else None, time.monotonic()
)
if not countdown_seen and countdown:
countdown_seen = True
self._summon_countdown_seen = False
logging.info("检测到倒计时 1，%.2f 秒后开始开火", COUNTDOWN_DELAY_SECONDS)
self.stage = "检测到倒计时 1，等待开火"
deadline = time.monotonic() + COUNTDOWN_DELAY_SECONDS
while time.monotonic() < deadline and not self.stop_event.is_set() and not self.timed_out():
self.input.release_user_left_button(self.settings.window_titles)
if time.monotonic() < getattr(self, "_reward_priority_until", 0.0):
wait_frame = self._capture_frame()
if self.stop_event.is_set():
return False
if self._check_reward(wait_frame, reward_confirmation, "奖励确认"):
return True
time.sleep(min(POLL_INTERVAL_SECONDS, max(0.001, deadline - time.monotonic())))
for _ in range(2 if reward_confirmation.hits else 0):
wait_frame = self._capture_frame()
if self.stop_event.is_set():
return False
if self._check_reward(wait_frame, reward_confirmation, "奖励确认"):
return True
if not reward_confirmation.hits:
break
if self.stop_event.is_set() or self.timed_out():
return False
self.input.release_user_left_button(self.settings.window_titles)
self.input.fire_down(self.settings.window_titles)
firing = True
self.stage = "已按住HID鼠标左键，等待击败文字确认"
if firing and defeat:
defeat_seen = True
defeat_at = time.monotonic()
self._reward_priority_until = defeat_at + REWARD_PRIORITY_SECONDS
logging.info("击败了文字已连续确认：%s帧，匹配分数=%.3f，位置=%s，停止开火",
defeat_confirmation.hits, self.detector.last_defeat_score,
self.detector.last_defeat_position)
self.stage = "检测到击败，停止开火"
self.input.fire_up()
firing = False
time.sleep(POST_DEFEAT_DELAY_SECONDS if self.live else 0.01)
self.input.reload_action(self.settings.window_titles, self.settings.arrow_hold)
if self.settings.operation_mode == "smart_card":
self._rapid_e_until = defeat_at + SUMMON_POST_DEFEAT_E_SECONDS
self.stage = "击败后已换弹，短按 E 查找召唤弹窗"
return True
return defeat_seen
time.sleep(POLL_INTERVAL_SECONDS)
return False
finally:
self.input.release_all()
def run(self) -> int:
self.started_at = time.monotonic()
completed = 0
try:
cycle = 1
while not self.stop_event.is_set():
if not self._run_cycle(cycle):
break
completed += 1
self.stage = f"已完成 {completed} 次"
cycle += 1
if self.stop_event.is_set():
return 0
return 1
finally:
self.input.release_all()
def setup_logging(path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", datefmt="%H:%M:%S", handlers=[logging.StreamHandler(sys.stdout), logging.FileHandler(path, encoding="utf-8")], force=True)
def main() -> int:
parser = argparse.ArgumentParser(description="CF 挑战助手 - 探险模式")
parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
parser.add_argument("--dry-run", action="store_true", help="仅记录动作，不控制游戏")
parser.add_argument("--check", action="store_true", help="检查模板和配置")
args = parser.parse_args()
setup_logging(ROOT / "logs" / "expedition_bot.log")
settings = build_settings(load_json(args.config))
logging.info("探险第二版配置有效：%sx%s，固定模板阈值 %.2f", *settings.resolution, MATCH_CONFIDENCE)
if args.check:
detector = TemplateDetector(settings)
for key in ("countdown", "defeat", "summon", "confirm"):
logging.info("模板 %s：%s", key, "已加载" if key in detector.templates else "缺失")
return 0
if not args.dry_run:
WindowGuard(settings).check()
bot = CFHelperCloneBot(settings, live=not args.dry_run)
signal.signal(signal.SIGINT, lambda *_: bot.stop())
return bot.run()
if __name__ == "__main__":
raise SystemExit(main())
challenge_bot.py
"""CF challenge workflow helper.
This project is intentionally template-driven: it does not inject into the game
process or attempt to bypass anti-cheat. Capture the UI buttons from your own
client and put them in ``templates/`` using the names in ``config.json``.
The default mode is a dry run. Pass ``--live`` only in an environment where
automation is permitted by the game/server rules.
"""
from __future__ import annotations
import argparse
import json
import logging
import os
import random
import signal
import sys
import time
from dataclasses import dataclass, field
from enum import Enum, auto
from pathlib import Path
from typing import Any, Iterable, Mapping, Optional
import cf_window
from score_reader import ScoreReader
ROOT = Path(__file__).resolve().parent
DEFAULT_CONFIG = ROOT / "challenge_config.json"
POLL_INTERVAL_SECONDS = 1.0
INITIAL_RIGHT_CLICK_MIN_SECONDS = 8.0
INITIAL_RIGHT_CLICK_MAX_SECONDS = 12.0
PERIODIC_CLICK_MIN_SECONDS = 10.0
PERIODIC_CLICK_MAX_SECONDS = 20.0
SCORE_CONFIRMATIONS = 3
QUICK_JOIN_SCROLL_SETTLE_SECONDS = 0.03
class BotError(RuntimeError):
"""An expected, actionable script error."""
class State(Enum):
HALL = auto()
MODE_SELECT = auto()
MAP_SELECT = auto()
ROOM = auto()
READY = auto()
IN_GAME = auto()
RESULT = auto()
UNKNOWN = auto()
@dataclass(frozen=True)
class TemplateSpec:
name: str
file: str
confidence: float = 0.86
region: Optional[tuple[int, int, int, int]] = None
grayscale: bool = False
@dataclass(frozen=True)
class KeyAction:
key: str
hold: float = 0.2
pause: float = 0.2
@dataclass
class Settings:
map_name: str
resolution: tuple[int, int]
window_titles: tuple[str, ...]
process_names: tuple[str, ...]
poll_interval: float
template_timeout: float
click_jitter: int
keep_alive_interval: float
score_region: Optional[tuple[int, int, int, int]] = None
bonus_region: Optional[tuple[int, int, int, int]] = None
score_target: int = 700000
score_poll_interval: float = 2.0
hid_dll: Optional[str] = "dll/HIDController.dll"
game_start_timeout: float = 180.0
inventory_scroll_attempts: int = 6
actions: list[KeyAction] = field(default_factory=list)
def load_json(path: Path) -> Mapping[str, Any]:
try:
with path.open("r", encoding="utf-8") as handle:
value = json.load(handle)
except FileNotFoundError as exc:
raise BotError(f"找不到配置文件：{path}") from exc
except json.JSONDecodeError as exc:
raise BotError(f"配置文件不是有效 JSON：{path}: {exc}") from exc
if not isinstance(value, dict):
raise BotError("配置文件根节点必须是对象")
return value
def build_settings(raw: Mapping[str, Any]) -> Settings:
def tuple_strings(key: str, default: Iterable[str]) -> tuple[str, ...]:
value = raw.get(key, list(default))
if not isinstance(value, list) or not all(isinstance(x, str) for x in value):
raise BotError(f"配置项 {key} 必须是字符串数组")
return tuple(value)
resolution = raw.get("resolution", [1024, 768])
if not isinstance(resolution, list) or len(resolution) != 2:
raise BotError("resolution 必须是 [宽, 高]")
score_region_value = raw.get("score_region", [345, 30, 85, 40])
if not isinstance(score_region_value, list) or len(score_region_value) != 4:
raise BotError("score_region 必须是 [x, y, width, height]")
score_region = tuple(int(x) for x in score_region_value)
bonus_region_value = raw.get("bonus_region", [0, 30, 95, 40])
if not isinstance(bonus_region_value, list) or len(bonus_region_value) != 4:
raise BotError("bonus_region 必须是 [x, y, width, height]")
bonus_region = tuple(int(x) for x in bonus_region_value)
action_values = raw.get("in_game_actions", [])
if not isinstance(action_values, list):
raise BotError("in_game_actions must be a list")
actions: list[KeyAction] = []
for action in action_values:
if not isinstance(action, dict) or not isinstance(action.get("key"), str):
raise BotError("每个 in_game_actions 项都需要 key 字段")
actions.append(
KeyAction(
key=action["key"],
hold=float(action.get("hold", 0.2)),
pause=float(action.get("pause", 0.2)),
)
)
settings = Settings(
map_name=str(raw.get("map_name", "新巨人成废墟")),
resolution=(int(resolution[0]), int(resolution[1])),
window_titles=tuple_strings("window_titles", ("穿越火线", "CrossFire")),
process_names=tuple_strings("process_names", ("crossfire.exe",)),
poll_interval=POLL_INTERVAL_SECONDS,
template_timeout=float(raw.get("template_timeout", 12.0)),
click_jitter=max(0, int(raw.get("click_jitter", 3))),
keep_alive_interval=max(0.0, float(raw.get("keep_alive_interval", 25))),
score_region=score_region,
bonus_region=bonus_region,
score_target=max(1, int(raw.get("score_target", 700000))),
score_poll_interval=max(0.5, float(raw.get("score_poll_interval", 2.0))),
hid_dll=str(raw["hid_dll"]) if raw.get("hid_dll") else None,
game_start_timeout=max(10.0, float(raw.get("game_start_timeout", 180.0))),
inventory_scroll_attempts=max(0, int(raw.get("inventory_scroll_attempts", 6))),
actions=actions,
)
return settings
class HIDController:
"""ctypes adapter for the project-local HIDController driver DLL."""
# HIDController.hpp expects WinUser.h mouse event flags, not button
# indexes. Passing 2 here would be an unrelated middle-button flag.
MOUSE_LEFT_DOWN = 0x0002
MOUSE_LEFT_UP = 0x0004
MOUSE_RIGHT_DOWN = 0x0008
MOUSE_RIGHT_UP = 0x0010
def __init__(self, path: Path):
import ctypes
if not path.is_file():
raise BotError(f"HID 驱动 DLL 不存在：{path}")
try:
self.dll = ctypes.WinDLL(str(path))
except OSError as exc:
raise BotError(f"无法加载 HID 驱动 DLL：{path}: {exc}") from exc
self._mouse_down = self._bind("MouseDown", [ctypes.c_uint16])
self._mouse_up = self._bind("MouseUp", [ctypes.c_uint16])
self._key_down = self._bind("KeyDown", [ctypes.c_uint16])
self._key_up = self._bind("KeyUp", [ctypes.c_uint16])
self._release_all = self._bind("release_all_keys", [])
def _bind(self, name: str, argtypes: list[Any], void: bool = False) -> Any:
import ctypes
try:
function = getattr(self.dll, name)
except AttributeError as exc:
raise BotError(f"HID 驱动缺少导出函数：{name}") from exc
function.argtypes = argtypes
function.restype = None if void or name == "release_all_keys" else ctypes.c_bool
return function
def _call(self, name: str, function: Any, *args: Any) -> bool:
try:
value = function(*args)
result = True if value is None else bool(value)
except Exception as exc:
logging.warning("HID %s 调用失败：%s", name, exc)
return False
if not result:
logging.warning("HID %s 返回失败", name)
return result
def mouse_down(self, button: int) -> bool:
return self._call("MouseDown", self._mouse_down, button)
def mouse_up(self, button: int) -> bool:
return self._call("MouseUp", self._mouse_up, button)
def key_down(self, vk: int) -> bool:
return self._call("KeyDown", self._key_down, vk)
def key_up(self, vk: int) -> bool:
return self._call("KeyUp", self._key_up, vk)
def release_all_keys(self) -> None:
self._call("release_all_keys", self._release_all)
class UI:
"""Small adapter around PyAutoGUI and optional Windows APIs."""
def __init__(self, templates_dir: Path, live: bool, jitter: int, window_titles: Iterable[str] = (), process_names: Iterable[str] = (), hid_dll: Optional[Path] = None):
try:
import pyautogui # type: ignore
except ImportError as exc:
raise BotError("缺少 PyAutoGUI，请先安装 requirements.txt 中的依赖") from exc
self.pag = pyautogui
self.pag.FAILSAFE = True
self.templates_dir = templates_dir
self.live = live
self.jitter = jitter
self.window_titles = tuple(window_titles)
self.process_names = tuple(name.lower() for name in process_names)
self.hid: Optional[HIDController] = None
self._template_images: dict[Path, Any] = {}
self.score_reader = ScoreReader(templates_dir / "score_digits_video")
self.game_origin = (0, 0)
if self.live:
if hid_dll is None:
raise BotError("live 模式需要配置 HID 驱动 DLL（hid_dll）")
self.hid = HIDController(hid_dll)
logging.info("已加载 HID 驱动：%s", hid_dll)
def activate_game_window(self) -> bool:
"""Bring CF to the foreground before sending input.
When the GUI panel is used, it remains the foreground window after the
Start button is pressed. The first click would otherwise only activate
CF and never reach the game's control.
"""
if not self.live or (not self.window_titles and not self.process_names):
return True
try:
import ctypes
user32 = ctypes.windll.user32
user32.GetForegroundWindow.restype = ctypes.c_void_p
hwnd = cf_window.find_window(self.window_titles, self.process_names)
if hwnd:
# OpenIcon restores clients that were parked at (-32000,-32000)
# by the game launcher. SetWindowPos with no move/resize keeps
# the user's normal client size while making it visible.
user32.OpenIcon(hwnd)
user32.ShowWindow(hwnd, 9)
user32.SetWindowPos(hwnd, 0, 0, 0, 0, 0, 0x0001 | 0x0002 | 0x0040)
user32.BringWindowToTop(hwnd)
user32.SetForegroundWindow(hwnd)
time.sleep(0.25)
if user32.GetForegroundWindow() == hwnd:
self._refresh_game_origin()
logging.debug("已激活游戏窗口句柄：%s", hwnd)
return True
# If Windows foreground-lock rules reject the direct request,
# briefly attach the caller and game input queues and retry.
foreground = user32.GetForegroundWindow()
foreground_thread = user32.GetWindowThreadProcessId(foreground, None) if foreground else 0
target_thread = user32.GetWindowThreadProcessId(hwnd, None)
attached = bool(foreground_thread and target_thread and foreground_thread != target_thread)
if attached:
user32.AttachThreadInput(foreground_thread, target_thread, True)
try:
user32.BringWindowToTop(hwnd)
user32.SetForegroundWindow(hwnd)
finally:
if attached:
user32.AttachThreadInput(foreground_thread, target_thread, False)
time.sleep(0.15)
if user32.GetForegroundWindow() == hwnd:
self._refresh_game_origin()
logging.debug("已激活游戏窗口句柄：%s", hwnd)
return True
logging.warning("未找到可激活的 CF 窗口：标题=%s，进程=%s", ", ".join(self.window_titles), ", ".join(self.process_names))
except Exception as exc:
logging.debug("激活 CF 窗口失败：%s", exc)
return False
def _refresh_game_origin(self) -> tuple[int, int]:
if not self.live:
return self.game_origin
rect = cf_window.window_rect(self.window_titles, self.process_names)
if rect is None:
raise BotError("未找到 CF 游戏窗口，暂不执行屏幕操作")
self.game_origin = (rect[0], rect[1])
return self.game_origin
def _virtual_key(self, key: str) -> int:
import ctypes
named = {
"esc": 0x1B,
"escape": 0x1B,
"enter": 0x0D,
"return": 0x0D,
"space": 0x20,
"tab": 0x09,
"shift": 0x10,
"ctrl": 0x11,
"alt": 0x12,
}
normalized = key.strip().lower()
if normalized in named:
return named[normalized]
if len(normalized) == 1:
# The upstream examples call KeyDown/KeyUp with uppercase ASCII
# letters (for example KeyDown('S')). Use the same value instead
# of relying on the keyboard-layout-dependent VkKeyScanW path.
if normalized.isalpha() and normalized.isascii():
return ord(normalized.upper())
value = ctypes.windll.user32.VkKeyScanW(ord(normalized))
if value != -1:
return value & 0xFF
raise BotError(f"不支持的按键：{key}")
def release_all_keys(self) -> None:
if self.hid is not None:
self.hid.release_all_keys()
def _template_path(self, spec: TemplateSpec) -> Path:
path = self.templates_dir / spec.file
if not path.is_file():
raise BotError(f"模板图片不存在：{path}")
return path
def locate(self, spec: TemplateSpec) -> Optional[tuple[int, int]]:
box = self.locate_box(spec)
if box is None:
return None
point = self.pag.center(box)
return int(point.x), int(point.y)
def locate_box(self, spec: TemplateSpec) -> Any:
path = self._template_path(spec)
try:
# Passing the Chinese project path directly to OpenCV-backed
# PyAutoGUI can fail on Windows. Load it through PIL first and
# pass the decoded image object to pyscreeze instead.
template = self._template_images.get(path)
if template is None:
from PIL import Image # type: ignore
with Image.open(path) as source:
template = source.copy()
self._template_images[path] = template
self._refresh_game_origin()
region = spec.region
if region is not None:
region = (region[0] + self.game_origin[0], region[1] + self.game_origin[1], region[2], region[3])
else:
region = (*self.game_origin, 1024, 768)
box = self.pag.locateOnScreen(
template,
confidence=spec.confidence,
region=region,
grayscale=spec.grayscale,
)
except Exception as exc: # PyAutoGUI raises different errors by version.
logging.debug("模板定位失败 %s：%s", spec.name, exc)
return None
if box is None:
return None
return box
def click(self, point: tuple[int, int], label: str) -> None:
matched_origin = self.game_origin
if self.live and not self.activate_game_window():
raise BotError("未能激活 CF 游戏窗口，取消点击")
x = point[0] + self.game_origin[0] - matched_origin[0]
y = point[1] + self.game_origin[1] - matched_origin[1]
if self.jitter:
x += random.randint(-self.jitter, self.jitter)
y += random.randint(-self.jitter, self.jitter)
logging.info("点击 %s：(%s, %s)%s", label, x, y, " [dry-run]" if not self.live else "")
if self.live:
self._human_move_to(x, y)
self.pag.mouseDown()
time.sleep(random.uniform(0.06, 0.14))
self.pag.mouseUp()
def double_click(self, point: tuple[int, int], label: str) -> None:
matched_origin = self.game_origin
if self.live and not self.activate_game_window():
raise BotError("未能激活 CF 游戏窗口，取消双击")
x = point[0] + self.game_origin[0] - matched_origin[0]
y = point[1] + self.game_origin[1] - matched_origin[1]
logging.info("双击 %s：(%s, %s)%s", label, x, y, " [dry-run]" if not self.live else "")
if self.live:
self._human_move_to(x, y)
self.pag.click(clicks=2, interval=random.uniform(0.10, 0.19))
def _human_move_to(self, x: int, y: int) -> None:
"""Move with a curved, variable-speed path and a small correction."""
try:
sx, sy = self.pag.position()
except Exception:
sx, sy = x, y
overshoot_x = x + random.randint(-10, 10)
overshoot_y = y + random.randint(-8, 8)
duration = random.uniform(0.18, 0.42)
self.pag.moveTo(overshoot_x, overshoot_y, duration=duration)
if random.random() < 0.7:
self.pag.moveTo(x + random.randint(-2, 2), y + random.randint(-2, 2), duration=random.uniform(0.06, 0.16))
self.pag.moveTo(x, y, duration=random.uniform(0.04, 0.12))
def right_click(self, label: str = "释放飞行器") -> None:
self.activate_game_window()
logging.info("右键 %s%s", label, " [dry-run]" if not self.live else "")
if self.live:
if self.hid is None:
raise BotError("局内右键需要 HID 驱动")
# Do not move the cursor to the screen center here. The HID driver
# sends the button event directly to the device path used by CF.
down_ok = self.hid.mouse_down(HIDController.MOUSE_RIGHT_DOWN)
time.sleep(random.uniform(0.12, 0.24))
up_ok = self.hid.mouse_up(HIDController.MOUSE_RIGHT_UP)
logging.info("已发送 HID 鼠标右键按下/释放（down=%s, up=%s）", down_ok, up_ok)
def left_click(self, label: str = "鼠标左键点按") -> None:
self.activate_game_window()
logging.info("左键 %s%s", label, " [dry-run]" if not self.live else "")
if not self.live:
return
if self.hid is None:
raise BotError("局内左键需要 HID 驱动")
down_ok = self.hid.mouse_down(HIDController.MOUSE_LEFT_DOWN)
try:
time.sleep(random.uniform(0.08, 0.16))
finally:
up_ok = self.hid.mouse_up(HIDController.MOUSE_LEFT_UP)
logging.info("已发送 HID 鼠标左键点按（down=%s, up=%s）", down_ok, up_ok)
def key_action(self, action: KeyAction) -> None:
logging.info("按键 %s，按住 %.2f 秒%s", action.key, action.hold, " [dry-run]" if not self.live else "")
if self.live:
# The game may lose focus while the loading/observer screen is
# shown. Bring it back to the foreground immediately before the
# key-down event so the diagnostic S hold reaches CF.
if not self.activate_game_window():
logging.warning("发送 HID 按键前未能确认 CF 窗口处于前台")
if self.hid is None:
raise BotError("局内键盘需要 HID 驱动")
vk = self._virtual_key(action.key)
logging.info("发送 HID 按键按下：%s（VK=0x%02X）", action.key.upper(), vk)
down_ok = self.hid.key_down(vk)
try:
time.sleep(action.hold)
finally:
up_ok = self.hid.key_up(vk)
logging.info("已发送 HID 按键：%s（VK=0x%02X，down=%s, up=%s）", action.key.upper(), vk, down_ok, up_ok)
else:
time.sleep(min(action.hold, 0.05))
time.sleep(action.pause)
def press_sequence(self, keys: Iterable[str], pause: float = 0.5) -> None:
logging.info("按键序列：%s%s", " ".join(keys), " [dry-run]" if not self.live else "")
if self.live:
for key in keys:
time.sleep(random.uniform(0.55, 1.25))
self.activate_game_window()
# Returning from the game is intentionally kept on the
# original PyAutoGUI path. The CF exit/menu layer accepts
# these foreground keyboard events even when gameplay input
# is handled by the HID driver.
self.pag.press(key)
logging.info("已发送原生退出按键：%s", key)
time.sleep(random.uniform(max(0.7, pause * 1.1), max(1.2, pause * 2.0)))
def click_lowest_population_channel(self) -> Optional[tuple[int, int]]:
"""Return a click point for the shortest visible channel population bar.
The channel page in the supplied 1024x768 client has 13 rows. The
blue population bars occupy the right side of the table; measuring
their blue-pixel count avoids hard-coding a particular channel number.
"""
if not self.live:
return (570, 200)
try:
origin_x, origin_y = self._refresh_game_origin()
left, top, width, height = 830, 185, 170, 370
screenshot = self.pag.screenshot(region=(left + origin_x, top + origin_y, width, height))
row_count = 13
row_height = height / row_count
scores: list[tuple[int, int]] = []
for index in range(row_count):
y0 = int(index * row_height)
y1 = int((index + 1) * row_height)
blue_pixels = 0
for y in range(y0, y1):
for x in range(width):
red, green, blue = screenshot.getpixel((x, y))[:3]
if blue > 120 and green > 75 and red < 130 and blue > red * 1.25:
blue_pixels += 1
scores.append((blue_pixels, index))
populated = [item for item in scores if item[0] > 0]
if not populated:
return None
_, index = min(populated)
row_y = int(top + origin_y + (index + 0.5) * row_height)
return (560 + origin_x, row_y)
except Exception as exc:
logging.debug("频道人数识别失败：%s", exc)
return None
def scroll_inventory(self, steps: int) -> None:
logging.info("道具列表滚轮翻页：%s%s", steps, " [dry-run]" if not self.live else "")
if self.live:
origin_x, origin_y = self._refresh_game_origin()
self.pag.moveTo(origin_x + 512, origin_y + 384, duration=0.1)
self.pag.scroll(steps)
def quick_join_mode_list_signature(self) -> Optional[bytes]:
"""Small snapshot used to detect that the mode list no longer scrolls."""
if not self.live:
return None
try:
origin_x, origin_y = self._refresh_game_origin()
image = self.pag.screenshot(region=(origin_x + 220, origin_y + 230, 177, 435))
return image.convert("L").resize((24, 48)).tobytes()
except Exception as exc:
logging.debug("快速加入模式列表截图失败：%s", exc)
return None
def _game_is_foreground(self) -> bool:
try:
import ctypes
ctypes.windll.user32.GetForegroundWindow.restype = ctypes.c_void_p
hwnd = cf_window.find_window(self.window_titles, self.process_names)
return bool(hwnd and ctypes.windll.user32.GetForegroundWindow() == hwnd)
except Exception:
return False
def scroll_quick_join_modes(self, steps: int) -> None:
"""Scroll over the mode list, not the adjacent map list or scrollbar."""
if not self.live:
return
if not self._game_is_foreground() and not self.activate_game_window():
raise BotError("未能激活 CF 游戏窗口，取消模式列表滚动")
origin_x, origin_y = self._refresh_game_origin()
self.pag.moveTo(origin_x + 300, origin_y + 450, _pause=False)
self.pag.scroll(steps, _pause=False)
def read_score(self, region: tuple[int, int, int, int]) -> Optional[int]:
"""Read the right-aligned score from one game screenshot."""
try:
import numpy as np # type: ignore
origin_x, origin_y = self._refresh_game_origin()
screen_region = (region[0] + origin_x, region[1] + origin_y, region[2], region[3])
screenshot = self.pag.screenshot(region=screen_region)
return self.score_reader.read(np.asarray(screenshot))
except Exception as exc:
logging.debug("分数模板识别失败：%s", exc)
return None
class WindowGuard:
"""Checks that the expected game window exists before live actions."""
def __init__(self, settings: Settings):
self.settings = settings
def check(self) -> None:
try:
cf_window.require_window(self.settings.window_titles, self.settings.process_names)
except RuntimeError as exc:
raise BotError(str(exc)) from exc
class ChallengeBot:
def __init__(self, settings: Settings, templates: Mapping[str, TemplateSpec], ui: UI):
self.settings = settings
self.templates = templates
self.ui = ui
self.stop_requested = False
self.started_at = time.monotonic()
self.in_game = False
self.stage = "未启动"
self.current_score: Optional[int] = None
self.current_bonus_percent: Optional[int] = None
self.score_confirmation_count = 0
self._last_accepted_score: Optional[int] = None
self.completed_rounds = 0
self._last_unknown_log_at = 0.0
def stop(self, *_args: Any) -> None:
self.stop_requested = True
self.ui.release_all_keys()
logging.warning("收到停止信号，将在当前步骤结束后退出")
def timed_out(self) -> bool:
return False
def find(self, name: str) -> Optional[tuple[int, int]]:
spec = self.templates.get(name)
if spec is None:
raise BotError(f"未配置模板：{name}")
return self.ui.locate(spec)
def wait_and_click(self, name: str, *, timeout: Optional[float] = None) -> bool:
deadline = time.monotonic() + (timeout or self.settings.template_timeout)
while not self.stop_requested and not self.timed_out() and time.monotonic() < deadline:
self.dismiss_popups()
point = self.find(name)
if point:
self.ui.click(point, name)
return True
time.sleep(self.settings.poll_interval)
logging.warning("等待模板超时：%s", name)
return False
def wait_and_double_click(self, name: str, *, timeout: Optional[float] = None) -> bool:
deadline = time.monotonic() + (timeout or self.settings.template_timeout)
while not self.stop_requested and not self.timed_out() and time.monotonic() < deadline:
self.dismiss_popups()
point = self.find(name)
if point:
self.ui.double_click(point, name)
return True
time.sleep(self.settings.poll_interval)
logging.warning("等待模板超时：%s", name)
return False
def wait_for(self, name: str, *, timeout: Optional[float] = None) -> bool:
deadline = time.monotonic() + (timeout or self.settings.template_timeout)
while not self.stop_requested and not self.timed_out() and time.monotonic() < deadline:
self.dismiss_popups()
if self.find(name):
return True
time.sleep(self.settings.poll_interval)
logging.warning("等待模板超时：%s", name)
return False
def wait_and_click_first(self, names: Iterable[str], *, timeout: Optional[float] = None) -> Optional[str]:
deadline = time.monotonic() + (timeout or self.settings.template_timeout)
while not self.stop_requested and not self.timed_out() and time.monotonic() < deadline:
self.dismiss_popups()
for name in names:
point = self.find(name)
if point:
self.ui.click(point, name)
return name
time.sleep(self.settings.poll_interval)
logging.warning("等待任一模板超时：%s", ", ".join(names))
return None
def close_recommended_popup(self) -> bool:
"""Close the occasional CF recommended-item dialog before continuing."""
if "recommended_header" not in self.templates or "recommended_close" not in self.templates:
return False
if self.find("recommended_header") is None:
return False
point = self.find("recommended_close")
if point is None:
return False
logging.info("检测到推荐道具弹窗，点击底部关闭按钮")
self.ui.click(point, "推荐道具弹窗关闭")
time.sleep(0.15 if self.ui.live else 0.01)
return True
def close_guide_popup(self) -> bool:
"""Close the post-room map guide that can cover the room HUD."""
if self.in_game or "guide_header" not in self.templates or "guide_close" not in self.templates:
return False
if self.find("guide_header") is None:
return False
point = self.find("guide_close")
if point is None:
return False
logging.info("检测到进房指南弹窗，点击底部关闭按钮")
self.ui.click(point, "指南弹窗关闭")
time.sleep(0.15 if self.ui.live else 0.01)
return True
def close_expired_item_popup(self) -> bool:
if self.in_game or "expired_item_header" not in self.templates or "expired_item_cancel" not in self.templates:
return False
if self.find("expired_item_header") is None:
return False
point = self.find("expired_item_cancel")
if point is None:
return False
logging.info("检测到选择过期道具弹窗，点击底部取消按钮")
self.ui.click(point, "过期道具弹窗取消")
time.sleep(0.15 if self.ui.live else 0.01)
return True
def dismiss_popups(self) -> None:
if self.close_expired_item_popup():
return
if self.close_guide_popup():
return
self.close_recommended_popup()
def select_quick_join_challenge_mode(self, *, timeout: Optional[float] = None) -> bool:
"""Find Challenge Mode despite an arbitrary initial list scroll position."""
deadline = time.monotonic() + (timeout or 120.0)
for steps, direction in ((-1, "下移"), (1, "上移")):
previous_signature: Optional[bytes] = None
unchanged = 0
for _ in range(25):
if self.stop_requested or self.timed_out() or time.monotonic() >= deadline:
return False
self.dismiss_popups()
point = self.find("challenge_mode") or self.find("challenge_mode_unselected")
if point:
self.ui.click(point, "挑战模式")
return True
signature = self.ui.quick_join_mode_list_signature()
if signature is not None and signature == previous_signature:
unchanged += 1
if unchanged >= 2:
logging.info("模式列表已无法继续%s，改为反向查找", direction)
break
else:
unchanged = 0
previous_signature = signature
logging.info("快速加入模式列表未找到挑战模式，滚轮%s一格", direction)
self.ui.scroll_quick_join_modes(steps)
time.sleep(QUICK_JOIN_SCROLL_SETTLE_SECONDS if self.ui.live else 0.01)
logging.warning("滚动模式列表后仍未找到挑战模式")
return False
def first_visible(self, names: Iterable[str]) -> Optional[str]:
for name in names:
if self.find(name):
return name
return None
def detect_stage(self) -> str:
"""Infer the current screen from distinctive, mutually exclusive UI anchors."""
# The top-right return icon is shared by several screens (including
# the home page), so it is an action target only and never a room
# state anchor. Home is recognized by its page-only banner first;
# the room-hall button alone is deliberately not enough evidence.
checks = (
("gameplay", ("in_game_hud",)),
("win_result", ("win_result",)),
("home", ("home_banner",)),
("room", ("join_game", "ready", "start_game", "cancel_ready")),
("quick_join_dialog", ("quick_join_title", "challenge_mode", "map_new_giant_ruins", "confirm")),
("room_list", ("quick_join_settings",)),
("channel_list", ("challenge_channel_header",)),
("server_list", ("challenge_mode_1",)),
# The room-hall entry is the final navigation fallback. It is
# checked after every more specific scene anchor so it cannot
# override room/channel/server recognition.
("home", ("room_hall",)),
)
for stage, names in checks:
for name in names:
try:
if self.find(name):
return stage
except BotError:
continue
now = time.monotonic()
if now - self._last_unknown_log_at >= 10.0:
self._last_unknown_log_at = now
try:
screen_size = self.ui.pag.size()
logging.warning(
"暂未识别当前场景，等待画面恢复：屏幕=%sx%s，已检查 %d 个场景模板",
screen_size.width,
screen_size.height,
sum(len(names) for _, names in checks),
)
except Exception:
logging.warning("暂未识别当前场景，等待画面恢复")
return "unknown"
def sync_stage(self) -> str:
detected = self.detect_stage()
labels = {
"gameplay": "游戏内挂机",
"win_result": "正常通关结算",
"room": "房间内",
"quick_join_dialog": "快速加入设置",
"room_list": "房间列表",
"channel_list": "频道列表",
"server_list": "大厅服务器列表",
"home": "游戏首页",
"unknown": "未知页面",
}
self.stage = labels[detected]
self.in_game = detected == "gameplay"
return detected
def enter_room_from_current(self) -> bool:
"""Handle room controls that vary between ready/join/start states."""
room_action = self.wait_and_click_first(("join_game", "start_game", "ready", "cancel_ready"), timeout=20)
if room_action is None:
return False
if room_action == "cancel_ready":
logging.info("房间已准备，等待 10 秒观察是否开局")
deadline = time.monotonic() + 10.0
while time.monotonic() < deadline and not self.stop_requested:
if self.find("in_game_hud"):
return self.run_gameplay()
time.sleep(0.5)
logging.info("等待超时，点击取消准备并退出房间列表")
if not self.wait_and_click("cancel_ready", timeout=5):
return False
return self.leave_room_to_list()
if room_action == "ready":
logging.info("已点击准备，等待 10 秒")
deadline = time.monotonic() + 10.0
while time.monotonic() < deadline and not self.stop_requested:
if self.find("in_game_hud"):
return self.run_gameplay()
time.sleep(0.5)
if self.find("cancel_ready"):
self.wait_and_click("cancel_ready", timeout=5)
return self.leave_room_to_list()
logging.info("已点击房间按钮: %s", room_action)
self.stage = "观战/加载中"
if not self.wait_for("in_game_hud", timeout=self.settings.game_start_timeout):
return False
self.in_game = True
self.stage = "游戏内挂机"
return self.run_gameplay()
def leave_room_to_list(self) -> bool:
if not self.wait_and_click("return_hall", timeout=10):
return False
self.stage = "房间列表"
return self.wait_for("quick_join_settings", timeout=10)
def run_round(self, round_number: int) -> bool:
logging.info("开始第 %d 轮，地图：%s", round_number, self.settings.map_name)
self.ui.activate_game_window()
self.dismiss_popups()
detected = self.sync_stage()
logging.info("当前场景：%s", self.stage)
if detected == "gameplay":
return self.run_gameplay()
if detected == "win_result":
return self.wait_and_click("result_confirm", timeout=10) and self.finish_round_from_room()
if detected == "room":
return self.enter_room_from_current()
if detected == "quick_join_dialog":
self.stage = "快速加入设置"
if not self.select_quick_join_challenge_mode() or not self.wait_and_click("map_new_giant_ruins") or not self.wait_and_click("confirm"):
return False
return self.enter_room_from_current()
if detected == "room_list":
if not self.wait_and_click("quick_join_settings"):
return False
return self.run_round(round_number)
if detected == "channel_list":
if not self.select_low_population_channel():
return False
return self.run_round(round_number)
if detected == "server_list":
if not self.wait_and_double_click("challenge_mode_1"):
return False
return self.run_round(round_number)
if detected == "unknown":
return False
if detected == "home":
if not self.wait_and_click("room_hall"):
return False
return self.run_round(round_number)
return False
def select_low_population_channel(self) -> bool:
"""Choose the shortest visible blue population bar on the channel page."""
try:
point = self.ui.click_lowest_population_channel()
except AttributeError:
logging.error("当前 UI 适配器不支持频道选择")
return False
if point is None:
logging.warning("未识别到频道人数条")
return False
self.ui.double_click(point, "低人数频道")
time.sleep(1.5 if self.ui.live else 0.05)
return True
def run_gameplay(self) -> bool:
last_keep_alive = time.monotonic()
last_score_read = 0.0
initial_right_clicked = False
next_periodic_click_at: Optional[float] = None
gameplay_started_at = time.monotonic()
initial_right_delay = random.uniform(INITIAL_RIGHT_CLICK_MIN_SECONDS, INITIAL_RIGHT_CLICK_MAX_SECONDS)
logging.info("进入局内后等待 %.1f 秒发送首次右键", initial_right_delay)
self.score_confirmation_count = 0
self._last_accepted_score = None
self.current_score = None
pending_jump: Optional[int] = None
while not self.stop_requested and not self.timed_out():
self.dismiss_popups()
if self.find("win_result"):
self.stage = "正常通关结算"
logging.info("检测到正常通关结算页面")
if not self.wait_and_click("result_confirm", timeout=10):
return False
return self.finish_round_from_room()
# A manual Esc/Enter exit changes the screen while this loop is
# still alive. Re-sample the HUD before logging it as gameplay.
if not self.find("in_game_hud"):
detected = self.detect_stage()
if detected == "room":
self.in_game = False
self.stage = "房间内"
logging.info("检测到已手动退出局内，当前场景：房间内")
return self.enter_room_from_current()
if detected != "gameplay":
self.stage = "局内画面暂未识别，等待恢复"
time.sleep(self.settings.poll_interval)
continue
now = time.monotonic()
if not initial_right_clicked and now - gameplay_started_at >= initial_right_delay:
self.ui.right_click()
initial_right_clicked = True
next_periodic_click_at = now + random.uniform(PERIODIC_CLICK_MIN_SECONDS, PERIODIC_CLICK_MAX_SECONDS)
self.stage = "游戏内挂机/周期点按"
logging.info("已发送首次右键，下一次左键+右键将在 %.1f 秒后执行", next_periodic_click_at - now)
elif initial_right_clicked and next_periodic_click_at is not None and now >= next_periodic_click_at:
self.ui.left_click("周期左键点按")
self.ui.right_click("周期右键点按")
next_periodic_click_at = now + random.uniform(PERIODIC_CLICK_MIN_SECONDS, PERIODIC_CLICK_MAX_SECONDS)
logging.info("已发送左键+右键，下一次将在 %.1f 秒后执行", next_periodic_click_at - now)
if self.settings.score_region and now - last_score_read >= self.settings.score_poll_interval:
score = self.ui.read_score(self.settings.score_region)
last_score_read = now
if score is not None and self._last_accepted_score is not None:
# Confirm an unusually large jump on a later sample so
# missed polls cannot permanently block a real high score.
previous = self._last_accepted_score
if score < max(0, previous - 120000):
logging.warning("忽略异常分数跳变：%s -> %s", previous, score)
score = None
elif score > previous * 3 + 300000:
if pending_jump is None or abs(score - pending_jump) > 120000:
logging.warning("分数大幅增加，等待下一次确认：%s -> %s", previous, score)
pending_jump = score
score = None
else:
pending_jump = None
else:
pending_jump = None
if score is not None:
self._last_accepted_score = score
self.current_score = score if score is not None else self._last_accepted_score
if score is not None:
logging.info("当前最终分数: %s/%s", score, self.settings.score_target)
if score >= self.settings.score_target:
self.score_confirmation_count += 1
self.stage = f"分数确认 {self.score_confirmation_count}/{SCORE_CONFIRMATIONS}"
if self.score_confirmation_count >= SCORE_CONFIRMATIONS:
self.stage = "退出局内确认"
self.ui.press_sequence(("esc", "enter", "enter"))
# Allow CF's menu transition to finish before
# searching for the room-page return icon.
time.sleep(random.uniform(1.5, 2.8) if self.ui.live else 0.05)
return self.finish_round_from_room()
else:
self.score_confirmation_count = 0
self.stage = "游戏内挂机"
else:
# A failed template sample is not a confirmation. Require a
# fresh uninterrupted run of valid above-threshold reads.
self.score_confirmation_count = 0
self.stage = "游戏内挂机/等待分数"
time.sleep(self.settings.poll_interval)
if self.settings.keep_alive_interval and time.monotonic() - last_keep_alive >= self.settings.keep_alive_interval:
logging.info("挑战仍在进行，未配置动作时仅保持监测")
last_keep_alive = time.monotonic()
return False
def finish_round_from_room(self) -> bool:
self.in_game = False
self.stage = "退出房间"
if not self.wait_and_click("return_hall", timeout=20):
logging.error("未找到房间页面右上角退出图标")
return False
# Crystal-box inventory tracking was removed from the workflow.
self.stage = "已回到房间列表"
return True
def run(self) -> int:
completed = 0
round_number = 1
try:
while not self.stop_requested:
try:
finished = self.run_round(round_number)
except Exception:
logging.exception("第 %d 轮操作异常，等待后重新检测场景", round_number)
finished = False
if self.stop_requested:
break
if finished:
completed += 1
self.completed_rounds = completed
round_number += 1
else:
self.stage = "等待场景恢复"
logging.warning("第 %d 轮未完成，保持运行并重新检测场景", round_number)
time.sleep(self.settings.poll_interval)
finally:
self.ui.release_all_keys()
logging.info("脚本结束：共完成 %d 轮", completed)
return 0
def read_templates(raw: Mapping[str, Any]) -> dict[str, TemplateSpec]:
values = raw.get("templates", {})
if not isinstance(values, dict):
raise BotError("templates 必须是对象")
result: dict[str, TemplateSpec] = {}
for name, value in values.items():
if not isinstance(value, dict) or not isinstance(value.get("file"), str):
raise BotError(f"模板 {name} 缺少 file")
region_value = value.get("region")
region = None
if region_value is not None:
if not isinstance(region_value, list) or len(region_value) != 4:
raise BotError(f"模板 {name} 的 region 必须是 [x,y,w,h]")
region = tuple(int(x) for x in region_value) # type: ignore[assignment]
result[name] = TemplateSpec(
name=name,
file=value["file"],
confidence=float(value.get("confidence", 0.86)),
region=region,
grayscale=bool(value.get("grayscale", False)),
)
return result
def setup_logging(log_path: Path) -> None:
log_path.parent.mkdir(parents=True, exist_ok=True)
# Windows consoles may use a legacy code page. Reconfigure the standard
# streams when possible so Chinese log messages remain readable both in
# the terminal and in the UTF-8 log file.
for stream in (sys.stdout, sys.stderr):
try:
stream.reconfigure(encoding="utf-8", errors="replace")
except (AttributeError, OSError):
pass
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
handlers=[logging.StreamHandler(sys.stdout), logging.FileHandler(log_path, encoding="utf-8")],
force=True,
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="CF 挑战模式模板识别挂机辅助")
parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
parser.add_argument("--dry-run", action="store_true", help="仅记录动作，不控制游戏")
parser.add_argument("--check", action="store_true", help="仅检查依赖、配置和模板文件")
return parser.parse_args()
def main() -> int:
args = parse_args()
# Configure UTF-8 output before loading user-controlled files so even
# configuration errors are readable in the console and log file.
setup_logging(ROOT / "logs" / "cf_challenge_bot.log")
raw = load_json(args.config)
settings = build_settings(raw)
templates = read_templates(raw)
templates_dir = (args.config.parent / str(raw.get("templates_dir", "challenge_templates"))).resolve()
missing = [str(templates_dir / spec.file) for spec in templates.values() if not (templates_dir / spec.file).is_file()]
if missing:
logging.warning("缺少 %d 个模板图片，请检查 challenge_templates 目录", len(missing))
for path in missing:
logging.warning("缺少模板: %s", path)
if args.check:
return 1
if args.check:
logging.info("配置检查通过，地图：%s，分辨率：%sx%s", settings.map_name, *settings.resolution)
return 0
guard = WindowGuard(settings)
if not args.dry_run:
guard.check()
ui = UI(
templates_dir,
live=not args.dry_run,
jitter=settings.click_jitter,
window_titles=settings.window_titles,
process_names=settings.process_names,
hid_dll=(args.config.parent / settings.hid_dll).resolve() if settings.hid_dll else None,
)
bot = ChallengeBot(settings, templates, ui)
signal.signal(signal.SIGINT, bot.stop)
signal.signal(signal.SIGTERM, bot.stop)
return bot.run()
if __name__ == "__main__":
try:
raise SystemExit(main())
except BotError as exc:
logging.error("%s", exc)
raise SystemExit(2)
项目实现思路参考于：https://wenku.csdn.net/column/71jnunp490n
七、免责声明
本软件仅依据屏幕画面进行识别，并模拟普通键盘、鼠标操作；不读取、注入或修改游戏进程及内存，不拦截网络通信，也不篡改游戏文件或游戏数据。
本软件代码完全免费、开源仅供交流探讨学习，请勿二次售卖、捆绑收费，或以本软件名义向他人收取费用。
画面识别和自动操作可能受到游戏更新、分辨率、电脑环境、网络状态、弹窗遮挡及其他不可控因素影响，无法保证始终准确。使用前请自行了解游戏规则、平台规定及相关风险，并自行决定是否使用本软件。
因使用本软件产生的误操作、游戏账号风险、封禁风险、数据损失或其他任何损失，由使用者自行承担。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
暂无评论内容