mirror of
https://github.com/aljazceru/chatgpt-telegram-bot.git
synced 2025-12-21 22:54:52 +01:00
Add IpLocationPlugin to PluginManager
This commit is contained in:
@@ -15,6 +15,7 @@ from plugins.deepl import DeeplTranslatePlugin
|
|||||||
from plugins.worldtimeapi import WorldTimeApiPlugin
|
from plugins.worldtimeapi import WorldTimeApiPlugin
|
||||||
from plugins.whois_ import WhoisPlugin
|
from plugins.whois_ import WhoisPlugin
|
||||||
from plugins.webshot import WebshotPlugin
|
from plugins.webshot import WebshotPlugin
|
||||||
|
from plugins.iplocation import IpLocationPlugin
|
||||||
|
|
||||||
|
|
||||||
class PluginManager:
|
class PluginManager:
|
||||||
@@ -40,6 +41,7 @@ class PluginManager:
|
|||||||
'auto_tts': AutoTextToSpeech,
|
'auto_tts': AutoTextToSpeech,
|
||||||
'whois': WhoisPlugin,
|
'whois': WhoisPlugin,
|
||||||
'webshot': WebshotPlugin,
|
'webshot': WebshotPlugin,
|
||||||
|
'iplocation': IpLocationPlugin,
|
||||||
}
|
}
|
||||||
self.plugins = [plugin_mapping[plugin]() for plugin in enabled_plugins if plugin in plugin_mapping]
|
self.plugins = [plugin_mapping[plugin]() for plugin in enabled_plugins if plugin in plugin_mapping]
|
||||||
|
|
||||||
@@ -69,4 +71,4 @@ class PluginManager:
|
|||||||
|
|
||||||
def __get_plugin_by_function_name(self, function_name):
|
def __get_plugin_by_function_name(self, function_name):
|
||||||
return next((plugin for plugin in self.plugins
|
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)
|
||||||
|
|||||||
44
bot/plugins/iplocation.py
Normal file
44
bot/plugins/iplocation.py
Normal file
@@ -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)}
|
||||||
Reference in New Issue
Block a user