From af927c6dca995aeda7553433f52119cceef6c2d8 Mon Sep 17 00:00:00 2001 From: eric Date: Sun, 29 Mar 2026 00:34:53 -0500 Subject: [PATCH] 's' --- src/config_governance.py | 695 ++++++++++++++++++ src/control_plane.py | 4 + src/youtube_ini_sync.py | 8 +- tests/test_config_governance.py | 49 ++ .../components/console/ConfigAdvisorPanel.tsx | 332 +++++++++ .../console/LiveConsoleWorkspace.tsx | 20 + .../live-product/LiveYoutubeUnmannedView.tsx | 75 +- web.py | 119 ++- 8 files changed, 1281 insertions(+), 21 deletions(-) create mode 100644 src/config_governance.py create mode 100644 tests/test_config_governance.py create mode 100644 web-console/components/console/ConfigAdvisorPanel.tsx diff --git a/src/config_governance.py b/src/config_governance.py new file mode 100644 index 0000000..1767461 --- /dev/null +++ b/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, + } diff --git a/src/control_plane.py b/src/control_plane.py index e21c771..31c8ae1 100644 --- a/src/control_plane.py +++ b/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(): diff --git a/src/youtube_ini_sync.py b/src/youtube_ini_sync.py index 8bb0bd9..90f3bf8 100644 --- a/src/youtube_ini_sync.py +++ b/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}" diff --git a/tests/test_config_governance.py b/tests/test_config_governance.py new file mode 100644 index 0000000..59c74a2 --- /dev/null +++ b/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() diff --git a/web-console/components/console/ConfigAdvisorPanel.tsx b/web-console/components/console/ConfigAdvisorPanel.tsx new file mode 100644 index 0000000..c6ca3c9 --- /dev/null +++ b/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: (path: string, init?: RequestInit) => Promise; + notify: (message: string) => void; + onChangeContent: (content: string) => void; + onRefreshListing?: () => Promise | 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(null); + const [loading, setLoading] = useState(false); + const [draftLoading, setDraftLoading] = useState(false); + const [actionBusy, setActionBusy] = useState(null); + const [restoreBusy, setRestoreBusy] = useState(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(`/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("/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 ( +
+ {t("选择文件后显示诊断、回滚和一键优化。", "Pick a file to see diagnostics, rollback, and one-click optimizations.")} +
+ ); + } + + const issues = insights?.issues || []; + const actions = insights?.actions || []; + const backups = insights?.backups || []; + const stats = insights?.stats || []; + + return ( +
+
+
+
+
+ +

{t("配置治理", "Config governance")}

+ {insights?.source ? ( + + {insights.source} + + ) : null} +
+

+ {insights?.summary || + t("校验配置风险、查看历史快照，并将常见优化应用到编辑器。", "Validate risks, inspect snapshots, and apply optimizations into the editor.")} +

+
+
+ + +
+
+ +
+ {stats.slice(0, 8).map((item) => ( +
+
{item.label}
+
{item.value}
+
+ ))} +
+
+ +
+
+
+ +

{t("风险与建议", "Risks & guidance")}

+
+ {issues.length ? ( +
+ {issues.map((item, index) => ( +
+
+ {item.message} + {item.line ? ` (L${item.line})` : ""} +
+
{item.code}
+
+ ))} +
+ ) : ( +

{t("当前没有发现明显配置风险。", "No obvious config risks were detected.")}

+ )} + +
+ {actions.length ? ( + actions.map((item) => ( + + )) + ) : ( + {t("当前文件没有可自动应用的预设。", "No auto-applicable presets for this file.")} + )} +
+
+ +
+
+ +

{t("历史快照", "Backup snapshots")}

+
+ {backups.length ? ( +
+ {backups.slice(0, 8).map((item) => ( +
+
+
+
{new Date(item.created_at).toLocaleString(lang === "zh" ? "zh-CN" : "en-US", { hour12: false })}
+
+ {item.reason} · {formatBytes(item.size)} + {item.source === "legacy" ? ` · ${t("旧备份", "legacy")}` : ""} +
+
+ +
+
+ ))} +
+ ) : ( +

{t("当前文件还没有 Web 侧快照。保存、删除或恢复后会自动累积历史。", "No Web snapshots yet. Saving, deleting, or restoring will accumulate history automatically.")}

+ )} +
+
+
+ ); +} diff --git a/web-console/components/console/LiveConsoleWorkspace.tsx b/web-console/components/console/LiveConsoleWorkspace.tsx index 24180af..85061d9 100644 --- a/web-console/components/console/LiveConsoleWorkspace.tsx +++ b/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("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(null); const [shotFetchErr, setShotFetchErr] = useState(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() {
{filePath ? `${rootId} / ${filePath}` : tr("请选择文件", "Pick a file")}
+
+ refreshFiles(rootId, filePath)} + /> +