buff3.0/steam_confirmation.py

#!/usr/bin/env python3
"""
Steam 订单确认 Python 客户端

使用说明：
1. 首次使用需要先登录（调用 login 函数）
2. 登录后会保存 session 到本地文件
3. 之后可以直接查询确认列表或执行确认操作

依赖安装：
    pip install requests

示例：
    python steam_confirmation.py
"""

import json
import os
import sys
import requests
from typing import Optional

# ============ 配置区域 ============
API_BASE_URL = "http://localhost:5000"
MA_FILE_PATH = r"C:\Users\Administrator\AppData\Local\steam-msg\tku24068.maFile"

# 如果需要代理，取消下面这行的注释并填写
# PROXY_URL = "http://127.0.0.1:7890"
PROXY_URL = None

# Session 保存路径
SESSION_FILE = "steam_session.json"
# ============ 配置区域结束 ============


def load_session() -> Optional[dict]:
    """从文件加载保存的 session"""
    if os.path.exists(SESSION_FILE):
        try:
            with open(SESSION_FILE, 'r', encoding='utf-8') as f:
                return json.load(f)
        except Exception as e:
            print(f"加载 session 失败: {e}")
    return None


def save_session(session_data: dict):
    """保存 session 到文件"""
    try:
        with open(SESSION_FILE, 'w', encoding='utf-8') as f:
            json.dump(session_data, f, indent=2, ensure_ascii=False)
        print("Session 已保存")
    except Exception as e:
        print(f"保存 session 失败: {e}")


def login(username: str, password: str) -> Optional[dict]:
    """
    登录 Steam 账号

    Args:
        username: Steam 用户名
        password: Steam 密码

    Returns:
        session 数据，包含 steamId, accessToken, refreshToken, sessionId
    """
    print(f"\n{'='*50}")
    print("正在登录 Steam...")
    print(f"用户名: {username}")
    print(f"MaFile: {MA_FILE_PATH}")
    print(f"{'='*50}\n")

    url = f"{API_BASE_URL}/login"
    payload = {
        "Username": username,
        "Password": password,
        "MaFilePath": MA_FILE_PATH
    }

    if PROXY_URL:
        payload["ProxyUrl"] = PROXY_URL

    try:
        response = requests.post(url, json=payload, timeout=60)
        print(f"响应状态码: {response.status_code}")

        if response.status_code == 200:
            data = response.json()
            session = data.get("session", {})
            print(f"\n登录成功!")
            print(f"SteamID: {session.get('steamId')}")
            print(f"SessionID: {session.get('sessionId', 'N/A')[:20]}...")

            # 保存 session
            save_session(session)

            return session
        else:
            print(f"登录失败: {response.text}")
            return None

    except requests.exceptions.ConnectionError:
        print(f"\n错误: 无法连接到 steamAPI 服务")
        print(f"请确保 steamAPI 服务正在运行 (默认: http://localhost:5000)")
        return None
    except requests.exceptions.Timeout:
        print("\n错误: 请求超时，请检查网络连接")
        return None
    except Exception as e:
        print(f"\n错误: {e}")
        return None


def get_confirmation_list(session: dict) -> Optional[list]:
    """
    获取待确认的交易/市场订单列表

    Args:
        session: 登录后获取的 session 数据

    Returns:
        确认列表
    """
    print(f"\n{'='*50}")
    print("正在获取待确认列表...")
    print(f"{'='*50}\n")

    url = f"{API_BASE_URL}/trade-confirmations/list"
    payload = {
        "MaFilePath": MA_FILE_PATH,
        "Session": {
            "SteamId": session["steamId"],
            "AccessToken": session["accessToken"],
            "RefreshToken": session["refreshToken"],
            "SessionId": session.get("sessionId")
        }
    }

    if PROXY_URL:
        payload["ProxyUrl"] = PROXY_URL

    try:
        response = requests.post(url, json=payload, timeout=30)
        print(f"响应状态码: {response.status_code}")

        if response.status_code == 200:
            data = response.json()
            confirmations = data.get("confirmations", [])
            print(f"\n找到 {len(confirmations)} 个待确认项:")

            # 更新 session
            if "session" in data:
                save_session(data["session"])

            if confirmations:
                print("\n" + "-" * 60)
                for i, conf in enumerate(confirmations, 1):
                    print(f"\n[{i}] 类型: {conf.get('type', 'Unknown')}")
                    print(f"    ID: {conf.get('id')}")
                    print(f"    Key: {conf.get('key')}")
                    print(f"    标题: {conf.get('headline', 'N/A')}")
                    print(f"    摘要: {conf.get('summary', 'N/A')}")
                print("\n" + "-" * 60)
            else:
                print("没有待确认的订单")

            return confirmations
        elif response.status_code == 401:
            print("Session 已过期，需要重新登录")
            return None
        else:
            print(f"获取失败: {response.text}")
            return None

    except requests.exceptions.ConnectionError:
        print(f"\n错误: 无法连接到 steamAPI 服务")
        return None
    except Exception as e:
        print(f"\n错误: {e}")
        return None


def operate_confirmation(
    session: dict,
    confirmation_id: int,
    operation: str = "accept",
    confirmation_key: Optional[int] = None
) -> bool:
    """
    执行确认操作（确认或取消）

    Args:
        session: 登录后获取的 session 数据
        confirmation_id: 确认项 ID（支持传入 creator_id 自动匹配）
        operation: 操作类型，"accept" 或 "cancel"
        confirmation_key: 可选，手动指定确认 key

    Returns:
        操作是否成功
    """
    if operation not in ("accept", "cancel"):
        print("操作类型错误，只能是 'accept' 或 'cancel'")
        return False

    op_text = "确认" if operation == "accept" else "取消"
    print(f"\n{'='*50}")
    print(f"正在执行 {op_text} 操作...")
    print(f"确认项 ID: {confirmation_id}")
    print(f"{'='*50}\n")

    url = f"{API_BASE_URL}/trade-confirmations/operate"
    payload = {
        "MaFilePath": MA_FILE_PATH,
        "Session": {
            "SteamId": session["steamId"],
            "AccessToken": session["accessToken"],
            "RefreshToken": session["refreshToken"],
            "SessionId": session.get("sessionId")
        },
        "ConfirmationId": confirmation_id,
        "Operation": operation
    }

    if confirmation_key:
        payload["ConfirmationKey"] = confirmation_key

    if PROXY_URL:
        payload["ProxyUrl"] = PROXY_URL

    try:
        response = requests.post(url, json=payload, timeout=30)
        print(f"响应状态码: {response.status_code}")

        if response.status_code == 200:
            data = response.json()
            success = data.get("success", False)
            print(f"\n{'='*50}")
            if success:
                print(f"{op_text}成功!")
            else:
                print(f"{op_text}失败!")
            print(f"{'='*50}")
            print(f"操作类型: {data.get('operation')}")
            print(f"确认ID: {data.get('confirmationId')}")
            print(f"Key: {data.get('confirmationKey')}")
            print(f"Key来源: {'列表匹配' if data.get('keyResolvedFromList') else '手动指定'}")
            print(f"ID来源: {data.get('matchedBy', 'N/A')}")

            # 更新 session
            if "session" in data:
                save_session(data["session"])

            return success
        elif response.status_code == 404:
            print("未找到对应的确认项，可能已过期或被处理")
            return False
        elif response.status_code == 401:
            print("Session 已过期，需要重新登录")
            return False
        else:
            print(f"操作失败: {response.text}")
            return False

    except requests.exceptions.ConnectionError:
        print(f"\n错误: 无法连接到 steamAPI 服务")
        return False
    except Exception as e:
        print(f"\n错误: {e}")
        return False


def confirm_all(session: dict) -> int:
    """
    确认所有待确认的订单

    Returns:
        成功确认的数量
    """
    confirmations = get_confirmation_list(session)
    if not confirmations:
        print("没有待确认的订单")
        return 0

    success_count = 0
    for conf in confirmations:
        conf_id = conf.get("id")
        conf_type = conf.get("type", "Unknown")
        headline = conf.get("headline", "N/A")

        print(f"\n正在确认: {headline} (类型: {conf_type})")
        if operate_confirmation(session, conf_id, "accept"):
            success_count += 1

    return success_count


def interactive_mode():
    """交互式模式"""
    print("\n" + "=" * 60)
    print(" Steam 订单确认工具 - 交互式模式")
    print("=" * 60)

    # 尝试加载已保存的 session
    session = load_session()

    while True:
        print("\n" + "-" * 40)
        print("请选择操作:")
        print("1. 查看待确认列表")
        print("2. 确认指定订单")
        print("3. 确认所有订单")
        print("4. 取消指定订单")
        print("5. 重新登录")
        print("0. 退出")
        print("-" * 40)

        choice = input("请输入选项 (0-5): ").strip()

        if choice == "0":
            print("退出程序")
            break

        if choice == "1":
            if not session:
                print("请先登录!")
                continue
            get_confirmation_list(session)
            # 更新 session
            new_session = load_session()
            if new_session:
                session = new_session

        elif choice == "2":
            if not session:
                print("请先登录!")
                continue
            confirm_id = input("请输入要确认的订单 ID: ").strip()
            try:
                confirm_id = int(confirm_id)
            except ValueError:
                print("ID 必须是数字")
                continue
            operate_confirmation(session, confirm_id, "accept")
            # 更新 session
            new_session = load_session()
            if new_session:
                session = new_session

        elif choice == "3":
            if not session:
                print("请先登录!")
                continue
            confirm("确认所有待确认的订单? (y/N): ")
            if confirm.lower() == 'y':
                success_count = confirm_all(session)
                print(f"\n成功确认 {success_count} 个订单")
                # 更新 session
                new_session = load_session()
                if new_session:
                    session = new_session

        elif choice == "4":
            if not session:
                print("请先登录!")
                continue
            confirm_id = input("请输入要取消的订单 ID: ").strip()
            try:
                confirm_id = int(confirm_id)
            except ValueError:
                print("ID 必须是数字")
                continue
            operate_confirmation(session, confirm_id, "cancel")
            # 更新 session
            new_session = load_session()
            if new_session:
                session = new_session

        elif choice == "5":
            username = input("请输入 Steam 用户名: ").strip()
            password = input("请输入 Steam 密码: ").strip()
            if username and password:
                session = login(username, password)
            else:
                print("用户名和密码不能为空")

        else:
            print("无效的选项")


def confirm(prompt: str) -> str:
    """简单的确认提示"""
    return input(prompt)


if __name__ == "__main__":
    # 检查 maFile 是否存在
    if not os.path.exists(MA_FILE_PATH):
        print(f"\n错误: MaFile 文件不存在!")
        print(f"路径: {MA_FILE_PATH}")
        print("\n请检查配置文件中的 MA_FILE_PATH 是否正确")
        sys.exit(1)

    # 检查 API 服务是否可用
    try:
        response = requests.get(f"{API_BASE_URL}/", timeout=5)
    except requests.exceptions.ConnectionError:
        print(f"\n警告: 无法连接到 steamAPI 服务 (http://localhost:5000)")
        print("请确保 steamAPI 服务正在运行")
        print("\n启动 steamAPI 的方法:")
        print("  cd steam-client/steamAPI")
        print("  dotnet run")
        print("\n如果需要指定端口:")
        print("  dotnet run --urls \"http://localhost:5000\"")
        print()

    # 检查命令行参数
    if len(sys.argv) > 1:
        if sys.argv[1] == "--login" and len(sys.argv) >= 4:
            # python steam_confirmation.py --login username password
            session = login(sys.argv[2], sys.argv[3])
            if session:
                print("\n登录成功，现在可以执行其他操作")
                confirmations = get_confirmation_list(session)
                if confirmations:
                    print(f"\n找到 {len(confirmations)} 个待确认订单")
        elif sys.argv[1] == "--list":
            session = load_session()
            if session:
                get_confirmation_list(session)
            else:
                print("没有保存的 session，请先登录")
        elif sys.argv[1] == "--confirm" and len(sys.argv) >= 3:
            session = load_session()
            if session:
                confirm_id = int(sys.argv[2])
                operate_confirmation(session, confirm_id, "accept")
            else:
                print("没有保存的 session，请先登录")
        elif sys.argv[1] == "--confirm-all":
            session = load_session()
            if session:
                success_count = confirm_all(session)
                print(f"\n成功确认 {success_count} 个订单")
            else:
                print("没有保存的 session，请先登录")
        else:
            print_usage()
    else:
        # 默认进入交互式模式
        interactive_mode()


def print_usage():
    """打印使用说明"""
    print("""
使用说明:

  1. 登录 (需要用户名和密码):
     python steam_confirmation.py --login <username> <password>

  2. 查看待确认列表:
     python steam_confirmation.py --list

  3. 确认指定订单:
     python steam_confirmation.py --confirm <confirmation_id>

  4. 确认所有订单:
     python steam_confirmation.py --confirm-all

  5. 进入交互式模式 (不带参数):
     python steam_confirmation.py

注意事项:
  - 首次使用必须先登录
  - Session 会保存到 steam_session.json
  - 如果 session 过期，需要重新登录
""")