Coverage for /opt/hostedtoolcache/Python/3.10.17/x64/lib/python3.10/site-packages/vfx_seqtools/actions/seqrm.py: 87%

1import logging 

2import pathlib 

3from typing import Annotated, Optional 

4 

5import fileseq 

6import typer 

7from rich.progress import ( 

8 BarColumn, 

9 MofNCompleteColumn, 

10 Progress, 

11 TaskProgressColumn, 

12 TextColumn, 

13 TimeRemainingColumn, 

14) 

15 

16from vfx_seqtools import common_options 

17from vfx_seqtools.decorators import attach_hook 

18from vfx_seqtools.parser import replace_hash_and_at_with_framenumber 

19 

20 

21def do_action(tupl: tuple) -> None: 

22 src, frame, is_dryrun, be_verbose, be_interactive, be_strict, logger = tupl 

23 substituted_src = replace_hash_and_at_with_framenumber(src, frame) 

24 if is_dryrun: 

25 logger.info(f"dry-run: DELETE {substituted_src}") 

26 else: 

27 if be_verbose: 

28 logger.info(f"DELETE: {substituted_src}") 

29 try: 

30 if be_interactive: 

31 confirm = input(f"delete {substituted_src}? (y/n): ") 

32 if confirm.lower() != "y": 

33 return 

34 pathlib.Path.unlink(pathlib.Path(substituted_src)) 

35 except Exception as e: 

36 if be_strict: 

37 raise 

38 logger.warning( 

39 f"Error deleting file: {substituted_src}, skipping it. Use --strict to stop on errors. {e}" 

40 ) 

41 

42 

43@attach_hook(common_options.dry_run_option, hook_output_kwarg="is_dryrun") 

44@attach_hook(common_options.frame_range_options, hook_output_kwarg="frame_range") 

45@attach_hook(common_options.frame_seq_options, hook_output_kwarg="frame_seq") 

46@attach_hook(common_options.logging_options, hook_output_kwarg="logger") 

47@attach_hook(common_options.threading_option, hook_output_kwarg="thread_count") 

48@attach_hook(common_options.interactive_option, hook_output_kwarg="be_interactive") 

49@attach_hook(common_options.verbose_option, hook_output_kwarg="be_verbose") 

50@attach_hook(common_options.strict_option, hook_output_kwarg="be_strict") 

51@attach_hook(common_options.version_option, hook_output_kwarg="show_version") 

52def seqrm( 

53 src: Annotated[str, typer.Argument(help="The files to remove.")], 

54 logger: logging.Logger, 

55 be_verbose: Optional[bool] = False, 

56 be_strict: Optional[bool] = False, 

57 be_interactive: Optional[bool] = False, 

58 frame_range: tuple[int, int, int] = (0, 0, 0), 

59 frame_seq: str = "", 

60 is_dryrun: Optional[bool] = False, 

61 show_version: Optional[bool] = False, 

62 thread_count: Optional[int] = 0, 

63) -> None: 

64 """ 

65 Remove files with pattern <SRC>, using the provided framerange. 

66 

67 Use 'seqrm file.####.exr -f 1-5' to remove file.0001.exr through file.0005.exr. 

68 

69 """ 

70 if frame_seq: 

71 frames = fileseq.FrameSet(frame_seq) 

72 elif frame_range[0] != 0 or frame_range[1] != 0: 

73 frames = fileseq.FrameSet(f"{frame_range[0]}-{frame_range[1]}x{frame_range[2]}") 

74 else: 

75 frames = [] 

76 

77 if is_dryrun: 

78 for frame in frames: 

79 do_action( 

80 (src, frame, is_dryrun, be_verbose, be_interactive, be_strict, logger) 

81 ) 

82 else: 

83 # skip progress bar if interactive 

84 if be_interactive: 

85 for frame in frames: 

86 do_action( 

87 ( 

88 src, 

89 frame, 

90 is_dryrun, 

91 be_verbose, 

92 be_interactive, 

93 be_strict, 

94 logger, 

95 ) 

96 ) 

97 else: 

98 from rich.logging import RichHandler 

99 

100 logging.basicConfig( 

101 level=logging.INFO, 

102 format="%(message)s", 

103 datefmt="[%X]", 

104 handlers=[RichHandler(rich_tracebacks=True)], 

105 ) 

106 logger = logging.getLogger("rich") 

107 progress = Progress( 

108 TextColumn("[progress.description]{task.description}"), 

109 MofNCompleteColumn(), 

110 BarColumn(), 

111 TaskProgressColumn(), 

112 TimeRemainingColumn(), 

113 ) 

114 with progress: 

115 for frame in progress.track(frames, description="Working..."): 

116 do_action( 

117 ( 

118 src, 

119 frame, 

120 is_dryrun, 

121 be_verbose, 

122 be_interactive, 

123 be_strict, 

124 logger, 

125 ) 

126 )