"""Config insights and optimize presets (from d2ypp2 config_governance, Windows/Electron adapted).""" from __future__ import annotations import configparser import json import re from collections import Counter from pathlib import Path from typing import Any from web2_config_governance import dedupe_url_config_content _URL_RE = re.compile(r"https?://[^\s\"'<>]+", re.IGNORECASE) _RECORDER_SECTION = "录制设置" def _issue(level: str, code: str, text: str, **extra: Any) -> dict[str, Any]: row = {"level": level, "code": code, "text": text} row.update(extra) return row def _action(action: str, label: str, description: str) -> dict[str, str]: return {"action": action, "label": label, "description": description} def _stat(key: str, label: str, value: Any) -> dict[str, Any]: return {"key": key, "label": label, "value": value} def _file_kind(rel_path: str) -> str: name = rel_path.replace("\\", "/").lower() if "url_config" in name or name.endswith("url_config.ini"): return "url_config" if name.endswith("youtube.ini") or "/youtube." in name: return "youtube_ini" if name.endswith("config.ini"): return "recorder_ini" if name.endswith(".json"): return "json" return "generic" def _extract_url_from_line(line: str) -> str | None: match = _URL_RE.search(line) return match.group(0).rstrip(")]") if match else None 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 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", "没有有效的直播源 URL")) if duplicates: issues.append(_issue("warning", "duplicate_urls", f"检测到 {len(duplicates)} 个重复 URL")) actions.append(_action("dedupe_urls", "智能去重", "去除重复源站地址，保留注释与顺序")) if invalid_lines: issues.append( _issue("warning", "invalid_lines", f"{len(invalid_lines)} 行非空内容不含有效 http(s) URL", line=invalid_lines[0]) ) stats.extend( [ _stat("active_urls", "有效 URL", len(urls)), _stat("disabled_urls", "注释行", disabled), _stat("duplicate_urls", "重复 URL", len(duplicates)), ] ) for provider, count in provider_counter.most_common(): stats.append(_stat(f"provider_{provider.lower()}", provider, count)) return { "summary": f"{len(urls)} 条有效 URL，{len(duplicates)} 个重复，{disabled} 行注释", "issues": issues, "stats": stats, "actions": actions, } def _parse_ini(content: str) -> tuple[configparser.ConfigParser | None, str | None]: 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", "平衡推流预设", "启用 RTMPS、4500k 码率、fast_audio，保留当前 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"无法解析 youtube.ini: {parse_error}")) return {"summary": "youtube.ini 解析失败", "issues": issues, "stats": stats, "actions": actions} key = parser.get("youtube", "key", fallback="").strip() if parser.has_section("youtube") else "" rtmps = parser.get("youtube", "rtmps", fallback="是").strip() bitrate = parser.get("youtube", "bitrate", fallback="").strip() fast_audio = parser.get("youtube", "fast_audio", fallback="1").strip() if not key: issues.append(_issue("error", "missing_stream_key", "YouTube stream key 为空")) if rtmps and rtmps.lower() in {"否", "0", "false", "no"}: issues.append(_issue("warning", "rtmps_disabled", "未启用 RTMPS，建议加密推流")) bitrate_num = int(bitrate) if bitrate.isdigit() else None if bitrate and bitrate_num is None: issues.append(_issue("warning", "bitrate_invalid", "bitrate 不是有效整数")) elif bitrate_num is not None and not (1500 <= bitrate_num <= 9000): issues.append(_issue("warning", "bitrate_out_of_range", "bitrate 建议在 1500–9000 kbps")) stats.extend( [ _stat("stream_key", "Stream key", "已配置" if key else "缺失"), _stat("rtmps", "RTMPS", rtmps or "是"), _stat("bitrate", "Bitrate", bitrate or "auto"), _stat("fast_audio", "fast_audio", fast_audio or "1"), ] ) return {"summary": "YouTube 推流配置已分析", "issues": issues, "stats": stats, "actions": actions} def _find_kv(content: str, section: str, key: str) -> tuple[str | None, int | None]: 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", "稳定录制预设", "TS 分段录制、30 分钟切片、磁盘阈值 2GB")] if f"[{_RECORDER_SECTION}]" not in content: issues.append(_issue("error", "missing_record_section", f"缺少 [{_RECORDER_SECTION}] 段")) return {"summary": "录制配置解析失败", "issues": issues, "stats": stats, "actions": actions} save_format, save_format_line = _find_kv(content, _RECORDER_SECTION, "视频保存格式ts|mkv|flv|mp4|mp3音频|m4a音频") threads, threads_line = _find_kv(content, _RECORDER_SECTION, "同一时间访问网络的线程数") split_enabled, split_enabled_line = _find_kv(content, _RECORDER_SECTION, "分段录制是否开启") threads_num = int(threads) if threads and threads.isdigit() else None if save_format and save_format.lower() == "mp4": issues.append(_issue("warning", "unsafe_container", "直接录 MP4 中断恢复较差，建议 ts/mkv", line=save_format_line)) if threads_num is not None and threads_num > 8: issues.append(_issue("warning", "threads_high", "并发线程数偏高", line=threads_line)) if split_enabled and split_enabled.strip() in {"否", "0", "false"}: issues.append(_issue("warning", "split_disabled", "未开启分段录制", line=split_enabled_line)) stats.extend( [ _stat("format", "保存格式", save_format or "unknown"), _stat("threads", "线程数", threads or "unknown"), _stat("split_recording", "分段录制", split_enabled or "unknown"), ] ) return {"summary": "录制轮询与容器安全性已分析", "issues": issues, "stats": stats, "actions": actions} def analyze_config_content(rel_path: str, content: str) -> dict[str, Any]: kind = _file_kind(rel_path) if kind == "url_config": report = analyze_url_config(content) elif kind == "youtube_ini": report = analyze_youtube_ini(content) elif kind == "recorder_ini": report = analyze_recorder_ini(content) elif kind == "json": try: json.loads(content or "{}") report = {"summary": "JSON 语法有效", "issues": [], "stats": [], "actions": []} except json.JSONDecodeError as exc: report = { "summary": "JSON 解析失败", "issues": [_issue("error", "json_parse_error", str(exc.msg), line=exc.lineno)], "stats": [], "actions": [], } else: report = { "summary": "文本文件", "issues": [], "stats": [_stat("lines", "行数", len(content.splitlines()))], "actions": [], } report["file"] = rel_path.replace("\\", "/") report["kind"] = kind return report def _set_value_in_section(content: str, section: str, key: str, value: str) -> str: header = f"[{section}]" lines = (content or "").splitlines() out: list[str] = [] in_section = False inserted = False section_found = False pattern = re.compile(rf"^\s*{re.escape(key)}\s*=", re.IGNORECASE) for raw in lines: stripped = raw.strip() if stripped == header: section_found = True in_section = True out.append(raw) continue if in_section and stripped.startswith("[") and stripped.endswith("]"): if not inserted: out.append(f"{key} = {value}") inserted = True in_section = False if in_section and pattern.match(raw): out.append(f"{key} = {value}") inserted = True continue out.append(raw) if section_found and in_section and not inserted: out.append(f"{key} = {value}") elif not section_found: if out and out[-1].strip(): out.append("") out.extend([header, f"{key} = {value}"]) return "\n".join(out).rstrip() + "\n" def optimize_config_content(action: str, rel_path: str, content: str) -> dict[str, Any]: kind = _file_kind(rel_path) if action == "dedupe_urls" and kind == "url_config": next_content, removed = dedupe_url_config_content(content) message = f"已去除 {removed} 条重复 URL" if removed else "未发现重复 URL" elif action == "youtube_balanced_preset" and kind == "youtube_ini": parser, _ = _parse_ini(content or "[youtube]\n") key = "" if parser and parser.has_section("youtube"): key = parser.get("youtube", "key", fallback="").strip() 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 = "已应用 YouTube 平衡推流预设" 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 = "已应用稳定录制预设" else: raise ValueError(f"Unsupported optimization action: {action}") return { "message": message, "content": next_content, "insights": analyze_config_content(rel_path, next_content), } def read_config_file(config_dir: Path, rel_path: str) -> str: path = config_dir / rel_path.replace("\\", "/").lstrip("/") if not path.is_file(): return "" return path.read_text(encoding="utf-8", errors="replace")