app.py

import asyncio.base_events as _asyncio_base_events


def _patch_asyncio_event_loop_del():
    original_del = getattr(_asyncio_base_events.BaseEventLoop, "__del__", None)
    if original_del is None:
        return
    def patched_del(self):
        try:
            original_del(self)
        except ValueError as e:
            if "Invalid file descriptor" not in str(e):
                raise
    _asyncio_base_events.BaseEventLoop.__del__ = patched_del


_patch_asyncio_event_loop_del()

import os
import gradio as gr
from civitai_to_hf import (
    search_civitai,
    download_civitai,
    smoke_test_civitai,
    add_civitai_item,
    get_civitai_tag,
    get_civitai_creator,
    refresh_civitai_creators,
    request_run_cancel,
    from_civitai_dropdown,
    update_civitai_selection,
    from_civitai_checkbox,
    select_civitai_all_item_fast,
    load_more_civitai,
    load_all_civitai,
    refresh_civitai_key_status,
    probe_civitai_api,
    update_civitai_gallery_mode,
    begin_probe_civitai,
    clear_retry_state,
    safe_retry_probe,
    safe_upload_verify_probe,
    safe_summary_probe,
    create_report_zip,
    CIVITAI_TYPE,
    CIVITAI_BASEMODEL,
    CIVITAI_SORT,
    CIVITAI_PERIOD,
    CIVITAI_FILETYPE,
    uniq_urls,
    HF_UPLOAD_RETRY_POLICY_CHOICES,
)


def update_upload_target_note(upload_type: str):
    note = "Bucket mode note: if the bucket does not exist, it will be created as private. The privacy checkbox is ignored for new buckets. Existing bucket visibility is kept as-is."
    return gr.update(value=note, visible=(upload_type == "bucket"))


def use_remaining_urls(remaining_urls: str, current_urls: str):
    next_urls = remaining_urls.strip() if remaining_urls and remaining_urls.strip() else current_urls
    return gr.update(value=uniq_urls(next_urls) if next_urls else "")


def use_failed_urls(failed_urls: str, current_urls: str):
    next_urls = failed_urls.strip() if failed_urls and failed_urls.strip() else current_urls
    return gr.update(value=uniq_urls(next_urls) if next_urls else "")


CIVITAI_TAGS = get_civitai_tag()
CIVITAI_CREATORS = get_civitai_creator()

def env_flag(name: str) -> bool:
    value = os.environ.get(name, "")
    return value.strip().lower() in {"1", "true", "yes", "on", "y", "t"}


CIVITAI_HF_DEBUG = env_flag("CIVITAI_HF_DEBUG")

css = """

.title { font-size: 3em; align-items: center; text-align: center; }

.info { align-items: center; text-align: center; }

.block.result { margin: 1em 0; padding: 1em; box-shadow: 0 0 3px 3px #664422, 0 0 3px 2px #664422 inset; border-radius: 6px; background: #665544; }

#gallery img { aspect-ratio: 3 / 4; object-fit: cover; }

#gallery { min-height: 620px; }

#civitai-detail img { object-fit: contain; background: rgba(0, 0, 0, 0.15); }

"""

with gr.Blocks(fill_width=True, delete_cache=(60, 3600)) as demo:
    with gr.Column():
        gr.Markdown("# CivitAI to HF🤗 Downloader & Uploader", elem_classes="title")
        state = gr.State(value={"civitai_search_preview_dir": "", "civitai_last_choices": [("", "")], "civitai_last_results": {}, "civitai_last_selects": [], "civitai_last_items": [], "civitai_visible_count": 0, "civitai_detail_url": "", "civitai_preview_cache": {}, "civitai_preview_fail_urls": [], "civitai_gallery_enabled": True})
        session_state = gr.State(value={})
        with gr.Accordion("Search Civitai", open=True):
            with gr.Row():
                search_civitai_type = gr.Dropdown(label="Type", choices=CIVITAI_TYPE, value=["LORA"], multiselect=True)
                search_civitai_sort = gr.Dropdown(label="Sort", choices=CIVITAI_SORT, value=CIVITAI_SORT[0])
                search_civitai_period = gr.Dropdown(label="Period", choices=CIVITAI_PERIOD, value="Month")
            with gr.Accordion("Advanced", open=False):
                search_civitai_basemodel = gr.CheckboxGroup(label="Base Model", choices=CIVITAI_BASEMODEL, value=[])
                with gr.Row():
                    search_civitai_filetype = gr.CheckboxGroup(label="File type", choices=CIVITAI_FILETYPE, value=["Model"])
                    search_civitai_limit = gr.Number(label="Limit", minimum=1, maximum=100, step=1, value=100)
                    search_civitai_page = gr.Number(label="Page", info="If 0, retrieve all pages", minimum=0, maximum=10, step=1, value=1)
            with gr.Row(equal_height=True):
                search_civitai_query = gr.Textbox(label="Query", placeholder="flux", lines=1)
                search_civitai_tag = gr.Dropdown(label="Tag", choices=CIVITAI_TAGS, value=CIVITAI_TAGS[0], allow_custom_value=True)
                search_civitai_user = gr.Dropdown(label="Username", choices=CIVITAI_CREATORS, value="", allow_custom_value=True)
                search_civitai_submit = gr.Button("Search on Civitai")
            with gr.Accordion("Results", open=True):
                with gr.Row(equal_height=False):
                    with gr.Column(scale=8):
                        with gr.Row():
                            search_civitai_page_label = gr.Markdown("Showing 0/0", elem_classes="info")
                            search_civitai_gallery_enabled = gr.Checkbox(label="Gallery", value=True)
                            search_civitai_load_more = gr.Button("Load More", variant="secondary", size="sm")
                            search_civitai_load_all = gr.Button("Load All", variant="secondary", size="sm")
                        search_civitai_gallery = gr.Gallery([], label="Results", allow_preview=False, columns=4, height=620, object_fit="cover", format="webp", elem_id="gallery", buttons=[], interactive=False)
                        search_civitai_result_info = gr.Markdown("Search result.", elem_classes="info")
                    with gr.Column(scale=5):
                        search_civitai_detail_image = gr.Image(label="Selected Preview", value=None, height=520, elem_id="civitai-detail", interactive=False, visible=False, format="webp")
                        search_civitai_desc = gr.Markdown(value="", visible=False, elem_classes="desc")
                        search_civitai_json = gr.JSON(value={}, visible=False)
                with gr.Row(equal_height=True):
                    with gr.Column(scale=9):
                        with gr.Accordion("Select by Checkbox", open=False):
                            search_civitai_result_checkbox = gr.CheckboxGroup(label="All Results", choices=[], value=[])
                        search_civitai_result = gr.Dropdown(label="Selected Results", choices=[("", "")], value=[], allow_custom_value=True, visible=True, multiselect=True)
                    with gr.Column(scale=1):
                        search_civitai_add = gr.Button("Add to download URLs")
                        search_civitai_select_all = gr.Button("Select All", variant="secondary", size="sm")
        with gr.Accordion("Civitai", open=True):
            dl_url = gr.Textbox(label="Download URL(s)", placeholder="https://civitai.com/api/download/models/28907\n...", value="", lines=3, max_lines=255)
            with gr.Column():
                civitai_key = gr.Textbox(label="Your Civitai Key(s)", value="", lines=2, max_lines=4)
                civitai_key_status = gr.Markdown(value=refresh_civitai_key_status(""), elem_classes="info")
                gr.Markdown("Your Civitai API key is available at [https://civitai.com/user/account](https://civitai.com/user/account). You can enter multiple keys separated by newline, comma, or space.", elem_classes="info")
        with gr.Accordion("Hugging Face", open=True):
            with gr.Row():
                with gr.Column():
                    hf_token = gr.Textbox(label="Your HF write token", placeholder="hf_...", value="", max_lines=1)
                    gr.Markdown("Your token is available at [hf.co/settings/tokens](https://huggingface.co/settings/tokens).", elem_classes="info")
                newrepo_id = gr.Textbox(label="Upload repo ID", placeholder="yourid/yourrepo", value="", max_lines=1)
            with gr.Row():
                newrepo_type = gr.Radio(label="Upload repo type", choices=["model", "dataset", "bucket"], value="model")
                is_private = gr.Checkbox(label="Create private repo", value=True)
                is_info = gr.Checkbox(label="Upload Civitai information files", value=False)
                is_rename = gr.Checkbox(label="Auto rename", value=True)
            hf_upload_retry_policy = gr.Dropdown(label="HF upload retry policy", choices=HF_UPLOAD_RETRY_POLICY_CHOICES, value="Auto", info="Auto is a cautious default. Use Patient for repeated HF 429/503/LFS commit errors.")
            bucket_mode_note = gr.Markdown(value="", visible=False, elem_classes="info")
        with gr.Row():
            run_button = gr.Button(value="Download and Upload", variant="primary")
            stop_button = gr.Button(value="Cancel", variant="stop")
        with gr.Accordion("Results", open=True):
            uploaded_urls = gr.CheckboxGroup(visible=False, choices=[], value=None)
            urls_md = gr.Markdown("<br><br>", elem_classes="result")
            with gr.Row():
                urls_remain = gr.Textbox(label="Remaining URLs", value="", buttons=["copy"], visible=False, scale=6)
                urls_failed = gr.Textbox(label="Failed URLs", value="", buttons=["copy"], visible=False, scale=6)
            with gr.Row():
                use_remaining_button = gr.Button(value="Use Remaining URLs", visible=True, variant="secondary")
                use_failed_button = gr.Button(value="Retry Failed Only", visible=True, variant="secondary")
                clear_remaining_button = gr.Button(value="Clear Remaining", visible=True, variant="secondary")
        with gr.Accordion("Probe / Experimental", open=False, visible=CIVITAI_HF_DEBUG):
            probe_md = gr.Markdown(value="", visible=False)
            with gr.Row():
                smoke_test_button = gr.Button(value="Smoke Test", variant="secondary")
                probe_button = gr.Button("Probe Civitai", variant="secondary")
                safe_retry_probe_button = gr.Button("Safe Retry Probe", variant="secondary")
                safe_upload_verify_probe_button = gr.Button("Safe Upload Verify Probe", variant="secondary")
                safe_summary_probe_button = gr.Button("Safe Summary Probe", variant="secondary")
            with gr.Row():
                report_zip_button = gr.Button("Create Report ZIP", variant="secondary")
                report_zip_file = gr.File(label="Report ZIP", visible=True)
        gr.DuplicateButton(value="Duplicate Space")

    heavy_outputs = [uploaded_urls, urls_md, urls_remain, urls_failed, civitai_key_status, session_state]
    search_outputs = [search_civitai_select_all, search_civitai_result, search_civitai_desc, search_civitai_json, search_civitai_gallery,
                      search_civitai_result_checkbox, search_civitai_result_info, search_civitai_page_label,
                      search_civitai_detail_image, civitai_key_status, state]

    download_evt = gr.on(
        triggers=[run_button.click],
        fn=download_civitai,
        inputs=[dl_url, civitai_key, hf_token, uploaded_urls, newrepo_id, newrepo_type, is_private, is_info, is_rename, session_state, hf_upload_retry_policy],
        outputs=heavy_outputs,
        queue=True,
        concurrency_limit=1,
        concurrency_id="heavy-run",
    )
    smoke_evt = gr.on(
        triggers=[smoke_test_button.click],
        fn=smoke_test_civitai,
        inputs=[civitai_key, hf_token, uploaded_urls, newrepo_type, is_private, is_info, is_rename, session_state, hf_upload_retry_policy],
        outputs=heavy_outputs,
        queue=True,
        api_visibility="undocumented",
        concurrency_limit=1,
        concurrency_id="heavy-run",
    )
    stop_button.click(request_run_cancel, [session_state], [session_state], queue=False, api_visibility="undocumented")
    stop_button.click(fn=None, cancels=[download_evt, smoke_evt], queue=False, api_visibility="undocumented")
    use_remaining_button.click(use_remaining_urls, [urls_remain, dl_url], [dl_url], queue=False, api_visibility="undocumented")
    use_failed_button.click(use_failed_urls, [urls_failed, dl_url], [dl_url], queue=False, api_visibility="undocumented")
    clear_remaining_button.click(clear_retry_state, [session_state], [urls_remain, urls_failed, session_state], queue=False, api_visibility="undocumented")

    gr.on(
        triggers=[search_civitai_submit.click, search_civitai_query.submit],
        fn=search_civitai,
        inputs=[search_civitai_query, search_civitai_type, search_civitai_basemodel, search_civitai_sort,
                search_civitai_period, search_civitai_tag, search_civitai_user, search_civitai_limit,
                search_civitai_page, search_civitai_filetype, civitai_key, state],
        outputs=search_outputs,
        queue=False,
        api_visibility="undocumented",
    )
    search_civitai_load_more.click(load_more_civitai, [state, civitai_key], search_outputs, queue=False, api_visibility="undocumented")
    search_civitai_load_all.click(load_all_civitai, [state, civitai_key], search_outputs, queue=True, api_visibility="undocumented")
    search_civitai_result.input(from_civitai_dropdown, [search_civitai_result, civitai_key, state], search_outputs, queue=False, api_visibility="undocumented")
    search_civitai_result_checkbox.input(from_civitai_checkbox, [search_civitai_result_checkbox, civitai_key, state], search_outputs, queue=False, api_visibility="undocumented")
    search_civitai_add.click(add_civitai_item, [search_civitai_result, dl_url], [dl_url], queue=False, api_visibility="undocumented")
    newrepo_type.input(update_upload_target_note, [newrepo_type], [bucket_mode_note], queue=False, api_visibility="undocumented")
    search_civitai_select_all.click(select_civitai_all_item_fast, [search_civitai_select_all, civitai_key, state], search_outputs, queue=False, api_visibility="undocumented")
    search_civitai_gallery.select(update_civitai_selection, [search_civitai_result, civitai_key, state], search_outputs, queue=False, api_visibility="undocumented")
    search_civitai_gallery_enabled.change(update_civitai_gallery_mode, [search_civitai_gallery_enabled, civitai_key, state], search_outputs, queue=False, api_visibility="undocumented")
    search_civitai_user.focus(refresh_civitai_creators, [search_civitai_user, civitai_key], [search_civitai_user], queue=False, api_visibility="undocumented")
    search_civitai_user.change(refresh_civitai_creators, [search_civitai_user, civitai_key], [search_civitai_user], queue=False, api_visibility="undocumented")
    civitai_key.change(refresh_civitai_key_status, [civitai_key], [civitai_key_status], queue=False, api_visibility="undocumented")
    probe_button.click(begin_probe_civitai, [], [probe_md], queue=False, api_visibility="undocumented").then(probe_civitai_api, [search_civitai_query, civitai_key], [probe_md, civitai_key_status], queue=False, api_visibility="undocumented")
    safe_retry_probe_button.click(safe_retry_probe, [hf_upload_retry_policy], [probe_md], queue=False, api_visibility="undocumented")
    safe_upload_verify_probe_button.click(safe_upload_verify_probe, [hf_upload_retry_policy], [probe_md], queue=False, api_visibility="undocumented")
    safe_summary_probe_button.click(safe_summary_probe, [], [probe_md], queue=False, api_visibility="undocumented")
    report_zip_button.click(create_report_zip, [session_state, state], [report_zip_file], queue=False, api_visibility="undocumented")


demo.queue()
demo.launch(theme="NoCrypt/miku@>=1.2.2", css=css, ssr_mode=False, mcp_server=False)