repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
public Clawd ADK gateway launch mirror
stars
latest
clone command
git clone gitlawb://did:key:z6Mkq5mY...iFZ5/my-project-publ...git clone gitlawb://did:key:z6Mkq5mY.../my-project-publ...2fa351d6docs: add automaton and perps launch sources16d ago| #1 | import hashlib |
| #2 | from io import StringIO |
| #3 | from urllib.parse import urlparse |
| #4 | |
| #5 | import requests |
| #6 | import yaml |
| #7 | |
| #8 | from embedchain.loaders.base_loader import BaseLoader |
| #9 | |
| #10 | |
| #11 | class OpenAPILoader(BaseLoader): |
| #12 | @staticmethod |
| #13 | def _get_file_content(content): |
| #14 | url = urlparse(content) |
| #15 | if all([url.scheme, url.netloc]) and url.scheme not in ["file", "http", "https"]: |
| #16 | raise ValueError("Not a valid URL.") |
| #17 | |
| #18 | if url.scheme in ["http", "https"]: |
| #19 | response = requests.get(content) |
| #20 | response.raise_for_status() |
| #21 | return StringIO(response.text) |
| #22 | elif url.scheme == "file": |
| #23 | path = url.path |
| #24 | return open(path) |
| #25 | else: |
| #26 | return open(content) |
| #27 | |
| #28 | @staticmethod |
| #29 | def load_data(content): |
| #30 | """Load yaml file of openapi. Each pair is a document.""" |
| #31 | data = [] |
| #32 | file_path = content |
| #33 | data_content = [] |
| #34 | with OpenAPILoader._get_file_content(content=content) as file: |
| #35 | yaml_data = yaml.load(file, Loader=yaml.SafeLoader) |
| #36 | for i, (key, value) in enumerate(yaml_data.items()): |
| #37 | string_data = f"{key}: {value}" |
| #38 | metadata = {"url": file_path, "row": i + 1} |
| #39 | data.append({"content": string_data, "meta_data": metadata}) |
| #40 | data_content.append(string_data) |
| #41 | doc_id = hashlib.sha256((content + ", ".join(data_content)).encode()).hexdigest() |
| #42 | return {"doc_id": doc_id, "data": data} |
| #43 |