refactor ui_models

Signed-off-by: Vladimir Mandic <mandic00@live.com>
2026-09-19 17:24:32 +02:00 · 2025-08-02 12:00:03 -04:00
parent 1d5dce1fb1
commit f9b585d983
10 changed files with 561 additions and 364 deletions
@@ -0,0 +1,191 @@
+import os
+import sys
+import json
+import time
+import logging
+import bs4
+
+
+debug = False
+logging.basicConfig(level = logging.INFO, format = '%(asctime)s %(levelname)s: %(message)s')
+log = logging.getLogger(__name__)
+
+
+class ModelImage(object):
+    def __init__(self, dct: dict):
+        if isinstance(dct, str):
+            dct = json.loads(dct)
+        self.dct: dict = dct
+        self.id: int = dct.get('id', 0)
+        self.url: str = dct.get('url', '')
+        self.width: int = dct.get('width', 0)
+        self.height: int = dct.get('height', 0)
+        self.type: str = dct.get('type', 'Unknown')
+
+    def __str__(self):
+        return f'ModelImage(id={self.id} url="{self.url}" width={self.width} height={self.height} type="{self.type}")'
+
+class ModelFile(object):
+    def __init__(self, dct: dict):
+        if isinstance(dct, str):
+            dct = json.loads(dct)
+        self.dct: dict = dct
+        self.id: int = dct.get('id', 0)
+        self.size: int = int(1024 * dct.get('sizeKB', 0))
+        self.name: str = dct.get('name', 'Unknown')
+        self.type: str = dct.get('type', 'Unknown')
+        self.hashes: list[str] = dct.get('hashes', {}).values()
+        self.url: str = dct.get('downloadUrl', '')
+
+    def __str__(self):
+        return f'ModelFile(id={self.id} name="{self.name}" size={self.size} type="{self.type}" url="{self.url}")'
+
+
+class ModelVersion(object):
+    def __init__(self, dct: dict):
+        if isinstance(dct, str):
+            dct = json.loads(dct)
+        self.dct = dct
+        self.id = dct.get('id', 0)
+        self.name = dct.get('name', 'Unknown')
+        self.base = dct.get('baseModel', 'Unknown')
+        self.mtime = dct.get('publishedAt', '')
+        self.downloads = dct.get('stats', {}).get('downloadCount', 0)
+        self.availability = dct.get('availability', 'Unknown')
+        self.html = dct.get('description', '') or ''
+        self.desc = bs4.BeautifulSoup(self.html, features="html.parser").get_text()
+        self.files = [ModelFile(f) for f in dct.get('files', [])]
+        self.images = [ModelImage(i) for i in dct.get('images', [])]
+
+    def __str__(self):
+        return f'ModelVersion(id={self.id} name="{self.name}" base="{self.base}" mtime="{self.mtime}" downloads={self.downloads} availability={self.availability} desc="{self.desc[:30]}...")'
+
+
+class Model(object):
+    def __init__(self, dct: dict):
+        if isinstance(dct, str):
+            dct = json.loads(dct)
+        self.id = dct.get('id', 0)
+        self.dct = dct
+        self.url = f'https://civitai.com/models/{self.id}'
+        self.type = dct.get('type', 'Unknown')
+        self.name = dct.get('name', 'Unknown')
+        self.html = dct.get('description', '')
+        self.desc = bs4.BeautifulSoup(self.html, features="html.parser").get_text()
+        self.tags = dct.get('tags', [])
+        self.nsfw = dct.get('nsfw', False)
+        self.level = dct.get('nsfwLevel', 0)
+        self.availability = dct.get('availability', 'Unknown')
+        self.downloads = dct.get('stats', {}).get('downloadCount', 0)
+        self.creator = dct.get('creator', {}).get('username', 'Unknown')
+        self.versions = [ModelVersion(v) for v in dct.get('modelVersions', [])]
+
+    def __str__(self):
+        return f'Model(id={self.id} type={self.type} name="{self.name}" versions={len(self.versions)} nsfw={self.nsfw}/{self.level} downloads={self.downloads} author="{self.creator}" tags={self.tags} desc="{self.desc[:30]}...")'
+
+
+def search_civitai(
+        query:str,
+        tag:str = '', # optional:tag name
+        types:str = '', # (Checkpoint, TextualInversion, Hypernetwork, AestheticGradient, LORA, Controlnet, Poses)
+        sort:str = '', # (Highest Rated, Most Downloaded, Newest)
+        period:str = '', # (AllTime, Year, Month, Week, Day)
+        nsfw:bool = None, # optional:bool
+        limit:int = 0,
+        base:list[str] = [], # list
+        token:str = None,
+        exact:bool = True,
+):
+    import requests
+    from urllib.parse import urlencode
+
+    if len(query) == 0:
+        log.error('CivitAI: empty query')
+        return []
+
+    t0 = time.time()
+    dct = { 'query': query }
+    if len(tag) > 0:
+        dct['tag'] = tag
+    if nsfw is not None:
+        dct['nsfw'] = 'true' if nsfw else 'false'
+    if limit > 0:
+        dct['limit'] = limit
+    if len(types) > 0:
+        dct['types'] = types
+    if len(sort) > 0:
+        dct['sort'] = sort
+    if len(period) > 0:
+        dct['period'] = period
+    if len(base) > 0:
+        dct['baseModels'] = ','.join(base)
+    encoded = urlencode(dct)
+
+    headers = {}
+    if token is None:
+        token = os.environ.get('CIVITAI_TOKEN', None)
+    if token is not None and len(token) > 0:
+        headers['Authorization'] = f'Bearer {token}'
+
+    url = 'https://civitai.com/api/v1/models'
+    uri = f'{url}?{encoded}'
+    log.info(f'CivitAI request: uri="{uri}" dct={dct} token={token is not None}')
+    result = requests.get(uri, headers=headers, timeout=60)
+
+    if result.status_code != 200:
+        log.error(f'CivitAI: code={result.status_code} reason={result.reason} uri={result.url}')
+        return []
+
+    models: list[Model] = []
+    exact_models: list[Model] = []
+    items = result.json().get('items', [])
+    for item in items:
+        models.append(Model(item))
+
+    if exact:
+        for model in models:
+            model_names = [model.name.lower()]
+            version_names = [v.name.lower() for v in model.versions]
+            file_names = [f.name.lower() for v in model.versions for f in v.files]
+            if any([query.lower() in name for name in model_names + version_names + file_names]):
+                exact_models.append(model)
+
+    t1 = time.time()
+    log.info(f'CivitAI result: code={result.status_code} exact={len(exact_models)} total={len(models)} time={t1-t0:.2f}')
+    return exact_models if len(exact_models) > 0 else models
+
+
+def print_models(models: list[Model]):
+    if debug:
+        from rich import print as dbg
+    else:
+        dbg = lambda *args, **kwargs: None # pylint: disable=unnecessary-lambda-assignment
+    for model in models:
+        log.info(f' {model}')
+        dbg('Model', model.dct)
+        for version in model.versions:
+            log.info(f'  {version}')
+            dbg('ModelVersion', version.dct)
+            for file in version.files:
+                log.info(f'   {file}')
+                dbg('ModelFile', file.dct)
+            for image in version.images:
+                log.info(f'   {image}')
+                dbg('ModelImage', image.dct)
+
+
+if __name__ == "__main__":
+    sys.argv.pop(0)
+    txt = ' '.join(sys.argv)
+    res = search_civitai(
+        query=txt,
+        # tag = '',
+        # types = '',
+        # sort = 'Most Downloaded',
+        # period = 'Year',
+        # nsfw = True,
+        # base = [],
+        # exact= True,
+        # limit=100,
+    )
+    print_models(res)
@@ -0,0 +1,293 @@
+import os
+import re
+import time
+import json
+import gradio as gr
+from modules.shared import log, opts, req, readfile, max_workers
+
+
+data = []
+selected_model = None
+update_data = []
+
+
+class CivitModel:
+    def __init__(self, name, fn, sha = None, meta = {}):
+        self.name = name
+        self.id = meta.get('id', 0)
+        self.fn = fn
+        self.sha = sha
+        self.meta = meta
+        self.versions = 0
+        self.vername = ''
+        self.latest = ''
+        self.latest_hashes = []
+        self.latest_name = ''
+        self.url = None
+        self.status = 'Not found'
+    def array(self):
+        return [self.id, self.fn, self.name, self.versions, self.vername, self.latest, self.status]
+
+
+def civit_update_metadata():
+    log.debug('CivitAI update metadata: models')
+    from modules import ui_extra_networks, modelloader
+    res = []
+    pages = ui_extra_networks.get_pages('Model')
+    if len(pages) == 0:
+        return 'CivitAI update metadata: no models found'
+    page: ui_extra_networks.ExtraNetworksPage = pages[0]
+    table_data = []
+    update_data.clear()
+    all_hashes = [(item.get('hash', None) or 'XXXXXXXX').upper()[:8] for item in page.list_items()]
+    for item in page.list_items():
+        model = CivitModel(name=item['name'], fn=item['filename'], sha=item.get('hash', None), meta=item.get('metadata', {}))
+        if model.sha is None or len(model.sha) == 0:
+            res.append(f'CivitAI skip search: name="{model.name}" hash=None')
+        else:
+            r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{model.sha}')
+            res.append(f'CivitAI search: name="{model.name}" hash={model.sha} status={r.status_code}')
+            if r.status_code == 200:
+                d = r.json()
+                model.id = d['modelId']
+                modelloader.download_civit_meta(model.fn, model.id)
+                fn = os.path.splitext(item['filename'])[0] + '.json'
+                model.meta = readfile(fn, silent=True)
+                model.name = model.meta.get('name', model.name)
+                model.versions = len(model.meta.get('modelVersions', []))
+        versions = model.meta.get('modelVersions', [])
+        if len(versions) > 0:
+            model.latest = versions[0].get('name', '')
+            model.latest_hashes.clear()
+            for v in versions[0].get('files', []):
+                for h in v.get('hashes', {}).values():
+                    model.latest_hashes.append(h[:8].upper())
+        for ver in versions:
+            for f in ver.get('files', []):
+                for h in f.get('hashes', {}).values():
+                    if h[:8].upper() == model.sha[:8].upper():
+                        model.vername = ver.get('name', '')
+                        model.url = f.get('downloadUrl', None)
+                        model.latest_name = f.get('name', '')
+                        if model.vername == model.latest:
+                            model.status = 'Latest'
+                        elif any(map(lambda v: v in model.latest_hashes, all_hashes)): # pylint: disable=cell-var-from-loop # noqa: C417
+                            model.status = 'Downloaded'
+                        else:
+                            model.status = 'Available'
+                        break
+        log.debug(res[-1])
+        update_data.append(model)
+        table_data.append(model.array())
+        yield gr.update(value=table_data), '<br>'.join([r for r in res if len(r) > 0])
+    return '<br>'.join([r for r in res if len(r) > 0])
+
+def civit_update_select(evt: gr.SelectData, in_data):
+    global selected_model # pylint: disable=global-statement
+    try:
+        selected_model = next([m for m in update_data if m.fn == in_data[evt.index[0]][1]])
+    except Exception:
+        selected_model = None
+    if selected_model is None or selected_model.url is None or selected_model.status != 'Available':
+        return [gr.update(value='Model update not available'), gr.update(visible=False)]
+    else:
+        return [gr.update(), gr.update(visible=True)]
+
+def civit_update_download():
+    if selected_model is None or selected_model.url is None or selected_model.status != 'Available':
+        return 'Model update not available'
+    if selected_model.latest_name is None or len(selected_model.latest_name) == 0:
+        model_name = f'{selected_model.name} {selected_model.latest}.safetensors'
+    else:
+        model_name = selected_model.latest_name
+    return civit_download_model(selected_model.url, model_name, model_path='', model_type='Model')
+
+
+def civit_search_model(name, tag, model_type):
+    # types = 'LORA' if model_type == 'LoRA' else 'Checkpoint'
+    url = 'https://civitai.com/api/v1/models?limit=25&Sort=Newest'
+    if model_type == 'Model':
+        url += '&types=Checkpoint'
+    elif model_type == 'LoRA':
+        url += '&types=LORA&types=DoRA&types=LoCon'
+    elif model_type == 'Embedding':
+        url += '&types=TextualInversion'
+    elif model_type == 'VAE':
+        url += '&types=VAE'
+    if name is not None and len(name) > 0:
+        url += f'&query={name}'
+    if tag is not None and len(tag) > 0:
+        url += f'&tag={tag}'
+    r = req(url)
+    log.debug(f'CivitAI search: type={model_type} name="{name}" tag={tag or "none"} url="{url}" status={r.status_code}')
+    if r.status_code != 200:
+        log.warning(f'CivitAI search: name="{name}" tag={tag} status={r.status_code}')
+        return [], gr.update(visible=False, value=[]), gr.update(visible=False, value=None), gr.update(visible=False, value=None)
+    try:
+        body = r.json()
+    except Exception as e:
+        log.error(f'CivitAI search: name="{name}" tag={tag} {e}')
+        return [], gr.update(visible=False, value=[]), gr.update(visible=False, value=None), gr.update(visible=False, value=None)
+    global data # pylint: disable=global-statement
+    data = body.get('items', [])
+    data1 = []
+    for model in data:
+        found = 0
+        if model_type == 'LoRA' and model['type'].lower() in ['lora', 'locon', 'dora', 'lycoris']:
+            found += 1
+        elif model_type == 'Embedding' and model['type'].lower() in ['textualinversion', 'embedding']:
+            found += 1
+        elif model_type == 'Model' and model['type'].lower() in ['checkpoint']:
+            found += 1
+        elif model_type == 'VAE' and model['type'].lower() in ['vae']:
+            found += 1
+        elif model_type == 'Other':
+            found += 1
+        if found > 0:
+            data1.append([
+                model['id'],
+                model['name'],
+                ', '.join(model['tags']),
+                model['stats']['downloadCount'],
+                model['stats']['rating']
+            ])
+    res = f'Search result: name={name} tag={tag or "none"} type={model_type} models={len(data1)}'
+    return res, gr.update(visible=len(data1) > 0, value=data1 if len(data1) > 0 else []), gr.update(visible=False, value=None), gr.update(visible=False, value=None)
+
+
+def civit_select1(evt: gr.SelectData, in_data):
+    model_id = in_data[evt.index[0]][0]
+    data2 = []
+    preview_img = None
+    for model in data:
+        if model['id'] == model_id:
+            for d in model['modelVersions']:
+                try:
+                    if d.get('images') is not None and len(d['images']) > 0 and len(d['images'][0]['url']) > 0:
+                        preview_img = d['images'][0]['url']
+                    data2.append([d.get('id', None), d.get('modelId', None) or model_id, d.get('name', None), d.get('baseModel', None), d.get('createdAt', None) or d.get('publishedAt', None)])
+                except Exception as e:
+                    log.error(f'CivitAI select: model="{in_data[evt.index[0]]}" {e}')
+                    log.error(f'CivitAI version data={type(d)}: {d}')
+    log.debug(f'CivitAI select: model="{in_data[evt.index[0]]}" versions={len(data2)}')
+    return data2, None, preview_img
+
+
+def civit_select2(evt: gr.SelectData, in_data):
+    variant_id = in_data[evt.index[0]][0]
+    model_id = in_data[evt.index[0]][1]
+    data3 = []
+    for model in data:
+        if model['id'] == model_id:
+            for variant in model['modelVersions']:
+                if variant['id'] == variant_id:
+                    for f in variant['files']:
+                        try:
+                            if os.path.splitext(f['name'])[1].lower() in ['.safetensors', '.ckpt', '.pt', '.pth', '.bin']:
+                                data3.append([f['name'], round(f['sizeKB']), json.dumps(f['metadata']), f['downloadUrl']])
+                        except Exception:
+                            pass
+    log.debug(f'CivitAI select: model="{in_data[evt.index[0]]}" files={len(data3)}')
+    return data3
+
+
+def civit_select3(evt: gr.SelectData, in_data):
+    log.debug(f'CivitAI select: variant={in_data[evt.index[0]]}')
+    return in_data[evt.index[0]][3], in_data[evt.index[0]][0], gr.update(interactive=True)
+
+
+def civit_download_model(model_url: str, model_name: str, model_path: str, model_type: str, token: str = None):
+    if model_url is None or len(model_url) == 0:
+        return 'No model selected'
+    try:
+        from modules.modelloader import download_civit_model
+        res = download_civit_model(model_url, model_name, model_path, model_type, token=token)
+    except Exception as e:
+        res = f"CivitAI model downloaded error: model={model_url} {e}"
+        log.error(res)
+        return res
+    from modules.sd_models import list_models  # pylint: disable=W0621
+    list_models()
+    return res
+
+
+def atomic_civit_search_metadata(item, res, rehash):
+    from modules.modelloader import download_civit_preview, download_civit_meta
+    if item is None:
+        return
+    meta = os.path.splitext(item['filename'])[0] + '.json'
+    has_meta = os.path.isfile(meta) and os.stat(meta).st_size > 0
+    if ('card-no-preview.png' in item['preview'] or not has_meta) and os.path.isfile(item['filename']):
+        sha = item.get('hash', None)
+        found = False
+        if sha is not None and len(sha) > 0:
+            r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{sha}')
+            log.debug(f'CivitAI search: name="{item["name"]}" hash={sha} status={r.status_code}')
+            if r.status_code == 200:
+                d = r.json()
+                res.append(download_civit_meta(item['filename'], d['modelId']))
+                if d.get('images') is not None:
+                    for i in d['images']:
+                        preview_url = i['url']
+                        img_res = download_civit_preview(item['filename'], preview_url)
+                        res.append(img_res)
+                        if 'error' not in img_res:
+                            found = True
+                            break
+        if not found and rehash and os.stat(item['filename']).st_size < (1024 * 1024 * 1024):
+            from modules import hashes
+            sha = hashes.calculate_sha256(item['filename'], quiet=True)[:10]
+            r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{sha}')
+            log.debug(f'CivitAI search: name="{item["name"]}" hash={sha} status={r.status_code}')
+            if r.status_code == 200:
+                d = r.json()
+                res.append(download_civit_meta(item['filename'], d['modelId']))
+                if d.get('images') is not None:
+                    for i in d['images']:
+                        preview_url = i['url']
+                        img_res = download_civit_preview(item['filename'], preview_url)
+                        res.append(img_res)
+                        if 'error' not in img_res:
+                            found = True
+                            break
+
+
+def civit_search_metadata(rehash, title):
+    log.debug(f'CivitAI search metadata: type={title if type(title) == str else "all"}')
+    from modules.ui_extra_networks import get_pages
+    res = []
+    scanned, skipped = 0, 0
+    t0 = time.time()
+    candidates = []
+    re_skip = [r.strip() for r in opts.extra_networks_scan_skip.split(',') if len(r.strip()) > 0]
+    log.debug(f'CivitAI search metadata: skip={re_skip}')
+    for page in get_pages():
+        if type(title) == str:
+            if page.title != title:
+                continue
+        if page.name == 'style':
+            continue
+        for item in page.list_items():
+            if item is None:
+                continue
+            if any(re.search(re_str, item.get('name', '') + item.get('filename', '')) for re_str in re_skip):
+                skipped += 1
+                continue
+            scanned += 1
+            candidates.append(item)
+            # atomic_civit_search_metadata(item, res, rehash)
+    import concurrent
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+        for fn in candidates:
+            executor.submit(atomic_civit_search_metadata, fn, res, rehash)
+    atomic_civit_search_metadata(None, res, rehash)
+    t1 = time.time()
+    log.debug(f'CivitAI search metadata: scanned={scanned} skipped={skipped} time={t1-t0:.2f}')
+    txt = '<br>'.join([r for r in res if len(r) > 0])
+    return txt
+
+
+def civitai_update_token(token):
+    log.debug('CivitAI update token')
+    opts.civitai_token = token
+    opts.save()
@@ -0,0 +1,42 @@
+import os
+import gradio as gr
+from modules.shared import log, opts
+
+
+def hf_init():
+    os.environ.setdefault('HF_HUB_DISABLE_EXPERIMENTAL_WARNING', '1')
+    os.environ.setdefault('HF_HUB_DISABLE_SYMLINKS_WARNING', '1')
+    os.environ.setdefault('HF_HUB_DISABLE_IMPLICIT_TOKEN', '1')
+    os.environ.setdefault('HUGGINGFACE_HUB_VERBOSITY', 'warning')
+
+
+def hf_search(keyword):
+    hf_init()
+    import huggingface_hub as hf
+    hf_api = hf.HfApi()
+    models = hf_api.list_models(model_name=keyword, full=True, library="diffusers", limit=50, sort="downloads", direction=-1)
+    data = []
+    for model in models:
+        tags = [t for t in model.tags if not t.startswith('diffusers') and not t.startswith('license') and not t.startswith('arxiv') and len(t) > 2]
+        data.append([model.id, model.pipeline_tag, tags, model.downloads, model.lastModified, f'https://huggingface.co/{model.id}'])
+    return data
+
+
+def hf_select(evt: gr.SelectData, data):
+    return data[evt.index[0]][0]
+
+
+def hf_download_model(hub_id: str, token, variant, revision, mirror, custom_pipeline):
+    hf_init()
+    from modules.modelloader import download_diffusers_model
+    download_diffusers_model(hub_id, cache_dir=opts.diffusers_dir, token=token, variant=variant, revision=revision, mirror=mirror, custom_pipeline=custom_pipeline)
+    from modules.sd_models import list_models  # pylint: disable=W0621
+    list_models()
+    log.info(f'Diffuser model downloaded: model="{hub_id}"')
+    return f'Diffuser model downloaded: model="{hub_id}"'
+
+
+def hf_update_token(token):
+    log.debug('Huggingface update token')
+    opts.huggingface_token = token
+    opts.save()
@@ -334,7 +334,7 @@ def parse_prompt_attention(text):
        whitespace = ''
    else:
        re_attention = re_attention_v2
-        if native and opts.sd_textencder_linebreak:
+        if opts.sd_textencder_linebreak:
            text = text.replace('\n', ' BREAK ')
        else:
            text = text.replace('\n', ' ')
@@ -127,7 +127,7 @@ def list_models():
    global checkpoints_list # pylint: disable=global-statement
    checkpoints_list.clear()
    checkpoint_aliases.clear()
-    ext_filter = [".safetensors"] if shared.opts.sd_disable_ckpt or shared.native else [".ckpt", ".safetensors"]
+    ext_filter = [".safetensors"]
    model_list = list(modelloader.load_models(model_path=model_path, model_url=None, command_path=shared.opts.ckpt_dir, ext_filter=ext_filter, download_name=None, ext_blacklist=[".vae.ckpt", ".vae.safetensors"]))
    safetensors_list = []
    for filename in sorted(model_list, key=str.lower):
@@ -136,21 +136,16 @@ def list_models():
        if checkpoint_info.name is not None:
            checkpoint_info.register()
    diffusers_list = []
-    if shared.native:
-        for repo in modelloader.load_diffusers_models(clear=True):
-            checkpoint_info = CheckpointInfo(repo['name'], sha=repo['hash'])
-            diffusers_list.append(checkpoint_info)
-            if checkpoint_info.name is not None:
-                checkpoint_info.register()
+    for repo in modelloader.load_diffusers_models(clear=True):
+        checkpoint_info = CheckpointInfo(repo['name'], sha=repo['hash'])
+        diffusers_list.append(checkpoint_info)
+        if checkpoint_info.name is not None:
+            checkpoint_info.register()
    if shared.cmd_opts.ckpt is not None:
-        if not os.path.exists(shared.cmd_opts.ckpt) and not shared.native:
-            if shared.cmd_opts.ckpt.lower() != "none":
-                shared.log.warning(f'Load model: path="{shared.cmd_opts.ckpt}" not found')
-        else:
-            checkpoint_info = CheckpointInfo(shared.cmd_opts.ckpt)
-            if checkpoint_info.name is not None:
-                checkpoint_info.register()
-                shared.opts.data['sd_model_checkpoint'] = checkpoint_info.title
+        checkpoint_info = CheckpointInfo(shared.cmd_opts.ckpt)
+        if checkpoint_info.name is not None:
+            checkpoint_info.register()
+            shared.opts.data['sd_model_checkpoint'] = checkpoint_info.title
    elif shared.cmd_opts.ckpt != shared.default_sd_model_file and shared.cmd_opts.ckpt is not None:
        shared.log.warning(f'Load model: path="{shared.cmd_opts.ckpt}" not found')
    shared.log.info(f'Available Models: safetensors="{shared.opts.ckpt_dir}":{len(safetensors_list)} diffusers="{shared.opts.diffusers_dir}":{len(diffusers_list)} items={len(checkpoints_list)} time={time.time()-t0:.2f}')
@@ -264,7 +264,7 @@ options_templates.update(options_section(('cuda', "Compute Settings"), {
    "diffusers_generator_device": OptionInfo("GPU", "Generator device", gr.Radio, {"choices": ["GPU", "CPU", "Unset"]}),

    "cross_attention_sep": OptionInfo("<h2>Cross Attention</h2>", "", gr.HTML),
-    "cross_attention_optimization": OptionInfo(startup_cross_attention, "Attention optimization method", gr.Radio, lambda: {"choices": shared_items.list_crossattention(native)}),
+    "cross_attention_optimization": OptionInfo(startup_cross_attention, "Attention optimization method", gr.Radio, lambda: {"choices": shared_items.list_crossattention()}),
    "sdp_options": OptionInfo(startup_sdp_options, "SDP options", gr.CheckboxGroup, {"choices": ['Flash attention', 'Memory attention', 'Math attention', 'Dynamic attention', 'CK Flash attention', 'Sage attention']}),
    "xformers_options": OptionInfo(['Flash attention'], "xFormers options", gr.CheckboxGroup, {"choices": ['Flash attention'] }),
    "dynamic_attention_slice_rate": OptionInfo(0.5, "Dynamic Attention slicing rate in GB", gr.Slider, {"minimum": 0.01, "maximum": max(gpu_memory,4), "step": 0.01}),
@@ -105,26 +105,15 @@ def refresh_te_list():
    modules.model_te.refresh_te_list()


-def list_crossattention(native:bool=True):
-    if native:
-        return [
-            "Disabled",
-            "Scaled-Dot-Product",
-            "xFormers",
-            "Batch matrix-matrix",
-            "Split attention",
-            "Dynamic Attention BMM"
-        ]
-    else:
-        return [
-            "Disabled",
-            "Scaled-Dot-Product",
-            "xFormers",
-            "Doggettx's",
-            "InvokeAI's",
-            "Sub-quadratic",
-            "Split attention"
-        ]
+def list_crossattention():
+    return [
+        "Disabled",
+        "Scaled-Dot-Product",
+        "xFormers",
+        "Batch matrix-matrix",
+        "Split attention",
+        "Dynamic Attention BMM"
+    ]

 def get_pipelines():
    from installer import log
@@ -13,7 +13,7 @@ supported_models = ['ldm', 'sd', 'sdxl']


 def list_embeddings(*dirs):
-    is_ext = extension_filter(['.SAFETENSORS', '.PT' ] + ( ['.PNG', '.WEBP', '.JXL', '.AVIF', '.BIN' ] if not shared.native else [] ))
+    is_ext = extension_filter(['.SAFETENSORS', '.PT' ])
    is_not_preview = lambda fp: not next(iter(os.path.splitext(fp))).upper().endswith('.PREVIEW') # pylint: disable=unnecessary-lambda-assignment
    return list(filter(lambda fp: is_ext(fp) and is_not_preview(fp) and os.stat(fp).st_size > 0, directory_files(*dirs)))

@@ -932,9 +932,8 @@ def create_ui(container, button_parent, tabname, skip_indexing = False):
        return pages

    def ui_scan_click(title):
-        from modules import ui_models
-        if ui_models.search_metadata_civit is not None:
-            ui_models.search_metadata_civit(True, title)
+        from modules.models_civitai import civit_search_metadata
+        civit_search_metadata(True, title)
        return ui_refresh_click(title)

    def ui_save_click():
@@ -1,21 +1,15 @@
 import os
-import re
-import time
 import json
 import inspect
 from datetime import datetime
 import gradio as gr
-from modules import errors, sd_models, sd_vae, extras, sd_samplers, ui_symbols, hashes
+from modules import errors, sd_models, sd_vae, extras, sd_samplers, ui_symbols
 from modules.ui_components import ToolButton
 from modules.ui_common import create_refresh_button
 from modules.call_queue import wrap_gradio_gpu_call
-from modules.shared import opts, log, req, readfile, max_workers, native
-from modules.merging import merge_methods
-from modules.merging.merge_utils import BETA_METHODS, TRIPLE_METHODS, interpolate
-from modules.merging.merge_presets import BLOCK_WEIGHTS_PRESETS, SDXL_BLOCK_WEIGHTS_PRESETS
+from modules.shared import opts, log


-search_metadata_civit = None
 extra_ui = []


@@ -67,6 +61,10 @@ def create_ui():
                ui_models_load.create_ui(models_outcome, models_file)

            with gr.Tab(label="Merge"):
+                from modules.merging import merge_methods
+                from modules.merging.merge_utils import BETA_METHODS, TRIPLE_METHODS, interpolate
+                from modules.merging.merge_presets import BLOCK_WEIGHTS_PRESETS, SDXL_BLOCK_WEIGHTS_PRESETS
+
                def sd_model_choices():
                    return ['None'] + sd_models.checkpoint_titles()

@@ -420,38 +418,7 @@ def create_ui():
                model_list_btn.click(fn=list_models, inputs=[], outputs=[model_table, models_outcome])

            with gr.Tab(label="Huggingface"):
-                data = []
-                os.environ.setdefault('HF_HUB_DISABLE_EXPERIMENTAL_WARNING', '1')
-                os.environ.setdefault('HF_HUB_DISABLE_SYMLINKS_WARNING', '1')
-                os.environ.setdefault('HF_HUB_DISABLE_IMPLICIT_TOKEN', '1')
-                os.environ.setdefault('HUGGINGFACE_HUB_VERBOSITY', 'warning')
-
-                def hf_search(keyword):
-                    import huggingface_hub as hf
-                    hf_api = hf.HfApi()
-                    models = hf_api.list_models(model_name=keyword, full=True, library="diffusers", limit=50, sort="downloads", direction=-1)
-                    data.clear()
-                    for model in models:
-                        tags = [t for t in model.tags if not t.startswith('diffusers') and not t.startswith('license') and not t.startswith('arxiv') and len(t) > 2]
-                        data.append([model.id, model.pipeline_tag, tags, model.downloads, model.lastModified, f'https://huggingface.co/{model.id}'])
-                    return data
-
-                def hf_select(evt: gr.SelectData, data):
-                    return data[evt.index[0]][0]
-
-                def hf_download_model(hub_id: str, token, variant, revision, mirror, custom_pipeline):
-                    from modules.modelloader import download_diffusers_model
-                    download_diffusers_model(hub_id, cache_dir=opts.diffusers_dir, token=token, variant=variant, revision=revision, mirror=mirror, custom_pipeline=custom_pipeline)
-                    from modules.sd_models import list_models  # pylint: disable=W0621
-                    list_models()
-                    log.info(f'Diffuser model downloaded: model="{hub_id}"')
-                    return f'Diffuser model downloaded: model="{hub_id}"'
-
-                def hf_update_token(token):
-                    log.debug('Huggingface update token')
-                    opts.huggingface_token = token
-                    opts.save()
-
+                from modules.models_hf import hf_search, hf_select, hf_download_model, hf_update_token
                with gr.Column(scale=6):
                    with gr.Row():
                        gr.HTML('<h2>&nbspDownload model from huggingface<br></h2>')
@@ -486,192 +453,7 @@ def create_ui():
                hf_token.change(fn=hf_update_token, inputs=[hf_token], outputs=[])

            with gr.Tab(label="CivitAI"):
-                data = []
-
-                def civit_search_model(name, tag, model_type):
-                    # types = 'LORA' if model_type == 'LoRA' else 'Checkpoint'
-                    url = 'https://civitai.com/api/v1/models?limit=25&Sort=Newest'
-                    if model_type == 'Model':
-                        url += '&types=Checkpoint'
-                    elif model_type == 'LoRA':
-                        url += '&types=LORA&types=DoRA&types=LoCon'
-                    elif model_type == 'Embedding':
-                        url += '&types=TextualInversion'
-                    elif model_type == 'VAE':
-                        url += '&types=VAE'
-                    if name is not None and len(name) > 0:
-                        url += f'&query={name}'
-                    if tag is not None and len(tag) > 0:
-                        url += f'&tag={tag}'
-                    r = req(url)
-                    log.debug(f'CivitAI search: type={model_type} name="{name}" tag={tag or "none"} url="{url}" status={r.status_code}')
-                    if r.status_code != 200:
-                        log.warning(f'CivitAI search: name="{name}" tag={tag} status={r.status_code}')
-                        return [], gr.update(visible=False, value=[]), gr.update(visible=False, value=None), gr.update(visible=False, value=None)
-                    try:
-                        body = r.json()
-                    except Exception as e:
-                        log.error(f'CivitAI search: name="{name}" tag={tag} {e}')
-                        return [], gr.update(visible=False, value=[]), gr.update(visible=False, value=None), gr.update(visible=False, value=None)
-                    nonlocal data
-                    data = body.get('items', [])
-                    data1 = []
-                    for model in data:
-                        found = 0
-                        if model_type == 'LoRA' and model['type'].lower() in ['lora', 'locon', 'dora', 'lycoris']:
-                            found += 1
-                        elif model_type == 'Embedding' and model['type'].lower() in ['textualinversion', 'embedding']:
-                            found += 1
-                        elif model_type == 'Model' and model['type'].lower() in ['checkpoint']:
-                            found += 1
-                        elif model_type == 'VAE' and model['type'].lower() in ['vae']:
-                            found += 1
-                        elif model_type == 'Other':
-                            found += 1
-                        if found > 0:
-                            data1.append([
-                                model['id'],
-                                model['name'],
-                                ', '.join(model['tags']),
-                                model['stats']['downloadCount'],
-                                model['stats']['rating']
-                            ])
-                    res = f'Search result: name={name} tag={tag or "none"} type={model_type} models={len(data1)}'
-                    return res, gr.update(visible=len(data1) > 0, value=data1 if len(data1) > 0 else []), gr.update(visible=False, value=None), gr.update(visible=False, value=None)
-
-                def civit_select1(evt: gr.SelectData, in_data):
-                    model_id = in_data[evt.index[0]][0]
-                    data2 = []
-                    preview_img = None
-                    for model in data:
-                        if model['id'] == model_id:
-                            for d in model['modelVersions']:
-                                try:
-                                    if d.get('images') is not None and len(d['images']) > 0 and len(d['images'][0]['url']) > 0:
-                                        preview_img = d['images'][0]['url']
-                                    data2.append([d.get('id', None), d.get('modelId', None) or model_id, d.get('name', None), d.get('baseModel', None), d.get('createdAt', None) or d.get('publishedAt', None)])
-                                except Exception as e:
-                                    log.error(f'CivitAI select: model="{in_data[evt.index[0]]}" {e}')
-                                    log.error(f'CivitAI version data={type(d)}: {d}')
-                    log.debug(f'CivitAI select: model="{in_data[evt.index[0]]}" versions={len(data2)}')
-                    return data2, None, preview_img
-
-                def civit_select2(evt: gr.SelectData, in_data):
-                    variant_id = in_data[evt.index[0]][0]
-                    model_id = in_data[evt.index[0]][1]
-                    data3 = []
-                    for model in data:
-                        if model['id'] == model_id:
-                            for variant in model['modelVersions']:
-                                if variant['id'] == variant_id:
-                                    for f in variant['files']:
-                                        try:
-                                            if os.path.splitext(f['name'])[1].lower() in ['.safetensors', '.ckpt', '.pt', '.pth', '.bin']:
-                                                data3.append([f['name'], round(f['sizeKB']), json.dumps(f['metadata']), f['downloadUrl']])
-                                        except Exception:
-                                            pass
-                    log.debug(f'CivitAI select: model="{in_data[evt.index[0]]}" files={len(data3)}')
-                    return data3
-
-                def civit_select3(evt: gr.SelectData, in_data):
-                    log.debug(f'CivitAI select: variant={in_data[evt.index[0]]}')
-                    return in_data[evt.index[0]][3], in_data[evt.index[0]][0], gr.update(interactive=True)
-
-                def civit_download_model(model_url: str, model_name: str, model_path: str, model_type: str, token: str = None):
-                    if model_url is None or len(model_url) == 0:
-                        return 'No model selected'
-                    try:
-                        from modules.modelloader import download_civit_model
-                        res = download_civit_model(model_url, model_name, model_path, model_type, token=token)
-                    except Exception as e:
-                        res = f"CivitAI model downloaded error: model={model_url} {e}"
-                        log.error(res)
-                        return res
-                    from modules.sd_models import list_models  # pylint: disable=W0621
-                    list_models()
-                    return res
-
-                def atomic_civit_search_metadata(item, res, rehash):
-                    from modules.modelloader import download_civit_preview, download_civit_meta
-                    if item is None:
-                        return
-                    meta = os.path.splitext(item['filename'])[0] + '.json'
-                    has_meta = os.path.isfile(meta) and os.stat(meta).st_size > 0
-                    if ('card-no-preview.png' in item['preview'] or not has_meta) and os.path.isfile(item['filename']):
-                        sha = item.get('hash', None)
-                        found = False
-                        if sha is not None and len(sha) > 0:
-                            r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{sha}')
-                            log.debug(f'CivitAI search: name="{item["name"]}" hash={sha} status={r.status_code}')
-                            if r.status_code == 200:
-                                d = r.json()
-                                res.append(download_civit_meta(item['filename'], d['modelId']))
-                                if d.get('images') is not None:
-                                    for i in d['images']:
-                                        preview_url = i['url']
-                                        img_res = download_civit_preview(item['filename'], preview_url)
-                                        res.append(img_res)
-                                        if 'error' not in img_res:
-                                            found = True
-                                            break
-                        if not found and rehash and os.stat(item['filename']).st_size < (1024 * 1024 * 1024):
-                            sha = hashes.calculate_sha256(item['filename'], quiet=True)[:10]
-                            r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{sha}')
-                            log.debug(f'CivitAI search: name="{item["name"]}" hash={sha} status={r.status_code}')
-                            if r.status_code == 200:
-                                d = r.json()
-                                res.append(download_civit_meta(item['filename'], d['modelId']))
-                                if d.get('images') is not None:
-                                    for i in d['images']:
-                                        preview_url = i['url']
-                                        img_res = download_civit_preview(item['filename'], preview_url)
-                                        res.append(img_res)
-                                        if 'error' not in img_res:
-                                            found = True
-                                            break
-
-                def civit_search_metadata(rehash, title):
-                    log.debug(f'CivitAI search metadata: type={title if type(title) == str else "all"}')
-                    from modules.ui_extra_networks import get_pages
-                    res = []
-                    scanned, skipped = 0, 0
-                    t0 = time.time()
-                    candidates = []
-                    re_skip = [r.strip() for r in opts.extra_networks_scan_skip.split(',') if len(r.strip()) > 0]
-                    log.debug(f'CivitAI search metadata: skip={re_skip}')
-                    for page in get_pages():
-                        if type(title) == str:
-                            if page.title != title:
-                                continue
-                        if page.name == 'style':
-                            continue
-                        for item in page.list_items():
-                            if item is None:
-                                continue
-                            if any(re.search(re_str, item.get('name', '') + item.get('filename', '')) for re_str in re_skip):
-                                skipped += 1
-                                continue
-                            scanned += 1
-                            candidates.append(item)
-                            # atomic_civit_search_metadata(item, res, rehash)
-                    import concurrent
-                    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
-                        for fn in candidates:
-                            executor.submit(atomic_civit_search_metadata, fn, res, rehash)
-                    atomic_civit_search_metadata(None, res, rehash)
-                    t1 = time.time()
-                    log.debug(f'CivitAI search metadata: scanned={scanned} skipped={skipped} time={t1-t0:.2f}')
-                    txt = '<br>'.join([r for r in res if len(r) > 0])
-                    return txt
-
-                global search_metadata_civit  # pylint: disable=global-statement
-                search_metadata_civit = civit_search_metadata
-
-                def civitai_update_token(token):
-                    log.debug('CivitAI update token')
-                    opts.civitai_token = token
-                    opts.save()
-
+                from modules.models_civitai import civitai_update_token, civit_search_model, civit_search_metadata, civit_select1, civit_select2, civit_select3, civit_download_model
                with gr.Row():
                    gr.HTML('<h2>&nbspCivitAI fetch metadata<br></h2>')
                    gr.HTML('Fetches preview and metadata information for all models with missing information<br>Models with existing previews and information are not updated<br>')
@@ -737,6 +519,7 @@ def create_ui():
                civit_previews_btn.click(fn=civit_search_metadata, inputs=[civit_previews_rehash, civit_previews_rehash], outputs=[models_outcome])

            with gr.Tab(label="Update"):
+                from modules.models_civitai import civit_update_metadata, civit_update_select, civit_update_download
                with gr.Row():
                    gr.HTML('<h2>&nbspScan CivitAI for information on latest available model versions<br></h2>')
                with gr.Row():
@@ -753,107 +536,12 @@ def create_ui():
                with gr.Row():
                    civit_update_download_btn = gr.Button(value="Download", variant='primary', visible=False)

-                class CivitModel:
-                    def __init__(self, name, fn, sha = None, meta = {}):
-                        self.name = name
-                        self.id = meta.get('id', 0)
-                        self.fn = fn
-                        self.sha = sha
-                        self.meta = meta
-                        self.versions = 0
-                        self.vername = ''
-                        self.latest = ''
-                        self.latest_hashes = []
-                        self.latest_name = ''
-                        self.url = None
-                        self.status = 'Not found'
-                    def array(self):
-                        return [self.id, self.fn, self.name, self.versions, self.vername, self.latest, self.status]
-
-                selected_model: CivitModel = None
-                update_data = []
-
-                def civit_update_metadata():
-                    nonlocal update_data
-                    log.debug('CivitAI update metadata: models')
-                    from modules import ui_extra_networks, modelloader
-                    res = []
-                    pages = ui_extra_networks.get_pages('Model')
-                    if len(pages) == 0:
-                        return 'CivitAI update metadata: no models found'
-                    page: ui_extra_networks.ExtraNetworksPage = pages[0]
-                    table_data = []
-                    update_data.clear()
-                    all_hashes = [(item.get('hash', None) or 'XXXXXXXX').upper()[:8] for item in page.list_items()]
-                    for item in page.list_items():
-                        model = CivitModel(name=item['name'], fn=item['filename'], sha=item.get('hash', None), meta=item.get('metadata', {}))
-                        if model.sha is None or len(model.sha) == 0:
-                            res.append(f'CivitAI skip search: name="{model.name}" hash=None')
-                        else:
-                            r = req(f'https://civitai.com/api/v1/model-versions/by-hash/{model.sha}')
-                            res.append(f'CivitAI search: name="{model.name}" hash={model.sha} status={r.status_code}')
-                            if r.status_code == 200:
-                                d = r.json()
-                                model.id = d['modelId']
-                                modelloader.download_civit_meta(model.fn, model.id)
-                                fn = os.path.splitext(item['filename'])[0] + '.json'
-                                model.meta = readfile(fn, silent=True)
-                                model.name = model.meta.get('name', model.name)
-                                model.versions = len(model.meta.get('modelVersions', []))
-                        versions = model.meta.get('modelVersions', [])
-                        if len(versions) > 0:
-                            model.latest = versions[0].get('name', '')
-                            model.latest_hashes.clear()
-                            for v in versions[0].get('files', []):
-                                for h in v.get('hashes', {}).values():
-                                    model.latest_hashes.append(h[:8].upper())
-                        for ver in versions:
-                            for f in ver.get('files', []):
-                                for h in f.get('hashes', {}).values():
-                                    if h[:8].upper() == model.sha[:8].upper():
-                                        model.vername = ver.get('name', '')
-                                        model.url = f.get('downloadUrl', None)
-                                        model.latest_name = f.get('name', '')
-                                        if model.vername == model.latest:
-                                            model.status = 'Latest'
-                                        elif any(map(lambda v: v in model.latest_hashes, all_hashes)): # pylint: disable=cell-var-from-loop # noqa: C417
-                                            model.status = 'Downloaded'
-                                        else:
-                                            model.status = 'Available'
-                                        break
-                        log.debug(res[-1])
-                        update_data.append(model)
-                        table_data.append(model.array())
-                        yield gr.update(value=table_data), '<br>'.join([r for r in res if len(r) > 0])
-                    return '<br>'.join([r for r in res if len(r) > 0])
-
-                def civit_update_select(evt: gr.SelectData, in_data):
-                    nonlocal selected_model, update_data
-                    try:
-                        selected_model = next([m for m in update_data if m.fn == in_data[evt.index[0]][1]])
-                    except Exception:
-                        selected_model = None
-                    if selected_model is None or selected_model.url is None or selected_model.status != 'Available':
-                        return [gr.update(value='Model update not available'), gr.update(visible=False)]
-                    else:
-                        return [gr.update(), gr.update(visible=True)]
-
-                def civit_update_download():
-                    if selected_model is None or selected_model.url is None or selected_model.status != 'Available':
-                        return 'Model update not available'
-                    if selected_model.latest_name is None or len(selected_model.latest_name) == 0:
-                        model_name = f'{selected_model.name} {selected_model.latest}.safetensors'
-                    else:
-                        model_name = selected_model.latest_name
-                    return civit_download_model(selected_model.url, model_name, model_path='', model_type='Model')
-
                civit_update_btn.click(fn=civit_update_metadata, inputs=[], outputs=[civit_results4, models_outcome])
                civit_results4.select(fn=civit_update_select, inputs=[civit_results4], outputs=[models_outcome, civit_update_download_btn])
                civit_update_download_btn.click(fn=civit_update_download, inputs=[], outputs=[models_outcome])

-            if native:
-                from modules.lora.lora_extract import create_ui as lora_extract_ui
-                lora_extract_ui()
+            from modules.lora.lora_extract import create_ui as lora_extract_ui
+            lora_extract_ui()

            for ui in extra_ui:
                if callable(ui):