Spaces:
Running
Running
import requests | |
import os | |
import gradio as gr | |
from huggingface_hub import update_repo_visibility, whoami, upload_folder, create_repo, upload_file, update_repo_visibility | |
from slugify import slugify | |
import gradio as gr | |
import re | |
import uuid | |
from typing import Optional | |
import json | |
from bs4 import BeautifulSoup | |
TRUSTED_UPLOADERS = ["KappaNeuro", "CiroN2022", "multimodalart", "Norod78", "joachimsallstrom", "blink7630", "e-n-v-y", "DoctorDiffusion", "RalFinger", "artificialguybr"] | |
def get_json_data(url): | |
url_split = url.split('/') | |
print("url_split: ", url_split) | |
api_url = f"https://civitai.com/api/v1/models/{url_split[4]}?token={os.environ['CIVITAI_API']}" | |
try: | |
response = requests.get(api_url) | |
response.raise_for_status() | |
return response.json() | |
except requests.exceptions.RequestException as e: | |
print(f"Error fetching JSON data: {e}") | |
return None | |
def check_nsfw(json_data, profile): | |
if json_data["nsfw"]: | |
return False | |
print(profile) | |
if(profile.username in TRUSTED_UPLOADERS): | |
return True | |
for model_version in json_data["modelVersions"]: | |
for image in model_version["images"]: | |
if image["nsfw"] > 5: | |
return False | |
return True | |
def get_prompts_from_image(image_id): | |
print("image_id: ", image_id) | |
url = f'https://civitai.com/images/{image_id}' | |
response = requests.get(url) | |
soup = BeautifulSoup(response.text, 'html.parser') | |
content = soup.find_all(class_='mantine-Code-root mantine-Code-block mantine-2v44jn') | |
print(content) | |
if(content): | |
if(len(content) > 1): | |
return content[0].text, content[1].text | |
else: | |
return content[0].text, "" | |
else: | |
return "", "" | |
def extract_info(json_data): | |
if json_data["type"] == "LORA": | |
for model_version in json_data["modelVersions"]: | |
if model_version["baseModel"] in ["SDXL 1.0", "SDXL 0.9", "SD 1.5", "SD 1.4", "SD 2.1", "SD 2.0", "SD 2.0 768", "SD 2.1 768"]: | |
for file in model_version["files"]: | |
print(file) | |
if "primary" in file: | |
# Start by adding the primary file to the list | |
urls_to_download = [{"url": file["downloadUrl"], "filename": file["name"], "type": "weightName"}] | |
# Then append all image URLs to the list | |
for image in model_version["images"]: | |
image_id = image["url"].split("/")[-1].split(".")[0] | |
prompt, negative_prompt = get_prompts_from_image(image_id) | |
if image["nsfwLevel"] > 5: | |
pass #ugly before checking the actual logic | |
else: | |
urls_to_download.append({ | |
"url": image["url"], | |
"filename": os.path.basename(image["url"]), | |
"type": "imageName", | |
"prompt": prompt, #if "meta" in image and "prompt" in image["meta"] else "" | |
"negative_prompt": negative_prompt | |
}) | |
model_mapping = { | |
"SDXL 1.0": "stabilityai/stable-diffusion-xl-base-1.0", | |
"SDXL 0.9": "stabilityai/stable-diffusion-xl-base-1.0", | |
"SD 1.5": "runwayml/stable-diffusion-v1-5", | |
"SD 1.4": "CompVis/stable-diffusion-v1-4", | |
"SD 2.1": "stabilityai/stable-diffusion-2-1-base", | |
"SD 2.0": "stabilityai/stable-diffusion-2-base", | |
"SD 2.1 768": "stabilityai/stable-diffusion-2-1", | |
"SD 2.0 768": "stabilityai/stable-diffusion-2" | |
} | |
base_model = model_mapping[model_version["baseModel"]] | |
print(json_data) | |
info = { | |
"urls_to_download": urls_to_download, | |
"id": model_version["id"], | |
"baseModel": base_model, | |
"modelId": model_version.get("modelId", ""), | |
"name": json_data["name"], | |
"description": json_data["description"], | |
"trainedWords": model_version["trainedWords"] if "trainedWords" in model_version else [], | |
"creator": json_data["creator"]["username"], | |
"tags": json_data["tags"], | |
"allowNoCredit": json_data["allowNoCredit"], | |
"allowCommercialUse": json_data["allowCommercialUse"], | |
"allowDerivatives": json_data["allowDerivatives"], | |
"allowDifferentLicense": json_data["allowDifferentLicense"] | |
} | |
return info | |
return None | |
def download_files(info, folder="."): | |
downloaded_files = { | |
"imageName": [], | |
"imagePrompt": [], | |
"imageNegativePrompt": [], | |
"weightName": [] | |
} | |
for item in info["urls_to_download"]: | |
download_file(item["url"], item["filename"], folder) | |
downloaded_files[item["type"]].append(item["filename"]) | |
if(item["type"] == "imageName"): | |
prompt_clean = re.sub(r'<.*?>', '', item["prompt"]) | |
negative_prompt_clean = re.sub(r'<.*?>', '', item["negative_prompt"]) | |
downloaded_files["imagePrompt"].append(prompt_clean) | |
downloaded_files["imageNegativePrompt"].append(negative_prompt_clean) | |
return downloaded_files | |
def download_file(url, filename, folder="."): | |
try: | |
response = requests.get(url) | |
response.raise_for_status() | |
with open(f"{folder}/{filename}", 'wb') as f: | |
f.write(response.content) | |
except requests.exceptions.RequestException as e: | |
raise gr.Error(f"Error downloading file: {e}") | |
def process_url(url, profile, do_download=True, folder="."): | |
json_data = get_json_data(url) | |
if json_data: | |
if check_nsfw(json_data, profile): | |
info = extract_info(json_data) | |
if info: | |
if(do_download): | |
downloaded_files = download_files(info, folder) | |
else: | |
downloaded_files = [] | |
return info, downloaded_files | |
else: | |
raise gr.Error("Only SDXL LoRAs are supported for now") | |
else: | |
raise gr.Error("This model has content tagged as unsafe by CivitAI") | |
else: | |
raise gr.Error("Something went wrong in fetching CivitAI API") | |
def create_readme(info, downloaded_files, user_repo_id, link_civit=False, is_author=True, folder="."): | |
readme_content = "" | |
original_url = f"https://civitai.com/models/{info['modelId']}" | |
link_civit_disclaimer = f'([CivitAI]({original_url}))' | |
non_author_disclaimer = f'This model was originally uploaded on [CivitAI]({original_url}), by [{info["creator"]}](https://civitai.com/user/{info["creator"]}/models). The information below was provided by the author on CivitAI:' | |
default_tags = ["text-to-image", "stable-diffusion", "lora", "diffusers", "template:sd-lora", "migrated"] | |
civit_tags = [t.replace(":", "") for t in info["tags"] if t not in default_tags] | |
tags = default_tags + civit_tags | |
unpacked_tags = "\n- ".join(tags) | |
trained_words = info['trainedWords'] if 'trainedWords' in info and info['trainedWords'] else [] | |
formatted_words = ', '.join(f'`{word}`' for word in trained_words) | |
if formatted_words: | |
trigger_words_section = f"""## Trigger words | |
You should use {formatted_words} to trigger the image generation. | |
""" | |
else: | |
trigger_words_section = "" | |
widget_content = "" | |
for index, (prompt, negative_prompt, image) in enumerate(zip(downloaded_files["imagePrompt"], downloaded_files["imageNegativePrompt"], downloaded_files["imageName"])): | |
escaped_prompt = prompt.replace("'", "''") | |
negative_prompt_content = f"""parameters: | |
negative_prompt: {negative_prompt} | |
""" if negative_prompt else "" | |
widget_content += f"""- text: '{escaped_prompt if escaped_prompt else ' ' }' | |
{negative_prompt_content} | |
output: | |
url: >- | |
{image} | |
""" | |
content = f"""--- | |
license: other | |
license_name: bespoke-lora-trained-license | |
license_link: https://multimodal.art/civitai-licenses?allowNoCredit={info["allowNoCredit"]}&allowCommercialUse={info["allowCommercialUse"][0]}&allowDerivatives={info["allowDerivatives"]}&allowDifferentLicense={info["allowDifferentLicense"]} | |
tags: | |
- {unpacked_tags} | |
base_model: {info["baseModel"]} | |
instance_prompt: {info['trainedWords'][0] if 'trainedWords' in info and len(info['trainedWords']) > 0 else ''} | |
widget: | |
{widget_content} | |
--- | |
# {info["name"]} | |
<Gallery /> | |
{non_author_disclaimer if not is_author else ''} | |
{link_civit_disclaimer if link_civit else ''} | |
## Model description | |
{info["description"]} | |
{trigger_words_section} | |
## Download model | |
Weights for this model are available in Safetensors format. | |
[Download](/{user_repo_id}/tree/main) them in the Files & versions tab. | |
## Use it with the [🧨 diffusers library](https://github.com/huggingface/diffusers) | |
```py | |
from diffusers import AutoPipelineForText2Image | |
import torch | |
pipeline = AutoPipelineForText2Image.from_pretrained('{info["baseModel"]}', torch_dtype=torch.float16).to('cuda') | |
pipeline.load_lora_weights('{user_repo_id}', weight_name='{downloaded_files["weightName"][0]}') | |
image = pipeline('{prompt if prompt else (formatted_words if formatted_words else 'Your custom prompt')}').images[0] | |
``` | |
For more details, including weighting, merging and fusing LoRAs, check the [documentation on loading LoRAs in diffusers](https://huggingface.co/docs/diffusers/main/en/using-diffusers/loading_adapters) | |
""" | |
#for index, (image, prompt) in enumerate(zip(downloaded_files["imageName"], downloaded_files["imagePrompt"])): | |
# if index == 1: | |
# content += f"## Image examples for the model:\n![Image {index}]({image})\n> {prompt}\n" | |
# elif index > 1: | |
# content += f"\n![Image {index}]({image})\n> {prompt}\n" | |
readme_content += content + "\n" | |
print(readme_content) | |
with open(f"{folder}/README.md", "w") as file: | |
file.write(readme_content) | |
def get_creator(username): | |
url = f"https://civitai.com/api/trpc/user.getCreator?input=%7B%22json%22%3A%7B%22username%22%3A%22{username}%22%2C%22authed%22%3Atrue%7D%7D" | |
headers = { | |
"authority": "civitai.com", | |
"accept": "*/*", | |
"accept-language": "en-BR,en;q=0.9,pt-BR;q=0.8,pt;q=0.7,es-ES;q=0.6,es;q=0.5,de-LI;q=0.4,de;q=0.3,en-GB;q=0.2,en-US;q=0.1,sk;q=0.1", | |
"content-type": "application/json", | |
"cookie": f'{os.environ["COOKIE_INFO"]}', | |
"if-modified-since": "Tue, 22 Aug 2023 07:18:52 GMT", | |
"referer": f"https://civitai.com/user/{username}/models", | |
"sec-ch-ua": "\"Not.A/Brand\";v=\"8\", \"Chromium\";v=\"114\", \"Google Chrome\";v=\"114\"", | |
"sec-ch-ua-mobile": "?0", | |
"sec-ch-ua-platform": "macOS", | |
"sec-fetch-dest": "empty", | |
"sec-fetch-mode": "cors", | |
"sec-fetch-site": "same-origin", | |
"user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36" | |
} | |
response = requests.get(url, headers=headers) | |
return response.json() | |
def extract_huggingface_username(username): | |
data = get_creator(username) | |
links = data.get('result', {}).get('data', {}).get('json', {}).get('links', []) | |
for link in links: | |
url = link.get('url', '') | |
if url.startswith('https://huggingface.co/') or url.startswith('https://www.huggingface.co/'): | |
username = url.split('/')[-1] | |
return username | |
return None | |
def check_civit_link(profile: Optional[gr.OAuthProfile], url): | |
info, _ = process_url(url, profile, do_download=False) | |
hf_username = extract_huggingface_username(info['creator']) | |
attributes_methods = dir(profile) | |
if(profile.username == "multimodalart"): | |
return '', gr.update(interactive=True), gr.update(visible=False), gr.update(visible=True) | |
if(not hf_username): | |
no_username_text = f'If you are {info["creator"]} on CivitAI, hi! Your CivitAI profile seems to not have information about your Hugging Face account. Please visit <a href="https://civitai.com/user/account" target="_blank">https://civitai.com/user/account</a> and include your 🤗 username there, here\'s mine:<br><img width="60%" src="https://i.imgur.com/hCbo9uL.png" /><br>(if you are not {info["creator"]}, you cannot submit their model at this time)' | |
return no_username_text, gr.update(interactive=False), gr.update(visible=True), gr.update(visible=False) | |
if(profile.username != hf_username): | |
unmatched_username_text = '<h4>Oops, the Hugging Face account in your CivitAI profile seems to be different than the one your are using here. Please visit <a href="https://civitai.com/user/account">https://civitai.com/user/account</a> and update it there to match your Hugging Face account<br><img src="https://i.imgur.com/hCbo9uL.png" /></h4>' | |
return unmatched_username_text, gr.update(interactive=False), gr.update(visible=True), gr.update(visible=False) | |
else: | |
return '', gr.update(interactive=True), gr.update(visible=False), gr.update(visible=True) | |
def swap_fill(profile: Optional[gr.OAuthProfile]): | |
if profile is None: | |
return gr.update(visible=True), gr.update(visible=False) | |
else: | |
return gr.update(visible=False), gr.update(visible=True) | |
def show_output(): | |
return gr.update(visible=True) | |
def list_civit_models(username): | |
url = f"https://civitai.com/api/v1/models?username={username}&limit=100" | |
json_models_list = [] | |
while url: | |
response = requests.get(url) | |
data = response.json() | |
# Add current page items to the list | |
json_models_list.extend(data.get('items', [])) | |
# Check if there is a nextPage URL in the metadata | |
metadata = data.get('metadata', {}) | |
url = metadata.get('nextPage', None) | |
urls = "" | |
for model in json_models_list: | |
urls += f'https://civitai.com/models/{model["id"]}/{slugify(model["name"])}\n' | |
return urls | |
def upload_civit_to_hf(profile: Optional[gr.OAuthProfile], oauth_token: gr.OAuthToken, url, link_civit=False): | |
if not profile.name: | |
return gr.Error("Are you sure you are logged in?") | |
folder = str(uuid.uuid4()) | |
os.makedirs(folder, exist_ok=False) | |
info, downloaded_files = process_url(url, profile, folder=folder) | |
#try: | |
#print(oauth_token) | |
#print(oauth_token.token) | |
#api = HfApi() | |
#except: | |
# raise gr.Error("logging into hf went wrong") | |
username = {profile.username} | |
slug_name = slugify(info["name"]) | |
user_repo_id = f"{profile.username}/{slug_name}" | |
create_readme(info, downloaded_files, user_repo_id, link_civit, folder=folder) | |
try: | |
#repo_id = f"{username}/{profile.preferred_username}-{slug_name}" | |
print("Creating repo...") | |
print(user_repo_id) | |
create_repo(repo_id=user_repo_id, private=False, exist_ok=True, token=oauth_token.token) | |
print("Uploading repo...") | |
upload_folder( | |
folder_path=folder, | |
repo_id=user_repo_id, | |
repo_type="model", | |
token=oauth_token.token | |
) | |
update_repo_visibility(repo_id=user_repo_id, private=False, token=oauth_token.token) | |
except: | |
raise gr.Error("Your Hugging Face Token expired. Log out and in again to upload your models.") | |
#transfer_repos = gr.load("multimodalart/transfer_repos", hf_token=os.environ["HUGGING_FACE_HUB_TOKEN"], src="spaces") | |
#response_code = transfer_repos(repo_id, user_repo_id) | |
#i = 0 | |
#while response_code != "200": | |
# message = None | |
# if response_code == "409": | |
# if i < 3: | |
# user_repo_id = f"{profile.preferred_username}/{slug_name}-{i}" | |
# response_code = transfer_repos(repo_id, user_repo_id) | |
# i += 1 | |
# else: | |
# message = "It seems this model has been uploaded already in your account." | |
# elif response_code == "404": | |
# message = "Something went wrong with the model upload. Try again." | |
# else: | |
# message = f"Unexpected response code: {response_code}." | |
# | |
# if message: | |
# api.delete_repo(repo_id=repo_id, repo_type="model") | |
# raise gr.Error(message) | |
return f'''# Model uploaded to 🤗! | |
## Access it here [{user_repo_id}](https://huggingface.co/{user_repo_id}) ''' | |
def bulk_upload(profile: Optional[gr.OAuthProfile], urls, link_civit=False, progress=gr.Progress(track_tqdm=True)): | |
urls = urls.split("\n") | |
print(urls) | |
for url in urls: | |
print(url) | |
if(url): | |
try: | |
upload_result = upload_civit_to_hf(profile, url, link_civit) | |
gr.Warning(f"Model {link_civit} uploaded") | |
except: | |
gr.Warning(f"Error uploading the model {link_civit}") | |
css = ''' | |
#login { | |
width: 100% !important; | |
margin: 0 auto; | |
} | |
#disabled_upload{ | |
opacity: 0.5; | |
pointer-events:none; | |
} | |
''' | |
with gr.Blocks(css=css) as demo: | |
gr.Markdown('''# Upload your CivitAI LoRA to Hugging Face 🤗 | |
By uploading your LoRAs to Hugging Face you get diffusers compatibility, a free GPU-based Inference Widget, and possibility to submit your model to the [LoRA the Explorer](https://huggingface.co/spaces/multimodalart/LoraTheExplorer) ✨ | |
''') | |
gr.LoginButton(elem_id="login") | |
with gr.Column(elem_id="disabled_upload") as disabled_area: | |
with gr.Row(): | |
submit_source_civit = gr.Textbox( | |
label="CivitAI model URL", | |
info="URL of the CivitAI LoRA", | |
) | |
submit_button_civit = gr.Button("Upload model to Hugging Face and submit", interactive=False) | |
with gr.Column(visible=False) as enabled_area: | |
with gr.Column(): | |
submit_source_civit = gr.Textbox( | |
label="CivitAI model URL", | |
info="URL of the CivitAI LoRA", | |
) | |
with gr.Accordion("Bulk upload (bring in multiple LoRAs)", open=False): | |
civit_username_to_bulk = gr.Textbox(label="CivitAI username (optional)", info="Type your CivitAI username here to automagically fill the bulk models URLs list below (optional, you can paste links down here directly)") | |
submit_bulk_civit = gr.Textbox( | |
label="CivitAI bulk models URLs", | |
info="Add one URL per line", | |
lines=6, | |
) | |
link_civit = gr.Checkbox(label="Link back to CivitAI?", value=False) | |
bulk_button = gr.Button("Bulk upload") | |
instructions = gr.HTML("") | |
try_again_button = gr.Button("I have added my HF profile to my account (it may take 1 minute to refresh)", visible=False) | |
submit_button_civit = gr.Button("Upload model to Hugging Face", interactive=False) | |
output = gr.Markdown(label="Output progress", visible=False) | |
demo.load(fn=swap_fill, outputs=[disabled_area, enabled_area], queue=False) | |
submit_source_civit.change(fn=check_civit_link, inputs=[submit_source_civit], outputs=[instructions, submit_button_civit, try_again_button, submit_button_civit], queue=False) | |
civit_username_to_bulk.change(fn=list_civit_models, inputs=[civit_username_to_bulk], outputs=[submit_bulk_civit], queue=False) | |
try_again_button.click(fn=check_civit_link, inputs=[submit_source_civit], outputs=[instructions, submit_button_civit, try_again_button, submit_button_civit], queue=False) | |
submit_button_civit.click(fn=show_output, inputs=[], outputs=[output]).then(fn=upload_civit_to_hf, inputs=[submit_source_civit, link_civit], outputs=[output], queue=False) | |
bulk_button.click(fn=show_output, inputs=[], outputs=[output]).then(fn=bulk_upload, inputs=[submit_bulk_civit, link_civit], outputs=[output], queue=False) | |
#gr.LogoutButton(elem_id="logout") | |
demo.queue() | |
demo.launch() |