mirror of
https://github.com/open-webui/open-webui.git
synced 2026-10-14 04:08:46 +00:00
520 lines
23 KiB
Python
520 lines
23 KiB
Python
"""Portable skill snapshots. Paths are virtual; packages are never extracted to disk."""
|
|
|
|
import asyncio
|
|
import base64
|
|
import io
|
|
import json
|
|
import re
|
|
import stat
|
|
import time
|
|
import zipfile
|
|
from pathlib import PurePosixPath
|
|
from typing import Literal
|
|
from urllib.parse import quote, unquote, urljoin, urlsplit
|
|
|
|
import aiohttp
|
|
import yaml
|
|
from pydantic import BaseModel, ConfigDict
|
|
|
|
MAX_FILE_BYTES = 10 * 1024 * 1024
|
|
MAX_SKILL_BYTES = 50 * 1024 * 1024
|
|
MAX_IMPORT_BYTES = 200 * 1024 * 1024
|
|
MAX_FILES = 1000
|
|
SKILL_CONTENT_MAX_CHARS = 20_000
|
|
SKILL_INJECTION_MAX_CHARS = 100_000
|
|
SKILL_PREVIEW_MAX_CHARS = 5_000
|
|
SKILL_OVERVIEW_LIST_MAX_CHARS = 2_500
|
|
SKILL_MANIFEST_MAX_ENTRIES = 50
|
|
SKILL_MANIFEST_MAX_CHARS = 5_000
|
|
SKILL_GREP_MAX_CHARS = 6_000
|
|
SKILL_GREP_EXCERPT_MAX_CHARS = 1_200
|
|
SKILL_GREP_LOCATIONS_MAX_CHARS = 2_000
|
|
|
|
|
|
def skill_content_page(content: str, offset: int = 0, max_chars: int = SKILL_CONTENT_MAX_CHARS) -> dict:
|
|
offset = max(0, offset)
|
|
end = offset + min(SKILL_CONTENT_MAX_CHARS, max(1, max_chars))
|
|
return {'content': content[offset:end], 'next_offset': end if end < len(content) else None}
|
|
|
|
|
|
def bounded_skill_manifest(entries: list, field: str = 'files', max_chars: int = SKILL_MANIFEST_MAX_CHARS) -> dict:
|
|
bounded, size = [], 2 # Include the JSON array brackets and separators in the budget.
|
|
for entry in entries[:SKILL_MANIFEST_MAX_ENTRIES]:
|
|
entry_size = len(json.dumps(entry, ensure_ascii=False)) + (2 if bounded else 0)
|
|
if size + entry_size > max_chars:
|
|
break
|
|
bounded.append(entry)
|
|
size += entry_size
|
|
result = {field: bounded}
|
|
if len(bounded) < len(entries):
|
|
result['notice'] = 'Additional supporting files omitted.'
|
|
return result
|
|
|
|
|
|
def iter_skill_headings(content: str):
|
|
offset = 0
|
|
frontmatter_match = re.match(r'\A---[ \t]*\r?\n.*?\r?\n---[ \t]*(?:\r?\n|$)', content, re.DOTALL)
|
|
body_start = frontmatter_match.end() if frontmatter_match else 0
|
|
fence = None
|
|
for line in content.splitlines(keepends=True):
|
|
line_offset = offset
|
|
offset += len(line)
|
|
if line_offset < body_start:
|
|
continue
|
|
text = line.rstrip('\r\n')
|
|
marker = re.match(r' {0,3}(`{3,}|~{3,})(.*)$', text)
|
|
if fence:
|
|
if marker and marker[1][0] == fence[0] and len(marker[1]) >= len(fence) and not marker[2].strip():
|
|
fence = None
|
|
continue
|
|
if marker and (marker[1][0] == '~' or '`' not in marker[2]):
|
|
fence = marker[1]
|
|
continue
|
|
heading = re.match(r' {0,3}(#{1,6})(?:[ \t]+(.*)|$)', text)
|
|
if heading:
|
|
title = re.sub(r'(?:^|[ \t]+)#+[ \t]*$', '', heading[2] or '').strip()
|
|
yield {'title': title, 'level': len(heading[1]), 'offset': line_offset}
|
|
|
|
|
|
def skill_heading_index(content: str) -> dict:
|
|
headings, size = [], 2
|
|
for entry in iter_skill_headings(content):
|
|
entry_size = len(json.dumps(entry, ensure_ascii=False)) + (2 if headings else 0)
|
|
if size + entry_size > SKILL_OVERVIEW_LIST_MAX_CHARS:
|
|
return {'headings': headings, 'headings_truncated': True}
|
|
headings.append(entry)
|
|
size += entry_size
|
|
return {'headings': headings, 'headings_truncated': False}
|
|
|
|
|
|
def _skill_match_window(content: str, start: int, end: int) -> tuple[int, int]:
|
|
left = max(0, start - 300)
|
|
right = min(len(content), end + 300, left + SKILL_GREP_EXCERPT_MAX_CHARS)
|
|
# Trim partial context words inward, but retain the match even when it is one long word.
|
|
if left and not content[left - 1].isspace():
|
|
boundary = next((i for i in range(left, start) if content[i].isspace()), None)
|
|
if boundary is not None:
|
|
left = boundary + 1
|
|
if right < len(content) and not content[right].isspace():
|
|
boundary = next((i for i in range(right - 1, end - 1, -1) if content[i].isspace()), None)
|
|
if boundary is not None:
|
|
right = boundary
|
|
return left, right
|
|
|
|
|
|
def grep_skill_content(files: list[dict], pattern: str, case_insensitive: bool = True, max_results: int = 10) -> str:
|
|
from open_webui.tools.knowledge_fs import MATCH_BUDGET_SECONDS, MatchBudgetExceeded, build_span_matcher
|
|
from open_webui.utils.json_codec import JSONCodec
|
|
|
|
deadline = time.monotonic() + MATCH_BUDGET_SECONDS
|
|
|
|
def check_budget():
|
|
if time.monotonic() >= deadline:
|
|
raise MatchBudgetExceeded(f'Search exceeded {MATCH_BUDGET_SECONDS:g}s, narrow the pattern')
|
|
|
|
matcher, error = build_span_matcher(pattern, case_insensitive, deadline=deadline)
|
|
if error:
|
|
raise ValueError(error)
|
|
max_results = min(30, max(1, max_results))
|
|
result = {'matches': [], 'total_matches': 0, 'truncated': False}
|
|
# Reserve room for locations so large excerpts cannot hide all the remaining matches.
|
|
notice = 'Excerpts omitted; see omitted_matches.'
|
|
accepting = True
|
|
accepting_locations = True
|
|
for file in sorted(files, key=lambda file: (file['path'] != 'SKILL.md', file['path'])):
|
|
check_budget()
|
|
if file.get('encoding'):
|
|
continue
|
|
content, offset = file['content'], 0
|
|
headings = iter_skill_headings(content)
|
|
heading, next_heading = None, next(headings, None) if accepting or accepting_locations else None
|
|
previous_window = None
|
|
for line in content.splitlines(keepends=True):
|
|
check_budget()
|
|
for start, end in matcher(line.rstrip('\r\n')):
|
|
check_budget()
|
|
result['total_matches'] += 1
|
|
if not accepting and not accepting_locations:
|
|
continue
|
|
start, end = offset + start, offset + end
|
|
while next_heading is not None and next_heading['offset'] <= start:
|
|
heading, next_heading = next_heading, next(headings, None)
|
|
check_budget()
|
|
if accepting:
|
|
left, right = _skill_match_window(content, start, end)
|
|
merge = (
|
|
previous_window is not None
|
|
and left < previous_window[1]
|
|
and max(right, previous_window[1]) - previous_window[0] <= SKILL_GREP_EXCERPT_MAX_CHARS
|
|
)
|
|
if merge:
|
|
left, right = previous_window[0], max(right, previous_window[1])
|
|
entry = {**result['matches'][-1], 'excerpt': content[left:right]}
|
|
candidates = [*result['matches'][:-1], entry]
|
|
else:
|
|
entry = {
|
|
'path': file['path'],
|
|
'offset': start,
|
|
'match_length': end - start,
|
|
'section': heading['title'] if heading else None,
|
|
'section_offset': heading['offset'] if heading else None,
|
|
'excerpt': content[left:right],
|
|
}
|
|
candidates = [*result['matches'], entry]
|
|
candidate = {
|
|
**result,
|
|
'matches': candidates,
|
|
'truncated': True,
|
|
'notice': notice,
|
|
'omitted_matches': [],
|
|
'omitted_matches_truncated': False,
|
|
}
|
|
if (
|
|
len(candidates) <= max_results
|
|
and len(JSONCodec.dumps(candidate, ensure_ascii=False))
|
|
<= SKILL_GREP_MAX_CHARS - SKILL_GREP_LOCATIONS_MAX_CHARS
|
|
):
|
|
result['matches'] = candidates
|
|
previous_window = left, right
|
|
continue
|
|
accepting = False
|
|
result.update(truncated=True, notice=notice, omitted_matches=[], omitted_matches_truncated=False)
|
|
locations = [
|
|
*result['omitted_matches'],
|
|
{
|
|
'path': file['path'],
|
|
'offset': start,
|
|
'section': heading['title'] if heading else None,
|
|
},
|
|
]
|
|
if (
|
|
len(locations) <= 30
|
|
and len(JSONCodec.dumps(locations, ensure_ascii=False)) <= SKILL_GREP_LOCATIONS_MAX_CHARS
|
|
):
|
|
result['omitted_matches'] = locations
|
|
else:
|
|
accepting_locations = False
|
|
result['omitted_matches_truncated'] = True
|
|
result['notice'] = 'More locations omitted; narrow the pattern or specify path.'
|
|
offset += len(line)
|
|
check_budget()
|
|
serialized = JSONCodec.dumps(result, ensure_ascii=False)
|
|
while len(serialized) > SKILL_GREP_MAX_CHARS:
|
|
result['omitted_matches'].pop()
|
|
result['omitted_matches_truncated'] = True
|
|
result['notice'] = 'More locations omitted; narrow the pattern or specify path.'
|
|
serialized = JSONCodec.dumps(result, ensure_ascii=False)
|
|
return serialized
|
|
|
|
|
|
def skill_overview(content: str, files: list, field: str = 'files', tools_enabled: bool = True) -> dict:
|
|
if not tools_enabled:
|
|
return {
|
|
'content': content[:SKILL_INJECTION_MAX_CHARS],
|
|
'next_offset': SKILL_INJECTION_MAX_CHARS if len(content) > SKILL_INJECTION_MAX_CHARS else None,
|
|
**bounded_skill_manifest(files, field),
|
|
}
|
|
large = len(content) > SKILL_CONTENT_MAX_CHARS
|
|
overview = {
|
|
**skill_content_page(content, max_chars=SKILL_PREVIEW_MAX_CHARS if large else SKILL_CONTENT_MAX_CHARS),
|
|
**bounded_skill_manifest(files, field, SKILL_OVERVIEW_LIST_MAX_CHARS),
|
|
}
|
|
if large:
|
|
overview.update(skill_heading_index(content))
|
|
return overview
|
|
|
|
|
|
def format_skill_content(page: dict, skill_id: str, tools_enabled: bool) -> str:
|
|
content = page['content']
|
|
if tools_enabled and 'headings' in page:
|
|
content += '\nSection index: ' + json.dumps(page['headings'], ensure_ascii=False)
|
|
if page.get('headings_truncated'):
|
|
content += '\nAdditional headings omitted.'
|
|
content += f'\nSkill ID: {json.dumps(skill_id)}.'
|
|
content += f'\nnext_offset: {page["next_offset"]}'
|
|
return content
|
|
if page['next_offset'] is not None:
|
|
content += '\nSkill instructions truncated.'
|
|
if tools_enabled:
|
|
content += f'\nread_skill_file(id={json.dumps(skill_id)}, path="SKILL.md", offset={page["next_offset"]}).'
|
|
return content
|
|
|
|
|
|
class SkillFile(BaseModel):
|
|
model_config = ConfigDict(extra='forbid')
|
|
path: str
|
|
content: str
|
|
encoding: Literal['base64'] | None = None
|
|
|
|
|
|
class SkillFileOperation(BaseModel):
|
|
op: Literal['put', 'move', 'delete']
|
|
path: str
|
|
content: str | None = None
|
|
encoding: Literal['base64'] | None = None
|
|
destination: str | None = None
|
|
|
|
|
|
def validate_path(path: str) -> str:
|
|
if not path or path.startswith('/') or '\\' in path or '\x00' in path or ':' in path:
|
|
raise ValueError('File paths must be relative POSIX paths')
|
|
if any(part in ('', '.', '..') for part in path.split('/')):
|
|
raise ValueError('Invalid file path')
|
|
return path
|
|
|
|
|
|
def file_bytes(file: dict) -> bytes:
|
|
if file.get('encoding') == 'base64':
|
|
return base64.b64decode(file['content'], validate=True)
|
|
return file['content'].encode('utf-8')
|
|
|
|
|
|
def encode_file(path: str, data: bytes) -> dict:
|
|
try:
|
|
content = data.decode('utf-8')
|
|
if '\x00' in content:
|
|
raise UnicodeError()
|
|
return {'path': path, 'content': content}
|
|
except UnicodeError:
|
|
return {'path': path, 'content': base64.b64encode(data).decode('ascii'), 'encoding': 'base64'}
|
|
|
|
|
|
def validate_files(files: list[dict], previous: list[dict] | None = None) -> list[dict]:
|
|
result, paths, total = [], set(), 0
|
|
previous = {f['path']: f for f in (previous or [])}
|
|
if len(files) > max(MAX_FILES, len(previous)):
|
|
raise ValueError(f'A skill may contain at most {MAX_FILES} files')
|
|
for item in files:
|
|
file = SkillFile.model_validate(item).model_dump(exclude_none=True)
|
|
path = validate_path(file['path'])
|
|
if path in paths:
|
|
raise ValueError(f'Duplicate file: {path}')
|
|
paths.add(path)
|
|
size = len(file_bytes(file))
|
|
old_size = len(file_bytes(previous[path])) if path in previous else 0
|
|
if size > max(MAX_FILE_BYTES, old_size):
|
|
raise ValueError(f'File exceeds 10 MiB: {path}')
|
|
total += size
|
|
result.append(file)
|
|
if total > max(MAX_SKILL_BYTES, sum(len(file_bytes(f)) for f in previous.values())):
|
|
raise ValueError('Skill exceeds 50 MiB')
|
|
for path in paths:
|
|
if any(str(parent) in paths for parent in PurePosixPath(path).parents if str(parent) != '.'):
|
|
raise ValueError(f'A file is also used as a directory: {path}')
|
|
root = next((f for f in result if f['path'] == 'SKILL.md'), None)
|
|
if root is None or root.get('encoding'):
|
|
raise ValueError('A skill requires a UTF-8 SKILL.md at its root')
|
|
return sorted(result, key=lambda f: f['path'])
|
|
|
|
|
|
def apply_operations(files: list[dict], operations: list[dict]) -> list[dict]:
|
|
result = {f['path']: dict(f) for f in files}
|
|
for raw in operations:
|
|
operation = SkillFileOperation.model_validate(raw)
|
|
path = validate_path(operation.path)
|
|
if operation.op == 'put':
|
|
if operation.content is None:
|
|
raise ValueError('put requires content')
|
|
result[path] = {'path': path, 'content': operation.content}
|
|
if operation.encoding:
|
|
result[path]['encoding'] = operation.encoding
|
|
else:
|
|
if path == 'SKILL.md':
|
|
raise ValueError('SKILL.md cannot be moved or deleted')
|
|
matched = [p for p in result if p == path or p.startswith(path + '/')]
|
|
if not matched:
|
|
raise ValueError(f'File or directory not found: {path}')
|
|
if operation.op == 'move':
|
|
destination = validate_path(operation.destination or '')
|
|
if destination.startswith(path + '/'):
|
|
raise ValueError('Cannot move a directory into itself')
|
|
moved = {destination + p[len(path) :]: result[p] for p in matched}
|
|
if any(p in result and p not in matched for p in moved):
|
|
raise ValueError('Move would overwrite an existing file')
|
|
for p in matched:
|
|
del result[p]
|
|
result.update({p: {**f, 'path': p} for p, f in moved.items()})
|
|
else:
|
|
for p in matched:
|
|
del result[p]
|
|
return validate_files(list(result.values()), files)
|
|
|
|
|
|
def frontmatter(content: str) -> dict:
|
|
match = re.match(r'\A---[ \t]*\r?\n(.*?)\r?\n---[ \t]*(?:\r?\n|$)', content, re.DOTALL)
|
|
if not match:
|
|
return {}
|
|
try:
|
|
value = yaml.safe_load(match.group(1))
|
|
return value if isinstance(value, dict) else {}
|
|
except yaml.YAMLError:
|
|
return {}
|
|
|
|
|
|
def file_summaries(files: list[dict]) -> list[dict]:
|
|
return [{'path': f['path'], 'size': len(file_bytes(f)), 'encoding': f.get('encoding')} for f in files]
|
|
|
|
|
|
def parse_import(
|
|
data: bytes, filename: str, *, discover_skills: bool = False, directory: str | None = None
|
|
) -> list[dict]:
|
|
if len(data) > MAX_IMPORT_BYTES:
|
|
raise ValueError('Import exceeds 200 MiB')
|
|
if filename.lower().endswith('.json'):
|
|
value = json.loads(data)
|
|
packages = value if isinstance(value, list) else [value]
|
|
elif filename.lower().endswith('.zip'):
|
|
entries = {}
|
|
links = []
|
|
total = 0
|
|
with zipfile.ZipFile(io.BytesIO(data)) as archive:
|
|
for info in archive.infolist():
|
|
path = validate_path(info.filename.rstrip('/'))
|
|
mode = info.external_attr >> 16
|
|
special = stat.S_IFMT(mode) not in (0, stat.S_IFREG, stat.S_IFDIR)
|
|
if special and not discover_skills:
|
|
raise ValueError('Archives cannot contain links or special files')
|
|
if info.is_dir():
|
|
continue
|
|
if directory is not None:
|
|
path = path.partition('/')[2]
|
|
if not path.startswith(directory):
|
|
continue
|
|
if special:
|
|
links.append(path)
|
|
continue
|
|
if path in entries:
|
|
raise ValueError(f'Duplicate archive path: {path}')
|
|
total += info.file_size
|
|
if info.file_size > MAX_FILE_BYTES or total > MAX_IMPORT_BYTES or len(entries) >= 10000:
|
|
raise ValueError('Archive exceeds import limits')
|
|
entries[path] = archive.read(info)
|
|
roots = (
|
|
[''] if 'SKILL.md' in entries else sorted(p[: -len('SKILL.md')] for p in entries if p.endswith('/SKILL.md'))
|
|
)
|
|
if not roots:
|
|
raise ValueError('Archive contains no SKILL.md')
|
|
roots = [root for root in roots if not any(root != parent and root.startswith(parent) for parent in roots)]
|
|
if any(path.startswith(root) for path in links for root in roots):
|
|
raise ValueError('Archives cannot contain links or special files')
|
|
if not discover_skills and any(not any(p.startswith(root) for root in roots) for p in entries):
|
|
raise ValueError('Archive contains files outside skill directories')
|
|
packages = [
|
|
{'files': [encode_file(p[len(root) :], value) for p, value in entries.items() if p.startswith(root)]}
|
|
for root in roots
|
|
]
|
|
elif filename.lower().endswith('.md'):
|
|
packages = [{'files': [encode_file('SKILL.md', data)]}]
|
|
else:
|
|
raise ValueError('Use JSON, ZIP, or Markdown')
|
|
total = 0
|
|
result = []
|
|
for package in packages:
|
|
if not isinstance(package, dict):
|
|
raise ValueError('Each imported skill must be an object')
|
|
files = validate_files(
|
|
package['files'] if 'files' in package else [{'path': 'SKILL.md', 'content': package.get('content', '')}]
|
|
)
|
|
total += sum(len(file_bytes(f)) for f in files)
|
|
if total > MAX_IMPORT_BYTES:
|
|
raise ValueError('Import exceeds 200 MiB decoded')
|
|
fm = frontmatter(next(f['content'] for f in files if f['path'] == 'SKILL.md'))
|
|
name = package.get('name') or fm.get('name') or 'Imported skill'
|
|
result.append(
|
|
{
|
|
'id': package.get('id')
|
|
or re.sub(r'[^a-z0-9_-]+', '-', str(name).lower()).strip('-')
|
|
or 'imported-skill',
|
|
'name': str(name),
|
|
'description': package.get('description', fm.get('description', '')),
|
|
'meta': package.get('meta') or {},
|
|
**({'is_active': package['is_active']} if 'is_active' in package else {}),
|
|
'files': files,
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def zip_export(packages: list[dict]) -> bytes:
|
|
output, roots = io.BytesIO(), set()
|
|
with zipfile.ZipFile(output, 'w', zipfile.ZIP_DEFLATED) as archive:
|
|
for package in packages:
|
|
root_file = next(f for f in package['files'] if f['path'] == 'SKILL.md')
|
|
name = frontmatter(root_file['content']).get('name')
|
|
root = name if isinstance(name, str) and re.fullmatch(r'[a-z0-9]+(?:-[a-z0-9]+)*', name) else package['id']
|
|
validate_path(root)
|
|
if root in roots:
|
|
raise ValueError('Selected skills have duplicate export directory names')
|
|
roots.add(root)
|
|
for file in package['files']:
|
|
archive.writestr(root + '/' + validate_path(file['path']), file_bytes(file))
|
|
return output.getvalue()
|
|
|
|
|
|
def skill_import_source(url: str) -> tuple[str, str | None]:
|
|
parsed = urlsplit(url.strip())
|
|
if parsed.scheme not in ('http', 'https') or not parsed.hostname or parsed.username or parsed.password:
|
|
raise ValueError('Use an HTTP(S) URL without embedded credentials')
|
|
if parsed.hostname.lower() != 'github.com':
|
|
return url.strip(), None
|
|
parts = parsed.path.strip('/').split('/')
|
|
if len(parts) < 2 or any(not re.fullmatch(r'[\w.-]+', part) or part in ('.', '..') for part in parts[:2]):
|
|
raise ValueError('Invalid GitHub repository URL')
|
|
owner, repo = parts[:2]
|
|
repo = repo.removesuffix('.git')
|
|
if not repo:
|
|
raise ValueError('Invalid GitHub repository URL')
|
|
ref, directory = 'HEAD', ''
|
|
if len(parts) > 2:
|
|
if parts[2] in ('archive', 'releases', 'raw'):
|
|
return url.strip(), None
|
|
if len(parts) < 4 or parts[2] not in ('tree', 'blob'):
|
|
raise ValueError('Use a GitHub repository, folder, or SKILL.md URL')
|
|
# shortcut: slash-containing refs must be URL-encoded; use a commit permalink otherwise.
|
|
ref = validate_path(unquote(parts[3]))
|
|
path = '/'.join(unquote(part) for part in parts[4:])
|
|
if parts[2] == 'blob':
|
|
if not path or path.split('/')[-1] != 'SKILL.md':
|
|
raise ValueError('Select a SKILL.md file or a skill folder')
|
|
path = path.removesuffix('SKILL.md').rstrip('/')
|
|
directory = validate_path(path) + '/' if path else ''
|
|
return f'https://codeload.github.com/{owner}/{repo}/zip/{quote(ref, safe="")}', directory
|
|
|
|
|
|
async def load_skill_from_url(url: str) -> list[dict]:
|
|
from open_webui.retrieval.web.utils import get_ssrf_safe_session, validate_url
|
|
|
|
url, directory = skill_import_source(url)
|
|
try:
|
|
async with asyncio.timeout(60), get_ssrf_safe_session(trust_env=False, store_cookies=False) as session:
|
|
for _ in range(6):
|
|
target = urlsplit(url)
|
|
if target.username or target.password:
|
|
raise ValueError('Use a URL without embedded credentials')
|
|
await asyncio.to_thread(validate_url, url)
|
|
async with session.get(url, allow_redirects=False) as response:
|
|
if response.status in (301, 302, 303, 307, 308):
|
|
url = urljoin(url, response.headers.get('Location', ''))
|
|
continue
|
|
if response.status != 200:
|
|
raise ValueError('Could not download the skill. Check that the URL is accessible.')
|
|
content_type = response.content_type
|
|
if content_type == 'text/html':
|
|
raise ValueError('The URL returned a web page. Use a raw skill file or a ZIP download link.')
|
|
data = bytearray()
|
|
async for chunk in response.content.iter_chunked(64 * 1024):
|
|
data.extend(chunk)
|
|
if len(data) > MAX_IMPORT_BYTES:
|
|
raise ValueError('Import exceeds 200 MiB')
|
|
filename = {
|
|
'application/json': 'skills.json',
|
|
'text/markdown': 'SKILL.md',
|
|
}.get(content_type, unquote(urlsplit(url).path))
|
|
if data.startswith(b'PK'):
|
|
filename = 'skills.zip'
|
|
return await asyncio.to_thread(
|
|
parse_import, bytes(data), filename, discover_skills=True, directory=directory
|
|
)
|
|
raise ValueError('Too many redirects while downloading the skill')
|
|
except (TimeoutError, aiohttp.ClientError) as error:
|
|
raise ValueError('Could not download the skill. Please try again.') from error
|