99 lines
2.6 KiB
Python
99 lines
2.6 KiB
Python
# Idk who made this i was provided it by louis
|
|
import asyncio
|
|
from typing import Dict
|
|
|
|
import httpx
|
|
|
|
"""
|
|
Minimal usage:
|
|
|
|
from dsql_shortlinks import generate_short_link_sync
|
|
|
|
result = generate_short_link_sync(
|
|
"https://example.com/product",
|
|
api_key="YOUR_DSQL_KEY",
|
|
)
|
|
|
|
print(result["short_url"])
|
|
"""
|
|
|
|
|
|
DEFAULT_BASE_URL = "https://api.profitlounge.us/link/affiliate"
|
|
|
|
|
|
class DsqlShortlinks:
|
|
def __init__(
|
|
self,
|
|
api_key: str,
|
|
*,
|
|
base_url: str = DEFAULT_BASE_URL,
|
|
timeout: int = 60,
|
|
) -> None:
|
|
self.api_key = api_key
|
|
self.base_url = base_url
|
|
self.timeout = timeout
|
|
|
|
async def create_short_link(self, url: str) -> Dict[str, str]:
|
|
headers = {"x-api-key": self.api_key}
|
|
payload = {"url": url}
|
|
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
self.base_url,
|
|
headers=headers,
|
|
json=payload,
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
if isinstance(data, dict) and data.get("short_url"):
|
|
return data
|
|
except Exception:
|
|
pass
|
|
|
|
return {"short_url": url, "long_url": url}
|
|
|
|
def create_short_link_sync(self, url: str) -> Dict[str, str]:
|
|
try:
|
|
running_loop = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
running_loop = None
|
|
|
|
if running_loop and running_loop.is_running():
|
|
raise RuntimeError(
|
|
"create_short_link_sync() cannot run inside an active event loop. "
|
|
"Use await create_short_link(...) instead."
|
|
)
|
|
|
|
return asyncio.run(self.create_short_link(url))
|
|
|
|
|
|
async def generate_short_link(
|
|
url: str, api_key: str, timeout: int = 60
|
|
) -> Dict[str, str]:
|
|
client = DsqlShortlinks(api_key, timeout=timeout)
|
|
return await client.create_short_link(url)
|
|
|
|
|
|
def generate_short_link_sync(
|
|
url: str, api_key: str, timeout: int = 60
|
|
) -> Dict[str, str]:
|
|
client = DsqlShortlinks(api_key, timeout=timeout)
|
|
return client.create_short_link_sync(url)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
EXAMPLE_URL = "https://woot.com/"
|
|
EXAMPLE_API_KEY = "YOUR_DSQL_KEY"
|
|
|
|
print("Example:")
|
|
print("from dsql_shortlinks import generate_short_link_sync")
|
|
print()
|
|
print("result = generate_short_link_sync(")
|
|
print(f' "{EXAMPLE_URL}",')
|
|
print(f' api_key="{EXAMPLE_API_KEY}",')
|
|
print(")")
|
|
print()
|
|
print('print(result["short_url"])')
|