Coverage for src / ptf_tools / tasks.py: 39%

1import logging 

2import traceback 

3 

4from celery import shared_task 

5from celery.utils.functional import chunks 

6from django.conf import settings 

7from ptf.cmds.ptf_cmds.archive_numdam_cmds import ( 

8 archiveNumdamResourcePtfCmd, 

9 get_numdam_collection_list, 

10 get_numdam_issues_list, 

11) 

12from ptf.models.classes.collection import Collection 

13from task.custom_task import TaskWithProgress 

14from task.tasks.archiving_tasks import ( 

15 _archive_collection_common, 

16 _archive_collections_common, 

17 _archive_resources_in_chunks, 

18 check_nfs_directories, 

19) 

20 

21from history.model_data import HistoryEventDict, HistoryEventStatus 

22from history.views import insert_history_event 

23 

24logger = logging.getLogger(__name__) 

25 

26 

27@shared_task( 

28 name="ptf_tools.tasks.archiving_tasks.archive_numdam_collections", 

29 queue="coordinator", 

30 bind=True, 

31 base=TaskWithProgress, 

32) 

33def archive_numdam_collections(self: "TaskWithProgress"): 

34 colids = get_numdam_collection_list() 

35 colids = [c for c in colids if c not in settings.MERSENNE_COLLECTIONS] 

36 _archive_collections_common(self, colids, archive_col_task=archive_numdam_collection) 

37 

38 

39@shared_task( 

40 name="ptf_tools.tasks.archiving_tasks.archive_numdam_collection", 

41 queue="coordinator", 

42 bind=True, 

43 base=TaskWithProgress, 

44) 

45def archive_numdam_collection( 

46 self: "TaskWithProgress", 

47 colid: str, 

48 username: str | None, 

49 mathdoc_archive: str = settings.MATHDOC_ARCHIVE_FOLDER, 

50 binary_files_folder: str | None = None, 

51 xml_only=False, 

52 needs_publication_date=False, # Unused but needed because Celery apply_async checks args 

53): 

54 check_nfs_directories( 

55 [ 

56 settings.NUMDAM_ISSUE_SRC_FOLDER, 

57 settings.NUMDAM_ARTICLE_SRC_FOLDER, 

58 settings.CEDRAM_TEX_FOLDER, 

59 ] 

60 ) 

61 

62 pids = sorted(get_numdam_issues_list(colid)) 

63 

64 task_id = self.request.id 

65 if not task_id: 

66 raise ValueError("Couldn't find current task id") 

67 

68 self.set_progress(current=0, total=len(pids), col=colid) 

69 

70 archive_numdam_resource(colid) 

71 # for pid in pids: 

72 # archive_numdam_resource(colid, pid) 

73 # increment_progress(task_id=task_id) 

74 # check_archive(mathdoc_archive, colid, xml_only=xml_only) 

75 arg_chunks = chunks(((colid, pid) for pid in pids), 10) 

76 _archive_resources_in_chunks( 

77 self, mathdoc_archive, colid, xml_only, arg_chunks, archive_numdam_resource 

78 ) 

79 

80 

81@shared_task( 

82 name="ptf_tools.tasks.archiving_tasks.archive_numdam_resource", 

83 queue="executor", 

84) 

85def archive_numdam_resource(colid, pid=None): 

86 params = {"colid": colid} 

87 if pid: 

88 params["pid"] = pid 

89 cmd = archiveNumdamResourcePtfCmd(params) 

90 cmd.do() 

91 

92 

93@shared_task( 

94 name="ptf_tools.tasks.archiving_tasks.archive_trammel_collection", 

95 queue="coordinator", 

96 bind=True, 

97 base=TaskWithProgress, 

98) 

99def archive_trammel_collection( 

100 self: "TaskWithProgress", 

101 colid: str, 

102 username: str | None = None, 

103 mathdoc_archive: str = settings.MATHDOC_ARCHIVE_FOLDER, 

104 binary_files_folder: str | None = None, 

105 xml_only=False, 

106 needs_publication_date=False, # Unused but needed because Celery apply_async checks args 

107): 

108 collection = Collection.objects.get(pid=colid) 

109 title = collection.title_html if collection is not None else "" 

110 event_dict: "HistoryEventDict" = { 

111 "type": "archive", 

112 "pid": f"archive-trammel-{colid}", 

113 "col": collection, 

114 "title": title, 

115 "status": HistoryEventStatus.PENDING, 

116 } 

117 

118 try: 

119 check_nfs_directories( 

120 [ 

121 settings.NUMDAM_ISSUE_SRC_FOLDER, 

122 settings.NUMDAM_ARTICLE_SRC_FOLDER, 

123 settings.CEDRAM_TEX_FOLDER, 

124 ] 

125 ) 

126 

127 _archive_collection_common( 

128 self, 

129 collection, 

130 binary_files_folder, 

131 mathdoc_archive, 

132 xml_only, 

133 batch_size=1, 

134 needs_publication_date=True, 

135 ) 

136 event_dict["status"] = HistoryEventStatus.OK 

137 except Exception: 

138 event_dict["status"] = HistoryEventStatus.ERROR 

139 event_dict["message"] = traceback.format_exc() 

140 logger.error(event_dict["message"]) 

141 raise 

142 # if isinstance(e, ExceptionGroup): 

143 # messages: "list[HistoryChildDict]" = [] 

144 # for exception in e.exceptions: 

145 # messages.append( 

146 # { 

147 # "type": "archive_resource_error", 

148 # "status": HistoryEventStatus.ERROR, 

149 # "message": "".join( 

150 # traceback.format_exception( 

151 # type(exception), exception, exception.__traceback__ 

152 # ) 

153 # ), 

154 # } 

155 # ) 

156 # event_dict["children"] = messages 

157 finally: 

158 insert_history_event(event_dict) 

159 

160 

161@shared_task( 

162 name="ptf_tools.tasks.archiving_tasks.archive_trammel_collections", 

163 queue="coordinator", 

164 bind=True, 

165 base=TaskWithProgress, 

166) 

167def archive_trammel_collections( 

168 self: "TaskWithProgress", 

169 *args, 

170 **kwargs, 

171): 

172 _archive_collections_common( 

173 self, 

174 *args, 

175 archive_col_task=archive_trammel_collection, 

176 **kwargs, 

177 )