Coverage for src/backend/InvenTree/part/stocktake.py: 13%

1"""Stock history functionality.""" 

2 

3from typing import Optional 

4 

5from django.core.files.base import ContentFile 

6 

7import structlog 

8import tablib 

9from djmoney.contrib.exchange.models import convert_money 

10from djmoney.money import Money 

11 

12import common.models 

13from InvenTree.helpers import current_date 

14 

15logger = structlog.get_logger('inventree') 

16 

17 

18def perform_stocktake( 

19 part_id: Optional[int] = None, 

20 category_id: Optional[int] = None, 

21 location_id: Optional[int] = None, 

22 exclude_external: Optional[bool] = None, 

23 generate_entry: bool = True, 

24 report_output_id: Optional[int] = None, 

25) -> None: 

26 """Capture a snapshot of stock-on-hand and stock value. 

27 

28 Arguments: 

29 part_id: Optional ID of a part to perform stocktake on. If not provided, all active parts will be processed. 

30 category_id: Optional category ID to use to filter parts 

31 location_id: Optional location ID to use to filter stock items 

32 exclude_external: If True, exclude external stock items from the stocktake 

33 generate_entry: If True, create stocktake entries in the database 

34 report_output_id: Optional ID of a DataOutput object for the stocktake report (e.g. for download) 

35 

36 The default implementation creates stocktake entries for all active parts, 

37 and writes these stocktake entries to the database. 

38 

39 Alternatively, the scope of the stocktake can be limited by providing a queryset of parts, 

40 or by providing a category ID or location ID to filter the parts/stock items. 

41 """ 

42 import InvenTree.helpers 

43 import part.models as part_models 

44 import part.serializers as part_serializers 

45 import stock.models as stock_models 

46 from common.currency import currency_code_default 

47 from common.settings import get_global_setting 

48 

49 if not get_global_setting('STOCKTAKE_ENABLE', False, cache=False): 49 ↛ 54line 49 didn't jump to line 54 because the condition on line 49 was always true

50 logger.info('Stocktake functionality is disabled - skipping') 

51 return 

52 

53 # If exclude_external is not provided, use global setting 

54 if exclude_external is None: 

55 exclude_external = get_global_setting( 

56 'STOCKTAKE_EXCLUDE_EXTERNAL', False, cache=False 

57 ) 

58 

59 # If a single part is provided, limit to that part 

60 # Otherwise, start with all active parts 

61 if part_id is not None: 

62 try: 

63 part = part_models.Part.objects.get(id=part_id) 

64 parts = part.get_descendants(include_self=True) 

65 except (ValueError, part_models.Part.DoesNotExist): 

66 parts = part_models.Part.objects.all() 

67 else: 

68 parts = part_models.Part.objects.all() 

69 

70 # Only use active parts 

71 parts = parts.filter(active=True) 

72 

73 # Prefetch related pricing information 

74 parts = parts.prefetch_related('pricing_data', 'stock_items') 

75 

76 # Filter part queryset by category, if provided 

77 if category_id is not None: 

78 # Filter parts by category (including subcategories) 

79 try: 

80 category = part_models.PartCategory.objects.get(id=category_id) 

81 parts = parts.filter( 

82 category__in=category.get_descendants(include_self=True) 

83 ) 

84 except (ValueError, part_models.PartCategory.DoesNotExist): 

85 pass 

86 

87 # Fetch location if provided 

88 if location_id is not None: 

89 try: 

90 location = stock_models.StockLocation.objects.get(id=location_id) 

91 except (ValueError, stock_models.StockLocation.DoesNotExist): 

92 location = None 

93 else: 

94 location = None 

95 

96 if location is not None: 

97 # Location limited, so we will disable saving of stocktake entries 

98 generate_entry = False 

99 

100 # New history entries to be created 

101 history_entries = [] 

102 

103 base_currency = currency_code_default() 

104 today = InvenTree.helpers.current_date() 

105 

106 logger.info('Creating new stock history entries for %s parts', parts.count()) 

107 

108 # Fetch report output object if provided 

109 if report_output_id is not None: 

110 try: 

111 report_output = common.models.DataOutput.objects.get(id=report_output_id) 

112 except (ValueError, common.models.DataOutput.DoesNotExist): 

113 report_output = None 

114 else: 

115 report_output = None 

116 

117 if report_output: 

118 # Initialize progress on the report output 

119 report_output.total = parts.count() 

120 report_output.progress = 0 

121 report_output.complete = False 

122 report_output.save() 

123 

124 for part in parts: 

125 # Is there a recent stock history record for this part? 

126 if ( 

127 generate_entry 

128 and part_models.PartStocktake.objects.filter( 

129 part=part, date__gte=today 

130 ).exists() 

131 ): 

132 continue 

133 

134 try: 

135 pricing = part.pricing_data 

136 except Exception: 

137 pricing = None 

138 

139 # Fetch all 'in stock' items for this part 

140 stock_items = part.stock_entries( 

141 location=location, 

142 in_stock=True, 

143 include_external=not exclude_external, 

144 include_variants=True, 

145 ) 

146 

147 total_cost_min = Money(0, base_currency) 

148 total_cost_max = Money(0, base_currency) 

149 

150 total_quantity = 0 

151 items_count = 0 

152 

153 if stock_items.count() == 0: 

154 # No stock items - skip this part if location is specified 

155 if location: 

156 continue 

157 

158 for item in stock_items: 

159 # Extract cost information 

160 

161 entry_cost_min = ( 

162 (pricing.overall_min or pricing.overall_max) if pricing else None 

163 ) 

164 entry_cost_max = ( 

165 (pricing.overall_max or pricing.overall_min) if pricing else None 

166 ) 

167 

168 if item.purchase_price is not None: 

169 entry_cost_min = item.purchase_price 

170 entry_cost_max = item.purchase_price 

171 

172 try: 

173 entry_cost_min = entry_cost_min * item.quantity 

174 entry_cost_max = entry_cost_max * item.quantity 

175 

176 if entry_cost_min.currency != base_currency: 

177 entry_cost_min = convert_money(entry_cost_min, base_currency) 

178 

179 if entry_cost_max.currency != base_currency: 

180 entry_cost_max = convert_money(entry_cost_max, base_currency) 

181 except Exception: 

182 entry_cost_min = Money(0, base_currency) 

183 entry_cost_max = Money(0, base_currency) 

184 

185 # Update total quantities 

186 items_count += 1 

187 total_quantity += item.quantity 

188 total_cost_min += entry_cost_min 

189 total_cost_max += entry_cost_max 

190 

191 if report_output: 

192 report_output.progress += 1 

193 

194 # Update report progress every few items, to avoid excessive database writes 

195 if report_output.progress % 50 == 0: 

196 report_output.save() 

197 

198 # Add a new stocktake entry for this part 

199 history_entries.append( 

200 part_models.PartStocktake( 

201 part=part, 

202 item_count=items_count, 

203 quantity=total_quantity, 

204 cost_min=total_cost_min, 

205 cost_max=total_cost_max, 

206 ) 

207 ) 

208 

209 if generate_entry: 

210 # Bulk-create PartStocktake entries 

211 part_models.PartStocktake.objects.bulk_create(history_entries, batch_size=250) 

212 

213 if report_output: 

214 # Save report data, and mark as complete 

215 

216 today = current_date() 

217 

218 serializer = part_serializers.PartStocktakeSerializer(exclude_pk=True) 

219 

220 headers = serializer.generate_headers() 

221 

222 # Export the data to a file 

223 dataset = tablib.Dataset(headers=list(headers.values())) 

224 

225 header_keys = list(headers.keys()) 

226 

227 for entry in history_entries: 

228 entry.date = today 

229 row = serializer.to_representation(entry) 

230 dataset.append([row.get(header, '') for header in header_keys]) 

231 

232 datafile = dataset.export('csv') 

233 

234 report_output.mark_complete( 

235 output=ContentFile(datafile, 'stocktake_report.csv') 

236 )