Spaces:
Runtime error
Runtime error
File size: 6,625 Bytes
c19ca42 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 |
import os
from datetime import datetime
import git
from modules import shared, errors
from modules.paths import extensions_dir, extensions_builtin_dir
extensions = []
if not os.path.exists(extensions_dir):
os.makedirs(extensions_dir)
def active():
if shared.opts.disable_all_extensions == "all":
return []
elif shared.opts.disable_all_extensions == "user":
return [x for x in extensions if x.enabled and x.is_builtin]
else:
return [x for x in extensions if x.enabled]
class Extension:
def __init__(self, name, path, enabled=True, is_builtin=False):
self.name = name
self.git_name = ''
self.path = path
self.enabled = enabled
self.status = ''
self.can_update = False
self.is_builtin = is_builtin
self.commit_hash = ''
self.commit_date = None
self.version = ''
self.description = ''
self.branch = None
self.remote = None
self.have_info_from_repo = False
self.mtime = 0
self.ctime = 0
def read_info(self, force=False):
if self.have_info_from_repo and not force:
return
self.have_info_from_repo = True
repo = None
self.mtime = datetime.fromtimestamp(os.path.getmtime(self.path)).isoformat() + 'Z'
self.ctime = datetime.fromtimestamp(os.path.getctime(self.path)).isoformat() + 'Z'
try:
if os.path.exists(os.path.join(self.path, ".git")):
repo = git.Repo(self.path)
except Exception as e:
errors.display(e, f'github info from {self.path}')
if repo is None or repo.bare:
self.remote = None
else:
try:
self.status = 'unknown'
if len(repo.remotes) == 0:
shared.log.debug(f"Extension: no remotes info repo={self.name}")
return
self.git_name = repo.remotes.origin.url.split('.git')[0].split('/')[-1]
self.description = repo.description
if self.description is None or self.description.startswith("Unnamed repository"):
self.description = "[No description]"
self.remote = next(repo.remote().urls, None)
head = repo.head.commit
self.commit_date = repo.head.commit.committed_date
try:
if repo.active_branch:
self.branch = repo.active_branch.name
except Exception:
pass
self.commit_hash = head.hexsha
self.version = f"<p>{self.commit_hash[:8]}</p><p>{datetime.fromtimestamp(self.commit_date).strftime('%a %b%d %Y %H:%M')}</p>"
except Exception as ex:
shared.log.error(f"Extension: failed reading data from git repo={self.name}: {ex}")
self.remote = None
def list_files(self, subdir, extension):
from modules import scripts
dirpath = os.path.join(self.path, subdir)
if not os.path.isdir(dirpath):
return []
res = []
for filename in sorted(os.listdir(dirpath)):
if not filename.endswith(".py") and not filename.endswith(".js") and not filename.endswith(".mjs"):
continue
priority = '50'
if os.path.isfile(os.path.join(dirpath, "..", ".priority")):
with open(os.path.join(dirpath, "..", ".priority"), "r", encoding="utf-8") as f:
priority = str(f.read().strip())
res.append(scripts.ScriptFile(self.path, filename, os.path.join(dirpath, filename), priority))
if priority != '50':
shared.log.debug(f'Extension priority override: {os.path.dirname(dirpath)}:{priority}')
res = [x for x in res if os.path.splitext(x.path)[1].lower() == extension and os.path.isfile(x.path)]
return res
def check_updates(self):
try:
repo = git.Repo(self.path)
except Exception:
self.can_update = False
return
for fetch in repo.remote().fetch(dry_run=True):
if fetch.flags != fetch.HEAD_UPTODATE:
self.can_update = True
self.status = "new commits"
return
try:
origin = repo.rev_parse('origin')
if repo.head.commit != origin:
self.can_update = True
self.status = "behind HEAD"
return
except Exception:
self.can_update = False
self.status = "unknown (remote error)"
return
self.can_update = False
self.status = "latest"
def git_fetch(self, commit='origin'):
repo = git.Repo(self.path)
# Fix: `error: Your local changes to the following files would be overwritten by merge`,
# because WSL2 Docker set 755 file permissions instead of 644, this results to the error.
repo.git.fetch(all=True)
repo.git.reset('origin', hard=True)
repo.git.reset(commit, hard=True)
self.have_info_from_repo = False
def list_extensions():
extensions.clear()
if not os.path.isdir(extensions_dir):
return
if shared.opts.disable_all_extensions == "all" or shared.opts.disable_all_extensions == "user":
shared.log.warning(f"Option set: Disable extensions: {shared.opts.disable_all_extensions}")
extension_paths = []
extension_names = []
extension_folders = [extensions_builtin_dir] if shared.cmd_opts.safe else [extensions_builtin_dir, extensions_dir]
for dirname in extension_folders:
if not os.path.isdir(dirname):
return
for extension_dirname in sorted(os.listdir(dirname)):
path = os.path.join(dirname, extension_dirname)
if not os.path.isdir(path):
continue
if extension_dirname in extension_names:
shared.log.info(f'Skipping conflicting extension: {path}')
continue
extension_names.append(extension_dirname)
extension_paths.append((extension_dirname, path, dirname == extensions_builtin_dir))
disabled_extensions = shared.opts.disabled_extensions + shared.temp_disable_extensions()
for dirname, path, is_builtin in extension_paths:
extension = Extension(name=dirname, path=path, enabled=dirname not in disabled_extensions, is_builtin=is_builtin)
extensions.append(extension)
shared.log.info(f'Disabled extensions: {[e.name for e in extensions if not e.enabled]}')
|