added shareus, bitly, tinyurl support release v0.0.5 ✨

RknDeveloper · RknDeveloper · commit 26e23d6c62b8 · 2025-08-29T21:05:57.000+05:45
diff --git a/README.md b/README.md
@@ -14,7 +14,7 @@ To Upgrade
 pip install --upgrade link-shortly
 ```
 
-## Quick Usage Example ( supports sync )
+## Quick Usage Example
 ```python
 from shortly import Shortly
 
@@ -28,24 +28,6 @@ if __name__ == "__main__":
     main()
 ```
 
-## Quick Usage Example ( supports async  )
-```python
-import asyncio
-from shortly import Shortly
-
-# Initialize Shortly
-shortly = Shortly(api_key='<YOUR API KEY>', base_url='<YOUR BASE SITE>')
-
-async def main():
-    # Async call to convert URL
-    link = await shortly.convert("https://example.com/long-url")
-    print(link)
-
-if __name__ == "__main__":
-    asyncio.run(main())
-```
-
-
 ## Error Handling
 
 The library comes with built-in exception handling to manage common errors such as invalid links, not found links, timeouts, or connection issues.
diff --git a/src/shortly/__init__.py b/src/shortly/__init__.py
@@ -9,7 +9,7 @@
 """
 
 
-__version__ = "0.0.4"
+__version__ = "0.0.5"
 __author__ = "RknDeveloper"
 __license__ = "MIT License "
 __copyright__ = "Copyright (C) 2025-present RknDeveloper <https://github.com/RknDeveloper>"
diff --git a/src/shortly/shortly.py b/src/shortly/shortly.py
@@ -10,7 +10,13 @@
 
 import asyncio
 import functools
-from .utils import convert as _utils_convert
+from .utils import (
+     adlinkfy_convert,
+     bitly_convert,
+     tinyurl_convert,
+     shareus_convert
+)
+
 from .exceptions import (
     ShortlyValueError
 )
@@ -34,12 +40,18 @@ def __init__(self, api_key=None, base_url=None):
         """
         if not base_url or not isinstance(base_url, str):
             raise ShortlyValueError("base_url must be a non-empty string")
-        if not api_key or not isinstance(api_key, str):
-            raise ShortlyValueError("api_key must be a non-empty string")
-            
-        self.api_key = api_key
+        
         self.base_url = (urlparse(base_url).netloc or urlparse(base_url).path).rstrip("/")
 
+        if self.base_url == "tinyurl.com":
+            # api_key optional
+            self.api_key = api_key
+        else:
+            # api_key required
+            if not api_key or not isinstance(api_key, str):
+                raise ShortlyValueError(f"api_key must be a non-empty string for {self.base_url}")
+            self.api_key = api_key
+            
     # Internal async method calling utils.convert
     async def _convert_async(self, link, alias=None, silently=False, timeout=10):
         """
@@ -54,7 +66,16 @@ async def _convert_async(self, link, alias=None, silently=False, timeout=10):
         Output:
             Returns shortened link or error response from utils.convert
         """
-        return await _utils_convert(self, link, alias, silently, timeout)
+        if self.base_url == "tinyurl.com":
+            self.shortner = await tinyurl_convert(self, link, alias, silently, timeout)
+        elif self.base_url == "shareus.io":
+            self.shortner = await shareus_convert(self, link, alias, silently, timeout)   
+        elif self.base_url == "bitly.com":
+            self.shortner = await bitly_convert(self, link, alias, silently, timeout)        
+        else:
+            self.shortner = await adlinkfy_convert(self, link, alias, silently, timeout)
+            
+        return self.shortner
 
 
 # -------------------------------
diff --git a/src/shortly/utils.py b/src/shortly/utils.py
@@ -20,7 +20,7 @@
 )
 
 
-async def convert(self, link, alias=None, silently=False, timeout=10):
+async def adlinkfy_convert(self, link, alias=None, silently=False, timeout=10):
     """
     Shorten a URL using Link Shortly/All Adlinkfy API.
 
@@ -86,4 +86,252 @@ async def convert(self, link, alias=None, silently=False, timeout=10):
     except aiohttp.ClientConnectionError:
         raise ShortlyConnectionError(f"Failed to connect to {self.base_url}.")
     except Exception as e:
-        raise ShortlyError(f"An unexpected error occurred: {e}")
+        raise ShortlyError(f"An unexpected error occurred: {e}")
+        
+async def shareus_convert(self, link, alias=None, silently=False, timeout=10):
+    """
+    Shorten a URL using Shareus.io API.
+
+    Parameters:
+        api_key (str): Your API key for the Shareus service.
+        base_url (str): The domain of the API (should be "shareus.io").
+        link (str): The long URL you want to shorten.
+        alias (str, optional): Custom alias for the short link. Default is None.
+        silently (bool): If True, the function will directly return the original URL without raising errors.
+        timeout (int, optional): Maximum seconds to wait for API response. Default is 10.
+
+    Returns:
+        str: The shortened URL returned by the API.
+
+    Raises:
+        ShortlyInvalidLinkError: If the provided link is invalid or malformed.
+        ShortlyLinkNotFoundError: If the short link does not exist or has expired.
+        ShortlyTimeoutError: If request exceeds the allowed timeout.
+        ShortlyConnectionError: If cannot connect to API server.
+        ShortlyError: For other API-related errors.
+    """
+    if silently:
+        return link
+
+    # Shareus.io API configuration
+    api_url = f"https://api.{self.base_url}/easy_api"
+    params = {"key": self.api_key, "link": link}
+    
+    if alias:
+        params["alias"] = alias
+
+    try:
+        async with aiohttp.ClientSession() as session:
+            headers = {
+                "User-Agent": (
+                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                    "AppleWebKit/537.36 (KHTML, like Gecko) "
+                    "Chrome/91.0.4472.124 Safari/537.36"
+                )
+            }
+            
+            async with session.get(api_url, params=params, headers=headers, timeout=timeout) as response:
+                if response.status != 200:
+                    raise ShortlyError(f"Failed to shorten your link (HTTP {response.status}).")
+                
+                # Shareus returns plain text response
+                data_text = await response.text()
+                
+                if data_text == "settings not saved":
+                    raise ShortlyError("Settings not saved or invalid API key.")
+                
+                # Check if response looks like an error
+                if any(error_indicator in data_text.lower() for error_indicator in ["error", "invalid", "not found", "failed"]):
+                    if "invalid" in data_text.lower():
+                        raise ShortlyInvalidLinkError(data_text)
+                    elif "not found" in data_text.lower() or "expired" in data_text.lower():
+                        raise ShortlyLinkNotFoundError(data_text)
+                    else:
+                        raise ShortlyError(data_text)
+                
+                # If we get here, it's likely a successful shortening
+                return data_text
+
+    except asyncio.TimeoutError:
+        raise ShortlyTimeoutError(f"Request timed out after {timeout} seconds.")
+    except aiohttp.ClientConnectionError:
+        raise ShortlyConnectionError(f"Failed to connect to {self.base_url}.")
+    except json.JSONDecodeError:
+        # This is expected for Shareus as it returns plain text, not JSON
+        raise ShortlyJsonDecodeError(f"Json Error: {e}")
+    except Exception as e:
+        raise ShortlyError(f"An unexpected error occurred: {e}")
+        
+async def tinyurl_convert(self, link, alias=None, silently=False, timeout=10):
+    """
+    Shorten a URL using TinyURL API - supports both old (no token) and new (with token) APIs
+    """
+    if silently:
+        return link
+
+    # Check if we have a token to decide which API to use
+    if hasattr(self, 'api_key') and self.api_key:
+        return await _tinyurl_new_api(self, link, alias, timeout)
+    else:
+        return await _tinyurl_old_api(self, link, alias, timeout)
+
+async def _tinyurl_old_api(self, link, alias=None, timeout=10):
+    """
+    Old TinyURL API (no token required)
+    Docs: http://tinyurl.com/api-create.php
+    """
+    api_url = f"http://{self.base_url}/api-create.php"
+    params = {"url": link}
+    
+    if alias:
+        params["alias"] = alias
+
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(api_url, params=params, timeout=timeout) as response:
+                if response.status != 200:
+                    raise ShortlyError(f"Failed to shorten link (TinyURL error: {response.status}).")
+                
+                shortened_url = await response.text()
+                
+                if "Error:" in shortened_url:
+                    raise ShortlyError(f"TinyURL error: {shortened_url}")
+                
+                return shortened_url.strip()
+                
+    except asyncio.TimeoutError:
+        raise ShortlyTimeoutError(f"TinyURL request timed out after {timeout} seconds.")
+    except aiohttp.ClientConnectionError:
+        raise ShortlyConnectionError("Failed to connect to TinyURL.")
+    except Exception as e:
+        raise ShortlyError(f"TinyURL unexpected error: {e}")
+
+async def _tinyurl_new_api(self, link, alias=None, timeout=10):
+    """
+    New TinyURL API (requires token)
+    Docs: https://tinyurl.com/app/dev/api
+    """
+    
+
+    api_url = f"https://api.{self.base_url}/create"  
+    headers = {  
+        "Authorization": f"Bearer {self.api_key}",  
+        "Content-Type": "application/json"  
+    }  
+    
+    payload = {  
+        "url": link,  
+        "domain": "tinyurl.com"  
+    }  
+    
+    if alias:  
+        payload["alias"] = alias  
+
+    try:  
+        async with aiohttp.ClientSession() as session:  
+            async with session.post(  
+                api_url, 
+                json=payload, 
+                headers=headers, 
+                timeout=timeout  
+            ) as response:  
+                
+                # 🔹 Safe JSON parsing with better error handling
+                try:  
+                    response_data = await response.json(content_type=None)  
+                except Exception:  
+                    response_text = await response.text()
+                    response_data = {"raw": response_text}
+                    
+                    # Handle case where alias already exists (returns string)
+                    if "alias already exists" in response_text.lower():
+                        raise ShortlyError("Custom alias already exists. Please choose a different one.")
+
+                if response.status == 200 and "data" in response_data:  
+                    return response_data["data"]["tiny_url"]  
+
+                elif response.status == 401:  
+                    raise ShortlyError("Invalid TinyURL API token.")  
+
+                elif response.status == 422:  
+                    error_msg = "Unknown validation error"  
+                    if isinstance(response_data, dict):  
+                        errors = response_data.get("errors")  
+                        if isinstance(errors, list) and errors:  
+                            error_msg = errors[0].get("message", error_msg)  
+                    elif isinstance(response_data, str):  
+                        error_msg = response_data  
+                    raise ShortlyError(f"TinyURL validation error: {error_msg}")  
+
+                elif response.status == 429:  
+                    raise ShortlyError("TinyURL rate limit exceeded. Please try again later.")  
+
+                else:  
+                    # Handle string responses gracefully
+                    if isinstance(response_data, str):
+                        error_message = response_data
+                    elif isinstance(response_data, dict) and "raw" in response_data:
+                        error_message = response_data["raw"]
+                    else:
+                        error_message = str(response_data)
+                    
+                    raise ShortlyError(f"TinyURL API error: {response.status} - {error_message}")  
+                
+    except asyncio.TimeoutError:  
+        raise ShortlyTimeoutError(f"TinyURL request timed out after {timeout} seconds.")
+    except aiohttp.ClientConnectionError:  
+        raise ShortlyConnectionError("Failed to connect to TinyURL API.")
+    except Exception as e:  
+        raise ShortlyError(f"TinyURL unexpected error: {e}")
+
+async def bitly_convert(self, link, alias=None, silently=False, timeout=10):
+    """
+    Bitly API का उपयोग करके URL को छोटा (shorten) करें।
+    Docs: https://dev.bitly.com/api-reference
+    """
+    if silently:
+        return link
+
+    api_url = f"https://api-ssl.{self.base_url}/v4/shorten"
+    headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
+    payload = {"long_url": link}
+    
+    # Correct way to add custom alias
+    if alias:      
+        payload["custom_bitlink"] = alias
+       
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.post(api_url, headers=headers, json=payload, timeout=timeout) as response:
+                
+                # Different error cases handle करें
+                if response.status == 401:
+                    raise ShortlyError("Invalid Bitly API key")
+                elif response.status == 400:
+                    error_text = await response.text()
+                    if "CUSTOM_BITLINK_ALREADY_EXISTS" in error_text:
+                        raise ShortlyError(f"Custom alias '{alias}' already exists on Bitly")
+                    elif "INVALID_CUSTOM_BITLINK" in error_text:
+                        raise ShortlyError(f"Invalid custom alias: {alias}")
+                    else:
+                        raise ShortlyError("Invalid URL provided to Bitly")
+                elif response.status == 403:
+                    raise ShortlyError("Bitly API permission denied")
+                elif response.status == 409:
+                    raise ShortlyError("Custom alias already exists")
+                elif response.status != 200 and response.status != 201:
+                    raise ShortlyError(f"Bitly error: {await response.text()}")
+                
+                try:
+                    data = await response.json()
+                except Exception as e:
+                    raise ShortlyJsonDecodeError(f"Invalid JSON from Bitly: {e}")
+                    
+                return data.get("link")
+                
+    except asyncio.TimeoutError:
+        raise ShortlyTimeoutError(f"Bitly request timed out after {timeout} seconds.")
+    except aiohttp.ClientConnectionError:
+        raise ShortlyConnectionError("Failed to connect to Bitly.")
+    except Exception as e:
+        raise ShortlyError(f"Bitly unexpected error: {e}")
