Coverage for documents/double_sided.py: 32%

1import datetime as dt 

2import logging 

3import os 

4import shutil 

5from pathlib import Path 

6from typing import Final 

7 

8from django.conf import settings 

9from pikepdf import Pdf 

10 

11from documents.consumer import ConsumerError 

12from documents.converters import convert_from_tiff_to_pdf 

13from documents.plugins.base import ConsumeTaskPlugin 

14from documents.plugins.base import NoCleanupPluginMixin 

15from documents.plugins.base import NoSetupPluginMixin 

16from documents.plugins.base import StopConsumeTaskError 

17 

18logger = logging.getLogger("paperless.double_sided") 

19 

20# Hardcoded for now, could be made a configurable setting if needed 

21TIMEOUT_MINUTES: Final[int] = 30 

22TIMEOUT_SECONDS: Final[int] = TIMEOUT_MINUTES * 60 

23 

24# Used by test cases 

25STAGING_FILE_NAME = "double-sided-staging.pdf" 

26 

27 

28class CollatePlugin(NoCleanupPluginMixin, NoSetupPluginMixin, ConsumeTaskPlugin): 

29 NAME: str = "CollatePlugin" 

30 

31 @property 

32 def able_to_run(self) -> bool: 

33 return ( 

34 settings.CONSUMER_ENABLE_COLLATE_DOUBLE_SIDED 

35 and settings.CONSUMER_COLLATE_DOUBLE_SIDED_SUBDIR_NAME 

36 in self.input_doc.original_file.parts 

37 ) 

38 

39 def run(self) -> str | None: 

40 """ 

41 Tries to collate pages from 2 single sided scans of a double sided 

42 document. 

43 

44 When called with a file, it checks whether or not a staging file 

45 exists, if not, the current file is turned into that staging file 

46 containing the odd numbered pages. 

47 

48 If a staging file exists, and it is not too old, the current file is 

49 considered to be the second part (the even numbered pages) and it will 

50 collate the pages of both, the pages of the second file will be added 

51 in reverse order, since the ADF will have scanned the pages from bottom 

52 to top. 

53 

54 Returns a status message on success, or raises a ConsumerError 

55 in case of failure. 

56 """ 

57 

58 if self.input_doc.mime_type == "application/pdf": 

59 pdf_file = self.input_doc.original_file 

60 elif ( 

61 self.input_doc.mime_type == "image/tiff" 

62 and settings.CONSUMER_COLLATE_DOUBLE_SIDED_TIFF_SUPPORT 

63 ): 

64 pdf_file = convert_from_tiff_to_pdf( 

65 self.input_doc.original_file, 

66 self.base_tmp_dir, 

67 ) 

68 self.input_doc.original_file.unlink() 

69 else: 

70 raise ConsumerError( 

71 "Unsupported file type for collation of double-sided scans", 

72 ) 

73 

74 staging: Path = settings.SCRATCH_DIR / STAGING_FILE_NAME 

75 

76 valid_staging_exists = False 

77 if staging.exists(): 

78 stats = staging.stat() 

79 # if the file is older than the timeout, we don't consider 

80 # it valid 

81 if (dt.datetime.now().timestamp() - stats.st_mtime) > TIMEOUT_SECONDS: 

82 logger.warning("Outdated double sided staging file exists, deleting it") 

83 staging.unlink() 

84 else: 

85 valid_staging_exists = True 

86 

87 if valid_staging_exists: 

88 try: 

89 # Collate pages from second PDF in reverse order 

90 with Pdf.open(staging) as pdf1, Pdf.open(pdf_file) as pdf2: 

91 pdf2.pages.reverse() 

92 try: 

93 for i, page in enumerate(pdf2.pages): 

94 pdf1.pages.insert(2 * i + 1, page) 

95 except IndexError: 

96 raise ConsumerError( 

97 "This second file (even numbered pages) contains more " 

98 "pages than the first/odd numbered one. This means the " 

99 "two uploaded files don't belong to the same double-" 

100 "sided scan. Please retry, starting with the odd " 

101 "numbered pages again.", 

102 ) 

103 # Merged file has the same path, but without the 

104 # double-sided subdir. Therefore, it is also in the 

105 # consumption dir and will be picked up for processing 

106 old_file = self.input_doc.original_file 

107 new_file = Path( 

108 *( 

109 part 

110 for part in old_file.with_name( 

111 f"{old_file.stem}-collated.pdf", 

112 ).parts 

113 if part 

114 != settings.CONSUMER_COLLATE_DOUBLE_SIDED_SUBDIR_NAME 

115 ), 

116 ) 

117 # If the user didn't create the subdirs yet, do it for them 

118 new_file.parent.mkdir(parents=True, exist_ok=True) 

119 pdf1.save(new_file) 

120 logger.info("Collated documents into new file %s", new_file) 

121 raise StopConsumeTaskError( 

122 "Success. Even numbered pages of double sided scan collated " 

123 "with odd pages", 

124 ) 

125 finally: 

126 # Delete staging and recently uploaded file no matter what. 

127 # If any error occurs, the user needs to be able to restart 

128 # the process from scratch; after all, the staging file 

129 # with the odd numbered pages might be the culprit 

130 pdf_file.unlink() 

131 staging.unlink() 

132 

133 else: 

134 shutil.move(pdf_file, staging) 

135 # update access to modification time so we know if the file 

136 # is outdated when another file gets uploaded 

137 timestamp = dt.datetime.now().timestamp() 

138 os.utime(staging, (timestamp, timestamp)) 

139 logger.info( 

140 "Got scan with odd numbered pages of double-sided scan, moved it to %s", 

141 staging, 

142 ) 

143 raise StopConsumeTaskError( 

144 "Received odd numbered pages of double sided scan, waiting up to " 

145 f"{TIMEOUT_MINUTES} minutes for even numbered pages", 

146 )