314 lines
13 KiB
Python
314 lines
13 KiB
Python
"""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")
|