From 23b5cde07fa5e6feac033d5b8e6e61742e711718 Mon Sep 17 00:00:00 2001 From: mac-zhou Date: Tue, 19 Mar 2024 19:46:44 +0800 Subject: [PATCH] Add IpLocationPlugin to PluginManager --- bot/plugin_manager.py | 4 +++- bot/plugins/iplocation.py | 44 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) create mode 100644 bot/plugins/iplocation.py diff --git a/bot/plugin_manager.py b/bot/plugin_manager.py index 370b1c3e..6cd62e80 100644 --- a/bot/plugin_manager.py +++ b/bot/plugin_manager.py @@ -15,6 +15,7 @@ from plugins.worldtimeapi import WorldTimeApiPlugin from plugins.whois_ import WhoisPlugin from plugins.webshot import WebshotPlugin +from plugins.iplocation import IpLocationPlugin class PluginManager: @@ -40,6 +41,7 @@ def __init__(self, config): 'auto_tts': AutoTextToSpeech, 'whois': WhoisPlugin, 'webshot': WebshotPlugin, + 'iplocation': IpLocationPlugin, } self.plugins = [plugin_mapping[plugin]() for plugin in enabled_plugins if plugin in plugin_mapping] @@ -69,4 +71,4 @@ def get_plugin_source_name(self, function_name) -> str: def __get_plugin_by_function_name(self, function_name): return next((plugin for plugin in self.plugins - if function_name in map(lambda spec: spec.get('name'), plugin.get_spec())), None) + if function_name in map(lambda spec: spec.get('name'), plugin.get_spec())), None) diff --git a/bot/plugins/iplocation.py b/bot/plugins/iplocation.py new file mode 100644 index 00000000..6933ef86 --- /dev/null +++ b/bot/plugins/iplocation.py @@ -0,0 +1,44 @@ +import requests +from typing import Dict + +from .plugin import Plugin + +class IpLocationPlugin(Plugin): + """ + A plugin to get geolocation and other information for a given IP address + """ + + def get_source_name(self) -> str: + return "IP.FM" + + def get_spec(self) -> [Dict]: + return [{ + "name": "iplocaion", + "description": "Get information for an IP address using the IP.FM API.", + "parameters": { + "type": "object", + "properties": { + "ip": {"type": "string", "description": "IP Address"} + }, + "required": ["ip"], + }, + }] + + async def execute(self, function_name, helper, **kwargs) -> Dict: + ip = kwargs.get('ip') + BASE_URL = "https://api.ip.fm/?ip={}" + url = BASE_URL.format(ip) + try: + response = requests.get(url) + response_data = response.json() + country = response_data.get('data', {}).get('country', "None") + subdivisions = response_data.get('data', {}).get('subdivisions', "None") + city = response_data.get('data', {}).get('city', "None") + location = ', '.join(filter(None, [country, subdivisions, city])) or "None" + + asn = response_data.get('data', {}).get('asn', "None") + as_name = response_data.get('data', {}).get('as_name', "None") + as_domain = response_data.get('data', {}).get('as_domain', "None") + return {"Location": location, "ASN": asn, "AS Name": as_name, "AS Domain": as_domain} + except Exception as e: + return {"Error": str(e)}