Byted Volcengine Tosutil

from __future__ import annotations

import argparse
import json
import sys
import time

from models import CommonOptions, CredentialMode, OperationRequest
from result_handler import parse_execution_result
from tosutil_service import (
    TosutilRunner,
    ValidationError,
    build_command,
    redact_argv,
    run_preflight,
    shell_join,
)


def _add_common_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--endpoint", help="TOS Endpoint（必须是 TOS 协议域名）")
    parser.add_argument("--region", help="Region，例如 cn-beijing")
    parser.add_argument("--access-key", help="Access Key（不会在输出中回显）")
    parser.add_argument("--secret-key", help="Secret Key（不会在输出中回显）")
    parser.add_argument("--security-token", help="STS Token（不会在输出中回显）")
    parser.add_argument("--conf-path", help="配置文件路径（-conf）")
    parser.add_argument("--bucket-type", help="桶类型，fns 或 hns")
    parser.add_argument("--output-dir", help="结果输出目录（-o）")
    parser.add_argument("--tosutil-binary", default="tosutil", help="tosutil 可执行文件路径")
    parser.add_argument(
        "--credential-mode",
        choices=[mode.value for mode in CredentialMode],
        default=CredentialMode.PERMANENT.value,
        help="凭证模式：permanent/sts/anonymous",
    )
    parser.add_argument("--timeout", type=int, default=300, help="单次执行超时秒数（默认 300）")
    parser.add_argument("--retries", type=int, default=0, help="网络/超时错误重试次数（默认 0）")
    parser.add_argument("--retry-backoff", type=float, default=0.8, help="重试退避基准秒数（默认 0.8）")


def _add_execution_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--run", action="store_true", help="是否真的执行（默认只预览命令）")
    parser.add_argument("--preflight", action="store_true", help="执行前进行版本/配置预检查")
    parser.add_argument("--connectivity", action="store_true", help="预检查时额外执行 ls 检查连通性")
    parser.add_argument("--yes", action="store_true", help="对破坏性操作跳过确认（高风险）")


def _build_subcommand_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="byted-volcengine-tosutil：tosutil 命令生成/执行/诊断器")
    sub = parser.add_subparsers(dest="subcommand", required=True)

    common_parent = argparse.ArgumentParser(add_help=False)
    _add_common_args(common_parent)
    execution_parent = argparse.ArgumentParser(add_help=False)
    _add_execution_args(execution_parent)
    parents = [common_parent, execution_parent]

    sub.add_parser("version", help="查看 tosutil 版本", parents=parents)

    p_help = sub.add_parser("help", help="查看 tosutil 帮助", parents=parents)
    p_help.add_argument("--help-command", help="具体命令名，例如 ls/cp/rm")

    sub.add_parser("config", help="查看或更新配置（默认仅查看配置路径）", parents=parents)

    p_ls = sub.add_parser("ls", help="列举桶或对象", parents=parents)
    p_ls.add_argument("--cloud-url", help="可选：tos://bucket 或 tos://bucket/prefix")

    p_mb = sub.add_parser("mb", help="创建桶", parents=parents)
    p_mb.add_argument("--cloud-url", required=True, help="tos://bucket")

    p_cp = sub.add_parser("cp", help="上传/下载/云上复制对象", parents=parents)
    p_cp.add_argument("--source", required=True, help="源路径：本地路径或 tos://")
    p_cp.add_argument("--target", required=True, help="目标路径：本地路径或 tos://")
    p_cp.add_argument("--recursive", action="store_true", help="递归复制目录/前缀")
    p_cp.add_argument("--jobs", type=int, help="批量任务并发数（-j）")
    p_cp.add_argument("--part-concurrency", type=int, help="分片任务并发数（-p）")
    p_cp.add_argument("--threshold", help="分片阈值（-threshold）")
    p_cp.add_argument("--part-size", help="分片大小（-ps）")

    p_rm = sub.add_parser("rm", help="删除桶/对象/前缀（破坏性）", parents=parents)
    p_rm.add_argument("--cloud-url", required=True, help="tos://bucket 或 tos://bucket/prefix")
    p_rm.add_argument("--recursive", action="store_true", help="递归删除（-r）")
    p_rm.add_argument("--force", action="store_true", help="强制删除（-f，跳过交互确认）")
    p_rm.add_argument("--jobs", type=int, help="批量删除并发数（-j）")

    p_du = sub.add_parser("du", help="统计对象/分片大小和数量", parents=parents)
    p_du.add_argument("--cloud-url", required=True, help="tos://bucket 或 tos://bucket/prefix")
    p_du.add_argument("--directory-mode", action="store_true", help="目录模式（-d）")
    p_du.add_argument("--include-versions", action="store_true", help="包含历史版本（-v）")
    p_du.add_argument("--include-multipart", action="store_true", help="包含分片任务（-m）")

    p_setmeta = sub.add_parser("setmeta", help="设置对象元数据", parents=parents)
    p_setmeta.add_argument("--cloud-url", required=True, help="tos://bucket/key 或 tos://bucket/prefix")
    p_setmeta.add_argument("--recursive", action="store_true", help="批量前缀模式（-r）")
    p_setmeta.add_argument("--jobs", type=int, help="批量并发（-j）")
    p_setmeta.add_argument("--meta", help="自定义元数据（-meta，格式 aaa:bbb#ccc:ddd）")
    p_setmeta.add_argument("--content-type", help="Content-Type（-contentType）")
    p_setmeta.add_argument("--expires", help="Expires（-expires，格式 YYYYMMDDHHmmSS）")

    p_stat = sub.add_parser("stat", help="查询桶/对象属性", parents=parents)
    p_stat.add_argument("--cloud-url", required=True, help="tos://bucket 或 tos://bucket/key")

    return parser


def _build_legacy_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="byted-volcengine-tosutil（legacy 模式）")
    _add_common_args(parser)
    parser.add_argument("--command", required=True, help="tosutil 命令名，例如 ls/cp/rm")
    parser.add_argument("--source", help="源路径，本地路径或 tos:// URI")
    parser.add_argument("--target", help="目标路径，本地路径或 tos:// URI")
    parser.add_argument("--cloud-url", help="单资源命令使用的 tos:// URI")
    parser.add_argument("--recursive", action="store_true", help="是否递归执行")
    parser.add_argument("--force", action="store_true", help="是否强制执行")
    parser.add_argument("--jobs", type=int, help="批量任务并发数")
    parser.add_argument("--part-concurrency", type=int, help="分片任务并发数")
    parser.add_argument("--threshold", help="分片阈值")
    parser.add_argument("--part-size", help="分片大小")
    parser.add_argument("--meta", help="自定义元数据")
    parser.add_argument("--content-type", help="Content-Type")
    parser.add_argument("--expires", help="Expires 时间")
    parser.add_argument("--directory-mode", action="store_true", help="du 目录模式")
    parser.add_argument("--include-versions", action="store_true", help="是否统计版本数据")
    parser.add_argument("--include-multipart", action="store_true", help="是否统计分片上传任务")
    parser.add_argument("--help-command", help="help 子命令名称")
    parser.add_argument("--assume-yes", action="store_true", help="跳过删除类命令确认")
    parser.add_argument("--run", action="store_true", help="是否真的执行命令")
    parser.add_argument("--preflight", action="store_true", help="执行前进行版本和配置预检查")
    parser.add_argument("--connectivity", action="store_true", help="预检查时额外执行 ls 检查连通性")
    return parser


def main() -> int:
    started_at = time.time()
    argv = sys.argv[1:]
    use_legacy = "--command" in argv or any(item.startswith("--command=") for item in argv)
    parser = _build_legacy_parser() if use_legacy else _build_subcommand_parser()
    args = parser.parse_args()

    common_options = CommonOptions(
        endpoint=args.endpoint,
        region=args.region,
        access_key=args.access_key,
        secret_key=args.secret_key,
        security_token=args.security_token,
        conf_path=args.conf_path,
        bucket_type=args.bucket_type,
        output_dir=args.output_dir,
        tosutil_binary=args.tosutil_binary,
        credential_mode=CredentialMode(args.credential_mode),
        timeout_seconds=args.timeout,
        retries=args.retries,
        retry_backoff_seconds=args.retry_backoff,
    )
    request = _build_request_from_args(args, common_options, legacy=use_legacy)

    try:
        if args.preflight:
            report = run_preflight(common_options, check_connectivity=args.connectivity)
            _emit_json(
                ok=report.ok,
                code="OK" if report.ok else "E_PREFLIGHT",
                message="预检查通过" if report.ok else "预检查失败",
                data={"preflight": _serialize_preflight(report)},
                started_at=started_at,
            )
            if not report.ok:
                return 2

        spec = build_command(request)
    except ValidationError as exc:
        _emit_json(
            ok=False,
            code="E_VALIDATION",
            message=str(exc),
            data={"input": _safe_input_summary(request)},
            started_at=started_at,
        )
        return 2

    safe_argv = redact_argv(spec.argv)
    preview = {
        "command": spec.command,
        "summary": spec.summary,
        "risk_level": spec.risk_level.value,
        "requires_confirmation": spec.requires_confirmation,
        "argv": safe_argv,
        "shell": shell_join(safe_argv),
        "hints": spec.hints,
    }

    if not args.run:
        _emit_json(ok=True, code="OK", message="已生成命令预览", data={"preview": preview}, started_at=started_at)
        return 0

    assume_yes = getattr(args, "yes", False) or getattr(args, "assume_yes", False)
    if spec.requires_confirmation and not assume_yes:
        _emit_json(
            ok=False,
            code="E_CONFIRM_REQUIRED",
            message="该操作为高风险/破坏性操作，默认不执行。请确认影响范围后重试。",
            data={
                "preview": preview,
                "next_actions": [
                    "先用 `ls` 验证影响范围。",
                    "确认无误后，重新执行并加 `--yes`（或 legacy 模式下加 `--assume-yes`）。",
                ],
            },
            started_at=started_at,
        )
        return 3

    runner = TosutilRunner(
        timeout_seconds=common_options.timeout_seconds,
        retries=common_options.retries,
        retry_backoff_seconds=common_options.retry_backoff_seconds,
    )
    result = runner.run(spec)
    parsed = parse_execution_result(spec, result)
    _emit_json(
        ok=parsed.success,
        code="OK" if parsed.success else parsed.advice.code if parsed.advice else "E_RUNTIME",
        message=parsed.summary,
        data={
            "preview": preview,
            "result": _safe_result_payload(parsed),
            "evidence": {"exit_code": result.exit_code, "duration_ms": result.duration_ms},
        },
        started_at=started_at,
    )
    return 0 if parsed.success else 1


def _serialize_preflight(report: object) -> dict[str, object]:
    checks = getattr(report, "checks", [])
    return {
        "ok": getattr(report, "ok", False),
        "checks": [
            {
                "name": check.name,
                "success": check.success,
                "details": check.details,
                "hints": check.hints,
            }
            for check in checks
        ],
    }


def _build_request_from_args(args: argparse.Namespace, common: CommonOptions, *, legacy: bool) -> OperationRequest:
    if legacy:
        return OperationRequest(
            command=args.command,
            common_options=common,
            source=args.source,
            target=args.target,
            cloud_url=args.cloud_url,
            recursive=args.recursive,
            force=args.force,
            jobs=args.jobs,
            part_concurrency=args.part_concurrency,
            threshold=args.threshold,
            part_size=args.part_size,
            meta=args.meta,
            content_type=args.content_type,
            expires=args.expires,
            directory_mode=args.directory_mode,
            include_versions=args.include_versions,
            include_multipart=args.include_multipart,
            help_command=args.help_command,
            assume_yes=args.assume_yes,
        )

    return OperationRequest(
        command=args.subcommand,
        common_options=common,
        source=getattr(args, "source", None),
        target=getattr(args, "target", None),
        cloud_url=getattr(args, "cloud_url", None),
        recursive=getattr(args, "recursive", False),
        force=getattr(args, "force", False),
        jobs=getattr(args, "jobs", None),
        part_concurrency=getattr(args, "part_concurrency", None),
        threshold=getattr(args, "threshold", None),
        part_size=getattr(args, "part_size", None),
        meta=getattr(args, "meta", None),
        content_type=getattr(args, "content_type", None),
        expires=getattr(args, "expires", None),
        directory_mode=getattr(args, "directory_mode", False),
        include_versions=getattr(args, "include_versions", False),
        include_multipart=getattr(args, "include_multipart", False),
        help_command=getattr(args, "help_command", None),
        assume_yes=False,
    )


def _emit_json(*, ok: bool, code: str, message: str, data: dict[str, object], started_at: float) -> None:
    print(json.dumps({"ok": ok, "code": code, "message": message, "data": data, "ts": int(started_at)}, ensure_ascii=False, indent=2))


def _safe_input_summary(request: OperationRequest) -> dict[str, object]:
    return {
        "command": request.command,
        "source": request.source,
        "target": request.target,
        "cloud_url": request.cloud_url,
        "recursive": request.recursive,
        "force": request.force,
    }


def _truncate(text: str | bytes, max_chars: int = 4000) -> dict[str, object]:
    if isinstance(text, bytes):
        text = text.decode("utf-8", errors="replace")
    if len(text) <= max_chars:
        return {"text": text, "truncated": False, "max_chars": max_chars}
    return {"text": text[:max_chars], "truncated": True, "max_chars": max_chars}


def _safe_result_payload(parsed: object) -> dict[str, object]:
    parsed_dict = getattr(parsed, "to_dict")()
    stdout = parsed_dict.pop("raw_stdout", "")
    stderr = parsed_dict.pop("raw_stderr", "")
    parsed_dict["stdout"] = _truncate(stdout)
    parsed_dict["stderr"] = _truncate(stderr)
    return parsed_dict


if __name__ == "__main__":
    sys.exit(main())

from __future__ import annotations

from dataclasses import asdict, dataclass, field
from enum import Enum
from typing import Any


class CredentialMode(str, Enum):
    PERMANENT = "permanent"
    STS = "sts"
    ANONYMOUS = "anonymous"


class ResourceKind(str, Enum):
    LOCAL_FILE = "local_file"
    LOCAL_DIR = "local_dir"
    TOS_URI = "tos_uri"
    UNKNOWN = "unknown"


class TransferDirection(str, Enum):
    UPLOAD = "upload"
    DOWNLOAD = "download"
    CLOUD_COPY = "cloud_copy"
    UNKNOWN = "unknown"


class RiskLevel(str, Enum):
    READONLY = "readonly"
    WRITE = "write"
    DESTRUCTIVE = "destructive"


@dataclass
class CommonOptions:
    endpoint: str | None = None
    region: str | None = None
    access_key: str | None = None
    secret_key: str | None = None
    security_token: str | None = None
    conf_path: str | None = None
    bucket_type: str | None = None
    output_dir: str | None = None
    tosutil_binary: str = "tosutil"
    credential_mode: CredentialMode = CredentialMode.PERMANENT
    timeout_seconds: int = 300
    retries: int = 0
    retry_backoff_seconds: float = 0.8

    def to_safe_dict(self) -> dict[str, Any]:
        data = asdict(self)
        for key in ("access_key", "secret_key", "security_token"):
            if data.get(key):
                data[key] = "***"
        return data


@dataclass
class TosResource:
    raw: str
    kind: ResourceKind
    exists: bool = False

    @property
    def is_tos(self) -> bool:
        return self.kind == ResourceKind.TOS_URI


@dataclass
class OperationRequest:
    command: str
    common_options: CommonOptions = field(default_factory=CommonOptions)
    source: str | None = None
    target: str | None = None
    cloud_url: str | None = None
    recursive: bool = False
    force: bool = False
    jobs: int | None = None
    part_concurrency: int | None = None
    threshold: str | None = None
    part_size: str | None = None
    meta: str | None = None
    content_type: str | None = None
    expires: str | None = None
    directory_mode: bool = False
    include_versions: bool = False
    include_multipart: bool = False
    help_command: str | None = None
    extra_args: list[str] = field(default_factory=list)
    assume_yes: bool = False


@dataclass
class CommandSpec:
    command: str
    argv: list[str]
    risk_level: RiskLevel
    summary: str
    requires_confirmation: bool = False
    hints: list[str] = field(default_factory=list)

    def shell_command(self) -> str:
        return " ".join(self.argv)


@dataclass
class ExecutionResult:
    exit_code: int
    stdout: str | bytes
    stderr: str | bytes
    duration_ms: int
    argv: list[str]
    timed_out: bool = False

    @property
    def combined_output(self) -> str:
        return "\n".join(
            _coerce_text(part)
            for part in (self.stdout, self.stderr)
            if part
        )


def _coerce_text(value: str | bytes) -> str:
    if isinstance(value, bytes):
        # 某些超时或底层调用场景仍可能返回 bytes，这里统一兼容处理。
        return value.decode("utf-8", errors="replace")
    return value


@dataclass
class ErrorAdvice:
    code: str
    category: str
    message: str
    probable_causes: list[str] = field(default_factory=list)
    next_actions: list[str] = field(default_factory=list)


@dataclass
class ParsedResult:
    success: bool
    command: str
    summary: str
    request_id: str | None = None
    task_id: str | None = None
    success_count: int | None = None
    failed_count: int | None = None
    bucket_count: int | None = None
    hints: list[str] = field(default_factory=list)
    advice: ErrorAdvice | None = None
    raw_stdout: str = ""
    raw_stderr: str = ""

    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        if self.advice is None:
            data.pop("advice", None)
        return data


@dataclass
class PreflightCheck:
    name: str
    success: bool
    details: str
    hints: list[str] = field(default_factory=list)


@dataclass
class PreflightReport:
    ok: bool
    checks: list[PreflightCheck] = field(default_factory=list)

    def add(self, check: PreflightCheck) -> None:
        self.checks.append(check)
        if not check.success:
            self.ok = False

from __future__ import annotations

import re

from models import CommandSpec, ErrorAdvice, ExecutionResult, ParsedResult

BUCKET_COUNT_RE = re.compile(r"Bucket number is:\s*(\d+)", re.IGNORECASE)
SUCCESS_COUNT_RE = re.compile(r"Succeed count is:\s*(\d+)", re.IGNORECASE)
FAILED_COUNT_RE = re.compile(r"Failed count is:\s*(\d+)", re.IGNORECASE)
TASK_ID_RE = re.compile(r"Task id is:\s*([A-Za-z0-9-]+)", re.IGNORECASE)
REQUEST_ID_RE = re.compile(r"request id(?:\s*\[|\s+)([A-Za-z0-9-]+)", re.IGNORECASE)


def map_error(command: str, result: ExecutionResult) -> ErrorAdvice | None:
    output = result.combined_output.lower()

    if "http status [403]" in output:
        return ErrorAdvice(
            code="E_PERMISSION_403",
            category="permission_error",
            message="命令执行失败，疑似访问凭证无效或无权限访问目标资源。",
            probable_causes=["AK/SK 配置错误。", "STS Token 已过期。", "目标桶或对象未授予当前身份访问权限。"],
            next_actions=["执行 `tosutil config` 检查当前配置。", "执行 `tosutil ls` 验证基础权限。", "确认目标桶策略、ACL 或 RAM 权限设置。"],
        )

    if "a connection attempt failed" in output:
        return ErrorAdvice(
            code="E_NETWORK_CONNECT",
            category="network_error",
            message="命令执行失败，当前环境无法连通 TOS Endpoint。",
            probable_causes=["网络异常或 DNS 解析失败。", "企业代理、防火墙或安全组限制。", "Endpoint 配置错误。"],
            next_actions=["确认 Endpoint 是否为正确的 TOS 协议域名。", "检查本机网络、代理与 VPN 配置。", "先执行 `tosutil ls` 或网络诊断相关命令定位连通性问题。"],
        )

    if "config file" in output and "not exist" in output:
        return ErrorAdvice(
            code="E_CONFIG_NOT_FOUND",
            category="config_error",
            message="命令执行失败，配置文件不存在或无法读取。",
            probable_causes=["传入的 `-conf` 路径不存在。", "当前用户没有权限访问配置文件。"],
            next_actions=["检查 `-conf` 参数路径是否正确。", "若需要默认配置，可先执行 `tosutil config` 初始化。"],
        )

    if "unknown flag" in output or "invalid argument" in output:
        return ErrorAdvice(
            code="E_INVALID_ARGUMENT",
            category="argument_error",
            message="命令执行失败，参数不合法或当前命令不支持该参数。",
            probable_causes=["命令拼装错误。", "tosutil 版本与参数能力不匹配。"],
            next_actions=["执行 `tosutil help <command>` 查看当前版本支持的参数。", "减少可选参数，先验证最小命令是否可执行。"],
        )

    if result.timed_out:
        return ErrorAdvice(
            code="E_TIMEOUT",
            category="timeout_error",
            message="命令执行超时。",
            probable_causes=["批量任务过大。", "网络波动导致任务执行缓慢。"],
            next_actions=["缩小操作范围后重试。", "针对 `cp`、`du` 等批量任务调整并发与扫描范围。"],
        )

    if result.exit_code != 0:
        return ErrorAdvice(
            code="E_RUNTIME",
            category="runtime_error",
            message=f"`{command}` 执行失败，请结合原始输出进一步排查。",
            probable_causes=["命令参数与资源状态不匹配。", "环境、权限或网络存在异常。"],
            next_actions=["查看 stdout/stderr 原始输出。", "先退回最小只读命令验证环境，例如 `version` 或 `ls`。"],
        )

    return None


def parse_execution_result(spec: CommandSpec, result: ExecutionResult) -> ParsedResult:
    output = result.combined_output
    bucket_count = _search_int(BUCKET_COUNT_RE, output)
    success_count = _search_int(SUCCESS_COUNT_RE, output)
    failed_count = _search_int(FAILED_COUNT_RE, output)
    task_id = _search_text(TASK_ID_RE, output)
    request_id = _search_text(REQUEST_ID_RE, output)
    success = result.exit_code == 0 and not result.timed_out
    advice = None if success else map_error(spec.command, result)

    hints = list(spec.hints)
    if bucket_count is not None:
        hints.append(f"当前列举结果包含 {bucket_count} 个桶。")
    if success_count is not None or failed_count is not None:
        hints.append(f"批量任务统计：成功 {success_count or 0}，失败 {failed_count or 0}。")
    if task_id:
        hints.append(f"任务标识：{task_id}")

    return ParsedResult(
        success=success,
        command=spec.command,
        summary=_build_summary(spec.command, success, bucket_count, success_count, failed_count),
        request_id=request_id,
        task_id=task_id,
        success_count=success_count,
        failed_count=failed_count,
        bucket_count=bucket_count,
        hints=hints,
        advice=advice,
        raw_stdout=result.stdout,
        raw_stderr=result.stderr,
    )


def _search_int(pattern: re.Pattern[str], text: str) -> int | None:
    match = pattern.search(text)
    if not match:
        return None
    return int(match.group(1))


def _search_text(pattern: re.Pattern[str], text: str) -> str | None:
    match = pattern.search(text)
    if not match:
        return None
    return match.group(1)


def _build_summary(
    command: str,
    success: bool,
    bucket_count: int | None,
    success_count: int | None,
    failed_count: int | None,
) -> str:
    if not success:
        return f"`{command}` 执行失败。"
    if command == "ls" and bucket_count is not None:
        return f"`ls` 执行成功，共列举到 {bucket_count} 个桶。"
    if success_count is not None or failed_count is not None:
        return f"`{command}` 执行完成，成功 {success_count or 0} 个，失败 {failed_count or 0} 个。"
    return f"`{command}` 执行成功。"

from __future__ import annotations

"""
集中放置 tosutil skill 的“服务层”逻辑：
- 命令构建
- 资源识别和公共参数校验
- 子进程执行与重试
- 预检查
- 输出脱敏与命令串拼接
"""

import os
import re
import shlex
import subprocess
import time
from pathlib import Path
from typing import Callable

from models import (
    CommandSpec,
    CommonOptions,
    ExecutionResult,
    OperationRequest,
    PreflightCheck,
    PreflightReport,
    ResourceKind,
    RiskLevel,
    TosResource,
    TransferDirection,
)

TOS_URI_PREFIX = "tos://"
_S3_ENDPOINT_HINT_RE = re.compile(r"(^|\.)(s3|tos-s3)(\.|$)", re.IGNORECASE)
_RETRYABLE_NETWORK_SNIPPET = "a connection attempt failed"


class ValidationError(ValueError):
    pass


# ----------------------------
# Resource detection
# ----------------------------
def is_tos_uri(value: str | None) -> bool:
    return bool(value) and value.startswith(TOS_URI_PREFIX)


def detect_resource(path_or_uri: str | None) -> TosResource:
    if not path_or_uri:
        return TosResource(raw="", kind=ResourceKind.UNKNOWN, exists=False)

    if is_tos_uri(path_or_uri):
        return TosResource(raw=path_or_uri, kind=ResourceKind.TOS_URI, exists=True)

    path = Path(path_or_uri).expanduser()
    if path.exists() and path.is_dir():
        return TosResource(raw=path_or_uri, kind=ResourceKind.LOCAL_DIR, exists=True)
    if path.exists() and path.is_file():
        return TosResource(raw=path_or_uri, kind=ResourceKind.LOCAL_FILE, exists=True)
    if path_or_uri.endswith(("/", os.sep)):
        return TosResource(raw=path_or_uri, kind=ResourceKind.LOCAL_DIR, exists=False)
    return TosResource(raw=path_or_uri, kind=ResourceKind.LOCAL_FILE, exists=False)


# ----------------------------
# Common validation & CLI helpers
# ----------------------------
def validate_tos_endpoint(endpoint: str | None) -> None:
    if endpoint and _S3_ENDPOINT_HINT_RE.search(endpoint):
        raise ValidationError("检测到疑似 S3 协议域名，请改用 TOS 协议 Endpoint。")


def validate_common_options(options: CommonOptions) -> list[str]:
    hints: list[str] = []
    validate_tos_endpoint(options.endpoint)

    if options.conf_path and not Path(options.conf_path).expanduser().exists():
        raise ValidationError("`-conf` 指向的配置文件不存在。")

    mode = options.credential_mode.value
    if mode == "anonymous":
        if options.access_key or options.secret_key or options.security_token:
            raise ValidationError("匿名访问模式下，不应再传入 AK/SK/Token。")
        hints.append("当前为匿名访问模式，仅适用于公共读或公共写场景。")
        return hints

    if mode == "sts" and not options.security_token:
        raise ValidationError("STS 模式缺少 security token。")

    return hints


def normalize_command_name(command: str) -> str:
    normalized = command.strip().lower()
    if not normalized:
        raise ValidationError("命令名不能为空。")
    return normalized


def ensure_required(value: str | None, field_name: str) -> str:
    if value:
        return value
    raise ValidationError(f"缺少必填参数：{field_name}")


def append_flag(argv: list[str], flag: str, enabled: bool) -> None:
    if enabled:
        argv.append(flag)


def append_option(argv: list[str], flag: str, value: str | int | None) -> None:
    if value is None or value == "":
        return
    argv.append(f"{flag}={value}")


def shell_join(argv: list[str]) -> str:
    join_fn = getattr(shlex, "join", None)
    if join_fn:
        return join_fn(argv)
    return " ".join(shlex.quote(arg) for arg in argv)


def redact_argv(argv: list[str]) -> list[str]:
    return [mask_sensitive_text(item) for item in argv]


def mask_sensitive_text(value: str) -> str:
    masked = value
    for pattern, replacement in (
        (r"(-i=)(\S+)", r"\1***"),
        (r"(-k=)(\S+)", r"\1***"),
        (r"(-t=)(\S+)", r"\1***"),
    ):
        masked = re.sub(pattern, replacement, masked)
    return masked


# ----------------------------
# Command building (public)
# ----------------------------
def build_command(request: OperationRequest) -> CommandSpec:
    command = normalize_command_name(request.command)
    hints = validate_common_options(request.common_options)

    builders: dict[str, Callable[[OperationRequest, list[str]], CommandSpec]] = {
        "version": _build_version,
        "help": _build_help,
        "config": _build_config,
        "ls": _build_ls,
        "mb": _build_mb,
        "cp": _build_cp,
        "rm": _build_rm,
        "du": _build_du,
        "setmeta": _build_setmeta,
        "stat": _build_stat,
    }

    builder = builders.get(command)
    if not builder:
        raise ValidationError(f"暂未实现命令：{command}")
    return builder(request, hints)


def _build_version(request: OperationRequest, hints: list[str]) -> CommandSpec:
    return CommandSpec(
        command="version",
        argv=[request.common_options.tosutil_binary, "version"],
        risk_level=RiskLevel.READONLY,
        summary="查看 tosutil 版本信息",
        hints=hints,
    )


def _build_help(request: OperationRequest, hints: list[str]) -> CommandSpec:
    argv = [request.common_options.tosutil_binary, "help"]
    if request.help_command:
        argv.append(request.help_command)
    _append_common_options(argv, request.common_options)
    return CommandSpec(command="help", argv=argv, risk_level=RiskLevel.READONLY, summary="查看 tosutil 帮助文档", hints=hints)


def _build_config(request: OperationRequest, hints: list[str]) -> CommandSpec:
    argv = [request.common_options.tosutil_binary, "config"]
    _append_common_options(argv, request.common_options, include_output=False)
    return CommandSpec(command="config", argv=argv, risk_level=RiskLevel.WRITE, summary="初始化或更新 tosutil 配置", hints=hints)


def _build_ls(request: OperationRequest, hints: list[str]) -> CommandSpec:
    argv = [request.common_options.tosutil_binary, "ls"]
    if request.cloud_url:
        argv.append(request.cloud_url)
    _append_common_options(argv, request.common_options)
    return CommandSpec(command="ls", argv=argv, risk_level=RiskLevel.READONLY, summary="列举桶或对象", hints=hints)


def _build_mb(request: OperationRequest, hints: list[str]) -> CommandSpec:
    cloud_url = ensure_required(request.cloud_url, "cloud_url")
    _ensure_tos_resource(cloud_url, "mb")
    argv = [request.common_options.tosutil_binary, "mb", cloud_url]
    _append_common_options(argv, request.common_options, include_output=False)
    return CommandSpec(command="mb", argv=argv, risk_level=RiskLevel.WRITE, summary="创建新桶", hints=hints)


def _build_cp(request: OperationRequest, hints: list[str]) -> CommandSpec:
    source = ensure_required(request.source, "source")
    target = ensure_required(request.target, "target")
    src_resource = detect_resource(source)
    dst_resource = detect_resource(target)
    direction = _detect_transfer_direction(src_resource, dst_resource)
    if direction == TransferDirection.UNKNOWN:
        raise ValidationError("无法识别 `cp` 的传输方向，请检查源和目标路径。")

    argv = [request.common_options.tosutil_binary, "cp", source, target]
    append_flag(argv, "-r", request.recursive or src_resource.kind == ResourceKind.LOCAL_DIR)
    append_option(argv, "-j", request.jobs)
    append_option(argv, "-p", request.part_concurrency)
    append_option(argv, "-threshold", request.threshold)
    append_option(argv, "-ps", request.part_size)
    _append_common_options(argv, request.common_options)

    summary_by_direction = {
        TransferDirection.UPLOAD: "上传本地文件或目录到 TOS",
        TransferDirection.DOWNLOAD: "从 TOS 下载对象到本地",
        TransferDirection.CLOUD_COPY: "在 TOS 内部复制对象",
    }
    return CommandSpec(
        command="cp",
        argv=argv,
        risk_level=RiskLevel.WRITE,
        summary=summary_by_direction[direction],
        hints=hints + _build_cp_hints(direction, request),
    )


def _build_rm(request: OperationRequest, hints: list[str]) -> CommandSpec:
    cloud_url = ensure_required(request.cloud_url, "cloud_url")
    _ensure_tos_resource(cloud_url, "rm")
    argv = [request.common_options.tosutil_binary, "rm", cloud_url]
    append_flag(argv, "-r", request.recursive)
    append_flag(argv, "-f", request.force or request.assume_yes)
    append_option(argv, "-j", request.jobs)
    _append_common_options(argv, request.common_options)
    return CommandSpec(
        command="rm",
        argv=argv,
        risk_level=RiskLevel.DESTRUCTIVE,
        summary="删除桶、对象或对象前缀",
        requires_confirmation=not (request.force or request.assume_yes),
        hints=hints + _build_rm_hints(request),
    )


def _build_du(request: OperationRequest, hints: list[str]) -> CommandSpec:
    cloud_url = ensure_required(request.cloud_url, "cloud_url")
    _ensure_tos_resource(cloud_url, "du")
    argv = [request.common_options.tosutil_binary, "du", cloud_url]
    append_flag(argv, "-d", request.directory_mode)
    append_flag(argv, "-v", request.include_versions)
    append_flag(argv, "-m", request.include_multipart)
    _append_common_options(argv, request.common_options, include_output=False)
    if not request.directory_mode:
        hints.append("如果对象规模较大，建议按目录拆分统计，避免一次性全量扫描。")
    return CommandSpec(command="du", argv=argv, risk_level=RiskLevel.READONLY, summary="统计对象与分片大小和数量", hints=hints)


def _build_setmeta(request: OperationRequest, hints: list[str]) -> CommandSpec:
    cloud_url = ensure_required(request.cloud_url, "cloud_url")
    _ensure_tos_resource(cloud_url, "setmeta")
    argv = [request.common_options.tosutil_binary, "setmeta", cloud_url]
    append_flag(argv, "-r", request.recursive)
    append_option(argv, "-j", request.jobs)
    append_option(argv, "-meta", request.meta)
    append_option(argv, "-contentType", request.content_type)
    append_option(argv, "-expires", request.expires)
    _append_common_options(argv, request.common_options)
    if request.recursive and request.jobs is None:
        hints.append("批量设置元数据时建议显式设置 `-j`，避免默认并发不可控。")
    return CommandSpec(command="setmeta", argv=argv, risk_level=RiskLevel.WRITE, summary="设置对象元数据", hints=hints)


def _build_stat(request: OperationRequest, hints: list[str]) -> CommandSpec:
    cloud_url = ensure_required(request.cloud_url, "cloud_url")
    _ensure_tos_resource(cloud_url, "stat")
    argv = [request.common_options.tosutil_binary, "stat", cloud_url]
    _append_common_options(argv, request.common_options, include_output=False)
    return CommandSpec(command="stat", argv=argv, risk_level=RiskLevel.READONLY, summary="查询桶或对象属性", hints=hints)


def _append_common_options(argv: list[str], options: CommonOptions, *, include_output: bool = True) -> None:
    append_option(argv, "-e", options.endpoint)
    append_option(argv, "-re", options.region)
    if options.credential_mode.value == "anonymous":
        argv.extend(["-i=", "-k=", "-t="])
    else:
        append_option(argv, "-i", options.access_key)
        append_option(argv, "-k", options.secret_key)
        append_option(argv, "-t", options.security_token)
    append_option(argv, "-conf", options.conf_path)
    append_option(argv, "-bt", options.bucket_type)
    if include_output:
        append_option(argv, "-o", options.output_dir)


def _detect_transfer_direction(source: TosResource, target: TosResource) -> TransferDirection:
    if source.is_tos and target.is_tos:
        return TransferDirection.CLOUD_COPY
    if source.is_tos and target.kind in (ResourceKind.LOCAL_DIR, ResourceKind.LOCAL_FILE):
        return TransferDirection.DOWNLOAD
    if target.is_tos and source.kind in (ResourceKind.LOCAL_DIR, ResourceKind.LOCAL_FILE):
        return TransferDirection.UPLOAD
    return TransferDirection.UNKNOWN


def _ensure_tos_resource(cloud_url: str, command: str) -> None:
    resource = detect_resource(cloud_url)
    if resource.kind != ResourceKind.TOS_URI:
        raise ValidationError(f"`{command}` 需要传入 `tos://` 资源地址。")


def _build_cp_hints(direction: TransferDirection, request: OperationRequest) -> list[str]:
    hints: list[str] = []
    if direction == TransferDirection.UPLOAD:
        hints.append("上传大文件时，建议结合 `-threshold`、`-p` 调整分片策略。")
    if direction == TransferDirection.DOWNLOAD:
        hints.append("下载目录或前缀时，请确认目标本地路径具备写权限。")
    if request.jobs and request.jobs > 50:
        hints.append("当前并发较高，建议根据机器资源和带宽情况评估是否下调。")
    return hints


def _build_rm_hints(request: OperationRequest) -> list[str]:
    hints = ["删除前建议先用 `ls` 检查影响范围。"]
    if request.recursive:
        hints.append("当前为递归删除，请确认前缀范围是否准确。")
    if request.force:
        hints.append("当前使用强制删除，命令将跳过交互确认。")
    return hints


# ----------------------------
# Runner
# ----------------------------
class TosutilRunner:
    def __init__(self, *, timeout_seconds: int = 300, retries: int = 0, retry_backoff_seconds: float = 0.8) -> None:
        self.timeout_seconds = timeout_seconds
        self.retries = max(0, retries)
        self.retry_backoff_seconds = max(0.0, retry_backoff_seconds)

    def run(self, spec: CommandSpec) -> ExecutionResult:
        last_result: ExecutionResult | None = None
        attempts = 1 + self.retries
        for attempt in range(1, attempts + 1):
            result = self._run_once(spec)
            last_result = result
            if result.exit_code == 0 and not result.timed_out:
                return result
            if attempt >= attempts or not self._is_retryable(result):
                return result
            # 指数退避：瞬时网络抖动下更稳定。
            time.sleep(self.retry_backoff_seconds * (2 ** (attempt - 1)))
        return last_result or self._run_once(spec)

    def _run_once(self, spec: CommandSpec) -> ExecutionResult:
        started_at = time.perf_counter()
        try:
            completed = subprocess.run(
                spec.argv,
                capture_output=True,
                text=True,
                check=False,
                timeout=self.timeout_seconds,
            )
            return ExecutionResult(
                exit_code=completed.returncode,
                stdout=completed.stdout,
                stderr=completed.stderr,
                duration_ms=int((time.perf_counter() - started_at) * 1000),
                argv=spec.argv,
                timed_out=False,
            )
        except subprocess.TimeoutExpired as exc:
            return ExecutionResult(
                exit_code=124,
                stdout=exc.stdout or "",
                stderr=exc.stderr or "",
                duration_ms=int((time.perf_counter() - started_at) * 1000),
                argv=spec.argv,
                timed_out=True,
            )

    def _is_retryable(self, result: ExecutionResult) -> bool:
        if result.timed_out:
            return True
        return _RETRYABLE_NETWORK_SNIPPET in result.combined_output.lower()


# ----------------------------
# Preflight
# ----------------------------
def run_preflight(common_options: CommonOptions, *, check_connectivity: bool = False) -> PreflightReport:
    runner = TosutilRunner(timeout_seconds=min(common_options.timeout_seconds, 30), retries=0)
    report = PreflightReport(ok=True)

    binary_check = _check_binary(common_options.tosutil_binary)
    report.add(binary_check)
    if not binary_check.success:
        return report

    report.add(_check_conf_path(common_options))
    report.add(_run_preflight_command(runner, OperationRequest(command="version", common_options=common_options)))
    if check_connectivity:
        report.add(_run_preflight_command(runner, OperationRequest(command="ls", common_options=common_options)))
    return report


def _run_preflight_command(runner: TosutilRunner, request: OperationRequest) -> PreflightCheck:
    spec = build_command(request)
    result = runner.run(spec)
    success = result.exit_code == 0 and not result.timed_out
    details = f"`{request.command}` 检查通过" if success else f"`{request.command}` 检查失败"
    hints: list[str] = []
    snippet = _compact_output(result.combined_output)
    if snippet and not success:
        hints.append(f"输出摘要：{snippet}")
    return PreflightCheck(name=request.command, success=success, details=details, hints=hints)


def _compact_output(text: str, max_chars: int = 200) -> str:
    cleaned = " ".join(text.split())
    if not cleaned:
        return ""
    if len(cleaned) <= max_chars:
        return cleaned
    return cleaned[:max_chars] + "..."


def _check_binary(binary_path: str) -> PreflightCheck:
    path = Path(binary_path).expanduser()
    if path.exists() and path.is_file() and os.access(str(path), os.X_OK):
        return PreflightCheck(name="binary", success=True, details=f"已找到可执行文件：{path}")
    return PreflightCheck(
        name="binary",
        success=False,
        details=f"未找到 tosutil 可执行文件：{binary_path}",
        hints=["请确认 `tosutil` 已下载并具备执行权限，例如 `chmod +x tosutil`。"],
    )


def _check_conf_path(common_options: CommonOptions) -> PreflightCheck:
    if not common_options.conf_path:
        return PreflightCheck(name="config_path", success=True, details="未显式指定 `-conf`，将使用默认配置文件路径。")
    conf_path = Path(common_options.conf_path).expanduser()
    if conf_path.exists():
        return PreflightCheck(name="config_path", success=True, details=f"配置文件存在：{conf_path}")
    return PreflightCheck(
        name="config_path",
        success=False,
        details=f"配置文件不存在：{conf_path}",
        hints=["请先创建配置文件，或移除 `-conf` 使用默认配置路径。"],
    )