optimize CLI and config
This commit is contained in:
@ -8,5 +8,14 @@
|
|||||||
".mp4",
|
".mp4",
|
||||||
".mkv"
|
".mkv"
|
||||||
],
|
],
|
||||||
"resolution": null
|
"resolution": "1920x1080",
|
||||||
|
"extra": [],
|
||||||
|
"manual": null,
|
||||||
|
"compress_dir_name": "compress",
|
||||||
|
"fps": 30,
|
||||||
|
"test_video_resolution": "1920x1080",
|
||||||
|
"test_video_fps": 30,
|
||||||
|
"test_video_input": "compress_video_test.mp4",
|
||||||
|
"test_video_output": "compressed_video_test.mp4",
|
||||||
|
"disable_hwaccel_when_fail": true
|
||||||
}
|
}
|
||||||
@ -7,12 +7,14 @@ from datetime import datetime
|
|||||||
from time import time
|
from time import time
|
||||||
from rich.logging import RichHandler
|
from rich.logging import RichHandler
|
||||||
from rich.progress import Progress
|
from rich.progress import Progress
|
||||||
|
from rich.console import Console
|
||||||
from pickle import dumps, loads
|
from pickle import dumps, loads
|
||||||
from typing import Optional, Callable, Literal, List, Any, TYPE_CHECKING
|
from typing import Optional, Callable, Literal, List, Any, TYPE_CHECKING
|
||||||
import atexit
|
import atexit
|
||||||
import re
|
import re
|
||||||
import get_frame
|
import get_frame
|
||||||
import json
|
import json
|
||||||
|
import argparse
|
||||||
|
|
||||||
try:
|
try:
|
||||||
from pydantic import BaseModel, Field, field_validator, model_validator
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
||||||
@ -250,22 +252,27 @@ def get_cmd(video_path: str | Path, output_file: str | Path) -> list[str]:
|
|||||||
|
|
||||||
|
|
||||||
# 配置logging
|
# 配置logging
|
||||||
def setup_logging():
|
def setup_logging(verbose: bool = False):
|
||||||
log_dir = Path("logs")
|
log_dir = Path("logs")
|
||||||
log_dir.mkdir(exist_ok=True)
|
log_dir.mkdir(exist_ok=True)
|
||||||
log_file = log_dir / f"video_compress_{datetime.now().strftime('%Y%m%d')}.log"
|
log_file = log_dir / f"video_compress_{datetime.now().strftime('%Y%m%d')}.log"
|
||||||
stream = RichHandler(rich_tracebacks=True, tracebacks_show_locals=True)
|
stream = RichHandler(level=logging.DEBUG if verbose else logging.INFO,
|
||||||
stream.setLevel(logging.INFO)
|
rich_tracebacks=True, tracebacks_show_locals=True)
|
||||||
|
# stream.setLevel(logging.DEBUG if verbose else logging.INFO)
|
||||||
stream.setFormatter(logging.Formatter("%(message)s"))
|
stream.setFormatter(logging.Formatter("%(message)s"))
|
||||||
|
|
||||||
file = logging.FileHandler(log_file, encoding="utf-8")
|
file = logging.FileHandler(log_file, encoding="utf-8")
|
||||||
file.setLevel(logging.DEBUG)
|
file.setLevel(logging.DEBUG)
|
||||||
|
|
||||||
|
# 清除现有的handlers,避免多次调用basicConfig无效
|
||||||
|
logging.getLogger().handlers.clear()
|
||||||
|
|
||||||
logging.basicConfig(
|
logging.basicConfig(
|
||||||
level=logging.DEBUG,
|
level=logging.DEBUG,
|
||||||
format="%(asctime)s - %(levelname) 7s - %(message)s",
|
format="%(asctime)s - %(levelname) 7s - %(message)s",
|
||||||
handlers=[file, stream],
|
handlers=[stream, file],
|
||||||
)
|
)
|
||||||
|
logging.debug("Logging is set up.")
|
||||||
|
|
||||||
|
|
||||||
def fmt_time(t: float | int) -> str:
|
def fmt_time(t: float | int) -> str:
|
||||||
@ -577,21 +584,25 @@ def main(_root=None):
|
|||||||
atexit.register(exit_pause)
|
atexit.register(exit_pause)
|
||||||
|
|
||||||
global root
|
global root
|
||||||
|
|
||||||
|
if _root is not None:
|
||||||
setup_logging()
|
setup_logging()
|
||||||
|
root = Path(_root)
|
||||||
|
else:
|
||||||
|
parser = argparse.ArgumentParser()
|
||||||
|
parser.add_argument("directory", nargs="?", help="目标目录路径")
|
||||||
|
parser.add_argument("--verbose", "-v", action="store_true", help="启用详细日志记录")
|
||||||
|
args = parser.parse_args()
|
||||||
|
if not args.directory:
|
||||||
|
print("Error termination via invalid input.")
|
||||||
|
sys.exit(1)
|
||||||
|
root = Path(args.directory)
|
||||||
|
setup_logging(args.verbose)
|
||||||
|
|
||||||
tot_bgn = time()
|
tot_bgn = time()
|
||||||
logging.info("-------------------------------")
|
logging.info("-------------------------------")
|
||||||
logging.info(datetime.now().strftime("Video Compress started at %Y/%m/%d %H:%M"))
|
logging.info(datetime.now().strftime("Video Compress started at %Y/%m/%d %H:%M"))
|
||||||
|
|
||||||
if _root is not None:
|
|
||||||
root = Path(_root)
|
|
||||||
else:
|
|
||||||
# 通过命令行参数传入需要遍历的目录
|
|
||||||
if len(sys.argv) < 2:
|
|
||||||
print(f"用法:python {__file__} <目标目录>")
|
|
||||||
logging.warning("Error termination via invalid input.")
|
|
||||||
sys.exit(1)
|
|
||||||
root = Path(sys.argv[1])
|
|
||||||
|
|
||||||
if root.name.lower() == CFG.compress_dir_name.lower():
|
if root.name.lower() == CFG.compress_dir_name.lower():
|
||||||
logging.critical("请修改目标目录名为非compress。")
|
logging.critical("请修改目标目录名为非compress。")
|
||||||
logging.error("Error termination via invalid input.")
|
logging.error("Error termination via invalid input.")
|
||||||
@ -601,7 +612,7 @@ def main(_root=None):
|
|||||||
test()
|
test()
|
||||||
|
|
||||||
if not root.is_dir():
|
if not root.is_dir():
|
||||||
print("提供的路径不是一个有效目录。")
|
logging.critical("提供的路径不是一个有效目录。")
|
||||||
logging.warning("Error termination via invalid input.")
|
logging.warning("Error termination via invalid input.")
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user