Coverage for src/backend/InvenTree/plugin/builtin/exporter/stocktake_exporter.py: 28%

1"""Custom data exporter for part stocktake data.""" 

2 

3from decimal import Decimal 

4 

5from django.utils.translation import gettext_lazy as _ 

6 

7from rest_framework import serializers 

8 

9from InvenTree.helpers import normalize 

10from part.models import Part 

11from part.serializers import PartSerializer 

12from plugin import InvenTreePlugin 

13from plugin.mixins import DataExportMixin 

14 

15 

16class PartStocktakeExportOptionsSerializer(serializers.Serializer): 

17 """Custom export options for the PartStocktakeExporter plugin.""" 

18 

19 export_pricing_data = serializers.BooleanField( 

20 default=True, label=_('Pricing Data'), help_text=_('Include part pricing data') 

21 ) 

22 

23 export_include_external_items = serializers.BooleanField( 

24 default=False, 

25 label=_('Include External Stock'), 

26 help_text=_('Include external stock in the stocktake data'), 

27 ) 

28 

29 export_include_variant_items = serializers.BooleanField( 

30 default=False, 

31 label=_('Include Variant Items'), 

32 help_text=_('Include part variant stock in stocktake data'), 

33 ) 

34 

35 export_exclude_zero_stock_entries = serializers.BooleanField( 

36 default=False, 

37 label=_('Exclude Zero Stock Entries'), 

38 help_text=_('Exclude parts with zero stock from the exported dataset'), 

39 ) 

40 

41 

42class PartStocktakeExporter(DataExportMixin, InvenTreePlugin): 

43 """Builtin plugin for exporting part stocktake data. 

44 

45 Extends the "part" export process, to include stocktake data. 

46 """ 

47 

48 NAME = 'Part Stocktake Exporter' 

49 SLUG = 'inventree-stocktake-exporter' 

50 TITLE = _('Part Stocktake Exporter') 

51 DESCRIPTION = _('Exporter for part stocktake data') 

52 VERSION = '1.1.1' 

53 AUTHOR = _('InvenTree contributors') 

54 

55 ExportOptionsSerializer = PartStocktakeExportOptionsSerializer 

56 

57 def supports_export( 

58 self, 

59 model_class: type, 

60 user=None, 

61 serializer_class=None, 

62 view_class=None, 

63 *args, 

64 **kwargs, 

65 ) -> bool: 

66 """Supported if the base model is Part.""" 

67 return model_class == Part and serializer_class == PartSerializer 

68 

69 def generate_filename(self, model_class, export_format: str) -> str: 

70 """Generate a filename for the exported part stocktake data.""" 

71 from InvenTree.helpers import current_date 

72 

73 date = current_date().isoformat() 

74 return f'InvenTree_Stocktake_{date}.{export_format}' 

75 

76 def update_headers(self, headers, context, **kwargs): 

77 """Define headers for the Stocktake export.""" 

78 export_pricing_data = context.get('export_pricing_data', True) 

79 include_external_items = context.get('export_include_external_items', True) 

80 include_variant_items = context.get('export_include_variant_items', False) 

81 

82 # Use only a subset of fields from the PartSerializer 

83 base_headers = [ 

84 'pk', 

85 'name', 

86 'IPN', 

87 'active', 

88 'component', 

89 'assembly', 

90 'description', 

91 'category', 

92 'allocated_to_build_orders', 

93 'allocated_to_sales_orders', 

94 'required_for_build_orders', 

95 'required_for_sales_orders', 

96 'ordering', 

97 'building', 

98 'scheduled_to_build', 

99 'external_stock', 

100 'variant_stock', 

101 'stock_item_count', 

102 'total_in_stock', 

103 ] 

104 

105 if not include_external_items: 

106 base_headers.remove('external_stock') 

107 

108 if not include_variant_items: 

109 base_headers.remove('variant_stock') 

110 

111 stocktake_headers = { 

112 key: headers[key] for key in base_headers if key in headers 

113 } 

114 

115 if export_pricing_data: 

116 stocktake_headers.update({ 

117 'pricing_min': _('Minimum Unit Cost'), 

118 'pricing_max': _('Maximum Unit Cost'), 

119 'pricing_min_total': _('Minimum Total Cost'), 

120 'pricing_max_total': _('Maximum Total Cost'), 

121 }) 

122 

123 return stocktake_headers 

124 

125 def prefetch_queryset(self, queryset): 

126 """Prefetch related data for the queryset.""" 

127 return queryset.prefetch_related('stock_items') 

128 

129 def export_data( 

130 self, queryset, serializer_class, headers, context, output, **kwargs 

131 ): 

132 """Export the data for the given queryset.""" 

133 export_pricing_data = context.get('export_pricing_data', True) 

134 include_external_items = context.get('export_include_external_items', False) 

135 include_variant_items = context.get('export_include_variant_items', False) 

136 exclude_zero_stock = context.get('export_exclude_zero_stock_entries', False) 

137 

138 data = super().export_data( 

139 queryset, serializer_class, headers, context, output, **kwargs 

140 ) 

141 

142 output_data = [] 

143 

144 for row in data: 

145 quantity = Decimal(row.get('total_in_stock', 0)) 

146 

147 if not include_external_items: 

148 quantity -= Decimal(row.get('external_stock', 0)) 

149 

150 if not include_variant_items: 

151 quantity -= Decimal(row.get('variant_stock', 0)) 

152 

153 if quantity < 0: 

154 quantity = Decimal(0) 

155 

156 if exclude_zero_stock: 

157 continue 

158 

159 # Update the 'total in stock' count for this row 

160 row['total_in_stock'] = float(quantity) 

161 

162 if export_pricing_data: 

163 pricing_min = row.get('pricing_min', None) or row.get( 

164 'pricing_max', None 

165 ) 

166 pricing_max = row.get('pricing_max', None) or row.get( 

167 'pricing_min', None 

168 ) 

169 

170 if pricing_min is not None: 

171 pricing_min = Decimal(pricing_min) 

172 row['pricing_min_total'] = normalize( 

173 pricing_min * quantity, rounding=10 

174 ) 

175 

176 if pricing_max is not None: 

177 pricing_max = Decimal(pricing_max) 

178 row['pricing_max_total'] = normalize( 

179 pricing_max * quantity, rounding=10 

180 ) 

181 

182 output_data.append(row) 

183 

184 return output_data