Create app.py

app.py

@@ -0,0 +1,434 @@
+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", "nevreal"]
+
+def get_json_data(url):
+    url_split = url.split('/')
+    api_url = f"https://civitai.com/api/v1/models/{url_split[4]}"
+    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["nsfwLevel"] > 5:
+                return False
+    return True
+
+def get_prompts_from_image(image_id):
+    url = f'https://civitai.com/api/trpc/image.getGenerationData?input={{"json":{{"id":{image_id}}}}}'
+    response = requests.get(url)
+    prompt = ""
+    negative_prompt = ""
+    if response.status_code == 200:
+        data = response.json()
+        result = data['result']['data']['json']
+        if "prompt" in result['meta']:
+            prompt = result['meta']['prompt']
+        if "negativePrompt" in result['meta']:
+            negative_prompt = result["meta"]["negativePrompt"]
+    
+    return prompt, negative_prompt
+
+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", "SD 3", "Flux.1 D", "Flux.1 S"]:
+                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)
+                            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",
+                            "SD 3": "stabilityai/stable-diffusion-3-medium-diffusers",
+                            "Flux.1 D": "black-forest-labs/FLUX.1-dev",
+                            "Flux.1 S": "black-forest-labs/FLUX.1-schnell"
+                        }
+                        base_model = model_mapping[model_version["baseModel"]]
+                        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="."):
+    headers = {}
+    try:
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()
+    except requests.exceptions.HTTPError as e:
+        print(e)
+        if response.status_code == 401: 
+            headers['Authorization'] = f'Bearer {os.environ["CIVITAI_API"]}'
+            try:
+                response = requests.get(url, headers=headers)
+                response.raise_for_status()
+            except requests.exceptions.RequestException as e:
+                raise gr.Error(f"Error downloading file: {e}")
+        else:
+            raise gr.Error(f"Error downloading file: {e}")
+    except requests.exceptions.RequestException as e:
+        raise gr.Error(f"Error downloading file: {e}")
+
+    with open(f"{folder}/{filename}", 'wb') as f:
+        f.write(response.content)
+
+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}
+"""
+    dtype = "torch.bfloat16" if info["baseModel"] == "black-forest-labs/FLUX.1-dev" or info["baseModel"] == "black-forest-labs/FLUX.1-schnell" else "torch.float16"
+        
+    content = f"""---
+license: other
+license_name: bespoke-lora-trained-license
+license_link: https://multimodal.art/civitai-licenses?allowNoCredit={info["allowNoCredit"]}&allowCommercialUse={info["allowCommercialUse"][0] if info["allowCommercialUse"] else 1}&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
+
+device = "cuda" if torch.cuda.is_available() else "cpu"
+
+pipeline = AutoPipelineForText2Image.from_pretrained('{info["baseModel"]}', torch_dtype={dtype}).to(device)
+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"
+    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)
+    gr.Info(f"Starting download of model {url}")
+    info, downloaded_files = process_url(url, profile, folder=folder)
+    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:
+        create_repo(repo_id=user_repo_id, private=True, exist_ok=True, token=oauth_token.token)
+        gr.Info(f"Starting to upload repo {user_repo_id} to Hugging Face...")
+        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)
+        gr.Info(f"Model uploaded!")
+    except Exception as e:
+        print(e)
+        raise gr.Error("Your Hugging Face Token expired. Log out and in again to upload your models.")
+        
+    return f'''# Model uploaded to 🤗!
+    ## Access it here [{user_repo_id}](https://huggingface.co/{user_repo_id}) '''
+
+def bulk_upload(profile: Optional[gr.OAuthProfile], oauth_token: gr.OAuthToken, urls, link_civit=False):
+    urls = urls.split("\n")
+    print(urls)
+    upload_results = ""
+    for url in urls:
+        if(url):
+            try:
+                upload_result = upload_civit_to_hf(profile, oauth_token, url, link_civit)
+                upload_results += upload_result+"\n"
+            except Exception as e:
+                gr.Warning(f"Error uploading the model {url}")
+    return upload_results
+
+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, you'll be listed in [LoRA Studio](https://lorastudio.co/models) after a short review, and get the 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(
+                placeholder="https://civitai.com/models/144684/pixelartredmond-pixel-art-loras-for-sd-xl",
+                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(
+                placeholder="https://civitai.com/models/144684/pixelartredmond-pixel-art-loras-for-sd-xl",
+                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])
+    civit_username_to_bulk.change(fn=list_civit_models, inputs=[civit_username_to_bulk], outputs=[submit_bulk_civit])
+    try_again_button.click(fn=check_civit_link, inputs=[submit_source_civit], outputs=[instructions, submit_button_civit, try_again_button, submit_button_civit])
+    
+    submit_button_civit.click(fn=show_output, inputs=[], outputs=[output]).then(fn=upload_civit_to_hf, inputs=[submit_source_civit, link_civit], outputs=[output])
+    bulk_button.click(fn=show_output, inputs=[], outputs=[output]).then(fn=bulk_upload, inputs=[submit_bulk_civit, link_civit], outputs=[output])
+    #gr.LogoutButton(elem_id="logout")
+    
+demo.queue(default_concurrency_limit=50)
+demo.launch()