's'

src/config_governance.py

@@ -0,0 +1,695 @@
+from __future__ import annotations
+
+import configparser
+import json
+import re
+import shutil
+from collections import Counter
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Optional
+
+from src.control_plane import resolve_managed_file
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+BACKUP_DIR = BASE_DIR / "backup_config"
+WEB_BACKUP_DIR = BACKUP_DIR / "__web__"
+MAX_BACKUPS_PER_FILE = 24
+
+_URL_RE = re.compile(r"(https?://[^\s,#]+)", re.IGNORECASE)
+_RECORDER_SECTION = "录制设置"
+
+
+def _now_stamp() -> str:
+    return datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+
+
+def _read_text(path: Path) -> str:
+    return path.read_text(encoding="utf-8", errors="replace")
+
+
+def _file_kind(path: Path) -> str:
+    name = path.name.lower()
+    if name == "url_config.ini":
+        return "url_config"
+    if name == "youtube.ini":
+        return "youtube_ini"
+    if name == "config.ini":
+        return "recorder_ini"
+    if name.endswith(".env"):
+        return "env"
+    if name.endswith(".json"):
+        return "json"
+    if name.endswith((".yaml", ".yml")):
+        return "yaml"
+    if name.endswith(".ini"):
+        return "ini"
+    return "text"
+
+
+def _history_dir(root_id: str, rel_path: Path) -> Path:
+    return WEB_BACKUP_DIR / root_id / rel_path.parent / f"{rel_path.name}.history"
+
+
+def _parse_backup_filename(name: str) -> tuple[str, str]:
+    stem = Path(name).stem
+    if "__" not in stem:
+        return stem, "snapshot"
+    stamp, _, reason = stem.partition("__")
+    return stamp, reason or "snapshot"
+
+
+def _legacy_backups(rel_path: Path) -> list[dict[str, Any]]:
+    prefix = f"{rel_path.name}_"
+    entries: list[dict[str, Any]] = []
+    if not BACKUP_DIR.is_dir():
+        return entries
+    for item in BACKUP_DIR.iterdir():
+        if not item.is_file():
+            continue
+        if not item.name.startswith(prefix):
+            continue
+        stamp = item.name[len(prefix) :]
+        stat = item.stat()
+        entries.append(
+            {
+                "id": f"legacy:{item.name}",
+                "source": "legacy",
+                "reason": "legacy",
+                "label": item.name,
+                "created_at": datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds"),
+                "size": stat.st_size,
+            }
+        )
+    entries.sort(key=lambda item: item["created_at"], reverse=True)
+    return entries
+
+
+def _managed_backups(root_id: str, rel_path: Path) -> list[dict[str, Any]]:
+    hist_dir = _history_dir(root_id, rel_path)
+    entries: list[dict[str, Any]] = []
+    if not hist_dir.is_dir():
+        return entries
+    for item in hist_dir.iterdir():
+        if not item.is_file():
+            continue
+        stamp, reason = _parse_backup_filename(item.name)
+        stat = item.stat()
+        created_at = datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds")
+        if re.fullmatch(r"\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}", stamp):
+            created_at = stamp.replace("_", "T", 1).replace("-", ":", 2)
+            created_at = datetime.strptime(stamp, "%Y-%m-%d_%H-%M-%S").isoformat(timespec="seconds")
+        entries.append(
+            {
+                "id": f"managed:{item.name}",
+                "source": "managed",
+                "reason": reason,
+                "label": item.name,
+                "created_at": created_at,
+                "size": stat.st_size,
+            }
+        )
+    entries.sort(key=lambda item: item["created_at"], reverse=True)
+    return entries
+
+
+def list_backups(root_id: str, relative_path: str, limit: int = 12) -> list[dict[str, Any]]:
+    _, rel_path, _ = resolve_managed_file(root_id, relative_path)
+    items = _managed_backups(root_id, rel_path) + _legacy_backups(rel_path)
+    items.sort(key=lambda item: item["created_at"], reverse=True)
+    return items[: max(1, min(limit, 50))]
+
+
+def _prune_history(hist_dir: Path) -> None:
+    items = sorted(
+        (item for item in hist_dir.iterdir() if item.is_file()),
+        key=lambda item: item.stat().st_mtime,
+        reverse=True,
+    )
+    for item in items[MAX_BACKUPS_PER_FILE:]:
+        item.unlink(missing_ok=True)
+
+
+def backup_managed_file(root_id: str, relative_path: str, *, reason: str) -> Optional[dict[str, Any]]:
+    _, rel_path, target = resolve_managed_file(root_id, relative_path)
+    if not target.is_file():
+        return None
+    hist_dir = _history_dir(root_id, rel_path)
+    hist_dir.mkdir(parents=True, exist_ok=True)
+    suffix = target.suffix or ".bak"
+    backup_name = f"{_now_stamp()}__{reason}{suffix}"
+    backup_path = hist_dir / backup_name
+    shutil.copy2(target, backup_path)
+    _prune_history(hist_dir)
+    stat = backup_path.stat()
+    return {
+        "id": f"managed:{backup_name}",
+        "source": "managed",
+        "reason": reason,
+        "label": backup_name,
+        "created_at": datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds"),
+        "size": stat.st_size,
+    }
+
+
+def _issue(level: str, code: str, message: str, *, line: Optional[int] = None) -> dict[str, Any]:
+    out: dict[str, Any] = {"level": level, "code": code, "message": message}
+    if line is not None:
+        out["line"] = line
+    return out
+
+
+def _stat(key: str, label: str, value: Any) -> dict[str, Any]:
+    return {"key": key, "label": label, "value": "" if value is None else str(value)}
+
+
+def _action(action_id: str, label: str, description: str) -> dict[str, str]:
+    return {"id": action_id, "label": label, "description": description}
+
+
+def _extract_url_from_line(line: str) -> Optional[str]:
+    match = _URL_RE.search(line)
+    if not match:
+        return None
+    return match.group(1).rstrip(")").rstrip("]")
+
+
+def _normalize_url(url: str) -> str:
+    return url.strip().rstrip("/").lower()
+
+
+def _analyze_url_config(content: str) -> dict[str, Any]:
+    issues: list[dict[str, Any]] = []
+    stats: list[dict[str, Any]] = []
+    actions: list[dict[str, str]] = []
+    urls: list[tuple[int, str]] = []
+    invalid_lines: list[int] = []
+    disabled = 0
+    provider_counter: Counter[str] = Counter()
+
+    for index, raw in enumerate(content.splitlines(), start=1):
+        line = raw.strip()
+        if not line:
+            continue
+        if line.startswith("#"):
+            disabled += 1
+            continue
+        url = _extract_url_from_line(line)
+        if not url:
+            invalid_lines.append(index)
+            continue
+        urls.append((index, url))
+        host = url.lower()
+        if "douyin.com" in host:
+            provider_counter["Douyin"] += 1
+        elif "tiktok.com" in host:
+            provider_counter["TikTok"] += 1
+        elif "youtube.com" in host or "youtu.be" in host:
+            provider_counter["YouTube"] += 1
+        elif "kuaishou.com" in host:
+            provider_counter["Kuaishou"] += 1
+        else:
+            provider_counter["Other"] += 1
+
+    normalized = [_normalize_url(url) for _, url in urls]
+    counts = Counter(normalized)
+    duplicates = [url for url, count in counts.items() if count > 1]
+
+    if not urls:
+        issues.append(_issue("error", "empty_active_urls", "No active livestream URLs found."))
+    if duplicates:
+        issues.append(
+            _issue(
+                "warning",
+                "duplicate_urls",
+                f"Detected {len(duplicates)} duplicated livestream URL(s); repeated polling wastes quota and threads.",
+            )
+        )
+        actions.append(
+            _action(
+                "dedupe_urls",
+                "Deduplicate URLs",
+                "Remove repeated active livestream URLs while preserving comments and order.",
+            )
+        )
+    if invalid_lines:
+        issues.append(
+            _issue(
+                "warning",
+                "invalid_lines",
+                f"{len(invalid_lines)} non-empty line(s) do not contain a valid http(s) URL.",
+                line=invalid_lines[0],
+            )
+        )
+
+    stats.extend(
+        [
+            _stat("active_urls", "Active URLs", len(urls)),
+            _stat("disabled_urls", "Disabled URLs", disabled),
+            _stat("duplicate_urls", "Duplicate URLs", len(duplicates)),
+        ]
+    )
+    for provider, count in provider_counter.most_common():
+        stats.append(_stat(f"provider_{provider.lower()}", provider, count))
+
+    summary = f"{len(urls)} active URL(s), {len(duplicates)} duplicate(s), {disabled} disabled line(s)."
+    return {"summary": summary, "issues": issues, "stats": stats, "actions": actions}
+
+
+def _parse_ini(content: str) -> tuple[Optional[configparser.ConfigParser], Optional[str]]:
+    parser = configparser.ConfigParser()
+    parser.optionxform = str
+    try:
+        parser.read_string(content)
+    except (configparser.Error, UnicodeDecodeError) as exc:
+        return None, str(exc)
+    return parser, None
+
+
+def _analyze_youtube_ini(content: str) -> dict[str, Any]:
+    issues: list[dict[str, Any]] = []
+    stats: list[dict[str, Any]] = []
+    actions = [
+        _action(
+            "youtube_balanced_preset",
+            "Apply balanced preset",
+            "Enable RTMPS, set a sane bitrate, and turn on fast_audio while keeping the current stream key.",
+        )
+    ]
+
+    parser, parse_error = _parse_ini(content or "[youtube]\n")
+    if parse_error or parser is None:
+        issues.append(_issue("error", "parse_error", f"Unable to parse youtube.ini: {parse_error or 'unknown error'}"))
+        return {"summary": "youtube.ini parse failed.", "issues": issues, "stats": stats, "actions": actions}
+
+    section = "youtube"
+    key = parser.get(section, "key", fallback="").strip() if parser.has_section(section) else ""
+    rtmps = parser.get(section, "rtmps", fallback="是").strip()
+    bitrate = parser.get(section, "bitrate", fallback="").strip()
+    fast_audio = parser.get(section, "fast_audio", fallback="0").strip()
+
+    if not key:
+        issues.append(_issue("error", "missing_stream_key", "YouTube stream key is empty."))
+    if rtmps and rtmps.lower() in {"否", "0", "false", "no"}:
+        issues.append(_issue("warning", "rtmps_disabled", "RTMPS is disabled; encrypted ingest is recommended."))
+    bitrate_num = int(bitrate) if bitrate.isdigit() else None
+    if bitrate and bitrate_num is None:
+        issues.append(_issue("warning", "bitrate_invalid", "Bitrate is set but not a valid integer."))
+    elif bitrate_num is not None and not (1500 <= bitrate_num <= 9000):
+        issues.append(_issue("warning", "bitrate_out_of_range", "Bitrate is outside the usual 1500-9000 kbps range."))
+
+    stats.extend(
+        [
+            _stat("stream_key", "Stream key", "configured" if key else "missing"),
+            _stat("rtmps", "RTMPS", rtmps or "是"),
+            _stat("bitrate", "Bitrate", bitrate or "auto"),
+            _stat("fast_audio", "fast_audio", fast_audio or "0"),
+        ]
+    )
+    summary = "YouTube relay settings analyzed."
+    return {"summary": summary, "issues": issues, "stats": stats, "actions": actions}
+
+
+def _find_kv(content: str, section: str, key: str) -> tuple[Optional[str], Optional[int]]:
+    current_section = ""
+    for index, raw in enumerate(content.splitlines(), start=1):
+        line = raw.strip()
+        if line.startswith("[") and line.endswith("]"):
+            current_section = line.strip("[]").strip()
+            continue
+        if current_section != section or "=" not in raw:
+            continue
+        left, _, right = raw.partition("=")
+        if left.strip() == key:
+            return right.strip(), index
+    return None, None
+
+
+def _analyze_recorder_ini(content: str) -> dict[str, Any]:
+    issues: list[dict[str, Any]] = []
+    stats: list[dict[str, Any]] = []
+    actions = [
+        _action(
+            "recording_stability_preset",
+            "Apply stability preset",
+            "Switch to safer segmented recording defaults for long-running livestream capture.",
+        )
+    ]
+
+    if f"[{_RECORDER_SECTION}]" not in content:
+        issues.append(_issue("error", "missing_record_section", "Missing [录制设置] section."))
+        return {"summary": "Recorder config parse failed.", "issues": issues, "stats": stats, "actions": actions}
+
+    save_format, save_format_line = _find_kv(content, _RECORDER_SECTION, "视频保存格式ts|mkv|flv|mp4|mp3音频|m4a音频")
+    quality, _ = _find_kv(content, _RECORDER_SECTION, "原画|超清|高清|标清|流畅")
+    threads, threads_line = _find_kv(content, _RECORDER_SECTION, "同一时间访问网络的线程数")
+    loop_seconds, loop_line = _find_kv(content, _RECORDER_SECTION, "循环时间(秒)")
+    proxy_enabled, proxy_enabled_line = _find_kv(content, _RECORDER_SECTION, "是否使用代理ip(是/否)")
+    proxy_addr, _ = _find_kv(content, _RECORDER_SECTION, "代理地址")
+    split_enabled, split_enabled_line = _find_kv(content, _RECORDER_SECTION, "分段录制是否开启")
+    split_time, split_time_line = _find_kv(content, _RECORDER_SECTION, "视频分段时间(秒)")
+    disk_limit, disk_limit_line = _find_kv(content, _RECORDER_SECTION, "录制空间剩余阈值(gb)")
+
+    threads_num = int(threads) if threads and threads.isdigit() else None
+    loop_num = int(loop_seconds) if loop_seconds and loop_seconds.isdigit() else None
+    split_num = int(split_time) if split_time and split_time.isdigit() else None
+    disk_num = float(disk_limit) if disk_limit and re.fullmatch(r"\d+(?:\.\d+)?", disk_limit) else None
+
+    if save_format and save_format.lower() == "mp4":
+        issues.append(
+            _issue(
+                "warning",
+                "unsafe_container",
+                "Recording directly to mp4 is less resilient to interruption than ts or mkv.",
+                line=save_format_line,
+            )
+        )
+    if threads_num is not None and threads_num > 8:
+        issues.append(
+            _issue(
+                "warning",
+                "threads_high",
+                "Concurrent network threads are high; aggressive polling can trigger bans or rate limits.",
+                line=threads_line,
+            )
+        )
+    if loop_num is not None and loop_num < 30:
+        issues.append(
+            _issue(
+                "warning",
+                "loop_too_low",
+                "Polling interval below 30s is unusually aggressive for long-running monitoring.",
+                line=loop_line,
+            )
+        )
+    if proxy_enabled and proxy_enabled.strip() in {"是", "1", "true", "True"} and not (proxy_addr or "").strip():
+        issues.append(
+            _issue(
+                "error",
+                "proxy_missing_address",
+                "Proxy recording is enabled but proxy address is empty.",
+                line=proxy_enabled_line,
+            )
+        )
+    if split_enabled and split_enabled.strip() in {"否", "0", "false", "False"}:
+        issues.append(
+            _issue(
+                "warning",
+                "split_disabled",
+                "Segmented recording is disabled; long recordings become harder to recover after failures.",
+                line=split_enabled_line,
+            )
+        )
+    if split_num is not None and split_num > 7200:
+        issues.append(
+            _issue(
+                "warning",
+                "split_too_large",
+                "Segment duration is very long; smaller chunks are easier to recover and upload.",
+                line=split_time_line,
+            )
+        )
+    if disk_num is not None and disk_num < 1.0:
+        issues.append(
+            _issue(
+                "warning",
+                "disk_threshold_low",
+                "Remaining disk threshold is below 1 GB; the recorder may run into disk exhaustion before stopping.",
+                line=disk_limit_line,
+            )
+        )
+
+    stats.extend(
+        [
+            _stat("format", "Save format", save_format or "unknown"),
+            _stat("quality", "Quality", quality or "unknown"),
+            _stat("threads", "Threads", threads or "unknown"),
+            _stat("loop_seconds", "Loop seconds", loop_seconds or "unknown"),
+            _stat("split_recording", "Segmented recording", split_enabled or "unknown"),
+            _stat("split_seconds", "Segment seconds", split_time or "unknown"),
+            _stat("disk_threshold_gb", "Disk threshold GB", disk_limit or "unknown"),
+        ]
+    )
+    summary = "Recorder config analyzed for polling, container, proxy, and segment safety."
+    return {"summary": summary, "issues": issues, "stats": stats, "actions": actions}
+
+
+def _analyze_env(content: str) -> dict[str, Any]:
+    issues: list[dict[str, Any]] = []
+    stats: list[dict[str, Any]] = []
+    data: dict[str, str] = {}
+    for index, raw in enumerate(content.splitlines(), start=1):
+        line = raw.strip()
+        if not line or line.startswith("#") or "=" not in raw:
+            continue
+        key, _, value = raw.partition("=")
+        data[key.strip()] = value.strip()
+    host = data.get("HOST", "")
+    port = data.get("PORT", "")
+    if host and host not in {"0.0.0.0", "::"}:
+        issues.append(_issue("warning", "host_not_public", "HOST is not 0.0.0.0; LAN access may fail."))
+    if port and not port.isdigit():
+        issues.append(_issue("warning", "port_invalid", "PORT is not a valid integer."))
+    stats.extend([_stat("HOST", "HOST", host or "unset"), _stat("PORT", "PORT", port or "unset")])
+    return {"summary": "Environment file analyzed.", "issues": issues, "stats": stats, "actions": []}
+
+
+def _analyze_json(content: str) -> dict[str, Any]:
+    issues: list[dict[str, Any]] = []
+    stats: list[dict[str, Any]] = []
+    try:
+        payload = json.loads(content) if content.strip() else {}
+    except json.JSONDecodeError as exc:
+        issues.append(_issue("error", "json_parse_error", f"JSON parse error: {exc.msg}", line=exc.lineno))
+        return {"summary": "JSON parse failed.", "issues": issues, "stats": stats, "actions": []}
+    shape = "array" if isinstance(payload, list) else "object" if isinstance(payload, dict) else type(payload).__name__
+    stats.append(_stat("json_shape", "JSON shape", shape))
+    if isinstance(payload, dict):
+        stats.append(_stat("json_keys", "Top-level keys", len(payload)))
+    return {"summary": "JSON syntax is valid.", "issues": issues, "stats": stats, "actions": []}
+
+
+def _analyze_generic(content: str) -> dict[str, Any]:
+    line_count = len(content.splitlines())
+    size = len(content.encode("utf-8", errors="replace"))
+    return {
+        "summary": "Text file analyzed.",
+        "issues": [],
+        "stats": [_stat("lines", "Lines", line_count), _stat("bytes", "Bytes", size)],
+        "actions": [],
+    }
+
+
+def analyze_content(root_id: str, relative_path: str, content: str, *, source: str = "saved") -> dict[str, Any]:
+    _, rel_path, target = resolve_managed_file(root_id, relative_path)
+    kind = _file_kind(rel_path)
+    if kind == "url_config":
+        detail = _analyze_url_config(content)
+    elif kind == "youtube_ini":
+        detail = _analyze_youtube_ini(content)
+    elif kind == "recorder_ini":
+        detail = _analyze_recorder_ini(content)
+    elif kind == "env":
+        detail = _analyze_env(content)
+    elif kind == "json":
+        detail = _analyze_json(content)
+    else:
+        detail = _analyze_generic(content)
+
+    all_stats = [
+        _stat("file_kind", "File type", kind),
+        _stat("line_count", "Line count", len(content.splitlines())),
+        _stat("size_bytes", "Bytes", len(content.encode("utf-8", errors="replace"))),
+        *detail["stats"],
+    ]
+    return {
+        "root_id": root_id,
+        "path": rel_path.as_posix(),
+        "exists": target.exists(),
+        "kind": kind,
+        "source": source,
+        "summary": detail["summary"],
+        "issues": detail["issues"],
+        "stats": all_stats,
+        "actions": detail["actions"],
+    }
+
+
+def inspect_file(root_id: str, relative_path: str, *, draft_content: Optional[str] = None) -> dict[str, Any]:
+    _, rel_path, target = resolve_managed_file(root_id, relative_path)
+    if draft_content is None:
+        content = _read_text(target) if target.is_file() else ""
+        source = "saved"
+    else:
+        content = draft_content
+        source = "draft"
+    result = analyze_content(root_id, rel_path.as_posix(), content, source=source)
+    result["backups"] = list_backups(root_id, rel_path.as_posix())
+    return result
+
+
+def _set_value_in_section(content: str, section: str, key: str, value: str) -> str:
+    lines = content.splitlines()
+    header = f"[{section}]"
+    out: list[str] = []
+    section_found = False
+    in_section = False
+    inserted = False
+    pattern = re.compile(rf"^\s*{re.escape(key)}\s*=")
+
+    for raw in lines:
+        stripped = raw.strip()
+        if stripped.startswith("[") and stripped.endswith("]"):
+            if in_section and not inserted:
+                out.append(f"{key} = {value}")
+                inserted = True
+            in_section = stripped == header
+            if in_section:
+                section_found = True
+            out.append(raw)
+            continue
+        if in_section and pattern.match(raw):
+            out.append(f"{key} = {value}")
+            inserted = True
+            continue
+        out.append(raw)
+
+    if not section_found:
+        if out and out[-1].strip():
+            out.append("")
+        out.append(header)
+        out.append(f"{key} = {value}")
+    elif in_section and not inserted:
+        out.append(f"{key} = {value}")
+
+    return "\n".join(out).rstrip() + "\n"
+
+
+def _dedupe_url_config(content: str) -> str:
+    seen: set[str] = set()
+    out: list[str] = []
+    for raw in content.splitlines():
+        line = raw.strip()
+        if not line or line.startswith("#"):
+            out.append(raw.rstrip())
+            continue
+        url = _extract_url_from_line(line)
+        if not url:
+            out.append(raw.rstrip())
+            continue
+        key = _normalize_url(url)
+        if key in seen:
+            continue
+        seen.add(key)
+        out.append(raw.rstrip())
+    return "\n".join(out).rstrip() + "\n"
+
+
+def optimize_content(root_id: str, relative_path: str, action: str, content: str) -> dict[str, Any]:
+    _, rel_path, _ = resolve_managed_file(root_id, relative_path)
+    kind = _file_kind(rel_path)
+
+    if action == "dedupe_urls" and kind == "url_config":
+        next_content = _dedupe_url_config(content)
+        message = "Removed duplicated active livestream URLs."
+    elif action == "youtube_balanced_preset" and kind == "youtube_ini":
+        parser, parse_error = _parse_ini(content or "[youtube]\n")
+        if parse_error or parser is None:
+            key = ""
+        else:
+            key = parser.get("youtube", "key", fallback="").strip() if parser.has_section("youtube") else ""
+        next_content = content or "[youtube]\n"
+        next_content = _set_value_in_section(next_content, "youtube", "key", key)
+        next_content = _set_value_in_section(next_content, "youtube", "rtmps", "是")
+        next_content = _set_value_in_section(next_content, "youtube", "bitrate", "4500")
+        next_content = _set_value_in_section(next_content, "youtube", "fast_audio", "1")
+        message = "Applied balanced YouTube relay preset."
+    elif action == "recording_stability_preset" and kind == "recorder_ini":
+        next_content = content
+        next_content = _set_value_in_section(
+            next_content,
+            _RECORDER_SECTION,
+            "视频保存格式ts|mkv|flv|mp4|mp3音频|m4a音频",
+            "ts",
+        )
+        next_content = _set_value_in_section(next_content, _RECORDER_SECTION, "分段录制是否开启", "是")
+        next_content = _set_value_in_section(next_content, _RECORDER_SECTION, "视频分段时间(秒)", "1800")
+        next_content = _set_value_in_section(next_content, _RECORDER_SECTION, "录制空间剩余阈值(gb)", "2.0")
+        message = "Applied long-running recording stability preset."
+    else:
+        raise ValueError(f"Unsupported optimization action: {action}")
+
+    return {
+        "message": message,
+        "content": next_content,
+        "insights": analyze_content(root_id, rel_path.as_posix(), next_content, source="draft"),
+    }
+
+
+def save_managed_file(root_id: str, relative_path: str, content: str) -> dict[str, Any]:
+    root, rel_path, target = resolve_managed_file(root_id, relative_path)
+    if not root.allow_create and not target.exists():
+        raise PermissionError(f"Creating files is disabled for {root_id}")
+    current = _read_text(target) if target.is_file() else None
+    if current == content:
+        return {
+            "root_id": root.root_id,
+            "path": rel_path.as_posix(),
+            "message": "No changes",
+            "backup": None,
+        }
+    backup = backup_managed_file(root_id, rel_path.as_posix(), reason="save")
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(content, encoding="utf-8")
+    return {
+        "root_id": root.root_id,
+        "path": rel_path.as_posix(),
+        "message": "Saved successfully",
+        "backup": backup,
+    }
+
+
+def delete_managed_file(root_id: str, relative_path: str) -> dict[str, Any]:
+    root, rel_path, target = resolve_managed_file(root_id, relative_path)
+    if not root.allow_delete:
+        raise PermissionError(f"Deleting files is disabled for {root_id}")
+    if not target.exists():
+        raise FileNotFoundError(relative_path)
+    backup = backup_managed_file(root_id, rel_path.as_posix(), reason="delete")
+    target.unlink()
+    return {
+        "root_id": root.root_id,
+        "path": rel_path.as_posix(),
+        "message": "Deleted successfully",
+        "backup": backup,
+    }
+
+
+def restore_backup(root_id: str, relative_path: str, backup_id: str) -> dict[str, Any]:
+    root, rel_path, target = resolve_managed_file(root_id, relative_path)
+    if backup_id.startswith("managed:"):
+        hist_dir = _history_dir(root_id, rel_path)
+        source = (hist_dir / backup_id.split(":", 1)[1]).resolve()
+        if hist_dir.resolve() not in (source, *source.parents):
+            raise PermissionError("Backup path escapes history directory")
+    elif backup_id.startswith("legacy:"):
+        source = (BACKUP_DIR / backup_id.split(":", 1)[1]).resolve()
+        if BACKUP_DIR.resolve() not in (source, *source.parents):
+            raise PermissionError("Backup path escapes backup directory")
+    else:
+        raise ValueError("Unsupported backup id")
+
+    if not source.is_file():
+        raise FileNotFoundError(backup_id)
+
+    backup = backup_managed_file(root_id, rel_path.as_posix(), reason="restore")
+    content = _read_text(source)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(content, encoding="utf-8")
+    return {
+        "root_id": root.root_id,
+        "path": rel_path.as_posix(),
+        "message": "Backup restored",
+        "content": content,
+        "backup": backup,
+    }

src/control_plane.py

@@ -214,6 +214,10 @@ def _resolve_managed_file(root_id: str, relative_path: str) -> tuple[ManagedRoot
     return root, rel, target
 
 
+def resolve_managed_file(root_id: str, relative_path: str) -> tuple[ManagedRoot, Path, Path]:
+    return _resolve_managed_file(root_id, relative_path)
+
+
 def list_root_summaries() -> list[dict]:
     summaries: list[dict] = []
     for root in managed_roots().values():

src/youtube_ini_sync.py

@@ -2,8 +2,11 @@
 from __future__ import annotations
 
 import configparser
+import io
 from pathlib import Path
 
+from src.config_governance import save_managed_file
+
 
 def write_youtube_ini_stream_key(repo_root: Path, stream_key_or_url: str) -> str:
     root = repo_root.resolve()
@@ -19,6 +22,7 @@ def write_youtube_ini_stream_key(repo_root: Path, stream_key_or_url: str) -> str
     if not parser.has_section("youtube"):
         parser.add_section("youtube")
     parser.set("youtube", "key", value)
-    with cfg_path.open("w", encoding="utf-8") as fh:
-        parser.write(fh)
+    buf = io.StringIO()
+    parser.write(buf)
+    save_managed_file("app-config", "youtube.ini", buf.getvalue())
     return f"wrote {cfg_path}"

tests/test_config_governance.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import unittest
+
+from src.config_governance import analyze_content, optimize_content
+
+
+class ConfigGovernanceTests(unittest.TestCase):
+    def test_url_config_detects_duplicates(self) -> None:
+        sample = "\n".join(
+            [
+                "https://live.douyin.com/123456,主播A",
+                "https://live.douyin.com/123456,主播A重复",
+                "# https://live.douyin.com/999999",
+                "https://www.tiktok.com/@demo/live",
+            ]
+        )
+        result = analyze_content("app-config", "URL_config.ini", sample, source="draft")
+        codes = {item["code"] for item in result["issues"]}
+        self.assertIn("duplicate_urls", codes)
+        self.assertTrue(any(item["id"] == "dedupe_urls" for item in result["actions"]))
+
+    def test_dedupe_urls_keeps_single_active_copy(self) -> None:
+        sample = "\n".join(
+            [
+                "# keep comment",
+                "https://live.douyin.com/123456",
+                "https://live.douyin.com/123456",
+                "https://live.douyin.com/654321",
+            ]
+        )
+        result = optimize_content("app-config", "URL_config.ini", "dedupe_urls", sample)
+        content = result["content"]
+        self.assertEqual(content.count("https://live.douyin.com/123456"), 1)
+        self.assertIn("# keep comment", content)
+        self.assertIn("https://live.douyin.com/654321", content)
+
+    def test_youtube_preset_preserves_key_and_enables_balanced_defaults(self) -> None:
+        sample = "[youtube]\nkey = demo-key\nrtmps = 否\nfast_audio = 0\n"
+        result = optimize_content("app-config", "youtube.ini", "youtube_balanced_preset", sample)
+        content = result["content"]
+        self.assertIn("key = demo-key", content)
+        self.assertIn("rtmps = 是", content)
+        self.assertIn("bitrate = 4500", content)
+        self.assertIn("fast_audio = 1", content)
+
+
+if __name__ == "__main__":
+    unittest.main()

web-console/components/console/ConfigAdvisorPanel.tsx

@@ -0,0 +1,332 @@
+"use client";
+
+import { AlertTriangle, History, Loader2, RefreshCw, RotateCcw, Sparkles } from "lucide-react";
+import { useCallback, useEffect, useState } from "react";
+
+type Lang = "zh" | "en";
+
+type ConfigIssue = {
+  level: string;
+  code: string;
+  message: string;
+  line?: number;
+};
+
+type ConfigStat = {
+  key: string;
+  label: string;
+  value: string;
+};
+
+type ConfigAction = {
+  id: string;
+  label: string;
+  description: string;
+};
+
+type ConfigBackup = {
+  id: string;
+  source: string;
+  reason: string;
+  label: string;
+  created_at: string;
+  size: number;
+};
+
+type ConfigInsights = {
+  kind?: string;
+  source?: string;
+  summary?: string;
+  issues?: ConfigIssue[];
+  stats?: ConfigStat[];
+  actions?: ConfigAction[];
+  backups?: ConfigBackup[];
+};
+
+type Props = {
+  lang: Lang;
+  rootId: string;
+  filePath: string;
+  fileContent: string;
+  refreshKey: number;
+  busy: string | null;
+  fetchJson: <T,>(path: string, init?: RequestInit) => Promise<T>;
+  notify: (message: string) => void;
+  onChangeContent: (content: string) => void;
+  onRefreshListing?: () => Promise<void> | void;
+};
+
+function tr(lang: Lang, zh: string, en: string) {
+  return lang === "zh" ? zh : en;
+}
+
+function formatBytes(size: number) {
+  if (!size) return "0 B";
+  const units = ["B", "KB", "MB", "GB"];
+  let value = size;
+  let unit = 0;
+  while (value >= 1024 && unit < units.length - 1) {
+    value /= 1024;
+    unit += 1;
+  }
+  return `${value.toFixed(value >= 10 || unit === 0 ? 0 : 1)} ${units[unit]}`;
+}
+
+function issueTone(level: string) {
+  if (level === "error") return "border-rose-500/30 bg-rose-500/10 text-rose-100";
+  if (level === "warning") return "border-amber-500/30 bg-amber-500/10 text-amber-100";
+  return "border-cyan-500/30 bg-cyan-500/10 text-cyan-100";
+}
+
+export function ConfigAdvisorPanel({
+  lang,
+  rootId,
+  filePath,
+  fileContent,
+  refreshKey,
+  busy,
+  fetchJson,
+  notify,
+  onChangeContent,
+  onRefreshListing,
+}: Props) {
+  const [insights, setInsights] = useState<ConfigInsights | null>(null);
+  const [loading, setLoading] = useState(false);
+  const [draftLoading, setDraftLoading] = useState(false);
+  const [actionBusy, setActionBusy] = useState<string | null>(null);
+  const [restoreBusy, setRestoreBusy] = useState<string | null>(null);
+
+  const t = useCallback((zh: string, en: string) => tr(lang, zh, en), [lang]);
+
+  const loadSavedInsights = useCallback(async () => {
+    if (!rootId || !filePath) {
+      setInsights(null);
+      return;
+    }
+    setLoading(true);
+    try {
+      const params = new URLSearchParams({ root: rootId, path: filePath });
+      const data = await fetchJson<ConfigInsights>(`/config_insights?${params.toString()}`);
+      setInsights(data);
+    } catch (reason) {
+      setInsights(null);
+      notify((reason as Error).message);
+    } finally {
+      setLoading(false);
+    }
+  }, [fetchJson, filePath, notify, rootId]);
+
+  useEffect(() => {
+    void loadSavedInsights();
+  }, [loadSavedInsights, refreshKey]);
+
+  const analyzeDraft = async () => {
+    if (!rootId || !filePath) return;
+    setDraftLoading(true);
+    try {
+      const data = await fetchJson<ConfigInsights>("/config_insights", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ root: rootId, path: filePath, content: fileContent }),
+      });
+      setInsights(data);
+      notify(t("已分析当前编辑内容", "Analyzed current editor draft"));
+    } catch (reason) {
+      notify((reason as Error).message);
+    } finally {
+      setDraftLoading(false);
+    }
+  };
+
+  const applyAction = async (actionId: string) => {
+    if (!rootId || !filePath) return;
+    setActionBusy(actionId);
+    try {
+      const data = await fetchJson<{ message?: string; content?: string; insights?: ConfigInsights }>(
+        "/config_optimize",
+        {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({ root: rootId, path: filePath, content: fileContent, action: actionId }),
+        },
+      );
+      if (typeof data.content === "string") onChangeContent(data.content);
+      if (data.insights) {
+        setInsights((prev) => ({ ...prev, ...data.insights, backups: prev?.backups || [] }));
+      }
+      notify(data.message || t("已应用优化", "Optimization applied"));
+    } catch (reason) {
+      notify((reason as Error).message);
+    } finally {
+      setActionBusy(null);
+    }
+  };
+
+  const restoreFromBackup = async (backupId: string) => {
+    if (!rootId || !filePath) return;
+    setRestoreBusy(backupId);
+    try {
+      const data = await fetchJson<{ message?: string; content?: string }>("/config_restore", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ root: rootId, path: filePath, backup_id: backupId }),
+      });
+      if (typeof data.content === "string") onChangeContent(data.content);
+      await Promise.resolve(onRefreshListing?.());
+      await loadSavedInsights();
+      notify(data.message || t("已恢复历史版本", "Backup restored"));
+    } catch (reason) {
+      notify((reason as Error).message);
+    } finally {
+      setRestoreBusy(null);
+    }
+  };
+
+  if (!filePath) {
+    return (
+      <div className="rounded-2xl border border-white/8 bg-black/20 p-4 text-sm text-slate-400">
+        {t("选择文件后显示诊断、回滚和一键优化。", "Pick a file to see diagnostics, rollback, and one-click optimizations.")}
+      </div>
+    );
+  }
+
+  const issues = insights?.issues || [];
+  const actions = insights?.actions || [];
+  const backups = insights?.backups || [];
+  const stats = insights?.stats || [];
+
+  return (
+    <div className="space-y-4">
+      <div className="rounded-2xl border border-cyan-500/15 bg-gradient-to-br from-cyan-500/[0.08] via-black/25 to-transparent p-4">
+        <div className="flex flex-wrap items-start justify-between gap-3">
+          <div>
+            <div className="flex items-center gap-2 text-white">
+              <Sparkles className="h-4 w-4 text-cyan-300" />
+              <h3 className="text-sm font-semibold">{t("配置治理", "Config governance")}</h3>
+              {insights?.source ? (
+                <span className="rounded-full border border-white/10 bg-black/25 px-2 py-0.5 text-[10px] uppercase tracking-wider text-slate-400">
+                  {insights.source}
+                </span>
+              ) : null}
+            </div>
+            <p className="mt-2 text-xs text-slate-400">
+              {insights?.summary ||
+                t("校验配置风险、查看历史快照，并将常见优化应用到编辑器。", "Validate risks, inspect snapshots, and apply optimizations into the editor.")}
+            </p>
+          </div>
+          <div className="flex flex-wrap gap-2">
+            <button
+              type="button"
+              disabled={!!busy || loading}
+              onClick={() => void loadSavedInsights()}
+              className="btn-focus inline-flex items-center gap-2 rounded-xl border border-white/10 bg-black/25 px-3 py-2 text-xs text-slate-100 disabled:opacity-40"
+            >
+              {loading ? <Loader2 className="h-3.5 w-3.5 animate-spin" /> : <RefreshCw className="h-3.5 w-3.5" />}
+              {t("刷新已保存状态", "Refresh saved state")}
+            </button>
+            <button
+              type="button"
+              disabled={!!busy || draftLoading}
+              onClick={() => void analyzeDraft()}
+              className="btn-focus inline-flex items-center gap-2 rounded-xl border border-cyan-500/30 bg-cyan-500/15 px-3 py-2 text-xs text-cyan-100 disabled:opacity-40"
+            >
+              {draftLoading ? <Loader2 className="h-3.5 w-3.5 animate-spin" /> : <AlertTriangle className="h-3.5 w-3.5" />}
+              {t("分析当前编辑器", "Analyze editor draft")}
+            </button>
+          </div>
+        </div>
+
+        <div className="mt-4 grid gap-2 sm:grid-cols-2 xl:grid-cols-4">
+          {stats.slice(0, 8).map((item) => (
+            <div key={item.key} className="rounded-xl border border-white/8 bg-black/20 px-3 py-2.5">
+              <div className="text-[10px] uppercase tracking-wider text-slate-500">{item.label}</div>
+              <div className="mt-1 truncate font-mono text-sm text-white">{item.value}</div>
+            </div>
+          ))}
+        </div>
+      </div>
+
+      <div className="grid gap-4 xl:grid-cols-2">
+        <div className="rounded-2xl border border-white/8 bg-black/20 p-4">
+          <div className="flex items-center gap-2 text-white">
+            <AlertTriangle className="h-4 w-4 text-amber-300" />
+            <h4 className="text-sm font-semibold">{t("风险与建议", "Risks & guidance")}</h4>
+          </div>
+          {issues.length ? (
+            <div className="mt-3 space-y-2">
+              {issues.map((item, index) => (
+                <div key={`${item.code}-${index}`} className={`rounded-xl border px-3 py-2 text-xs ${issueTone(item.level)}`}>
+                  <div className="font-medium">
+                    {item.message}
+                    {item.line ? ` (L${item.line})` : ""}
+                  </div>
+                  <div className="mt-1 font-mono opacity-70">{item.code}</div>
+                </div>
+              ))}
+            </div>
+          ) : (
+            <p className="mt-3 text-xs text-emerald-300/90">{t("当前没有发现明显配置风险。", "No obvious config risks were detected.")}</p>
+          )}
+
+          <div className="mt-4 flex flex-wrap gap-2">
+            {actions.length ? (
+              actions.map((item) => (
+                <button
+                  key={item.id}
+                  type="button"
+                  disabled={!!busy || !!actionBusy}
+                  onClick={() => void applyAction(item.id)}
+                  className="btn-focus rounded-xl border border-violet-500/30 bg-violet-500/15 px-3 py-2 text-xs text-violet-100 disabled:opacity-40"
+                  title={item.description}
+                >
+                  {actionBusy === item.id ? t("处理中...", "Working...") : item.label}
+                </button>
+              ))
+            ) : (
+              <span className="text-xs text-slate-500">{t("当前文件没有可自动应用的预设。", "No auto-applicable presets for this file.")}</span>
+            )}
+          </div>
+        </div>
+
+        <div className="rounded-2xl border border-white/8 bg-black/20 p-4">
+          <div className="flex items-center gap-2 text-white">
+            <History className="h-4 w-4 text-cyan-300" />
+            <h4 className="text-sm font-semibold">{t("历史快照", "Backup snapshots")}</h4>
+          </div>
+          {backups.length ? (
+            <div className="mt-3 space-y-2">
+              {backups.slice(0, 8).map((item) => (
+                <div key={item.id} className="rounded-xl border border-white/8 bg-black/25 px-3 py-2.5">
+                  <div className="flex flex-wrap items-center justify-between gap-2">
+                    <div>
+                      <div className="text-xs text-white">{new Date(item.created_at).toLocaleString(lang === "zh" ? "zh-CN" : "en-US", { hour12: false })}</div>
+                      <div className="mt-1 text-[11px] text-slate-500">
+                        {item.reason} · {formatBytes(item.size)}
+                        {item.source === "legacy" ? ` · ${t("旧备份", "legacy")}` : ""}
+                      </div>
+                    </div>
+                    <button
+                      type="button"
+                      disabled={!!busy || !!restoreBusy}
+                      onClick={() => void restoreFromBackup(item.id)}
+                      className="btn-focus inline-flex items-center gap-1 rounded-lg border border-white/10 bg-white/5 px-3 py-1.5 text-xs text-slate-100 disabled:opacity-40"
+                    >
+                      {restoreBusy === item.id ? (
+                        <Loader2 className="h-3.5 w-3.5 animate-spin" />
+                      ) : (
+                        <RotateCcw className="h-3.5 w-3.5" />
+                      )}
+                      {t("恢复", "Restore")}
+                    </button>
+                  </div>
+                </div>
+              ))}
+            </div>
+          ) : (
+            <p className="mt-3 text-xs text-slate-500">{t("当前文件还没有 Web 侧快照。保存、删除或恢复后会自动累积历史。", "No Web snapshots yet. Saving, deleting, or restoring will accumulate history automatically.")}</p>
+          )}
+        </div>
+      </div>
+    </div>
+  );
+}

web-console/components/console/LiveConsoleWorkspace.tsx

@@ -3,6 +3,7 @@
 import { type ChangeEvent, type MouseEvent, useCallback, useEffect, useMemo, useState } from "react";
 
 import OverviewCharts from "@/components/OverviewCharts";
+import { ConfigAdvisorPanel } from "@/components/console/ConfigAdvisorPanel";
 import { normalizeBrowserReachableHttpUrl } from "@/lib/panelUrls";
 
 type Lang = "zh" | "en";
@@ -236,6 +237,7 @@ export function LiveConsoleWorkspace() {
   const [previewMode, setPreviewMode] = useState<PreviewMode>("tap");
   const [previewPoint, setPreviewPoint] = useState<{ x: number; y: number } | null>(null);
   const [fileImportKey, setFileImportKey] = useState(0);
+  const [configInsightVersion, setConfigInsightVersion] = useState(0);
   const [shotObjectUrl, setShotObjectUrl] = useState<string | null>(null);
   const [shotFetchErr, setShotFetchErr] = useState<string | null>(null);
   const [shotLoading, setShotLoading] = useState(false);
@@ -642,6 +644,7 @@ export function LiveConsoleWorkspace() {
       });
       notify(tr("文件已保存", "File saved"));
       await refreshFiles(rootId, filePath);
+      setConfigInsightVersion((value) => value + 1);
     } catch (reason) {
       notify((reason as Error).message);
     } finally {
@@ -665,6 +668,7 @@ export function LiveConsoleWorkspace() {
       });
       notify(tr("文件已创建", "File created"));
       await refreshFiles(rootId, nextPath);
+      setConfigInsightVersion((value) => value + 1);
     } catch (reason) {
       notify((reason as Error).message);
     } finally {
@@ -687,6 +691,7 @@ export function LiveConsoleWorkspace() {
       );
       notify(tr("文件已删除", "File deleted"));
       await refreshFiles(rootId);
+      setConfigInsightVersion((value) => value + 1);
     } catch (reason) {
       notify((reason as Error).message);
     } finally {
@@ -711,6 +716,7 @@ export function LiveConsoleWorkspace() {
       notify(tr("导入完成，已替换同名文件", "Import finished and replaced matching file"));
       setFilePath(upload.name);
       await refreshFiles(rootId, upload.name);
+      setConfigInsightVersion((value) => value + 1);
     } catch (reason) {
       notify((reason as Error).message);
     } finally {
@@ -1518,6 +1524,20 @@ export function LiveConsoleWorkspace() {
                   </div>
                 </div>
                 <div className="mt-4 rounded-2xl border border-white/8 bg-black/20 px-4 py-3 font-mono text-[11px] text-slate-500">{filePath ? `${rootId} / ${filePath}` : tr("请选择文件", "Pick a file")}</div>
+                <div className="mt-4">
+                  <ConfigAdvisorPanel
+                    lang={lang}
+                    rootId={rootId}
+                    filePath={filePath}
+                    fileContent={fileContent}
+                    refreshKey={configInsightVersion}
+                    busy={busy}
+                    fetchJson={fetchJson}
+                    notify={notify}
+                    onChangeContent={setFileContent}
+                    onRefreshListing={() => refreshFiles(rootId, filePath)}
+                  />
+                </div>
                 <textarea className="log-scroll mt-4 min-h-[32rem] w-full resize-y rounded-2xl border border-white/10 bg-black/45 p-4 font-mono text-xs leading-relaxed text-slate-200" spellCheck={false} value={fileContent} onChange={(event) => setFileContent(event.target.value)} />
               </article>
             </section>

web-console/components/live-product/LiveYoutubeUnmannedView.tsx

@@ -71,6 +71,7 @@ export function LiveYoutubeUnmannedView({
   const [proUrlDraft, setProUrlDraft] = useState("");
   const [urlAutoRefresh, setUrlAutoRefresh] = useState(true);
   const [urlReloading, setUrlReloading] = useState(false);
+  const [urlOptimizing, setUrlOptimizing] = useState(false);
   const [urlFetchNote, setUrlFetchNote] = useState<string | null>(null);
   const urlListDirtyRef = useRef(false);
 
@@ -260,6 +261,32 @@ export function LiveYoutubeUnmannedView({
     await saveYoutubeIniToServer(body);
   };
 
+  const applyYoutubeBalancedPreset = async () => {
+    setYtIniLoading(true);
+    setYtIniStatus(null);
+    try {
+      const source = youtubeIniText || buildYoutubeIniFromFields(ytFields);
+      const data = await fetchJson<{ message?: string; content?: string }>("/config_optimize", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          root: "app-config",
+          path: "youtube.ini",
+          content: source,
+          action: "youtube_balanced_preset",
+        }),
+      });
+      const next = data.content ?? source;
+      setYoutubeIniText(next);
+      setYtFields(parseYoutubeIniFields(next));
+      setYtIniStatus(data.message || t("已应用 YouTube 预设", "Applied YouTube preset"));
+    } catch (e) {
+      setYtIniStatus((e as Error).message);
+    } finally {
+      setYtIniLoading(false);
+    }
+  };
+
   const syncFieldsFromText = () => {
     setYtFields(parseYoutubeIniFields(youtubeIniText));
     setYtIniStatus(t("已从下方原文同步到表单", "Synced from raw"));
@@ -292,6 +319,36 @@ export function LiveYoutubeUnmannedView({
     }
   };
 
+  const dedupeUrlDraft = async () => {
+    const source = mode === "single" ? urlConfig : proUrlDraft;
+    setUrlOptimizing(true);
+    setUrlFetchNote(null);
+    try {
+      const data = await fetchJson<{ message?: string; content?: string }>("/config_optimize", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          root: "app-config",
+          path: "URL_config.ini",
+          content: source,
+          action: "dedupe_urls",
+        }),
+      });
+      const next = data.content ?? source;
+      if (mode === "single") setUrlConfig(next);
+      else {
+        setProUrlDraft(next);
+        if (active) updateActive({ urlLines: next });
+      }
+      urlListDirtyRef.current = true;
+      setUrlFetchNote(data.message || t("已去重当前 URL 列表", "Deduplicated current URL list"));
+    } catch (e) {
+      setUrlFetchNote((e as Error).message);
+    } finally {
+      setUrlOptimizing(false);
+    }
+  };
+
   const pushProStreamKeyToServer = async () => {
     if (!active) return;
     const base = parseYoutubeIniFields(youtubeIniText || buildYoutubeIniFromFields(ytFields));
@@ -306,7 +363,7 @@ export function LiveYoutubeUnmannedView({
     updateActive({ streamKey: f.key });
   };
 
-  const lock = busy || ytIniLoading || urlReloading;
+  const lock = busy || ytIniLoading || urlReloading || urlOptimizing;
 
   const urlToolbar = (
     <div className="mt-2 flex flex-wrap items-center gap-4 text-xs text-zinc-400">
@@ -327,6 +384,14 @@ export function LiveYoutubeUnmannedView({
       >
         {t("手动重新读取", "Reload from server")}
       </button>
+      <button
+        type="button"
+        disabled={lock}
+        onClick={() => void dedupeUrlDraft()}
+        className="rounded-lg border border-cyan-500/20 bg-cyan-500/10 px-3 py-1.5 text-cyan-100 hover:bg-cyan-500/15"
+      >
+        {urlOptimizing ? t("去重中...", "Deduplicating...") : t("智能去重", "Deduplicate")}
+      </button>
     </div>
   );
 
@@ -564,6 +629,14 @@ export function LiveYoutubeUnmannedView({
               {ytIniLoading ? <Loader2 className="h-4 w-4 animate-spin" /> : null}
               {t("保存 youtube.ini", "Save youtube.ini")}
             </button>
+            <button
+              type="button"
+              disabled={lock}
+              onClick={() => void applyYoutubeBalancedPreset()}
+              className="rounded-xl border border-cyan-500/20 bg-cyan-500/10 px-4 py-2 text-sm text-cyan-100"
+            >
+              {t("平衡预设", "Balanced preset")}
+            </button>
             <button
               type="button"
               disabled={lock}

web.py

@@ -35,6 +35,14 @@ from src.control_plane import (
     system_snapshot,
     write_root_file,
 )
+from src.config_governance import (
+    delete_managed_file as governed_delete_managed_file,
+    inspect_file,
+    list_backups,
+    optimize_content,
+    restore_backup,
+    save_managed_file as governed_save_managed_file,
+)
 from src.hub_routes import configure_hub, router as hub_router
 from src.web_process_backend import WebProcessBackend, force_kill_ffmpeg, strip_ansi_codes
 
@@ -186,6 +194,35 @@ app.add_middleware(
 )
 
 
+class ProcessNameBody(BaseModel):
+    process: str = Field(..., min_length=1, max_length=160)
+
+
+class ServiceActionPostBody(BaseModel):
+    service: str = Field(..., min_length=1, max_length=64)
+    action: str = Field(..., pattern="^(start|stop|restart)$")
+
+
+class ManagedFileBody(BaseModel):
+    root: str = Field(..., min_length=1, max_length=64)
+    path: str = Field(..., min_length=1, max_length=240)
+    content: str = ""
+
+
+class ConfigAnalyzeBody(ManagedFileBody):
+    pass
+
+
+class ConfigOptimizeBody(ManagedFileBody):
+    action: str = Field(..., min_length=1, max_length=64)
+
+
+class ConfigRestoreBody(BaseModel):
+    root: str = Field(..., min_length=1, max_length=64)
+    path: str = Field(..., min_length=1, max_length=240)
+    backup_id: str = Field(..., min_length=1, max_length=320)
+
+
 async def read_log_path(path: Optional[str], lines: int) -> str:
     if not path:
         return "无日志路径\n"
@@ -370,8 +407,8 @@ async def get_managed_file(
 @app.post("/managed_file")
 async def save_managed_file(request: Request):
     try:
-        data = await request.json()
-        return write_root_file(data["root"], data["path"], data.get("content", ""))
+        data = ManagedFileBody(**(await request.json()))
+        return governed_save_managed_file(data.root, data.path, data.content)
     except KeyError as exc:
         return JSONResponse({"error": f"Missing field: {exc}"}, status_code=400)
     except (PermissionError, ValueError) as exc:
@@ -384,7 +421,7 @@ async def delete_managed_file(
     path: str = Query(..., description="相对路径"),
 ):
     try:
-        return delete_root_file(root, path)
+        return governed_delete_managed_file(root, path)
     except KeyError as exc:
         return JSONResponse({"error": str(exc)}, status_code=404)
     except FileNotFoundError:
@@ -393,6 +430,65 @@ async def delete_managed_file(
         return JSONResponse({"error": str(exc)}, status_code=400)
 
 
+@app.get("/config_insights")
+async def get_config_insights(
+    root: str = Query(..., description="管理目录 id"),
+    path: str = Query(..., description="相对路径"),
+):
+    try:
+        return inspect_file(root, path)
+    except KeyError as exc:
+        return JSONResponse({"error": str(exc)}, status_code=404)
+    except (PermissionError, ValueError) as exc:
+        return JSONResponse({"error": str(exc)}, status_code=400)
+
+
+@app.post("/config_insights")
+async def post_config_insights(body: ConfigAnalyzeBody):
+    try:
+        return inspect_file(body.root, body.path, draft_content=body.content)
+    except KeyError as exc:
+        return JSONResponse({"error": str(exc)}, status_code=404)
+    except (PermissionError, ValueError) as exc:
+        return JSONResponse({"error": str(exc)}, status_code=400)
+
+
+@app.get("/config_backups")
+async def get_config_backups(
+    root: str = Query(..., description="管理目录 id"),
+    path: str = Query(..., description="相对路径"),
+    limit: int = Query(12, ge=1, le=50),
+):
+    try:
+        return {"backups": list_backups(root, path, limit=limit)}
+    except KeyError as exc:
+        return JSONResponse({"error": str(exc)}, status_code=404)
+    except (PermissionError, ValueError) as exc:
+        return JSONResponse({"error": str(exc)}, status_code=400)
+
+
+@app.post("/config_optimize")
+async def post_config_optimize(body: ConfigOptimizeBody):
+    try:
+        return optimize_content(body.root, body.path, body.action, body.content)
+    except KeyError as exc:
+        return JSONResponse({"error": str(exc)}, status_code=404)
+    except (PermissionError, ValueError) as exc:
+        return JSONResponse({"error": str(exc)}, status_code=400)
+
+
+@app.post("/config_restore")
+async def post_config_restore(body: ConfigRestoreBody):
+    try:
+        return restore_backup(body.root, body.path, body.backup_id)
+    except KeyError as exc:
+        return JSONResponse({"error": str(exc)}, status_code=404)
+    except FileNotFoundError:
+        return JSONResponse({"error": f"Backup not found: {body.backup_id}"}, status_code=404)
+    except (PermissionError, ValueError) as exc:
+        return JSONResponse({"error": str(exc)}, status_code=400)
+
+
 @app.get("/hardware_profile")
 async def hardware_profile(refresh: bool = Query(False, description="重新探测硬件")):
     return await load_hardware_profile(force_refresh=refresh)
@@ -436,10 +532,8 @@ async def save_config(request: Request, process: str = Query(..., description="
     except Exception:
         return JSONResponse({"error": "无效的 JSON 数据"}, status_code=400)
 
-    config_file = CONFIG_DIR / "youtube.ini"
     try:
-        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
-        config_file.write_text(content, encoding="utf-8")
+        governed_save_managed_file("app-config", "youtube.ini", content)
         return {"message": "配置保存成功"}
     except OSError as e:
         return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
@@ -480,24 +574,13 @@ async def save_url_config(request: Request, process: str = Query(..., descriptio
     except Exception:
         return JSONResponse({"error": "无效的 JSON 数据"}, status_code=400)
 
-    config_file = CONFIG_DIR / "URL_config.ini"
     try:
-        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
-        config_file.write_text(content, encoding="utf-8")
+        governed_save_managed_file("app-config", "URL_config.ini", content)
         return {"message": "配置保存成功"}
     except OSError as e:
         return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
 
 
-class ProcessNameBody(BaseModel):
-    process: str = Field(..., min_length=1, max_length=160)
-
-
-class ServiceActionPostBody(BaseModel):
-    service: str = Field(..., min_length=1, max_length=64)
-    action: str = Field(..., pattern="^(start|stop|restart)$")
-
-
 def _not_found_hint(process: str) -> str:
     if process_backend.mode == "local":
         return (