mirror of https://github.com/langgenius/dify.git
52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
import json
|
|
|
|
import httpx
|
|
|
|
from services.auth.api_key_auth_base import ApiKeyAuthBase
|
|
|
|
|
|
class FirecrawlAuth(ApiKeyAuthBase):
|
|
def __init__(self, credentials: dict):
|
|
super().__init__(credentials)
|
|
auth_type = credentials.get("auth_type")
|
|
if auth_type != "bearer":
|
|
raise ValueError("Invalid auth type, Firecrawl auth type must be Bearer")
|
|
self.api_key = credentials.get("config", {}).get("api_key", None)
|
|
self.base_url = credentials.get("config", {}).get("base_url", "https://api.firecrawl.dev")
|
|
|
|
if not self.api_key:
|
|
raise ValueError("No API key provided")
|
|
|
|
def validate_credentials(self):
|
|
headers = self._prepare_headers()
|
|
options = {
|
|
"url": "https://example.com",
|
|
"includePaths": [],
|
|
"excludePaths": [],
|
|
"limit": 1,
|
|
"scrapeOptions": {"onlyMainContent": True},
|
|
}
|
|
response = self._post_request(self._build_url("v1/crawl"), options, headers)
|
|
if response.status_code == 200:
|
|
return True
|
|
else:
|
|
self._handle_error(response)
|
|
|
|
def _prepare_headers(self):
|
|
return {"Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}"}
|
|
|
|
def _build_url(self, path: str) -> str:
|
|
# ensure exactly one slash between base and path, regardless of user-provided base_url
|
|
return f"{self.base_url.rstrip('/')}/{path.lstrip('/')}"
|
|
|
|
def _post_request(self, url, data, headers):
|
|
return httpx.post(url, headers=headers, json=data)
|
|
|
|
def _handle_error(self, response):
|
|
try:
|
|
payload = response.json()
|
|
except json.JSONDecodeError:
|
|
payload = {}
|
|
error_message = payload.get("error") or payload.get("message") or (response.text or "Unknown error occurred")
|
|
raise Exception(f"Failed to authorize. Status code: {response.status_code}. Error: {error_message}")
|