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, }