Public/removeads
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add a dedicated class for temporary files management.

Browse Source
This commit is contained in:
Frédéric Tronel
2026-08-31 11:12:47 +02:00
parent d1df44c82c
commit 5b2afd29b3
7 changed files with 62 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tscut/cli.py
+6 -1
View File
@@ -15,6 +15,7 @@ import coloredlogs
from tscut.exceptions import TSCutError
from tscut.models import ProcessingOptions
from tscut.pipeline import process_recording
from tscut.temporaries import TemporaryFiles
from tscut.tools.discovery import check_required_tools
from tscut.tools.timeframe import compare_time_interval, parse_time_interval
@@ -110,9 +111,13 @@ def main() -> int:
logger.debug("Configuration: %s", config)
try:
process_recording(config)
temporaries = TemporaryFiles(config.keep_temporaries)
process_recording(config, temporaries)
except TSCutError as exc:
logger.error("%s", exc)
return 1
finally:
logger.info("Cleaning temporary files")
temporaries.cleanup()
return 0
src/tscut/coarse.py
+2 -2
View File
@@ -5,8 +5,8 @@
import logging
from typing import IO
from tscut.tools.mkvtoolnix import extract_mkv_part
from tscut.exceptions import TemporaryFileError
from tscut.tools.mkvtoolnix import extract_mkv_part
logger = logging.getLogger(__name__)
@@ -27,4 +27,4 @@ def do_coarse_processing(ffmpeg_path:str, ffprobe_path:str, mkvmerge_path:str,
extract_mkv_part(mkvmerge_path=mkvmerge_path, input_file=input_file, output_file=internal_mkv,
begin=begin, end=end)
temporaries.append(internal_mkv)
temporaries.add(internal_mkv)
src/tscut/cutting.py
+11 -10
View File
@@ -12,7 +12,7 @@ from os import (
write,
)
from shutil import copyfile
from typing import IO, Any, BinaryIO, Sequence, TextIO
from typing import BinaryIO, Sequence, TextIO
import hexdump
from tqdm import tqdm
@@ -22,6 +22,7 @@ from tscut.exceptions import InvalidMediaError, TemporaryFileError
from tscut.h264.avc import get_avc_config_from_h264
from tscut.matroska.codec import dump_codec_private_data
from tscut.models import CutResult, PreparedMedia, ProcessingOptions
from tscut.temporaries import TemporaryFiles
from tscut.tools.ffmpeg import extract_all_streams
from tscut.tools.ffprobe import (
get_nearest_iframe,
@@ -94,7 +95,7 @@ def concatenate_h264_ts_parts(h264_ts_parts: Sequence[TextIO], output: TextIO) -
first = False
def cut_recording(media: PreparedMedia, options: ProcessingOptions,
temporaries: list[IO[Any]]) -> CutResult:
temporaries: TemporaryFiles) -> CutResult:
# Pour chaque portion
partnum = 0
@@ -243,10 +244,10 @@ def cut_recording(media: PreparedMedia, options: ProcessingOptions,
remove_video_tracks_from_mkv(mkvmerge_path=options.tools_paths['mkvmerge'],
input_file=internal_mkv, output_file=internal_novideo_mkv)
temporaries.append(internal_mkv)
temporaries.append(internal_h264)
temporaries.append(internal_h264_ts)
temporaries.append(internal_novideo_mkv)
temporaries.add(internal_mkv)
temporaries.add(internal_h264)
temporaries.add(internal_h264_ts)
temporaries.add(internal_novideo_mkv)
h264parts.append(internal_h264)
h264_ts.append(internal_h264_ts)
@@ -280,7 +281,7 @@ def cut_recording(media: PreparedMedia, options: ProcessingOptions,
part = merge_mkvs(mkvmerge_path=options.tools_paths['mkvmerge'], inputs=subparts,
output_name=f'part-{partnum:d}.mkv', concatenate=True)
mkvparts.append(part)
temporaries.append(part)
temporaries.add(part)
pos = pos+tail_iframe_ts-ts1
@@ -301,7 +302,7 @@ def cut_recording(media: PreparedMedia, options: ProcessingOptions,
logger.info('Merging all H264 tracks')
concatenate_h264_parts(h264parts=h264parts, output=full_h264)
temporaries.append(full_h264)
temporaries.add(full_h264)
try:
full_h264_ts = open(f'{media.basename}-ts.txt', 'w+', encoding='utf8')
@@ -310,7 +311,7 @@ def cut_recording(media: PreparedMedia, options: ProcessingOptions,
logger.info('Merging H264 timestamps')
concatenate_h264_ts_parts(h264_ts_parts=h264_ts, output=full_h264_ts)
temporaries.append(full_h264_ts)
temporaries.add(full_h264_ts)
final_novideo_name = f'{media.basename}-novideo.mkv'
final_with_video_name = f'{media.basename}-video.mkv'
@@ -331,7 +332,7 @@ def cut_recording(media: PreparedMedia, options: ProcessingOptions,
except OSError:
raise TemporaryFileError(f"Impossible to open file: {final_novideo_name}")
temporaries.append(final_novideo)
temporaries.add(final_novideo)
full_h264_ts.seek(0)
src/tscut/models.py
+1 -1
View File
@@ -57,5 +57,5 @@ class PreparedMedia:
@dataclass
class CutResult:
filename: str
movie: BinaryIO
movie: IO[bytes]
check_positions: list[timedelta]
src/tscut/pipeline.py
+9 -19
View File
@@ -5,9 +5,8 @@
import logging
import os
from datetime import datetime, timedelta
from os import unlink
from shutil import copyfile, move
from typing import IO, Any, BinaryIO
from typing import BinaryIO
import hexdump
@@ -17,6 +16,7 @@ from tscut.h264.avc import parse_codec_private
from tscut.matroska.codec import dump_codec_private_data
from tscut.models import PreparedMedia, ProcessingOptions, SupportedFormat
from tscut.subtitles.ocr import do_ocr, extract_srt, get_tesseract_supported_lang
from tscut.temporaries import TemporaryFiles
from tscut.tools.ffmpeg import ffmpeg_convert
from tscut.tools.ffprobe import (
find_subtitles_tracks,
@@ -42,10 +42,8 @@ def detect_supported_format(format_name: str) -> SupportedFormat:
f"Unsupported media format: {format_name}"
)
def process_recording(options: ProcessingOptions) -> None:
def process_recording(options: ProcessingOptions, temporaries: TemporaryFiles) -> None:
nb_parts = len(options.parts)
temporaries : list[IO[Any]] = []
basename = os.path.splitext(os.path.basename(options.input_file))[0]
mp4filename = basename+'.mp4'
mkvfilename = basename+'.mkv'
@@ -88,7 +86,7 @@ def process_recording(options: ProcessingOptions) -> None:
with open(mp4filename, 'wb+') as mp4:
ffmpeg_convert(options.tools_paths['ffmpeg'], options.tools_paths['ffprobe'],
input_file, 'mpegts', mp4, 'mp4', duration)
temporaries.append(mp4)
temporaries.add(mp4)
logger.info("Converting MP4 to MKV.")
try:
mkv = open(mkvfilename, 'wb+')
@@ -98,7 +96,7 @@ def process_recording(options: ProcessingOptions) -> None:
ffmpeg_convert(options.tools_paths['ffmpeg'], options.tools_paths['ffprobe'],
mp4, 'mp4', mkv, 'matroska', duration)
if nb_parts > 0:
temporaries.append(mkv)
temporaries.add(mkv)
except OSError:
raise TemporaryFileError(f"Impossible to create: {mp4filename}")
@@ -112,7 +110,7 @@ def process_recording(options: ProcessingOptions) -> None:
ffmpeg_convert(options.tools_paths['ffmpeg'], options.tools_paths['ffprobe'],
input_file, 'mp4', mkv, 'matroska', duration)
if nb_parts > 0:
temporaries.append(mkv)
temporaries.add(mkv)
else:
logger.info("Already in MKV")
mkv = input_file
@@ -198,7 +196,7 @@ def process_recording(options: ProcessingOptions) -> None:
move(final_with_video_name, options.output_file)
else:
# Final cut is not any more the final step.
temporaries.append(final_with_video)
temporaries.add(final_with_video)
duration = get_movie_duration(options.tools_paths['ffprobe'], final_with_video)
supported_langs = get_tesseract_supported_lang(options.tools_paths['tesseract'])
logger.info('Supported lang: %s', supported_langs)
@@ -238,8 +236,8 @@ def process_recording(options: ProcessingOptions) -> None:
except OSError:
raise TemporaryFileError(f"Impossible to open {sub_name}")
temporaries.append(idx)
temporaries.append(sub)
temporaries.add(idx)
temporaries.add(sub)
ocr = do_ocr(options.tools_paths['vobsubocr'], list_of_subtitles, duration,
temporaries, options.dump_memory)
@@ -253,14 +251,6 @@ def process_recording(options: ProcessingOptions) -> None:
else:
move(final_with_video_name, options.output_file)
if not options.keep_temporaries:
logger.info("Cleaning temporary files")
for tmp in temporaries:
path = os.path.realpath(tmp.name)
logger.info("Removing: %s", path)
tmp.close()
unlink(path)
d = datetime(1,1,1)
for c in checks:
logger.info("Please check cut smoothness at %s", (c+d).strftime("%H:%M:%S"))
src/tscut/subtitles/ocr.py
+4 -3
View File
@@ -17,7 +17,7 @@ from os import (
write,
)
from subprocess import PIPE, Popen
from typing import IO, Any, Sequence
from typing import Sequence
from iso639 import Lang
from iso639.exceptions import InvalidLanguageValue
@@ -25,6 +25,7 @@ from tqdm import tqdm
from typeguard import typechecked
from tscut.exceptions import ExternalToolError
from tscut.temporaries import TemporaryFiles
logger = logging.getLogger(__name__)
@@ -130,7 +131,7 @@ def extract_srt(mkvextract:str, filename:str, subtitles:dict[str, list[int]],
@typechecked
def do_ocr(vobsubocr:str, idxs: Sequence[tuple[str,str,str,str]], duration:timedelta,
temporaries:list[IO[Any]], dump_mem_fd:bool=False):
temporaries:TemporaryFiles, dump_mem_fd:bool=False):
res = []
for idx_name, _, lang, iso in idxs:
@@ -177,7 +178,7 @@ def do_ocr(vobsubocr:str, idxs: Sequence[tuple[str,str,str,str]], duration:timed
pos = 0
while pos < srt_length:
pos+=write(outfd, buf[pos:])
temporaries.append(dump_srt)
temporaries.add(dump_srt)
except OSError:
logger.error('Impossible to create file: %s', srtname)
return None
src/tscut/temporaries.py
+29
View File
@@ -0,0 +1,29 @@
# SPDX-License-Identifier: GPL-2.0-or-later
#
# Copyright (C) 2026 Frédéric Tronel
"""A class to manage (and clean) all temporary files created during conversion"""
import logging
import os
from typing import IO, Any
logger = logging.getLogger(__name__)
class TemporaryFiles:
def __init__(self, keep: bool = False) -> None:
self._files: list[IO[Any]] = []
self._keep = keep
def add(self, file: IO[Any]) -> None:
self._files.append(file)
def cleanup(self) -> None:
if self._keep:
return
for f in self._files:
path = os.path.realpath(f.name)
logger.info("Removing: %s", path)
f.close()
os.unlink(path)