Coverage for /home/runner/work/netbox-data-import-plugin/netbox-data-import-plugin/netbox-data-import-plugin/netbox_data

1# SPDX-License-Identifier: Apache-2.0

3from django.contrib import messages

4from django.contrib.auth.mixins import LoginRequiredMixin

5from django.shortcuts import get_object_or_404, redirect, render

6from django.urls import reverse

7from django.utils.http import url_has_allowed_host_and_scheme

8from django.views import View

9from netbox.views import generic

10from .models import (

11 ImportProfile,

12 ColumnMapping,

13 ClassRoleMapping,

14 DeviceTypeMapping,

15 ImportJob,

16 ColumnTransformRule,

17 DeviceExistingMatch,

18 ManufacturerMapping,

19)

20from .forms import (

21 ImportProfileForm,

22 ColumnMappingForm,

23 ClassRoleMappingForm,

24 DeviceTypeMappingForm,

25 ColumnTransformRuleForm,

26 ImportSetupForm,

27)

28from .tables import (

29 ImportProfileTable,

30 ColumnMappingTable,

31 ClassRoleMappingTable,

32 DeviceTypeMappingTable,

33 ColumnTransformRuleTable,

34)

35from .filters import ImportProfileFilterSet

38def _safe_next_url(request, fallback: str) -> str:

39 """Return a validated same-host redirect URL from POST or the fallback view name."""

40 url = request.POST.get("next", "")

41 if url and url_has_allowed_host_and_scheme(

42 url, allowed_hosts={request.get_host()}, require_https=request.is_secure()

43 ):

44 return url

45 return reverse(fallback)

48# ---------------------------------------------------------------------------

49# ImportProfile

50# ---------------------------------------------------------------------------

53class ImportProfileListView(generic.ObjectListView):

54 """List all import profiles with their mapping counts."""

56 queryset = ImportProfile.objects.prefetch_related("column_mappings", "class_role_mappings", "device_type_mappings")

57 table = ImportProfileTable

58 filterset = ImportProfileFilterSet

59 template_name = "netbox_data_import/importprofile_list.html"

62class ImportProfileView(generic.ObjectView):

63 """Detail view for a single import profile, with inline mapping tables."""

65 queryset = ImportProfile.objects.prefetch_related("column_mappings", "class_role_mappings", "device_type_mappings")

67 def get_extra_context(self, request, instance):

68 """Inject inline mapping tables into the template context."""

69 column_table = ColumnMappingTable(instance.column_mappings.all())

70 class_role_table = ClassRoleMappingTable(instance.class_role_mappings.all())

71 device_type_table = DeviceTypeMappingTable(instance.device_type_mappings.all())

72 transform_table = ColumnTransformRuleTable(instance.column_transform_rules.all())

73 return {

74 "column_table": column_table,

75 "class_role_table": class_role_table,

76 "device_type_table": device_type_table,

77 "transform_table": transform_table,

78 }

81class ImportProfileEditView(generic.ObjectEditView):

82 """Create or edit an ImportProfile."""

84 queryset = ImportProfile.objects.all()

85 form = ImportProfileForm

88class ImportProfileDeleteView(generic.ObjectDeleteView):

89 """Delete an ImportProfile and all its child mappings."""

91 queryset = ImportProfile.objects.all()

94# ---------------------------------------------------------------------------

95# ColumnMapping CRUD

96# ---------------------------------------------------------------------------

99class ColumnMappingAddView(LoginRequiredMixin, View):

100 """Add a column mapping to an existing ImportProfile."""

101

102 def get(self, request, profile_pk):

103 """Render the add form for a new column mapping."""

104 profile = get_object_or_404(ImportProfile, pk=profile_pk)

105 form = ColumnMappingForm(initial={"profile": profile})

106 return render(request, "netbox_data_import/columnmapping_edit.html", {"form": form, "profile": profile})

107

108 def post(self, request, profile_pk):

109 """Save a new column mapping or re-render the form with errors."""

110 profile = get_object_or_404(ImportProfile, pk=profile_pk)

111 form = ColumnMappingForm(request.POST)

112 if form.is_valid():

113 form.save()

114 messages.success(request, "Column mapping added.")

115 return redirect(profile.get_absolute_url())

116 return render(request, "netbox_data_import/columnmapping_edit.html", {"form": form, "profile": profile})

117

118

119class ColumnMappingEditView(LoginRequiredMixin, View):

120 """Edit an existing column mapping."""

121

122 def get(self, request, pk):

123 """Render the edit form for an existing column mapping."""

124 obj = get_object_or_404(ColumnMapping, pk=pk)

125 form = ColumnMappingForm(instance=obj)

126 return render(

127 request, "netbox_data_import/columnmapping_edit.html", {"form": form, "profile": obj.profile, "object": obj}

128 )

129

130 def post(self, request, pk):

131 """Save edits to an existing column mapping or re-render with errors."""

132 obj = get_object_or_404(ColumnMapping, pk=pk)

133 form = ColumnMappingForm(request.POST, instance=obj)

134 if form.is_valid():

135 form.save()

136 messages.success(request, "Column mapping updated.")

137 return redirect(obj.profile.get_absolute_url())

138 return render(

139 request, "netbox_data_import/columnmapping_edit.html", {"form": form, "profile": obj.profile, "object": obj}

140 )

141

142

143class ColumnMappingDeleteView(LoginRequiredMixin, View):

144 """Delete a column mapping."""

145

146 def get(self, request, pk):

147 """Render the delete confirmation page for a column mapping."""

148 obj = get_object_or_404(ColumnMapping, pk=pk)

149 return render(

150 request,

151 "netbox_data_import/confirm_delete.html",

152 {"object": obj, "return_url": obj.profile.get_absolute_url()},

153 )

154

155 def post(self, request, pk):

156 """Delete the column mapping and redirect to the parent profile."""

157 obj = get_object_or_404(ColumnMapping, pk=pk)

158 profile_url = obj.profile.get_absolute_url()

159 obj.delete()

160 messages.success(request, "Column mapping deleted.")

161 return redirect(profile_url)

162

163

164# ---------------------------------------------------------------------------

165# ClassRoleMapping CRUD

166# ---------------------------------------------------------------------------

167

168

169class ClassRoleMappingAddView(LoginRequiredMixin, View):

170 """Add a class→role mapping to an existing ImportProfile."""

171

172 def get(self, request, profile_pk):

173 """Render the add form for a new class→role mapping."""

174 profile = get_object_or_404(ImportProfile, pk=profile_pk)

175 form = ClassRoleMappingForm(initial={"profile": profile})

176 return render(request, "netbox_data_import/classrolemapping_edit.html", {"form": form, "profile": profile})

177

178 def post(self, request, profile_pk):

179 """Save a new class→role mapping or re-render with errors."""

180 profile = get_object_or_404(ImportProfile, pk=profile_pk)

181 form = ClassRoleMappingForm(request.POST)

182 if form.is_valid():

183 form.save()

184 messages.success(request, "Class→Role mapping added.")

185 return redirect(profile.get_absolute_url())

186 return render(request, "netbox_data_import/classrolemapping_edit.html", {"form": form, "profile": profile})

187

188

189class ClassRoleMappingEditView(LoginRequiredMixin, View):

190 """Edit an existing class→role mapping."""

191

192 def get(self, request, pk):

193 """Render the edit form for an existing class→role mapping."""

194 obj = get_object_or_404(ClassRoleMapping, pk=pk)

195 form = ClassRoleMappingForm(instance=obj)

196 return render(

197 request,

198 "netbox_data_import/classrolemapping_edit.html",

199 {"form": form, "profile": obj.profile, "object": obj},

200 )

201

202 def post(self, request, pk):

203 """Save edits to an existing class→role mapping or re-render with errors."""

204 obj = get_object_or_404(ClassRoleMapping, pk=pk)

205 form = ClassRoleMappingForm(request.POST, instance=obj)

206 if form.is_valid():

207 form.save()

208 messages.success(request, "Class→Role mapping updated.")

209 return redirect(obj.profile.get_absolute_url())

210 return render(

211 request,

212 "netbox_data_import/classrolemapping_edit.html",

213 {"form": form, "profile": obj.profile, "object": obj},

214 )

215

216

217class ClassRoleMappingDeleteView(LoginRequiredMixin, View):

218 """Delete a class→role mapping."""

219

220 def get(self, request, pk):

221 """Render the delete confirmation page for a class→role mapping."""

222 obj = get_object_or_404(ClassRoleMapping, pk=pk)

223 return render(

224 request,

225 "netbox_data_import/confirm_delete.html",

226 {"object": obj, "return_url": obj.profile.get_absolute_url()},

227 )

228

229 def post(self, request, pk):

230 """Delete the class→role mapping and redirect to the parent profile."""

231 obj = get_object_or_404(ClassRoleMapping, pk=pk)

232 profile_url = obj.profile.get_absolute_url()

233 obj.delete()

234 messages.success(request, "Class→Role mapping deleted.")

235 return redirect(profile_url)

236

237

238# ---------------------------------------------------------------------------

239# DeviceTypeMapping CRUD

240# ---------------------------------------------------------------------------

241

242

243class DeviceTypeMappingAddView(LoginRequiredMixin, View):

244 """Add a device type mapping to an existing ImportProfile."""

245

246 def get(self, request, profile_pk):

247 """Render the add form for a new device type mapping."""

248 profile = get_object_or_404(ImportProfile, pk=profile_pk)

249 form = DeviceTypeMappingForm(initial={"profile": profile})

250 return render(request, "netbox_data_import/devicetypemapping_edit.html", {"form": form, "profile": profile})

251

252 def post(self, request, profile_pk):

253 """Save a new device type mapping or re-render with errors."""

254 profile = get_object_or_404(ImportProfile, pk=profile_pk)

255 form = DeviceTypeMappingForm(request.POST)

256 if form.is_valid():

257 form.save()

258 messages.success(request, "Device type mapping added.")

259 return redirect(profile.get_absolute_url())

260 return render(request, "netbox_data_import/devicetypemapping_edit.html", {"form": form, "profile": profile})

261

262

263class DeviceTypeMappingEditView(LoginRequiredMixin, View):

264 """Edit an existing device type mapping."""

265

266 def get(self, request, pk):

267 """Render the edit form for an existing device type mapping."""

268 obj = get_object_or_404(DeviceTypeMapping, pk=pk)

269 form = DeviceTypeMappingForm(instance=obj)

270 return render(

271 request,

272 "netbox_data_import/devicetypemapping_edit.html",

273 {"form": form, "profile": obj.profile, "object": obj},

274 )

275

276 def post(self, request, pk):

277 """Save edits to an existing device type mapping or re-render with errors."""

278 obj = get_object_or_404(DeviceTypeMapping, pk=pk)

279 form = DeviceTypeMappingForm(request.POST, instance=obj)

280 if form.is_valid():

281 form.save()

282 messages.success(request, "Device type mapping updated.")

283 return redirect(obj.profile.get_absolute_url())

284 return render(

285 request,

286 "netbox_data_import/devicetypemapping_edit.html",

287 {"form": form, "profile": obj.profile, "object": obj},

288 )

289

290

291class DeviceTypeMappingDeleteView(LoginRequiredMixin, View):

292 """Delete a device type mapping."""

293

294 def get(self, request, pk):

295 """Render the delete confirmation page for a device type mapping."""

296 obj = get_object_or_404(DeviceTypeMapping, pk=pk)

297 return render(

298 request,

299 "netbox_data_import/confirm_delete.html",

300 {"object": obj, "return_url": obj.profile.get_absolute_url()},

301 )

302

303 def post(self, request, pk):

304 """Delete the device type mapping and redirect to the parent profile."""

305 obj = get_object_or_404(DeviceTypeMapping, pk=pk)

306 profile_url = obj.profile.get_absolute_url()

307 obj.delete()

308 messages.success(request, "Device type mapping deleted.")

309 return redirect(profile_url)

310

311

312# ---------------------------------------------------------------------------

313# Import Wizard — Phase 2 (setup + preview)

314# ---------------------------------------------------------------------------

315

316

317class ImportSetupView(LoginRequiredMixin, View):

318 """Step 1: select profile, upload file, choose site/location/tenant."""

319

320 def get(self, request):

321 """Render the import setup form."""

322 initial = {}

323 if profile_pk := request.GET.get("profile"):

324 initial["profile"] = profile_pk

325 form = ImportSetupForm(initial=initial)

326 return render(request, "netbox_data_import/import_setup.html", {"form": form})

327

328 def post(self, request):

329 """Parse the uploaded file and redirect to the preview step."""

330 form = ImportSetupForm(request.POST, request.FILES)

331 if not form.is_valid():

332 return render(request, "netbox_data_import/import_setup.html", {"form": form})

333

334 from . import engine

335

336 profile = form.cleaned_data["profile"]

337 excel_file = form.cleaned_data["excel_file"]

338 site = form.cleaned_data["site"]

339 location = form.cleaned_data.get("location")

340 tenant = form.cleaned_data.get("tenant")

341

342 try:

343 rows = engine.parse_file(excel_file, profile)

344 except engine.ParseError as exc:

345 messages.error(request, f"Failed to parse file: {exc}")

346 return render(request, "netbox_data_import/import_setup.html", {"form": form})

347

348 context = {"site": site, "location": location, "tenant": tenant}

349 result = engine.run_import(rows, profile, context, dry_run=True)

350

351 # Store result + raw rows + context in session for the preview/execute steps

352 # Rows need JSON-safe serialization (handle datetime from Excel)

353 request.session["import_result"] = result.to_session_dict()

354 request.session["import_rows"] = _serialize_rows(rows)

355 request.session["import_context"] = {

356 "profile_id": profile.pk,

357 "site_id": site.pk,

358 "location_id": location.pk if location else None,

359 "tenant_id": tenant.pk if tenant else None,

360 "filename": excel_file.name,

361 }

362 return redirect(reverse("plugins:netbox_data_import:import_preview"))

363

364

365class ImportPreviewView(LoginRequiredMixin, View):

366 """Step 2: show dry-run results, let user confirm or go back."""

367

368 def get(self, request):

369 """Re-run the dry-run import and render the preview template."""

370 rows = request.session.get("import_rows")

371 ctx = request.session.get("import_context", {})

372 if not rows or not ctx:

373 messages.warning(request, "No import in progress. Please start a new import.")

374 return redirect(reverse("plugins:netbox_data_import:import_setup"))

375

376 from . import engine

377 from dcim.models import Site, Location

378 from tenancy.models import Tenant

379

380 profile = ImportProfile.objects.filter(pk=ctx.get("profile_id")).first()

381 if not profile:

382 messages.warning(request, "Import profile not found.")

383 return redirect(reverse("plugins:netbox_data_import:import_setup"))

384

385 site = Site.objects.filter(pk=ctx.get("site_id")).first()

386 location = Location.objects.filter(pk=ctx.get("location_id")).first() if ctx.get("location_id") else None

387 tenant = Tenant.objects.filter(pk=ctx.get("tenant_id")).first() if ctx.get("tenant_id") else None

388

389 context_obj = {"site": site, "location": location, "tenant": tenant}

390 # Always re-run so any new mappings/matches are immediately reflected

391 result = engine.run_import(rows, profile, context_obj, dry_run=True)

392 request.session["import_result"] = result.to_session_dict()

393

394 # Build existing resolutions map for the split-name modal preview

395 from .models import SourceResolution

396 import json as _json

397

398 existing_resolutions = {}

399 for res in SourceResolution.objects.filter(profile=profile):

400 existing_resolutions.setdefault(str(res.source_id), {})[res.source_column] = {

401 "original_value": res.original_value,

402 "resolved_fields": res.resolved_fields,

403 }

404

405 view_mode = request.GET.get("view", profile.preview_view_mode)

406 return render(

407 request,

408 "netbox_data_import/import_preview.html",

409 {

410 "result": result,

411 "filename": ctx.get("filename", ""),

412 "profile_id": ctx.get("profile_id"),

413 "profile": profile,

414 "view_mode": view_mode,

415 "existing_resolutions_json": _json.dumps(existing_resolutions),

416 },

417 )

418

419

420class ImportRunView(LoginRequiredMixin, View):

421 """Step 3: run the real import (dry_run=False)."""

422

423 def post(self, request):

424 """Execute the real import and redirect to the results page."""

425 rows = request.session.get("import_rows")

426 ctx_data = request.session.get("import_context")

427 if not rows or not ctx_data:

428 messages.warning(request, "No import in progress.")

429 return redirect(reverse("plugins:netbox_data_import:import_setup"))

430

431 from django.db import transaction

432 from dcim.models import Site, Location

433 from tenancy.models import Tenant

434 from . import engine

435

436 profile = get_object_or_404(ImportProfile, pk=ctx_data["profile_id"])

437 site = get_object_or_404(Site, pk=ctx_data["site_id"])

438 location = get_object_or_404(Location, pk=ctx_data["location_id"]) if ctx_data.get("location_id") else None

439 tenant = get_object_or_404(Tenant, pk=ctx_data["tenant_id"]) if ctx_data.get("tenant_id") else None

440

441 context = {"site": site, "location": location, "tenant": tenant}

442

443 with transaction.atomic():

444 result = engine.run_import(rows, profile, context, dry_run=False)

445

446 # Persist job record

447 from .models import ImportJob

448

449 job = ImportJob.objects.create(

450 profile=profile,

451 input_filename=ctx_data.get("filename", ""),

452 dry_run=False,

453 site_name=site.name,

454 result_counts=result.counts,

455 result_rows=[r.to_dict() for r in result.rows],

456 )

457

458 request.session["import_result"] = result.to_session_dict()

459 request.session["import_job_id"] = job.pk

460 messages.success(

461 request,

462 f"Import complete: {result.counts.get('devices_created', 0)} devices created, "

463 f"{result.counts.get('racks_created', 0)} racks created.",

464 )

465 return redirect(reverse("plugins:netbox_data_import:import_results"))

466

467

468class ImportResultsView(LoginRequiredMixin, View):

469 """Step 4: show final results with links to created objects."""

470

471 def get(self, request):

472 """Render the results page for the most recent import."""

473 session_data = request.session.get("import_result")

474 if not session_data:

475 return redirect(reverse("plugins:netbox_data_import:import_setup"))

476

477 from . import engine

478

479 result = engine.ImportResult.from_session_dict(session_data)

480 job_id = request.session.get("import_job_id")

481 return render(request, "netbox_data_import/import_results.html", {"result": result, "job_id": job_id})

482

483

484# ---------------------------------------------------------------------------

485# Import Job history

486# ---------------------------------------------------------------------------

487

488

489class ImportJobListView(LoginRequiredMixin, View):

490 """List all past import jobs for audit / history."""

491

492 def get(self, request):

493 """Render the import job history list."""

494 jobs = ImportJob.objects.select_related("profile").all()

495 return render(request, "netbox_data_import/importjob_list.html", {"jobs": jobs})

496

497

498# ---------------------------------------------------------------------------

499# ColumnTransformRule CRUD

500# ---------------------------------------------------------------------------

501

502

503class ColumnTransformRuleAddView(LoginRequiredMixin, View):

504 """Add a column transform rule to an existing ImportProfile."""

505

506 def get(self, request, profile_pk):

507 """Render the add form for a new column transform rule."""

508 profile = get_object_or_404(ImportProfile, pk=profile_pk)

509 form = ColumnTransformRuleForm(initial={"profile": profile})

510 return render(request, "netbox_data_import/columntransformrule_edit.html", {"form": form, "profile": profile})

511

512 def post(self, request, profile_pk):

513 """Save a new column transform rule or re-render with errors."""

514 profile = get_object_or_404(ImportProfile, pk=profile_pk)

515 form = ColumnTransformRuleForm(request.POST)

516 if form.is_valid():

517 form.save()

518 messages.success(request, "Column transform rule added.")

519 return redirect(profile.get_absolute_url())

520 return render(request, "netbox_data_import/columntransformrule_edit.html", {"form": form, "profile": profile})

521

522

523class ColumnTransformRuleEditView(LoginRequiredMixin, View):

524 """Edit an existing column transform rule."""

525

526 def get(self, request, pk):

527 """Render the edit form for an existing column transform rule."""

528 obj = get_object_or_404(ColumnTransformRule, pk=pk)

529 form = ColumnTransformRuleForm(instance=obj)

530 return render(

531 request,

532 "netbox_data_import/columntransformrule_edit.html",

533 {"form": form, "profile": obj.profile, "object": obj},

534 )

535

536 def post(self, request, pk):

537 """Save edits to an existing column transform rule or re-render with errors."""

538 obj = get_object_or_404(ColumnTransformRule, pk=pk)

539 form = ColumnTransformRuleForm(request.POST, instance=obj)

540 if form.is_valid():

541 form.save()

542 messages.success(request, "Column transform rule updated.")

543 return redirect(obj.profile.get_absolute_url())

544 return render(

545 request,

546 "netbox_data_import/columntransformrule_edit.html",

547 {"form": form, "profile": obj.profile, "object": obj},

548 )

549

550

551class ColumnTransformRuleDeleteView(LoginRequiredMixin, View):

552 """Delete a column transform rule."""

553

554 def get(self, request, pk):

555 """Render the delete confirmation page for a column transform rule."""

556 obj = get_object_or_404(ColumnTransformRule, pk=pk)

557 return render(

558 request,

559 "netbox_data_import/confirm_delete.html",

560 {"object": obj, "return_url": obj.profile.get_absolute_url()},

561 )

562

563 def post(self, request, pk):

564 """Delete the column transform rule and redirect to the parent profile."""

565 obj = get_object_or_404(ColumnTransformRule, pk=pk)

566 profile_url = obj.profile.get_absolute_url()

567 obj.delete()

568 messages.success(request, "Column transform rule deleted.")

569 return redirect(profile_url)

570

571

572# ---------------------------------------------------------------------------

573# Ignore / Unignore device

574# ---------------------------------------------------------------------------

575

576

577class IgnoreDeviceView(LoginRequiredMixin, View):

578 """Mark a specific device (by source_id) as ignored for a profile."""

579

580 def post(self, request):

581 """Add the specified device to the profile's ignore list."""

582 from .models import IgnoredDevice

583

584 profile_id = request.POST.get("profile_id")

585 source_id = request.POST.get("source_id")

586 device_name = request.POST.get("device_name", "")

587 next_url = _safe_next_url(request, "plugins:netbox_data_import:import_preview")

588

589 if profile_id and source_id:

590 profile = get_object_or_404(ImportProfile, pk=profile_id)

591 IgnoredDevice.objects.get_or_create(

592 profile=profile,

593 source_id=source_id,

594 defaults={"device_name": device_name},

595 )

596 messages.success(request, f"Device '{device_name or source_id}' added to ignore list.")

597 return redirect(next_url)

598

599

600class UnignoreDeviceView(LoginRequiredMixin, View):

601 """Remove a device from the ignore list."""

602

603 def post(self, request):

604 """Remove the specified device from the profile's ignore list."""

605 from .models import IgnoredDevice

606

607 profile_id = request.POST.get("profile_id")

608 source_id = request.POST.get("source_id")

609 next_url = _safe_next_url(request, "plugins:netbox_data_import:import_preview")

610

611 if profile_id and source_id:

612 IgnoredDevice.objects.filter(

613 profile_id=profile_id,

614 source_id=source_id,

615 ).delete()

616 messages.success(request, "Device removed from ignore list.")

617 return redirect(next_url)

618

619

620# ---------------------------------------------------------------------------

621# Save resolution (rerere)

622# ---------------------------------------------------------------------------

623

624

625class SaveResolutionView(LoginRequiredMixin, View):

626 """Save a manual field resolution for rerere replay."""

627

628 def post(self, request):

629 """Persist a manual field resolution for rerere replay."""

630 import json

631 from .models import SourceResolution

632

633 profile_id = request.POST.get("profile_id")

634 source_id = request.POST.get("source_id")

635 source_column = request.POST.get("source_column")

636 original_value = request.POST.get("original_value")

637 resolved_fields_json = request.POST.get("resolved_fields", "{}")

638 next_url = _safe_next_url(request, "plugins:netbox_data_import:import_preview")

639

640 try:

641 resolved_fields = json.loads(resolved_fields_json)

642 except (json.JSONDecodeError, TypeError):

643 resolved_fields = {}

644

645 if profile_id and source_id and source_column:

646 profile = get_object_or_404(ImportProfile, pk=profile_id)

647 SourceResolution.objects.update_or_create(

648 profile=profile,

649 source_id=source_id,

650 source_column=source_column,

651 defaults={

652 "original_value": original_value or "",

653 "resolved_fields": resolved_fields,

654 },

655 )

656 messages.success(request, "Resolution saved. Re-run the import to apply it.")

657 return redirect(next_url)

658

659

660# ---------------------------------------------------------------------------

661# Device type analysis view

662# ---------------------------------------------------------------------------

663

664

665class DeviceTypeAnalysisView(LoginRequiredMixin, View):

666 """Show all unique (make, model) pairs across import jobs and profiles.

667

668 Highlights which ones have explicit DeviceTypeMapping vs auto-slugified.

669 """

670

671 def get(self, request, profile_pk=None):

672 """Render the device type analysis page for the given profile."""

673 profile = get_object_or_404(ImportProfile, pk=profile_pk) if profile_pk else None

674 profiles = ImportProfile.objects.all()

675

676 # Collect unique (make, model) pairs from all recent import jobs

677 job_qs = ImportJob.objects.all()

678 if profile:

679 job_qs = job_qs.filter(profile=profile)

680

681 for job in job_qs.order_by("-created")[:50]:

682 for row in job.result_rows:

683 if row.get("object_type") in ("manufacturer", "device_type"):

684 continue

685 # We don't store make/model in result_rows, so skip

686 pass

687

688 # Better: build analysis from DeviceTypeMapping + auto-slugify check

689 if profile:

690 dt_mappings = DeviceTypeMapping.objects.filter(profile=profile)

691 else:

692 dt_mappings = DeviceTypeMapping.objects.select_related("profile").all()

693

694 # Collect entries: explicit mappings

695 entries = []

696 for dtm in dt_mappings:

697 entries.append(

698 {

699 "profile": dtm.profile,

700 "source_make": dtm.source_make,

701 "source_model": dtm.source_model,

702 "manufacturer_slug": dtm.netbox_manufacturer_slug,

703 "device_type_slug": dtm.netbox_device_type_slug,

704 "mapping_type": "explicit",

705 "mapping_pk": dtm.pk,

706 }

707 )

708

709 # Check which mapped device types exist in NetBox

710 from dcim.models import DeviceType

711

712 for entry in entries:

713 entry["exists_in_netbox"] = DeviceType.objects.filter(

714 manufacturer__slug=entry["manufacturer_slug"],

715 slug=entry["device_type_slug"],

716 ).exists()

717

718 return render(

719 request,

720 "netbox_data_import/analysis.html",

721 {

722 "profile": profile,

723 "profiles": profiles,

724 "entries": entries,

725 },

726 )

727

728

729# ---------------------------------------------------------------------------

730# Bulk YAML import for mappings

731# ---------------------------------------------------------------------------

732

733

734class BulkYamlImportView(LoginRequiredMixin, View):

735 """Accept a YAML file and bulk-create ClassRoleMappings or DeviceTypeMappings for a profile.

736

737 Useful for bootstrapping from contrib/ definition files.

738 """

739

740 def get(self, request, profile_pk):

741 """Render the bulk YAML import form."""

742 profile = get_object_or_404(ImportProfile, pk=profile_pk)

743 return render(request, "netbox_data_import/bulk_yaml_import.html", {"profile": profile})

744

745 def post(self, request, profile_pk):

746 """Parse the uploaded YAML file and create mappings in bulk."""

747 profile = get_object_or_404(ImportProfile, pk=profile_pk)

748 yaml_file = request.FILES.get("yaml_file")

749 mapping_type = request.POST.get("mapping_type", "class_role")

750

751 if not yaml_file:

752 messages.error(request, "No YAML file uploaded.")

753 return render(request, "netbox_data_import/bulk_yaml_import.html", {"profile": profile})

754

755 try:

756 import yaml

757

758 data = yaml.safe_load(yaml_file.read())

759 except Exception as exc:

760 messages.error(request, f"Failed to parse YAML: {exc}")

761 return render(request, "netbox_data_import/bulk_yaml_import.html", {"profile": profile})

762

763 if not isinstance(data, list):

764 messages.error(request, "YAML must be a list of mapping objects.")

765 return render(request, "netbox_data_import/bulk_yaml_import.html", {"profile": profile})

766

767 created = 0

768 skipped = 0

769 errors = []

770

771 if mapping_type == "class_role":

772 for item in data:

773 try:

774 _, was_created = ClassRoleMapping.objects.get_or_create(

775 profile=profile,

776 source_class=item["source_class"],

777 defaults={

778 "creates_rack": item.get("creates_rack", False),

779 "role_slug": item.get("role_slug", ""),

780 "ignore": item.get("ignore", False),

781 },

782 )

783 if was_created:

784 created += 1

785 else:

786 skipped += 1

787 except Exception as exc:

788 errors.append(str(exc))

789 elif mapping_type == "device_type":

790 for item in data:

791 try:

792 _, was_created = DeviceTypeMapping.objects.get_or_create(

793 profile=profile,

794 source_make=item["source_make"],

795 source_model=item["source_model"],

796 defaults={

797 "netbox_manufacturer_slug": item["netbox_manufacturer_slug"],

798 "netbox_device_type_slug": item["netbox_device_type_slug"],

799 },

800 )

801 if was_created:

802 created += 1

803 else:

804 skipped += 1

805 except Exception as exc:

806 errors.append(str(exc))

807

808 if errors:

809 messages.warning(

810 request, f"Created {created}, skipped {skipped}, {len(errors)} errors: {'; '.join(errors[:3])}"

811 )

812 else:

813 messages.success(request, f"Bulk import complete: {created} created, {skipped} already existed.")

814 return redirect(profile.get_absolute_url())

815

816

817# ---------------------------------------------------------------------------

818# Profile YAML export / full-profile YAML import

819# ---------------------------------------------------------------------------

820

821

822class ExportProfileYamlView(LoginRequiredMixin, View):

823 """Download all profile configuration as a single YAML file."""

824

825 def get(self, request, pk):

826 """Serialize the profile and all its mappings to YAML and return as a file download."""

827 from django.http import HttpResponse

828 import yaml

829

830 profile = get_object_or_404(ImportProfile, pk=pk)

831

832 data = {

833 "profile": {

834 "name": profile.name,

835 "description": profile.description,

836 "sheet_name": profile.sheet_name,

837 "source_id_column": profile.source_id_column,

838 "custom_field_name": profile.custom_field_name,

839 "update_existing": profile.update_existing,

840 "create_missing_device_types": profile.create_missing_device_types,

841 "preview_view_mode": profile.preview_view_mode,

842 },

843 "column_mappings": [

844 {"source_column": cm.source_column, "target_field": cm.target_field}

845 for cm in profile.column_mappings.all()

846 ],

847 "class_role_mappings": [

848 {

849 "source_class": m.source_class,

850 "creates_rack": m.creates_rack,

851 "role_slug": m.role_slug,

852 "ignore": m.ignore,

853 }

854 for m in profile.class_role_mappings.all()

855 ],

856 "device_type_mappings": [

857 {

858 "source_make": m.source_make,

859 "source_model": m.source_model,

860 "netbox_manufacturer_slug": m.netbox_manufacturer_slug,

861 "netbox_device_type_slug": m.netbox_device_type_slug,

862 }

863 for m in profile.device_type_mappings.all()

864 ],

865 "manufacturer_mappings": [

866 {

867 "source_make": m.source_make,

868 "netbox_manufacturer_slug": m.netbox_manufacturer_slug,

869 }

870 for m in profile.manufacturer_mappings.all()

871 ],

872 "column_transform_rules": [

873 {

874 "source_column": r.source_column,

875 "pattern": r.pattern,

876 "group_1_target": r.group_1_target,

877 "group_2_target": r.group_2_target,

878 }

879 for r in profile.column_transform_rules.all()

880 ],

881 }

882

883 yaml_str = yaml.dump(data, allow_unicode=True, default_flow_style=False, sort_keys=False)

884 safe_name = profile.name.lower().replace(" ", "_").replace("/", "-")

885 filename = f"profile_{safe_name}.yaml"

886 return HttpResponse(

887 yaml_str,

888 content_type="application/x-yaml",

889 headers={"Content-Disposition": f'attachment; filename="{filename}"'},

890 )

891

892

893class ImportProfileYamlView(LoginRequiredMixin, View):

894 """Import a full profile YAML (as exported by ExportProfileYamlView).

895

896 If the profile already exists (by name), merges/updates its mappings.

897 """

898

899 def get(self, request):

900 """Render the profile YAML import form."""

901 return render(request, "netbox_data_import/import_profile_yaml.html")

902

903 def post(self, request):

904 """Parse the uploaded YAML and create or update the profile and its mappings."""

905 import yaml

906

907 yaml_file = request.FILES.get("yaml_file")

908 if not yaml_file:

909 messages.error(request, "No YAML file uploaded.")

910 return render(request, "netbox_data_import/import_profile_yaml.html")

911

912 try:

913 data = yaml.safe_load(yaml_file.read())

914 except Exception as exc:

915 messages.error(request, f"Failed to parse YAML: {exc}")

916 return render(request, "netbox_data_import/import_profile_yaml.html")

917

918 if not isinstance(data, dict) or "profile" not in data:

919 messages.error(request, "YAML must contain a top-level 'profile' key.")

920 return render(request, "netbox_data_import/import_profile_yaml.html")

921

922 pdata = data["profile"]

923 profile, _ = ImportProfile.objects.update_or_create(

924 name=pdata["name"],

925 defaults={

926 "description": pdata.get("description", ""),

927 "sheet_name": pdata.get("sheet_name", "Data"),

928 "source_id_column": pdata.get("source_id_column", ""),

929 "custom_field_name": pdata.get("custom_field_name", ""),

930 "update_existing": pdata.get("update_existing", True),

931 "create_missing_device_types": pdata.get("create_missing_device_types", True),

932 "preview_view_mode": pdata.get("preview_view_mode", "rows"),

933 },

934 )

935

936 stats = {}

937 for cm in data.get("column_mappings", []):

938 _, c = ColumnMapping.objects.update_or_create(

939 profile=profile,

940 target_field=cm["target_field"],

941 defaults={"source_column": cm["source_column"]},

942 )

943 stats["column_mappings"] = stats.get("column_mappings", 0) + 1

944

945 for m in data.get("class_role_mappings", []):

946 ClassRoleMapping.objects.update_or_create(

947 profile=profile,

948 source_class=m["source_class"],

949 defaults={

950 "creates_rack": m.get("creates_rack", False),

951 "role_slug": m.get("role_slug", ""),

952 "ignore": m.get("ignore", False),

953 },

954 )

955 stats["class_role_mappings"] = stats.get("class_role_mappings", 0) + 1

956

957 for m in data.get("device_type_mappings", []):

958 DeviceTypeMapping.objects.update_or_create(

959 profile=profile,

960 source_make=m["source_make"],

961 source_model=m["source_model"],

962 defaults={

963 "netbox_manufacturer_slug": m["netbox_manufacturer_slug"],

964 "netbox_device_type_slug": m["netbox_device_type_slug"],

965 },

966 )

967 stats["device_type_mappings"] = stats.get("device_type_mappings", 0) + 1

968

969 for m in data.get("manufacturer_mappings", []):

970 ManufacturerMapping.objects.update_or_create(

971 profile=profile,

972 source_make=m["source_make"],

973 defaults={"netbox_manufacturer_slug": m["netbox_manufacturer_slug"]},

974 )

975 stats["manufacturer_mappings"] = stats.get("manufacturer_mappings", 0) + 1

976

977 from .models import ColumnTransformRule

978

979 for r in data.get("column_transform_rules", []):

980 ColumnTransformRule.objects.update_or_create(

981 profile=profile,

982 source_column=r["source_column"],

983 defaults={

984 "pattern": r["pattern"],

985 "group_1_target": r.get("group_1_target", ""),

986 "group_2_target": r.get("group_2_target", ""),

987 },

988 )

989 stats["column_transform_rules"] = stats.get("column_transform_rules", 0) + 1

990

991 summary = ", ".join(f"{v} {k.replace('_', ' ')}" for k, v in stats.items())

992 messages.success(request, f"Profile '{profile.name}' imported/updated. {summary}.")

993 return redirect(profile.get_absolute_url())

994

995

996# ---------------------------------------------------------------------------

997

998

999class CheckDeviceNameView(LoginRequiredMixin, View):

1000 """AJAX endpoint: check if a device with the given name exists in NetBox.

1001

1002 Returns JSON: {"exists": bool, "url": str|null, "id": int|null}.

1003 """

1004

1005 def get(self, request):

1006 """Return JSON indicating whether a device with the given name exists."""

1007 from django.http import JsonResponse

1008 from dcim.models import Device

1009

1010 name = request.GET.get("name", "").strip()

1011 if not name:

1012 return JsonResponse({"exists": False, "url": None, "id": None})

1013

1014 try:

1015 device = Device.objects.get(name=name)

1016 return JsonResponse(

1017 {

1018 "exists": True,

1019 "url": request.build_absolute_uri(device.get_absolute_url()),

1020 "id": device.pk,

1021 }

1022 )

1023 except Device.DoesNotExist:

1024 return JsonResponse({"exists": False, "url": None, "id": None})

1025 except Device.MultipleObjectsReturned:

1026 devices = Device.objects.filter(name=name)

1027 first = devices.first()

1028 return JsonResponse(

1029 {

1030 "exists": True,

1031 "url": request.build_absolute_uri(first.get_absolute_url()),

1032 "id": first.pk,

1033 "count": devices.count(),

1034 }

1035 )

1036

1037

1038# ---------------------------------------------------------------------------

1039# Source Resolutions list view (per profile)

1040# ---------------------------------------------------------------------------

1041

1042

1043class SourceResolutionListView(LoginRequiredMixin, View):

1044 """List all saved name-split resolutions for a profile."""

1045

1046 def get(self, request, profile_pk):

1047 """Render the list of saved source resolutions for the given profile."""

1048 from .models import SourceResolution

1049

1050 profile = get_object_or_404(ImportProfile, pk=profile_pk)

1051 resolutions = SourceResolution.objects.filter(profile=profile).order_by("source_id")

1052 return render(

1053 request,

1054 "netbox_data_import/source_resolution_list.html",

1055 {

1056 "profile": profile,

1057 "resolutions": resolutions,

1058 },

1059 )

1060

1061

1062class SourceResolutionDeleteView(LoginRequiredMixin, View):

1063 """Delete a saved source resolution."""

1064

1065 def get(self, request, pk):

1066 """Render the delete confirmation page for a source resolution."""

1067 from .models import SourceResolution

1068

1069 obj = get_object_or_404(SourceResolution, pk=pk)

1070 return render(

1071 request,

1072 "netbox_data_import/confirm_delete.html",

1073 {

1074 "object": obj,

1075 "return_url": reverse("plugins:netbox_data_import:source_resolution_list", args=[obj.profile_id]),

1076 },

1077 )

1078

1079 def post(self, request, pk):

1080 """Delete the source resolution and redirect to the profile's resolution list."""

1081 from .models import SourceResolution

1082

1083 obj = get_object_or_404(SourceResolution, pk=pk)

1084 profile_pk = obj.profile_id

1085 obj.delete()

1086 messages.success(request, "Resolution deleted.")

1087 return redirect(reverse("plugins:netbox_data_import:source_resolution_list", args=[profile_pk]))

1088

1089

1090# ---------------------------------------------------------------------------

1091# Quick-resolve views (inline fixes from preview page)

1092# ---------------------------------------------------------------------------

1093

1094

1095class QuickCreateManufacturerView(LoginRequiredMixin, View):

1096 """Immediately create a Manufacturer in NetBox from the preview page.

1097

1098 Redirects back to preview so the row changes from 'create' to a device action.

1099 """

1100

1101 def post(self, request):

1102 """Create the manufacturer in NetBox and redirect back to preview."""

1103 from dcim.models import Manufacturer

1104

1105 mfg_name = request.POST.get("mfg_name", "").strip()

1106 mfg_slug = request.POST.get("mfg_slug", "").strip()

1107 if not mfg_name or not mfg_slug:

1108 messages.error(request, "Manufacturer name and slug are required.")

1109 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1110 mfg, created = Manufacturer.objects.get_or_create(

1111 slug=mfg_slug,

1112 defaults={"name": mfg_name},

1113 )

1114 if created:

1115 messages.success(request, f"Manufacturer '{mfg.name}' created.")

1116 else:

1117 messages.info(request, f"Manufacturer '{mfg.name}' already existed.")

1118 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1119

1120

1121class QuickResolveManufacturerView(LoginRequiredMixin, View):

1122 """Save a ManufacturerMapping (source make → NetBox manufacturer slug) from the preview page.

1123

1124 Used when a source has inconsistent naming (e.g. 'Dell EMC' → 'dell').

1125 Redirects back to preview which re-runs with the mapping applied.

1126 """

1127

1128 def post(self, request):

1129 """Save the manufacturer mapping and redirect back to preview."""

1130 profile_id = request.POST.get("profile_id")

1131 profile = get_object_or_404(ImportProfile, pk=profile_id)

1132 source_make = " ".join(request.POST.get("source_make", "").split())

1133 netbox_mfg_slug = request.POST.get("netbox_mfg_slug", "").strip()

1134 if not source_make or not netbox_mfg_slug:

1135 messages.error(request, "Source make and NetBox manufacturer slug are required.")

1136 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1137 _, created = ManufacturerMapping.objects.update_or_create(

1138 profile=profile,

1139 source_make=source_make,

1140 defaults={"netbox_manufacturer_slug": netbox_mfg_slug},

1141 )

1142 verb = "Created" if created else "Updated"

1143 messages.success(request, f"{verb} manufacturer mapping: '{source_make}' → {netbox_mfg_slug}")

1144 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1145

1146

1147class QuickResolveDeviceTypeView(LoginRequiredMixin, View):

1148 """Save a DeviceTypeMapping (source make/model → NetBox slugs) from the preview page.

1149

1150 Optionally also creates the manufacturer and/or device type in NetBox right now.

1151 Redirects back to preview which re-runs and shows the resolved rows.

1152 """

1153

1154 def post(self, request):

1155 """Save the device type mapping (and optionally create objects) then redirect."""

1156 from dcim.models import Manufacturer, DeviceType

1157 from django.utils.text import slugify

1158

1159 profile_id = request.POST.get("profile_id")

1160 profile = get_object_or_404(ImportProfile, pk=profile_id)

1161 source_make = " ".join(request.POST.get("source_make", "").split())

1162 source_model = " ".join(request.POST.get("source_model", "").split())

1163 netbox_mfg_slug = request.POST.get("netbox_mfg_slug", "").strip()

1164 netbox_dt_slug = request.POST.get("netbox_dt_slug", "").strip()

1165 action = request.POST.get("action", "map") # "map" or "create_now"

1166

1167 if not source_make or not source_model:

1168 messages.error(request, "Source make and model are required.")

1169 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1170

1171 if not netbox_mfg_slug:

1172 netbox_mfg_slug = slugify(source_make)

1173 if not netbox_dt_slug:

1174 netbox_dt_slug = slugify(source_model)

1175

1176 # Save/update DeviceTypeMapping

1177 dtm, created = DeviceTypeMapping.objects.update_or_create(

1178 profile=profile,

1179 source_make=source_make,

1180 source_model=source_model,

1181 defaults={

1182 "netbox_manufacturer_slug": netbox_mfg_slug,

1183 "netbox_device_type_slug": netbox_dt_slug,

1184 },

1185 )

1186

1187 if action == "create_now":

1188 mfg, _ = Manufacturer.objects.get_or_create(

1189 slug=netbox_mfg_slug,

1190 defaults={"name": source_make},

1191 )

1192 dt_name = request.POST.get("netbox_dt_name", source_model).strip() or source_model

1193 try:

1194 u_height = max(1, int(request.POST.get("u_height", "1")))

1195 except ValueError:

1196 u_height = 1

1197 DeviceType.objects.get_or_create(

1198 manufacturer=mfg,

1199 slug=netbox_dt_slug,

1200 defaults={"model": dt_name, "u_height": u_height},

1201 )

1202 messages.success(

1203 request, f"Mapping saved and device type '{source_make} / {source_model}' created in NetBox."

1204 )

1205 else:

1206 verb = "created" if created else "updated"

1207 messages.success(

1208 request,

1209 f"DeviceType mapping {verb}: '{source_make} / {source_model}' → {netbox_mfg_slug}/{netbox_dt_slug}",

1210 )

1211

1212 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1213

1214

1215class QuickAddClassRoleMappingView(LoginRequiredMixin, View):

1216 """Quickly add a ClassRoleMapping (ignore / role) directly from an error row in preview.

1217

1218 Redirects back to preview; error rows for that class disappear on re-run.

1219 """

1220

1221 def post(self, request):

1222 """Save the class→role mapping and redirect back to preview."""

1223 profile_id = request.POST.get("profile_id")

1224 profile = get_object_or_404(ImportProfile, pk=profile_id)

1225 source_class = request.POST.get("source_class", "").strip()

1226 mapping_action = request.POST.get("mapping_action", "ignore") # "ignore" or "role"

1227 role_slug = request.POST.get("role_slug", "").strip()

1228 creates_rack = request.POST.get("creates_rack") == "1"

1229

1230 if not source_class:

1231 messages.error(request, "Source class is required.")

1232 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1233

1234 _, created = ClassRoleMapping.objects.update_or_create(

1235 profile=profile,

1236 source_class=source_class,

1237 defaults={

1238 "ignore": mapping_action == "ignore",

1239 "creates_rack": creates_rack,

1240 "role_slug": role_slug if mapping_action == "role" else "",

1241 },

1242 )

1243 verb = "Created" if created else "Updated"

1244 if mapping_action == "ignore":

1245 action_label = "ignore"

1246 elif mapping_action == "rack":

1247 action_label = "creates rack"

1248 else:

1249 action_label = f"role '{role_slug}'"

1250 messages.success(request, f"{verb} mapping: class '{source_class}' → {action_label}")

1251 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1252

1253

1254class MatchExistingDeviceView(LoginRequiredMixin, View):

1255 """Link a source row to an existing NetBox device (by device ID).

1256

1257 Saves a DeviceExistingMatch; on next preview re-run the row shows action='update'.

1258 """

1259

1260 def post(self, request):

1261 """Save the device match and redirect back to preview."""

1262 from dcim.models import Device

1263

1264 profile_id = request.POST.get("profile_id")

1265 profile = get_object_or_404(ImportProfile, pk=profile_id)

1266 source_id = request.POST.get("source_id", "").strip()

1267 netbox_device_id = request.POST.get("netbox_device_id", "").strip()

1268

1269 if not source_id or not netbox_device_id:

1270 messages.error(request, "source_id and netbox_device_id are required.")

1271 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1272

1273 try:

1274 device = Device.objects.get(pk=int(netbox_device_id))

1275 except (Device.DoesNotExist, ValueError):

1276 messages.error(request, f"Device #{netbox_device_id} not found.")

1277 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1278

1279 DeviceExistingMatch.objects.update_or_create(

1280 profile=profile,

1281 source_id=source_id,

1282 defaults={

1283 "netbox_device_id": device.pk,

1284 "device_name": device.name,

1285 "source_asset_tag": request.POST.get("source_asset_tag", "").strip(),

1286 },

1287 )

1288 messages.success(request, f"Source '{source_id}' linked to existing device '{device.name}'.")

1289 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1290

1291

1292class SearchNetBoxObjectsView(LoginRequiredMixin, View):

1293 """AJAX search endpoint for NetBox objects used in preview quick-fix modals.

1294

1295 GET params: type (manufacturer|device_type|device|role), q (search string).

1296 Returns JSON list of {id, name, slug, url} dicts.

1297 """

1298

1299 def get(self, request):

1300 """Return a JSON list of matching NetBox objects for the given type and query."""

1301 from django.http import JsonResponse

1302 from dcim.models import Manufacturer, DeviceType, Device, DeviceRole

1303

1304 obj_type = request.GET.get("type", "device")

1305 q = request.GET.get("q", "").strip()

1306 limit = 20

1307

1308 if not q:

1309 return JsonResponse({"results": []})

1310

1311 results = []

1312 if obj_type == "manufacturer":

1313 for mfg in Manufacturer.objects.filter(name__icontains=q)[:limit]:

1314 results.append(

1315 {

1316 "id": mfg.pk,

1317 "name": mfg.name,

1318 "slug": mfg.slug,

1319 "url": request.build_absolute_uri(mfg.get_absolute_url()),

1320 }

1321 )

1322 elif obj_type == "device_type":

1323 mfg_filter = request.GET.get("mfg_slug", "")

1324 qs = DeviceType.objects.select_related("manufacturer")

1325 if mfg_filter:

1326 qs = qs.filter(manufacturer__slug=mfg_filter)

1327 for dt in qs.filter(model__icontains=q)[:limit]:

1328 results.append(

1329 {

1330 "id": dt.pk,

1331 "name": f"{dt.manufacturer.name} / {dt.model}",

1332 "slug": dt.slug,

1333 "mfg_slug": dt.manufacturer.slug,

1334 "url": request.build_absolute_uri(dt.get_absolute_url()),

1335 }

1336 )

1337 elif obj_type == "device":

1338 for dev in Device.objects.filter(name__icontains=q).select_related("site")[:limit]:

1339 results.append(

1340 {

1341 "id": dev.pk,

1342 "name": dev.name,

1343 "site": dev.site.name if dev.site else "",

1344 "url": request.build_absolute_uri(dev.get_absolute_url()),

1345 }

1346 )

1347 elif obj_type == "role":

1348 for role in DeviceRole.objects.filter(name__icontains=q)[:limit]:

1349 results.append(

1350 {

1351 "id": role.pk,

1352 "name": role.name,

1353 "slug": role.slug,

1354 "url": request.build_absolute_uri(role.get_absolute_url()),

1355 }

1356 )

1357

1358 return JsonResponse({"results": results})

1359

1360

1361def _auto_match_single_device(device_model, device_name, serial, asset_tag):

1362 """Try to match a single device row to an existing NetBox device.

1363

1364 Returns (device_or_None, is_ambiguous). Matching priority: serial →

1365 asset_tag → exact name. Multiple matches on any field → ambiguous.

1366 """

1367 device = None

1368 if serial:

1369 results = list(device_model.objects.filter(serial=serial)[:2])

1370 if len(results) == 1:

1371 device = results[0]

1372 elif len(results) > 1:

1373 return None, True

1374

1375 if device is None and asset_tag:

1376 results = list(device_model.objects.filter(asset_tag=asset_tag)[:2])

1377 if len(results) == 1:

1378 device = results[0]

1379 elif len(results) > 1:

1380 return None, True

1381

1382 if device is None and device_name:

1383 results = list(device_model.objects.filter(name=device_name)[:2])

1384 if len(results) == 1:

1385 device = results[0]

1386 elif len(results) > 1:

1387 return None, True

1388

1389 return device, False

1390

1391

1392class AutoMatchDevicesView(LoginRequiredMixin, View):

1393 """Scan all device rows in the session and auto-match to existing NetBox devices.

1394

1395 Priority: serial > asset_tag > exact name match.

1396 Name substring matches are recorded as probable_matches only (not auto-linked).

1397 """

1398

1399 def post(self, request):

1400 """Run auto-matching and redirect back to preview with a summary message."""

1401 from dcim.models import Device

1402

1403 profile_id = request.POST.get("profile_id")

1404 profile = get_object_or_404(ImportProfile, pk=profile_id)

1405 rows = request.session.get("import_rows", [])

1406

1407 matched = 0

1408 ambiguous = 0

1409 already = 0

1410 probable = 0

1411

1412 for row in rows:

1413 source_id = str(row.get("source_id", "")).strip()

1414 device_name = str(row.get("device_name", "")).strip()

1415 serial = str(row.get("serial", "")).strip()

1416 asset_tag = str(row.get("asset_tag", "")).strip()

1417 if not source_id:

1418 continue

1419 if profile.device_matches.filter(source_id=source_id).exists():

1420 already += 1

1421 continue

1422

1423 device, is_ambiguous = _auto_match_single_device(Device, device_name, serial, asset_tag)

1424 if is_ambiguous:

1425 ambiguous += 1

1426 continue

1427

1428 if device is not None:

1429 DeviceExistingMatch.objects.create(

1430 profile=profile,

1431 source_id=source_id,

1432 netbox_device_id=device.pk,

1433 device_name=device.name,

1434 source_asset_tag=asset_tag,

1435 )

1436 matched += 1

1437 elif device_name:

1438 # Substring name match → probable only (no auto-link)

1439 short_name = device_name.split(" - ")[-1].strip() if " - " in device_name else device_name

1440 if Device.objects.filter(name__icontains=short_name).exists():

1441 probable += 1

1442

1443 msg_parts = []

1444 if matched:

1445 msg_parts.append(f"{matched} auto-matched (serial/asset_tag/name)")

1446 if probable:

1447 msg_parts.append(f"{probable} probable name match(es) — use Link button to confirm")

1448 if ambiguous:

1449 msg_parts.append(f"{ambiguous} ambiguous (multiple devices)")

1450 if already:

1451 msg_parts.append(f"{already} already matched")

1452 messages.success(request, f"Auto-match: {', '.join(msg_parts) or 'nothing found'}.")

1453 return redirect(reverse("plugins:netbox_data_import:import_preview"))

1454

1455

1456# ---------------------------------------------------------------------------

1457# Session helpers

1458# ---------------------------------------------------------------------------

1459

1460

1461def _serialize_rows(rows: list) -> list:

1462 """Convert parsed rows to JSON-serializable format (handle Excel datetime values)."""

1463 import datetime

1464

1465 safe_rows = []

1466 for row in rows:

1467 safe_row = {}

1468 for k, v in row.items():

1469 if isinstance(v, (datetime.datetime, datetime.date)):

1470 safe_row[k] = v.isoformat()

1471 else:

1472 safe_row[k] = v

1473 safe_rows.append(safe_row)

1474 return safe_rows

Coverage for / home / runner / work / netbox-data-import-plugin / netbox-data-import-plugin / netbox-data-import-plugin / netbox_data_import / views.py: 98%

657 statements