Coverage for src/backend/InvenTree/company/migrations/0019_auto_20200413_0642.py: 9%

1# Generated by Django 2.2.10 on 2020-04-13 06:42 

2 

3import sys 

4import os 

5from rapidfuzz import fuzz 

6 

7from django.db import migrations, connection 

8from django.db.utils import OperationalError, ProgrammingError 

9 

10""" 

11When this migration is tested by CI, it cannot accept user input. 

12So a simplified version of the migration is implemented. 

13""" 

14TESTING = 'test' in sys.argv 

15 

16def clear(): 

17 if not TESTING: # pragma: no cover 

18 os.system('cls' if os.name == 'nt' else 'clear') # ty:ignore[deprecated] 

19 

20 

21def reverse_association(apps, schema_editor): # pragma: no cover 

22 """ 

23 This is the 'reverse' operation of the manufacturer reversal. 

24 This operation is easier: 

25 

26 For each SupplierPart object, copy the name of the 'manufacturer' field 

27 into the 'manufacturer_name' field. 

28 """ 

29 

30 cursor = connection.cursor() 

31 

32 response = cursor.execute('select id, "MPN" from part_supplierpart;') 

33 supplier_parts = cursor.fetchall() 

34 

35 # Exit if there are no SupplierPart objects 

36 # This crucial otherwise the unit test suite fails! 

37 if len(supplier_parts) == 0: 

38 return 

39 

40 print("Reversing migration for manufacturer association") 

41 

42 for (index, row) in enumerate(supplier_parts): 

43 supplier_part_id, MPN = row 

44 

45 print(f"Checking SupplierPart [{supplier_part_id}]:") 

46 

47 # Grab the manufacturer ID from the part 

48 response = cursor.execute(f"SELECT manufacturer_id FROM part_supplierpart WHERE id={supplier_part_id};") 

49 

50 manufacturer_id = None 

51 

52 row = cursor.fetchone() 

53 

54 if row and len(row) > 0: 

55 try: 

56 manufacturer_id = int(row[0]) 

57 except (TypeError, ValueError): 

58 pass 

59 

60 if manufacturer_id is None: 

61 print(" - Manufacturer ID not set: Skipping") 

62 continue 

63 

64 print(" - Manufacturer ID: [{id}]".format(id=manufacturer_id)) 

65 

66 # Now extract the "name" for the manufacturer 

67 response = cursor.execute(f"SELECT name from company_company where id={manufacturer_id};") 

68 

69 row = cursor.fetchone() 

70 if row: 

71 name = row[0] 

72 

73 print(" - Manufacturer name: '{name}'".format(name=name)) 

74 

75 response = cursor.execute("UPDATE part_supplierpart SET manufacturer_name='{name}' WHERE id={ID};".format(name=name, ID=supplier_part_id)) 

76 

77def associate_manufacturers(apps, schema_editor): 

78 """ 

79 This migration is the "middle step" in migration of the "manufacturer" field for the SupplierPart model. 

80 

81 Previously the "manufacturer" field was a simple text field with the manufacturer name. 

82 This is quite insufficient. 

83 The new "manufacturer" field is a link to Company object which has the "is_manufacturer" parameter set to True 

84 

85 This migration requires user interaction to create new "manufacturer" Company objects, 

86 based on the text value in the "manufacturer_name" field (which was created in the previous migration). 

87 

88 It uses fuzzy pattern matching to help the user out as much as possible. 

89 """ 

90 

91 def get_manufacturer_name(part_id): 

92 """ 

93 THIS IS CRITICAL! 

94 

95 Once the pythonic representation of the model has removed the 'manufacturer_name' field, 

96 it is NOT ACCESSIBLE by calling SupplierPart.manufacturer_name. 

97 

98 However, as long as the migrations are applied in order, then the table DOES have a field called 'manufacturer_name'. 

99 

100 So, we just need to request it using dirty SQL. 

101 """ 

102 

103 query = "SELECT manufacturer_name from part_supplierpart where id={ID};".format(ID=part_id) 

104 

105 cursor = connection.cursor() 

106 response = cursor.execute(query) 

107 row = cursor.fetchone() 

108 

109 if row and len(row) > 0: 

110 return row[0] 

111 return '' # pragma: no cover 

112 

113 cursor = connection.cursor() 

114 

115 response = cursor.execute(f'select id, "MPN" from part_supplierpart;') 

116 supplier_parts = cursor.fetchall() 

117 

118 # Exit if there are no SupplierPart objects 

119 # This crucial otherwise the unit test suite fails! 

120 if len(supplier_parts) == 0: 120 ↛ 124line 120 didn't jump to line 124 because the condition on line 120 was always true

121 return 

122 

123 # Link a 'manufacturer_name' to a 'Company' 

124 links = {} 

125 

126 # Map company names to company objects 

127 companies = {} 

128 

129 # Iterate through each company object 

130 response = cursor.execute("select id, name from company_company;") 

131 results = cursor.fetchall() 

132 

133 for index, row in enumerate(results): 

134 pk, name = row 

135 

136 companies[name] = pk 

137 

138 def link_part(part_id, name): 

139 """ Attempt to link Part to an existing Company """ 

140 

141 # Matches a company name directly 

142 if name in companies.keys(): # pragma: no cover 

143 print(" - Part[{pk}]: '{n}' maps to existing manufacturer".format(pk=part_id, n=name)) 

144 

145 manufacturer_id = companies[name] 

146 

147 query = f"update part_supplierpart set manufacturer_id={manufacturer_id} where id={part_id};" 

148 result = cursor.execute(query) 

149 

150 return True 

151 

152 # Have we already mapped this 

153 if name in links.keys(): # pragma: no cover 

154 print(" - Part[{pk}]: Mapped '{n}' - manufacturer <{c}>".format(pk=part_id, n=name, c=links[name])) 

155 

156 manufacturer_id = links[name] 

157 

158 query = f"update part_supplierpart set manufacturer_id={manufacturer_id} where id={part_id};" 

159 result = cursor.execute(query) 

160 return True 

161 

162 # Mapping not possible 

163 return False 

164 

165 def create_manufacturer(part_id, input_name, company_name): 

166 """ Create a new manufacturer """ 

167 

168 Company = apps.get_model('company', 'company') 

169 

170 manufacturer = Company.objects.create( 

171 name=company_name, 

172 description=company_name, 

173 is_manufacturer=True 

174 ) 

175 

176 # Map both names to the same company 

177 links[input_name] = manufacturer.pk 

178 links[company_name] = manufacturer.pk 

179 

180 companies[company_name] = manufacturer.pk 

181 

182 print(" - Part[{pk}]: Created new manufacturer: '{name}'".format(pk=part_id, name=company_name)) 

183 

184 # Update SupplierPart object in the database 

185 cursor.execute(f"update part_supplierpart set manufacturer_id={manufacturer.pk} where id={part_id};") 

186 

187 def find_matches(text, threshold=65): 

188 """ 

189 Attempt to match a 'name' to an existing Company. 

190 A list of potential matches will be returned. 

191 """ 

192 

193 matches = [] 

194 

195 for name in companies.keys(): 

196 # Case-insensitive matching 

197 ratio = fuzz.partial_ratio(name.lower(), text.lower()) 

198 

199 if ratio > threshold: # pragma: no cover 

200 matches.append({'name': name, 'match': ratio}) 

201 

202 if len(matches) > 0: # pragma: no cover 

203 return [match['name'] for match in sorted(matches, key=lambda item: item['match'], reverse=True)] 

204 else: 

205 return [] 

206 

207 

208 def map_part_to_manufacturer(part_id, idx, total): 

209 

210 cursor = connection.cursor() 

211 

212 name = get_manufacturer_name(part_id) 

213 

214 # Skip empty names 

215 if not name or len(name) == 0: # pragma: no cover 

216 print(" - Part[{pk}]: No manufacturer_name provided, skipping".format(pk=part_id)) 

217 return 

218 

219 # Can be linked to an existing manufacturer 

220 if link_part(part_id, name): # pragma: no cover 

221 return 

222 

223 # Find a list of potential matches 

224 matches = find_matches(name) 

225 

226 clear() 

227 

228 # Present a list of options 

229 if not TESTING: # pragma: no cover 

230 print("----------------------------------") 

231 

232 print("Checking part [{pk}] ({idx} of {total})".format(pk=part_id, idx=idx+1, total=total)) 

233 

234 if not TESTING: # pragma: no cover 

235 print("Manufacturer name: '{n}'".format(n=name)) 

236 print("----------------------------------") 

237 print("Select an option from the list below:") 

238 

239 print("0) - Create new manufacturer '{n}'".format(n=name)) 

240 print("") 

241 

242 for i, m in enumerate(matches[:10]): 

243 print("{i}) - Use manufacturer '{opt}'".format(i=i+1, opt=m)) 

244 

245 print("") 

246 print("OR - Type a new custom manufacturer name") 

247 

248 while True: 

249 if TESTING: 

250 # When running unit tests, simply select the name of the part 

251 response = '0' 

252 else: # pragma: no cover 

253 response = str(input("> ")).strip() 

254 

255 # Attempt to parse user response as an integer 

256 try: 

257 n = int(response) 

258 

259 # Option 0) is to create a new manufacturer with the current name 

260 if n == 0: 

261 

262 create_manufacturer(part_id, name, name) 

263 return 

264 

265 # Options 1) - n) select an existing manufacturer 

266 else: # pragma: no cover 

267 n = n - 1 

268 

269 if n < len(matches): 

270 # Get the company which matches the selected options 

271 company_name = matches[n] 

272 company_id = companies[company_name] 

273 

274 # Ensure the company is designated as a manufacturer 

275 cursor.execute(f"update company_company set is_manufacturer=true where id={company_id};") 

276 

277 # Link the company to the part 

278 cursor.execute(f"update part_supplierpart set manufacturer_id={company_id} where id={part_id};") 

279 

280 # Link the name to the company 

281 links[name] = company_id 

282 links[company_name] = company_id 

283 

284 print(" - Part[{pk}]: Linked '{n}' to manufacturer '{m}'".format(pk=part_id, n=name, m=company_name)) 

285 

286 return 

287 else: 

288 print("Please select a valid option") 

289 

290 except ValueError: # pragma: no cover 

291 # User has typed in a custom name! 

292 

293 if not response or len(response) == 0: 

294 # Response cannot be empty! 

295 print("Please select an option") 

296 

297 # Double-check if the typed name corresponds to an existing item 

298 elif response in companies.keys(): 

299 link_part(part_id, companies[response]) 

300 return 

301 

302 elif response in links.keys(): 

303 link_part(part_id, links[response]) 

304 return 

305 

306 # No match, create a new manufacturer 

307 else: 

308 create_manufacturer(part_id, name, response) 

309 return 

310 

311 clear() 

312 print("") 

313 clear() 

314 

315 if not TESTING: # pragma: no cover 

316 print("---------------------------------------") 

317 print("The SupplierPart model needs to be migrated,") 

318 print("as the new 'manufacturer' field maps to a 'Company' reference.") 

319 print("The existing 'manufacturer_name' field will be used to match") 

320 print("against possible companies.") 

321 print("This process requires user input.") 

322 print("") 

323 print("Note: This process MUST be completed to migrate the database.") 

324 print("---------------------------------------") 

325 print("") 

326 

327 input("Press <ENTER> to continue.") 

328 

329 clear() 

330 

331 # Extract all SupplierPart objects from the database 

332 cursor = connection.cursor() 

333 response = cursor.execute('select id, "MPN", "SKU", manufacturer_id, manufacturer_name from part_supplierpart;') 

334 results = cursor.fetchall() 

335 

336 part_count = len(results) 

337 

338 # Create a unique set of manufacturer names 

339 for index, row in enumerate(results): 

340 pk, MPN, SKU, manufacturer_id, manufacturer_name = row 

341 

342 if manufacturer_id is not None: # pragma: no cover 

343 print(f" - SupplierPart <{pk}> already has a manufacturer associated (skipping)") 

344 continue 

345 

346 map_part_to_manufacturer(pk, index, part_count) 

347 

348 print("Done!") 

349 

350class Migration(migrations.Migration): 

351 

352 atomic = False 

353 

354 dependencies = [ 

355 ('company', '0018_supplierpart_manufacturer'), 

356 ] 

357 

358 operations = [ 

359 migrations.RunPython(associate_manufacturers, reverse_code=reverse_association) 

360 ]