使用n8n构建自动化哈希解密机器人

S0cke3t2025-07-182025-07-18

前言

在前段时间某哈希解密大站更新了他们的付费API使用规则,更新的说明如下

新规定说白了就是除非充值最高金额,否则你的api只能使用10天,对新规定笔者不得已重新设计解密机器人实现方案.

经过一系列的测试，笔者最终决定使用如下方案实现.该方案优点是完全可以私有化部署

消息监听	使用青龙面板的定时任务功能,设置固定时间拉取wx消息(此方式效率较慢,可使用其他方式代替)
数据处理	使用n8n工作流实现消息的接收和网站的自动化登录,查询功能.最后通过wx框架将结果推送给用户
WxPad	wx08算法框架,主要提供消息轮询和消息推送等接口
Redis	主要用来维护执行工作流时产生的页面状态等临时数据

方案实现

注: 该文章只作为一个设计方案参考,具体的工作流文件和框架不会提供.

为方便理解,先从Wxpad开始.至于什么是Wxpad可以这么理解.某些大手子把wx的ipad,pc,移动端等通讯协议进行了逆向并封装成了api接口,这样我们就可以通过调用api接口的方式实现操作wx的目的.下面是该框架搭建起来大致的样子(不同版本会有些许的不同)

框架中提供了各种各样的接口,可根据实际需要进行调用

这里我们主要用到两个接口一个消息轮询,一个文本消息发送

其中消息轮询支持http和ws两种协议,笔者这里使用的为http方式

下面来看青龙面板的定时任务(关于如何青龙面板如何使用可自行百度)

我们需要写一个python或者nodejs脚本实现定时的去请求wxpad消息轮询接口，来抓取最新的消息

笔者这里使用python实现,hash_bot.py关键代码如下

SYNC_MSG_URL = f"http://192.168.0.5:8848/message/HttpSyncMsg?key={API_KEY}"
SELF_HOSTED_API_URL = "http://192.168.0.5:5678/webhook/727d2c8e-b1a9-46e3-baee-2e76183f3319"

def send_message(to_user, content):
    print(f"Sending message to {to_user}: {content}")
    payload = {
        "MsgItem": [
            {
                "AtWxIDList": [],
                "ImageContent": "",
                "MsgType": 0,
                "TextContent": content,
                "ToUserName": to_user
            }
        ]
    }
    try:
        response = requests.post(SEND_MSG_URL, json=payload)
        return response.status_code == 200
    except Exception as e:
        print(f"Error sending message: {e}")
        return False

def process_hash(hash_value, touser):
    """
    提交哈希值和用户信息到工作流，无需处理响应
    """
    payload = {'hash': hash_value, 'touser': touser}
    try:
        requests.post(SELF_HOSTED_API_URL, data=payload)
    except requests.RequestException as e:
        print(f"请求工作流时发生错误: {e}")
        return f"提交哈希值时发生错误: {e}"
    except Exception as e:
        print(f"提交哈希值时发生未知错误: {e}")
        return f"提交哈希值时发生未知错误: SYNC_MSG_URL = f"http://192.168.0.5:8848/message/HttpSyncMsg?key={API_KEY}"
SELF_HOSTED_API_URL = "http://192.168.0.5:5678/webhook/727d2c8e-b1a9-46e3-baee-2e76183f3319"

def send_message(to_user, content):
    print(f"Sending message to {to_user}: {content}")
    payload = {
        "MsgItem": [
            {
                "AtWxIDList": [],
                "ImageContent": "",
                "MsgType": 0,
                "TextContent": content,
                "ToUserName": to_user
            }
        ]
    }
    try:
        response = requests.post(SEND_MSG_URL, json=payload)
        return response.status_code == 200
    except Exception as e:
        print(f"Error sending message: {e}")
        return False

def process_hash(hash_value, touser):
    """
    提交哈希值和用户信息到工作流，无需处理响应
    """
    payload = {'hash': hash_value, 'touser': touser}
    try:
        requests.post(SELF_HOSTED_API_URL, data=payload)
    except requests.RequestException as e:
        print(f"请求工作流时发生错误: {e}")
        return f"提交哈希值时发生错误: {e}"
    except Exception as e:
        print(f"提交哈希值时发生未知错误: {e}")
        return f"提交哈希值时发生未知错误: {e}"