Month: September 2025

class XianYuGameBot:
def __init__(self):
# 基础配置
self.GAME_TITLE = "咸鱼之王"
self.LOG_FILE = "game_bot.log"
self.IMAGE_DIR = "captures"

# 关键词配置
self.KEYWORDS = {
"咸将": "咸将",
"装备": "装备",
"副本": "副本",
"宝箱": "木质宝箱",
"战斗": "俱乐部",
"活动": "限时活动",
"订单": "订单支付",
"解锁": ["滑动", "解锁"]
}

# 正则表达式
self.level_re = re.compile(r'第\s*(\d+(?:\s*/\s*\d+)?)\s*关', re.S)
self.power_re = re.compile(r'战力[:：]\s*(\d+(?:\.\d+)?)\s*万', re.S)

# 初始化组件
self.setup_logging()
self.init_ocr()
self.init_window()
self.create_directories()

# 操作延迟配置
self.DELAYS = {
"click": (0.1, 0.3),
"page_switch": (0.8, 1.5),
"upgrade": (0.3, 0.5),
"battle": (2, 4)
}

def setup_logging(self):
"""配置日志系统"""
logging.basicConfig(
filename=self.LOG_FILE,
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
encoding="utf-8"
)
console = logging.StreamHandler()
console.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(message)s')
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)

def init_ocr(self):
"""初始化OCR引擎 - 添加备用方案"""
try:
# 主要OCR引擎
self.ocr = PaddleOCR(
lang="ch",
use_textline_orientation=True,
text_det_thresh=0.3,
text_det_box_thresh=0.5
)
logging.info("PaddleOCR初始化成功")

# 备用OCR引擎 - 使用更简单的配置
try:
self.ocr_backup = PaddleOCR(lang="ch")
logging.info("备用PaddleOCR初始化成功")
except Exception as e:
logging.warning(f"备用PaddleOCR初始化失败: {e}")
self.ocr_backup = None

except Exception as e:
logging.error(f"主要PaddleOCR初始化失败: {e}")
# 尝试使用最简配置
try:
self.ocr = PaddleOCR(lang="ch")
logging.info("使用最简配置初始化PaddleOCR成功")
self.ocr_backup = None
except Exception as e2:
logging.error(f"最简配置也失败: {e2}")
raise

def init_window(self):
"""初始化游戏窗口"""
self.window = self.get_game_window()

def create_directories(self):
"""创建必要的目录"""
if not os.path.exists(self.IMAGE_DIR):
os.makedirs(self.IMAGE_DIR)

def get_game_window(self):
"""获取游戏窗口"""
try:
windows = gw.getWindowsWithTitle(self.GAME_TITLE)
if windows:
logging.info(f"找到游戏窗口: {windows[0]}")
return windows[0]
else:
logging.warning("未找到游戏窗口")
return None
except Exception as e:
logging.error(f"获取游戏窗口失败: {e}")
return None

def activate_window(self):
"""激活游戏窗口"""
if self.window:
try:
self.window.activate()
time.sleep(0.5)
return True
except Exception as e:
logging.error(f"激活窗口失败: {e}")
return False
return False

def grab_screenshot(self):
"""截取游戏画面"""
if not self.activate_window():
return None

try:
if self.window:
region = (self.window.left + 15, self.window.top,
self.window.width, self.window.height)
screenshot = pyautogui.screenshot(region=region)
logging.debug(f"截图成功，尺寸: {screenshot.size}")
return screenshot
except Exception as e:
logging.error(f"截图失败: {e}")
return None
return None

def preprocess_image(self, img):
"""图像预处理 - 添加更多调试信息"""
try:
# 转换为OpenCV格式
img_cv = cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR)

logging.debug(f"原始图像尺寸: {img_cv.shape}")

# 保存原始图像用于调试
self.save_debug_image(img_cv, "original_cv")

# 转换为灰度图
gray = cv2.cvtColor(img_cv, cv2.COLOR_BGR2GRAY)

# 检查图像质量
mean_brightness = np.mean(gray)
logging.debug(f"图像平均亮度: {mean_brightness}")

# 方法1：简单二值化
_, binary_simple = cv2.threshold(gray, 127, 255, cv2.THRESH_BINARY)

# 方法2：自适应阈值
binary_adaptive = cv2.adaptiveThreshold(
gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
cv2.THRESH_BINARY, 11, 2
)

# 方法3：Otsu自动阈值
_, binary_otsu = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)

# 方法4：反转图像
binary_inverted = cv2.bitwise_not(binary_simple)

# 方法5：边缘检测
edges = cv2.Canny(gray, 50, 150)

# 方法6：形态学操作
kernel = np.ones((2,2), np.uint8)
morph = cv2.morphologyEx(binary_simple, cv2.MORPH_CLOSE, kernel)

# 保存所有预处理结果
methods = [
("binary_simple", binary_simple),
("binary_adaptive", binary_adaptive),
("binary_otsu", binary_otsu),
("binary_inverted", binary_inverted),
("edges", edges),
("morph", morph)
]

for name, img_method in methods:
self.save_debug_image(img_method, name)

# 选择最佳预处理结果
processed_img = binary_inverted

# 转换为3通道图像
processed_3ch = cv2.cvtColor(processed_img, cv2.COLOR_GRAY2BGR)

logging.debug(f"预处理完成，最终图像尺寸: {processed_3ch.shape}")
return processed_3ch

except Exception as e:
logging.error(f"图像预处理失败: {e}")
import traceback
logging.error(f"预处理详细错误: {traceback.format_exc()}")
# 如果预处理失败，返回原始图像
try:
return cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR)
except:
return np.array(img)

def extract_text(self, result):
"""从OCR结果中提取文本 - 增强版本"""
if not result:
return ""

text_lines = []

try:
# 处理不同的返回格式
if isinstance(result, str):
return result.strip()

if isinstance(result, (list, tuple)):
for item in result:
if isinstance(item, (list, tuple)):
for sub_item in item:
if isinstance(sub_item, (list, tuple)) and len(sub_item) >= 2:
if isinstance(sub_item[1], (list, tuple)) and len(sub_item[1]) >= 1:
text = str(sub_item[1][0]).strip()
if text:
text_lines.append(text)
# 记录置信度
if len(sub_item[1]) >= 2:
confidence = sub_item[1][1]
logging.debug(f"识别文本: '{text}' (置信度: {confidence})")
elif isinstance(sub_item, str):
text = sub_item.strip()
if text:
text_lines.append(text)
elif isinstance(item, str):
text = item.strip()
if text:
text_lines.append(text)
except Exception as e:
logging.debug(f"文本提取出错: {e}")

# 过滤空行并合并
filtered_lines = [line for line in text_lines if line.strip()]
result_text = '\n'.join(filtered_lines)

logging.debug(f"提取到 {len(filtered_lines)} 行文本")
return result_text

def ocr_text(self, img):
"""OCR文本识别 - 增强错误处理和调试"""
try:
# 预处理图像
processed_img = self.preprocess_image(img)

# 确保图像是3通道的
if len(processed_img.shape) == 2:
processed_img = cv2.cvtColor(processed_img, cv2.COLOR_GRAY2BGR)

logging.debug(f"OCR输入图像形状: {processed_img.shape}")

# 尝试多种OCR方法
methods = [
("主要PaddleOCR", self.ocr, processed_img),
("备用PaddleOCR", self.ocr_backup, processed_img) if self.ocr_backup else None,
("原始图像", self.ocr, cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR)),
("灰度图像", self.ocr, cv2.cvtColor(cv2.cvtColor(np.array(img), cv2.COLOR_RGB2GRAY), cv2.COLOR_GRAY2BGR)),
("反转图像", self.ocr, cv2.bitwise_not(cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR)))
]

# 过滤掉None值
methods = [m for m in methods if m is not None]

best_result = ""
best_text = ""

for method_name, ocr_engine, method_img in methods:
try:
logging.info(f"尝试OCR方法: {method_name}")

# 检查图像是否为空
if method_img is None or method_img.size == 0:
logging.warning(f"方法 {method_name} 的图像为空")
continue

# 检查图像尺寸
if method_img.shape[0] < 10 or method_img.shape[1] < 10: logging.warning(f"方法 {method_name} 的图像尺寸过小: {method_img.shape}") continue # OCR识别 logging.debug(f"开始OCR识别，图像尺寸: {method_img.shape}") result = ocr_engine.predict(method_img) logging.debug(f"OCR原始返回结果: {result}") # 提取文本 text = self.extract_text(result) if text and len(text.strip()) > len(best_text.strip()):
best_result = result
best_text = text
logging.info(f"方法 {method_name} 获得更好结果: {text[:100]}...")

# 如果识别到关键游戏文本，直接返回
if any(keyword in text for keyword in ["关卡", "通关", "滑动", "解锁", "咸将", "装备"]):
logging.info(f"方法 {method_name} 识别到游戏关键词")
return text

except Exception as e:
logging.error(f"方法 {method_name} 失败: {e}")
import traceback
logging.error(f"方法 {method_name} 详细错误: {traceback.format_exc()}")
continue

# 返回最佳结果
if best_text:
logging.info(f"使用最佳OCR结果: {best_text[:100]}...")
return best_text
else:
logging.warning("所有OCR方法都失败")

# 最后的备用方案：尝试使用PIL图像
try:
logging.info("尝试使用PIL图像进行OCR")
pil_img = img.convert('L') # 转换为灰度图
pil_img = pil_img.resize((pil_img.width * 2, pil_img.height * 2)) # 放大2倍
pil_array = np.array(pil_img)
pil_3ch = cv2.cvtColor(pil_array, cv2.COLOR_GRAY2BGR)

result_pil = self.ocr.predict(pil_3ch)
text_pil = self.extract_text(result_pil)

if text_pil:
logging.info(f"PIL备用方案成功: {text_pil[:100]}...")
return text_pil

except Exception as e:
logging.error(f"PIL备用方案也失败: {e}")

return ""

except Exception as e:
logging.error(f"OCR识别失败: {e}")
import traceback
logging.error(f"详细错误: {traceback.format_exc()}")
return ""

def save_debug_image(self, img, prefix="debug"):
"""保存调试图像 - 增强版本"""
try:
timestamp = time.strftime("%Y%m%d_%H%M%S")
filename = os.path.join(self.IMAGE_DIR, f"{prefix}_{timestamp}.png")

if isinstance(img, np.ndarray):
# 如果是numpy数组，转换为PIL图像
if len(img.shape) == 2: # 灰度图
img_pil = Image.fromarray(img)
else: # 彩色图
img_pil = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
img_pil.save(filename)
else:
img.save(filename)

logging.debug(f"保存调试图像: {filename}")
return filename
except Exception as e:
logging.error(f"保存图像失败: {str(e)}")
return None

def click_position(self, x, y, random_offset=True):
"""点击指定位置"""
if not self.activate_window():
return False

try:
if random_offset:
x += random.randint(-3, 3)
y += random.randint(-3, 3)

pyautogui.moveTo(x, y, duration=random.uniform(*self.DELAYS["click"]))
print("点击")
pyautogui.click()

logging.debug(f"点击位置: ({x}, {y})")
return True
except Exception as e:
logging.error(f"点击失败: {e}")
return False

def click_center(self):
"""点击游戏窗口中心"""
if self.window:
center_x = self.window.left + self.window.width // 2
center_y = self.window.top + self.window.height // 2
return self.click_position(center_x, center_y)
return False

def swipe(self):
"""滑动屏幕"""
if not self.activate_window():
return False

try:
if self.window:
x0 = self.window.left + self.window.width // 3
y = self.window.top + self.window.height // 2
x1 = self.window.left + self.window.width * 2 // 3

pyautogui.moveTo(x0, y, duration=0.1)
pyautogui.dragTo(x1, y, duration=0.5, button="left")
logging.info("滑动完成")
return True
except Exception as e:
logging.error(f"滑动失败: {e}")
return False

def navigate_to_tab(self, tab_name):
"""导航到指定标签页"""
tab_positions = {
"咸将": (0.1, 0.9),
"装备": (0.2, 0.9),
"战斗": (0.3, 0.9),
"宝箱": (0.4, 0.9),
"副本": (0.5, 0.9)
}

if tab_name not in tab_positions:
logging.error(f"未知标签页: {tab_name}")
return False

if not self.window:
return False

rel_x, rel_y = tab_positions[tab_name]
abs_x = self.window.left + int(self.window.width * rel_x)
abs_y = self.window.top + int(self.window.height * rel_y)

if self.click_position(abs_x, abs_y):
time.sleep(random.uniform(*self.DELAYS["page_switch"]))
logging.info(f"已导航到标签页: {tab_name}")
return True

return False

def auto_open_box(self):
"""自动开箱"""
logging.info("开始自动开箱")
try:
if not self.navigate_to_tab("宝箱"):
return False

# 开箱按钮位置
box_positions = [(0.3, 0.6), (0.5, 0.6), (0.7, 0.6)]

for pos in box_positions:
if self.window:
abs_x = self.window.left + int(self.window.width * pos[0])
abs_y = self.window.top + int(self.window.height * pos[1])
self.click_position(abs_x, abs_y)
time.sleep(random.uniform(*self.DELAYS["page_switch"]))

logging.info("自动开箱完成")
return True
except Exception as e:
logging.error(f"自动开箱失败: {e}")
return False

def upgrade_hero(self):
"""咸将升级"""
logging.info("开始咸将升级")
try:
if not self.navigate_to_tab("咸将"):
return False

upgrade_pos = (0.5, 0.8)

if self.window:
abs_x = self.window.left + int(self.window.width * upgrade_pos[0])
abs_y = self.window.top + int(self.window.height * upgrade_pos[1])

for _ in range(5):
self.click_position(abs_x, abs_y)
time.sleep(random.uniform(*self.DELAYS["upgrade"]))

logging.info("咸将升级完成")
return True
except Exception as e:
logging.error(f"咸将升级失败: {e}")
return False

def upgrade_equipment(self):
"""装备升级"""
logging.info("开始装备升级")
try:
if not self.navigate_to_tab("装备"):
return False

upgrade_pos = (0.7, 0.8)

if self.window:
abs_x = self.window.left + int(self.window.width * upgrade_pos[0])
abs_y = self.window.top + int(self.window.height * upgrade_pos[1])
self.click_position(abs_x, abs_y)
time.sleep(random.uniform(*self.DELAYS["page_switch"]))

logging.info("装备升级完成")
return True
except Exception as e:
logging.error(f"装备升级失败: {e}")
return False

def auto_dungeon(self):
"""自动副本"""
logging.info("开始自动副本")
try:
if not self.navigate_to_tab("副本"):
return False

dungeon_positions = [(0.4, 0.5), (0.6, 0.5)]

for pos in dungeon_positions:
if self.window:
abs_x = self.window.left + int(self.window.width * pos[0])
abs_y = self.window.top + int(self.window.height * pos[1])
self.click_position(abs_x, abs_y)
time.sleep(random.uniform(*self.DELAYS["page_switch"]))

logging.info("自动副本完成")
return True
except Exception as e:
logging.error(f"自动副本失败: {e}")
return False

def handle_page(self, text):
"""处理不同页面的逻辑 - 针对游戏失败页面优化"""
if not text.strip():
logging.debug("文本为空，跳过页面处理")
return False

lines = text.split('\n')
logging.info(f"处理页面文本，共 {len(lines)} 行")

# 打印所有行用于调试
for i, line in enumerate(lines):
if line.strip():
logging.debug(f"第{i+1}行: {line}")

# 游戏失败页面 - 检查关键信息
if "通关失败" in text or "关卡" in text:
logging.info("检测到游戏失败页面")

# 查找关卡信息
for line in lines:
if "关卡" in line:
logging.info(f"找到关卡信息: {line}")
break

# 检查是否有滑动解锁提示
if "滑动解锁" in text:
logging.info("检测到滑动解锁提示")
if self.swipe():
logging.info("滑动解锁完成")
time.sleep(1) # 等待解锁动画
return True

return True

# 咸将页面
if any("咸将" in line for line in lines):
logging.info("检测到咸将页面")
self.upgrade_hero()
return True

# 战斗页面
if any("俱乐部" in line for line in lines) and any("客厅" in line for line in lines):
level_info = ""
power_info = ""

for line in lines:
level_match = self.level_re.search(line)
if level_match:
level_info = level_match.group(0)
logging.info(f"找到关卡信息: {level_info}")

power_match = self.power_re.search(line)
if power_match:
power_info = power_match.group(0)
logging.info(f"找到战力信息: {power_info}")

if level_info and power_info:
logging.info(f"战斗页面确认: {level_info} | {power_info}")
self.auto_open_box()
return True

# 其他页面处理
elif "点击任意区域返回主界面" in text:
logging.info("检测到未锁定状态")
self.click_center()
self.auto_open_box()
return True
elif self.KEYWORDS["宝箱"] in text:
logging.info("检测到宝箱页面")
self.auto_dungeon()
return True
elif self.KEYWORDS["副本"] in text:
logging.info("检测到副本页面")
self.upgrade_equipment()
return True
elif self.KEYWORDS["装备"] in text:
logging.info("检测到装备页面")
self.upgrade_hero()
self.click_center()
return True
elif self.KEYWORDS["活动"] in text:
logging.info("检测到活动页面")
self.click_back_button()
self.upgrade_hero()
self.click_center()
return True
elif self.KEYWORDS["订单"] in text:
logging.info("检测到订单页面")
self.click_back_button()
self.upgrade_hero()
self.click_center()
return True

# 滑动解锁 - 检查所有行
elif any(any(k in line for k in self.KEYWORDS["解锁"]) for line in lines):
logging.info("检测到解锁提示")
if self.swipe():
logging.info("滑动解锁完成")
return True

logging.debug("未识别到特定页面")
return False

def click_back_button(self):
"""点击返回按钮"""
if self.window:
back_x = self.window.left + int(self.window.width * 0.1)
back_y = self.window.top + int(self.window.height * 0.1)
self.click_position(back_x, back_y)

def run(self):
"""主运行循环"""
logging.info("游戏机器人启动")

# 测试OCR功能
test_img = self.grab_screenshot()
if test_img:
test_text = self.ocr_text(test_img)
logging.info(f"OCR测试结果: {test_text[:100]}...")

while True:
try:
# 刷新窗口引用
self.window = self.get_game_window()

# 截图并识别
img = self.grab_screenshot()
if img is None:
logging.warning("未找到游戏窗口")
time.sleep(3)
continue

text = self.ocr_text(img)
print(f"识别文本: {text[:100]}...")

# 处理页面
if not self.handle_page(text):
logging.debug("未识别到特定页面")

time.sleep(2)

except KeyboardInterrupt:
logging.info("用户中断程序")
break
except Exception as e:
logging.error(f"主循环出错: {e}")
time.sleep(3)

if __name__ == "__main__":
try:
import paddleocr
print("PaddleOCR已安装")
except ImportError:
print("请先安装PaddleOCR: pip install paddleocr")
exit(1)