151 lines
4.3 KiB
Python
151 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
|
|
DEFAULT_TEST_TXT = """# 频道,URL
|
|
# 示例:
|
|
# CCTV-1,http://example.com/live/index.m3u8
|
|
"""
|
|
|
|
DEFAULT_PASS_TEST_TXT = "# channel,url\n"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RuntimePaths:
|
|
app_dir: Path
|
|
settings_file: Path
|
|
test_file: Path
|
|
pass_test_file: Path
|
|
input_dir: Path
|
|
output_dir: Path
|
|
logs_dir: Path
|
|
reports_dir: Path
|
|
del_line_dir: Path
|
|
download_dir: Path
|
|
|
|
|
|
def get_runtime_dir() -> Path:
|
|
if getattr(sys, "frozen", False):
|
|
return Path(sys.executable).resolve().parent
|
|
return Path(__file__).resolve().parent
|
|
|
|
|
|
def get_runtime_paths() -> RuntimePaths:
|
|
app_dir = get_runtime_dir()
|
|
return RuntimePaths(
|
|
app_dir=app_dir,
|
|
settings_file=app_dir / "videoconvert_settings.json",
|
|
test_file=app_dir / "test.txt",
|
|
pass_test_file=app_dir / "pass_test.txt",
|
|
input_dir=app_dir / "input",
|
|
output_dir=app_dir / "output",
|
|
logs_dir=app_dir / "logs",
|
|
reports_dir=app_dir / "reports",
|
|
del_line_dir=app_dir / "delLine",
|
|
download_dir=app_dir / "download",
|
|
)
|
|
|
|
|
|
def build_default_settings(paths: RuntimePaths) -> dict:
|
|
cpu_count = 8
|
|
try:
|
|
import os
|
|
|
|
cpu_count = os.cpu_count() or 8
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"probe_input_file": str(paths.test_file),
|
|
"probe_output_file": str(paths.pass_test_file),
|
|
"probe_workers": min(96, max(24, cpu_count * 2)),
|
|
"probe_host_limit": 3,
|
|
"probe_timeout": 8.0,
|
|
"probe_min_speed": 400.0,
|
|
"probe_sample_bytes": 384,
|
|
"probe_sample_segments": 3,
|
|
"probe_ffmpeg_fallback": True,
|
|
"convert_root_dir": str(paths.app_dir),
|
|
"convert_parallel_jobs": min(4, max(2, (cpu_count + 2) // 6)),
|
|
"convert_resolution": "1280:720",
|
|
"convert_video_bitrate": "0",
|
|
"convert_audio_bitrate": "96k",
|
|
"convert_hls_time": 10,
|
|
"convert_preset": "p3",
|
|
"convert_episode_mode": False,
|
|
"shuffle_file": str(paths.pass_test_file),
|
|
"shuffle_keep_header": True,
|
|
"trim_dir": str(paths.del_line_dir),
|
|
"trim_remove_segments": 40,
|
|
"trim_min_segments": 120,
|
|
"trim_recursive": True,
|
|
}
|
|
|
|
|
|
def _ensure_text_file(path: Path, content: str) -> None:
|
|
if not path.exists():
|
|
path.write_text(content, encoding="utf-8")
|
|
|
|
|
|
def ensure_runtime_workspace() -> RuntimePaths:
|
|
paths = get_runtime_paths()
|
|
for directory in (
|
|
paths.app_dir,
|
|
paths.input_dir,
|
|
paths.output_dir,
|
|
paths.logs_dir,
|
|
paths.reports_dir,
|
|
paths.del_line_dir,
|
|
paths.download_dir,
|
|
):
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
|
|
_ensure_text_file(paths.test_file, DEFAULT_TEST_TXT)
|
|
_ensure_text_file(paths.pass_test_file, DEFAULT_PASS_TEST_TXT)
|
|
|
|
if not paths.settings_file.exists():
|
|
defaults = build_default_settings(paths)
|
|
paths.settings_file.write_text(json.dumps(defaults, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
return paths
|
|
|
|
|
|
def initialize_runtime_workspace() -> dict[str, object]:
|
|
paths = ensure_runtime_workspace()
|
|
created = {
|
|
"app_dir": str(paths.app_dir),
|
|
"settings_file": str(paths.settings_file),
|
|
"test_file": str(paths.test_file),
|
|
"pass_test_file": str(paths.pass_test_file),
|
|
"input_dir": str(paths.input_dir),
|
|
"output_dir": str(paths.output_dir),
|
|
"logs_dir": str(paths.logs_dir),
|
|
"reports_dir": str(paths.reports_dir),
|
|
"del_line_dir": str(paths.del_line_dir),
|
|
"download_dir": str(paths.download_dir),
|
|
}
|
|
return created
|
|
|
|
|
|
def load_settings_with_defaults(paths: RuntimePaths | None = None) -> dict:
|
|
runtime_paths = paths or ensure_runtime_workspace()
|
|
defaults = build_default_settings(runtime_paths)
|
|
|
|
if runtime_paths.settings_file.exists():
|
|
try:
|
|
loaded = json.loads(runtime_paths.settings_file.read_text(encoding="utf-8"))
|
|
defaults.update(loaded)
|
|
except Exception:
|
|
pass
|
|
return defaults
|
|
|
|
|
|
def save_settings(paths: RuntimePaths, data: dict) -> None:
|
|
merged = build_default_settings(paths)
|
|
merged.update(data)
|
|
paths.settings_file.write_text(json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8")
|