244 lines
8.8 KiB
Python
244 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Flash 游戏批量使用物品脚本
|
||
基于抓包分析：ItemAction.useBatchItem
|
||
"""
|
||
|
||
import struct
|
||
import requests
|
||
import sys
|
||
import time
|
||
|
||
# ==================== 配置 ====================
|
||
SERVER_URL = "http://47.121.191.22:8012/gateway"
|
||
COOKIE_FILE = "/Users/amos/Downloads/game/7q/latest_cookie.txt"
|
||
# ==============================================
|
||
|
||
class AMF0:
|
||
"""AMF0 协议编码器"""
|
||
|
||
NUMBER = 0x00
|
||
STRING = 0x02
|
||
NULL = 0x05
|
||
STRICT_ARRAY = 0x0A
|
||
|
||
@staticmethod
|
||
def encode_amf_request(target_uri, response_uri, params):
|
||
"""编码 AMF 请求"""
|
||
data = b''
|
||
data += struct.pack('!H', 0x0000) # AMF 版本
|
||
data += struct.pack('!H', 0x0000) # Header count
|
||
data += struct.pack('!H', 0x0001) # Message count
|
||
|
||
# Target URI
|
||
target_bytes = target_uri.encode('utf-8')
|
||
data += struct.pack('!H', len(target_bytes)) + target_bytes
|
||
|
||
# Response URI
|
||
response_bytes = response_uri.encode('utf-8')
|
||
data += struct.pack('!H', len(response_bytes)) + response_bytes
|
||
|
||
# Length
|
||
data += struct.pack('!I', 0xFFFFFFFF)
|
||
|
||
# Body - Strict array
|
||
data += struct.pack('!BI', AMF0.STRICT_ARRAY, len(params))
|
||
for item in params:
|
||
if isinstance(item, (int, float)):
|
||
data += struct.pack('!Bd', AMF0.NUMBER, float(item))
|
||
elif isinstance(item, str):
|
||
s_bytes = item.encode('utf-8')
|
||
data += struct.pack('!BH', AMF0.STRING, len(s_bytes)) + s_bytes
|
||
|
||
return data
|
||
|
||
|
||
def read_cookie():
|
||
"""从文件读取 Cookie"""
|
||
try:
|
||
with open(COOKIE_FILE, 'r') as f:
|
||
cookie = f.read().strip()
|
||
if cookie:
|
||
return cookie
|
||
except FileNotFoundError:
|
||
print(f"⚠️ Cookie 文件不存在: {COOKIE_FILE}")
|
||
except Exception as e:
|
||
print(f"⚠️ 读取 Cookie 失败: {e}")
|
||
return None
|
||
|
||
|
||
def use_batch_item(session, cookie, item_instance_id, item_type_id, quantity, request_id):
|
||
"""批量使用物品
|
||
|
||
Args:
|
||
item_instance_id: 物品实例ID（背包中的物品ID）
|
||
item_type_id: 物品类型ID
|
||
quantity: 使用数量
|
||
"""
|
||
|
||
amf_data = AMF0.encode_amf_request(
|
||
target_uri="com.huzi.item.action.ItemAction.useBatchItem",
|
||
response_uri=f"/{request_id}",
|
||
params=[item_instance_id, item_type_id, quantity]
|
||
)
|
||
|
||
headers = {
|
||
'Host': '47.121.191.22:8012',
|
||
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 14.0; rv:102.0) Gecko/20100101 Firefox/102.0',
|
||
'Accept': '*/*',
|
||
'Accept-Language': 'en-US,en;q=0.5',
|
||
'Accept-Encoding': 'gzip, deflate',
|
||
'Cookie': cookie,
|
||
'Origin': 'http://47.121.191.22:8012',
|
||
'Connection': 'keep-alive',
|
||
'Referer': 'http://47.121.191.22:8012/swf13/Load.swf',
|
||
'Content-Type': 'application/x-amf'
|
||
}
|
||
|
||
print(f"\n 📤 请求: ItemAction.useBatchItem({item_instance_id}, {item_type_id}, {quantity})")
|
||
print(f" 📍 Response URI: /{request_id}")
|
||
|
||
try:
|
||
response = session.post(SERVER_URL, data=amf_data, headers=headers, timeout=10)
|
||
|
||
print(f" 📥 响应状态: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
data = response.content
|
||
print(f" 📦 响应大小: {len(data)} 字节")
|
||
|
||
# 检查是否是错误响应
|
||
if b'onStatus' in data:
|
||
print(f" ⚠️ 检测到 onStatus (错误响应)")
|
||
if b'error' in data:
|
||
print(f" ⚠️ 检测到 error 字段")
|
||
if b'NullPointerException' in data:
|
||
print(f" ⚠️ NullPointerException")
|
||
return False, "服务器错误"
|
||
|
||
# 查找 result 字段
|
||
result_pos = data.find(b'result')
|
||
|
||
if result_pos > 0:
|
||
value_pos = result_pos + 6
|
||
|
||
if value_pos < len(data):
|
||
value_type = data[value_pos]
|
||
|
||
if value_type == 0x00: # Number
|
||
if value_pos + 9 <= len(data):
|
||
result_value = struct.unpack('!d', data[value_pos+1:value_pos+9])[0]
|
||
result = int(result_value)
|
||
print(f" 📊 result = {result}")
|
||
|
||
if result == 0:
|
||
return True, "成功"
|
||
else:
|
||
# 查找 msg
|
||
msg = None
|
||
msg_pos = data.find(b'msg')
|
||
if msg_pos > 0:
|
||
msg_value_pos = msg_pos + 3
|
||
if msg_value_pos < len(data):
|
||
msg_type = data[msg_value_pos]
|
||
if msg_type == 0x02: # String
|
||
try:
|
||
if msg_value_pos + 3 <= len(data):
|
||
msg_len = struct.unpack('!H', data[msg_value_pos+1:msg_value_pos+3])[0]
|
||
if msg_len > 0 and msg_value_pos + 3 + msg_len <= len(data):
|
||
msg = data[msg_value_pos+3:msg_value_pos+3+msg_len].decode('utf-8', errors='ignore')
|
||
print(f" 💬 msg = {msg}")
|
||
except:
|
||
pass
|
||
return False, msg if msg else f"失败(result={result})"
|
||
|
||
# 如果没有 result 字段，可能是成功的（某些接口不返回 result）
|
||
print(f" ℹ️ 未找到 result 字段，可能已成功")
|
||
return True, "可能成功"
|
||
else:
|
||
return False, f"HTTP {response.status_code}"
|
||
except Exception as e:
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False, f"异常: {e}"
|
||
|
||
|
||
def main():
|
||
print("🎮 Flash 游戏批量使用物品脚本")
|
||
print("="*60)
|
||
|
||
# 从文件读取 Cookie
|
||
cookie = read_cookie()
|
||
if not cookie:
|
||
print("❌ 无法读取 Cookie，请确保游戏已登录")
|
||
sys.exit(1)
|
||
|
||
print(f"🔑 Cookie: {cookie}")
|
||
print("⚠️ 需要在游戏中保持登录状态")
|
||
|
||
# 物品实例ID（从命令行参数获取）
|
||
if len(sys.argv) >= 2:
|
||
item_instance_id = int(sys.argv[1])
|
||
else:
|
||
print("\n❌ 缺少参数")
|
||
print("用法: python3 use_item.py <物品实例ID> <物品类型ID> <使用数量>")
|
||
print("示例: python3 use_item.py 1951 61 10")
|
||
print("\n说明:")
|
||
print(" 物品实例ID: 背包中物品的唯一ID（需要从游戏中获取）")
|
||
print(" 物品类型ID: 物品的类型ID")
|
||
print(" 使用数量: 一次性使用的数量（不能超过背包中的数量）")
|
||
sys.exit(1)
|
||
|
||
# 物品类型ID
|
||
if len(sys.argv) >= 3:
|
||
item_type_id = int(sys.argv[2])
|
||
else:
|
||
print("\n❌ 缺少物品类型ID")
|
||
print("用法: python3 use_item.py <物品实例ID> <物品类型ID> <使用数量>")
|
||
sys.exit(1)
|
||
|
||
# 使用数量
|
||
if len(sys.argv) >= 4:
|
||
quantity = int(sys.argv[3])
|
||
else:
|
||
print("\n💡 请输入使用数量")
|
||
while True:
|
||
try:
|
||
count_input = input("使用数量: ").strip()
|
||
quantity = int(count_input) if count_input else 1
|
||
if quantity > 0:
|
||
break
|
||
print("❌ 请输入大于0的数字")
|
||
except ValueError:
|
||
print("❌ 请输入有效的数字")
|
||
|
||
print("\n" + "="*60)
|
||
print(f"🎁 物品实例ID: {item_instance_id}")
|
||
print(f"🏷️ 物品类型ID: {item_type_id}")
|
||
print(f"🔢 使用数量: {quantity}")
|
||
print("="*60)
|
||
print("\n⚠️ 注意：这是批量使用，会一次性使用指定数量的物品")
|
||
print("⚠️ 如果背包中物品数量不足，服务端会返回错误")
|
||
|
||
# 创建 Session
|
||
session = requests.Session()
|
||
|
||
request_id = 1
|
||
|
||
print("\n" + "="*60)
|
||
print("🚀 开始使用物品")
|
||
print("="*60)
|
||
|
||
success, message = use_batch_item(session, cookie, item_instance_id, item_type_id, quantity, request_id)
|
||
|
||
if success:
|
||
print(f"\n✅ 使用成功！已使用 {quantity} 个物品")
|
||
else:
|
||
print(f"\n❌ 使用失败：{message}")
|
||
|
||
print("\n" + "="*60)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|