"""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_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
    return {
        **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),
    }


def format_skill_content(page: dict, skill_id: str, tools_enabled: bool) -> str:
    content = page['content']
    if page['next_offset'] is not None:
        content += '\nSkill instructions truncated.'
        if tools_enabled:
            content += f'\nSkill ID: {json.dumps(skill_id)}.'
            content += f'\nnext_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', '.skill')):
        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')
                if info.flag_bits & 1:
                    raise ValueError('Encrypted skill archives are not supported')
                try:
                    entries[path] = archive.read(info)
                except NotImplementedError as error:
                    raise ValueError('Skill archive uses an unsupported compression method') from error
        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, .skill, 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/.skill 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