PagerMaid_Plugins_Pyro/gfw/main.py

from pagermaid.listener import listener
from pagermaid.enums import Message
from pagermaid.services import client as requests

import socket


def is_ip(ip):
    try:
        socket.inet_aton(ip)
        return True
    except Exception:
        return False


def get_ip(domain):
    try:
        ip = socket.gethostbyname(domain)
    except Exception:
        return None
    return ip


async def post(host):
    url = "https://api.potatonet.idc.wiki/network/simple_health_check/scripts/gfw_check"
    data = {"host": host}
    return await requests.post(url, data=data)


@listener(command="gfw", parameters="[text]", description="查询是否被墙")
async def gfw(message: Message):
    text = None
    if message.arguments:
        text = message.arguments
    elif message.reply_to_message:
        text = message.reply_to_message.text
    if not text:
        return await message.edit("请输入或回复一个IP或域名。")
    try:
        if not is_ip(text):
            if get_ip(text) is None:
                return await message.edit("出错了呜呜呜 ~ 无效的参数。")
            else:
                text = get_ip(text)
        try:
            message: Message = await message.edit(f"{text} 查询中...")
        except Exception:
            return await message.edit("出错了呜呜呜 ~ 无效的参数。")
        data = await post(text)
        data = data.json()
    except Exception as e:
        return await message.edit(f"出错了呜呜呜 ~ 查询失败。{e}")

    if data["success"]:
        if (
            data["data"]["tcp"]["cn"] == data["data"]["tcp"]["!cn"]
            and data["data"]["icmp"]["cn"] == data["data"]["icmp"]["!cn"]
        ):
            if (
                data["data"]["tcp"]["cn"] == False
                and data["data"]["icmp"]["cn"] == False
            ):
                reply = f"IP: {text}\n状态: 全球不通，不能判断是否被墙"
            else:
                reply = f"IP: {text}\n状态: 未被墙"
        else:
            reply = f"IP: {text}\n状态: 被墙"
    else:
        reply = f"IP: {text}\n状态: 查询失败"
    return await message.edit(reply)