Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

_ #19

Merged
merged 4 commits into from
Nov 18, 2024
Merged

_ #19

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
39a4c14
Merge pull request #18 from TeamKillerX/test
xtsea Nov 14, 2024
b098372
Fix removed
xtsea Nov 14, 2024
05851dd
Fix return api
xtsea Nov 18, 2024
7476263
Fix missing
xtsea Nov 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 56 additions & 23 deletions akenoai/akeno.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ def __init__(self, dictionary):
def __repr__(self):
return f"{self.__dict__}"


class AkenoPlus:
def __init__(self, key=..., api_endpoint: str = "https://private-akeno.randydev.my.id"):
if key is Ellipsis:
Expand All @@ -32,6 +33,11 @@ def __init__(self, key=..., api_endpoint: str = "https://private-akeno.randydev.
self.api_endpoint = api_endpoint
self.headers = {"x-akeno-key": str(self.key)}

def api_akenoai(self, method):
if not method:
raise ValueError("Method parameter cannot be None or empty")
return f"{self.api_endpoint}/{method}"

def set_key(self, new_key: str):
self.key = new_key

Expand All @@ -48,54 +54,64 @@ async def clean(self, file_path: str):
return f"Error removing file {file_path}: {e}"

async def terabox(self, link=None):
url = self.api_akenoai(f"akeno/terabox-v1?link={link}")
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/terabox-v1?link={link}", headers=self.headers) as response:
async with session.get(url, headers=self.headers) as response:
return await response.json()

async def terabox_v2(self, link=None):
url = self.api_akenoai(f"akeno/terabox-v2?link={link}")
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/terabox-v2?link={link}", headers=self.headers) as response:
async with session.get(url, headers=self.headers) as response:
return await response.json()

async def chatgpt_old(self, query=None):
url = self.api_akenoai("ryuzaki/chatgpt-old")
payload = {"query": query}
async with aiohttp.ClientSession() as session:
async with session.post(f"{self.api_endpoint}/ryuzaki/chatgpt-old", json=payload) as response:
async with session.post(url, json=payload) as response:
return await response.json()

async def chatgpt_mode_web(self, query=None, **params):
url = self.api_akenoai("api/akeno-ai-web")
combined_params = {"query": query}
combined_params.update(params)
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/api/akeno-ai-web", params=combined_params) as response:
async with session.get(url, params=combined_params) as response:
return await response.json()

async def sites_torrens_all(self):
url = self.api_akenoai("akeno/sites_torrens_all")
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/sites_torrens_all") as response:
async with session.get(url) as response:
return await response.json()

async def search_for_torrents(self, **params):
url = self.api_akenoai("akeno/search_for_torrents")
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/search_for_torrents", params=params) as response:
async with session.get(url, params=params) as response:
return await response.json()

async def get_torrent_from_url(self, **params):
url = self.api_akenoai("akeno/get_torrent_from_url")
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/get_torrent_from_url", params=params) as response:
async with session.get(url, params=params) as response:
return await response.json()

async def get_recent(self, **params):
url = self.api_akenoai("akeno/get_recent")
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/get_recent", params=params) as response:
async with session.get(url, params=params) as response:
return await response.json()

async def get_category(self, **params):
url = self.api_akenoai("akeno/get_category")
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/get_category", params=params) as response:
async with session.get(url, params=params) as response:
return await response.json()

async def paal_see(self, files_open=None, **params):
url = self.api_akenoai("akeno/paal-see")
async with aiohttp.ClientSession() as session:
form_data = aiohttp.FormData()
form_data.add_field(
Expand All @@ -104,12 +120,13 @@ async def paal_see(self, files_open=None, **params):
filename=os.path.basename(files_open),
content_type='application/octet-stream'
)
async with session.post(f"{self.api_endpoint}/akeno/paal-see", data=form_data, params=params) as response:
async with session.post(url, data=form_data, params=params) as response:
if response.status != 200:
raise Exception(f"Error occurred: {response.status}")
return await response.json()

async def remini_enhancer(self, files_open=None):
url = self.api_akenoai("api/v2/remini/enhancer")
async with aiohttp.ClientSession() as session:
form_data = aiohttp.FormData()
form_data.add_field(
Expand All @@ -118,7 +135,7 @@ async def remini_enhancer(self, files_open=None):
filename=os.path.basename(files_open),
content_type='application/octet-stream'
)
async with session.post(f"{self.api_endpoint}/api/v2/remini/enhancer", data=form_data) as response:
async with session.post(url, data=form_data) as response:
if response.status != 200:
raise Exception(f"Error occurred: {response.status}")
file_path = "enchancer.jpg"
Expand All @@ -127,11 +144,13 @@ async def remini_enhancer(self, files_open=None):
return file_path

async def paal_text_to_image(self, **params):
url = self.api_akenoai("akeno/paal-text-to-image")
async with aiohttp.ClientSession() as session:
async with session.post(f"{self.api_endpoint}/akeno/paal-text-to-image", params=params, headers=self.headers) as response:
async with session.post(url, params=params, headers=self.headers) as response:
return await response.json()

async def google_video_to_text(self, files_open=None, **params):
url = self.api_akenoai("api/v2/google/video-to-text")
async with aiohttp.ClientSession() as session:
form_data = aiohttp.FormData()
form_data.add_field(
Expand All @@ -140,12 +159,13 @@ async def google_video_to_text(self, files_open=None, **params):
filename=os.path.basename(files_open),
content_type='application/octet-stream'
)
async with session.post(f"{self.api_endpoint}/api/v2/google/video-to-text", data=form_data, params=params) as response:
async with session.post(url, data=form_data, params=params) as response:
if response.status != 200:
raise Exception(f"Error occurred: {response.status}")
return await response.json()

async def google_image_to_text(self, files_open=None, **params):
url = self.api_akenoai("api/v2/google/image-to-text")
async with aiohttp.ClientSession() as session:
form_data = aiohttp.FormData()
form_data.add_field(
Expand All @@ -154,12 +174,13 @@ async def google_image_to_text(self, files_open=None, **params):
filename=os.path.basename(files_open),
content_type='application/octet-stream'
)
async with session.post(f"{self.api_endpoint}/api/v2/google/image-to-text", data=form_data, params=params) as response:
async with session.post(url, data=form_data, params=params) as response:
if response.status != 200:
raise Exception(f"Error occurred: {response.status}")
return await response.json()

async def google_audio_to_text(self, files_open=None, **params):
url = self.api_akenoai("api/v2/google/audio-to-text")
async with aiohttp.ClientSession() as session:
form_data = aiohttp.FormData()
form_data.add_field(
Expand All @@ -168,46 +189,58 @@ async def google_audio_to_text(self, files_open=None, **params):
filename=os.path.basename(files_open),
content_type='application/octet-stream'
)
async with session.post(f"{self.api_endpoint}/api/v2/google/audio-to-text", data=form_data, params=params) as response:
async with session.post(url, data=form_data, params=params) as response:
if response.status != 200:
raise Exception(f"Error occurred: {response.status}")
return await response.json()

async def blackbox(self, **payload):
params = {"query": query}
url = self.api_akenoai("ryuzaki/blackbox")
async with aiohttp.ClientSession() as session:
async with session.post(f"{self.api_endpoint}/ryuzaki/blackbox", json=payload, headers=self.headers) as response:
async with session.post(url, json=payload, headers=self.headers) as response:
return await response.json()

async def hentai(self):
url = self.api_akenoai("akeno/hentai")
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/hentai", headers=self.headers) as response:
async with session.get(url, headers=self.headers) as response:
return await response.json()

async def fbdown(self, link=None):
url = self.api_akenoai("akeno/fbdown-v2")
params = {"link": link}
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/fbdown-v2", params=params, headers=self.headers) as response:
async with session.get(url, params=params, headers=self.headers) as response:
return await response.json()

async def pinterest(self, **params):
url = self.api_akenoai("akeno/pinterest-v2")
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, headers=self.headers) as response:
return await response.json()

async def igdl(self, version=False, **params):
url_ig2 = self.api_akenoai("akeno/fastdl-ig-v2")
url = self.api_akenoai("akeno/fastdl-ig")
async with aiohttp.ClientSession() as session:
if version:
async with session.get(f"{self.api_endpoint}/akeno/fastdl-ig-v2", params=params) as response:
async with session.get(url_ig2, params=params) as response:
return await response.json()
else:
async with session.get(f"{self.api_endpoint}/akeno/fastdl-ig", params=params) as response:
async with session.get(url, params=params) as response:
return await response.json()

async def fdownloader(self, **params):
url = self.api_akenoai("akeno/fdownloader")
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/fdownloader", params=params, headers=self.headers) as response:
async with session.get(url, params=params, headers=self.headers) as response:
return await response.json()

async def capcut(self, link=None):
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (bug_risk): The url variable is used but never defined using api_akenoai

self.api_akenoai("akeno/capcut-v1")
params = {"link": link}
async with aiohttp.ClientSession() as session:
async with session.get(f"{self.api_endpoint}/akeno/capcut-v1", params=params, headers=self.headers) as response:
async with session.get(url, params=params, headers=self.headers) as response:
return await response.json()

async def add_ipblock(self, ip=None):
Expand Down