Create civitai.py
Browse files- backend/civitai.py +29 -0
backend/civitai.py
ADDED
|
@@ -0,0 +1,29 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
import requests
|
| 2 |
+
import os
|
| 3 |
+
import hashlib
|
| 4 |
+
import shutil
|
| 5 |
+
|
| 6 |
+
def download_civitai_model(model_version_id: str, api_key: str) -> str:
|
| 7 |
+
api_url = f"https://civitai.com/api/v1/model-versions/{model_version_id}"
|
| 8 |
+
headers = {"Authorization": f"Bearer {api_key}"}
|
| 9 |
+
resp = requests.get(api_url, headers=headers)
|
| 10 |
+
resp.raise_for_status()
|
| 11 |
+
data = resp.json()
|
| 12 |
+
download_url = data["downloadUrl"]
|
| 13 |
+
cache_dir = "/data/civitai_cache"
|
| 14 |
+
os.makedirs(cache_dir, exist_ok=True)
|
| 15 |
+
filename = os.path.basename(download_url)
|
| 16 |
+
local_path = os.path.join(cache_dir, filename)
|
| 17 |
+
if not os.path.exists(local_path):
|
| 18 |
+
with requests.get(download_url, headers=headers, stream=True) as r:
|
| 19 |
+
r.raise_for_status()
|
| 20 |
+
with open(local_path, "wb") as f:
|
| 21 |
+
shutil.copyfileobj(r.raw, f)
|
| 22 |
+
# For safetensors or zip, extract if zip?
|
| 23 |
+
if local_path.endswith(".zip"):
|
| 24 |
+
import zipfile
|
| 25 |
+
extract_dir = local_path + "_extracted"
|
| 26 |
+
with zipfile.ZipFile(local_path, 'r') as zip_ref:
|
| 27 |
+
zip_ref.extractall(extract_dir)
|
| 28 |
+
return extract_dir
|
| 29 |
+
return local_path
|