"""Small opt-in development fixture. Never repairs or overwrites changed records.""" from datetime import date, datetime, timezone from uuid import uuid5, NAMESPACE_URL from django.conf import settings from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.core.exceptions import ValidationError from django.db import connection, transaction from apps.organization.models import Company, Region, ServiceCenter, Department, UserOrganizationAssignment from apps.organization.assignment_services import create_assignment from apps.access.models import Role, UserRoleAssignment from apps.access.services import set_role_permissions, create_role_assignment from apps.catalog.models import Brand, ProductCategory, ProductModel, DeviceIdentificationPolicy from apps.catalog.identification import set_identification_policy from apps.service_catalog.models import ComplaintSymptom, FaultDiagnosis, RepairAction from apps.service_catalog import services as taxonomy from apps.customers.models import Customer from apps.customers.services import create_customer from apps.devices.models import DeviceIdentifier from apps.devices.services import register_device, assign_device_owner from apps.parts.models import PartCategory, SparePart from apps.parts import services as parts from apps.inventory.models import InventoryLocation from apps.inventory.services import create_location from apps.service.models import ServiceCase from apps.service.services import create_service_case, add_service_case_complaint from apps.communications.models import NotificationTemplate from apps.communications.services import save_template from apps.sla.models import SlaPolicy from apps.sla.services import save_policy, enroll CODE="CCARE-DEMO" USERNAME="ccare-demo-operator" MARKER="C-CARE development fixture v1; never production data." PERMISSIONS=( "customers.view_customer","customers.add_customer","devices.view_device", "service.view_servicecase","service.add_servicecase","service.handle_servicecase", "parts.view_sparepart","inventory.view_stock","inventory.manage_inventory", "commercial.view_servicequotation","commercial.view_serviceinvoice","commercial.view_servicepayment", "frontdesk.view_appointment","frontdesk.view_queue","frontdesk.manage_slots","frontdesk.create_appointment", "frontdesk.manage_queue","frontdesk.check_in_appointment","frontdesk.cancel_appointment", "communications.view_notification","communications.manage_templates", "sla.view_sla","sla.monitor_sla","sla.manage_slapolicy","reporting.view_operational_dashboard", ) def identifier(name): return uuid5(NAMESPACE_URL,"https://ccare.invalid/development-fixture/v1/"+name) def require_development(): if getattr(settings,"SETTINGS_MODULE","")!="config.settings.development" or not settings.DEBUG: raise ValidationError("Demo seeding requires config.settings.development with DEBUG=True.") database=connection.settings_dict if database["ENGINE"]!="django.db.backends.postgresql" or database.get("HOST") not in ("localhost","127.0.0.1","::1"): raise ValidationError("Demo seeding requires a local PostgreSQL database.") def _verify(row,expected): if any(getattr(row,key)!=value for key,value in expected.items()): raise ValidationError("A reserved demo record was changed or collides with other data; nothing is overwritten.") return row def _master(model,key,**fields): expected={"id":identifier(model._meta.label_lower),**key,**fields} row=model.objects.filter(**key).first() if row: return _verify(row,expected) return model.objects.create(**expected) def _service_record(model,key,expected,create,*,initial=False): matches=list(model.objects.filter(**key)[:2]) if len(matches)>1 or (matches and initial): raise ValidationError("Reserved demo identity already exists; seed will not adopt unrelated records.") return _verify(matches[0],expected) if matches else create() def seed_demo(*,password=None): require_development() with transaction.atomic(): # Serialize seed runs without adding a bootstrap/marker table. Ordinary # writes still use each domain's own validation and locking services. with connection.cursor() as cursor: cursor.execute("SELECT pg_advisory_xact_lock(%s)",[4846800275428132173]) initial=not Company.objects.filter(pk=identifier("organization.company")).exists() company=_master(Company,{"code":CODE},name="C-CARE Demo Company",legal_name=MARKER,is_active=True) region=_master(Region,{"company":company,"code":CODE},name="Demo Region",is_active=True) center=_master(ServiceCenter,{"company":company,"code":CODE},region=region,name="Demo Service Center",center_type="OWN",is_active=True) _master(Department,{"company":company,"code":CODE},name="Demo Operations",is_active=True) User=get_user_model() user=User.objects.filter(username=USERNAME).first() if user: _verify(user,dict(id=identifier("operator"),is_active=True,is_staff=False,is_superuser=False)) else: user=User.objects.create_user(id=identifier("operator"),username=USERNAME,password=None,is_staff=False,is_superuser=False) if password is not None: from django.contrib.auth.password_validation import validate_password validate_password(password,user) user.set_password(password) user.save(update_fields=["password"]) role=_master(Role,{"code":CODE},name="Demo operational capabilities",is_active=True) permissions=[] for value in PERMISSIONS: app,code=value.split(".") permissions.append(Permission.objects.get(content_type__app_label=app,codename=code)) if initial: set_role_permissions(role=role,permissions=permissions) elif set(role.permissions.values_list("pk",flat=True))!={p.pk for p in permissions}: raise ValidationError("Demo role permissions changed; they will not be reset.") assignment=_service_record(UserOrganizationAssignment,{"user":user,"company":company}, dict(region_id=None,service_center_id=None,department_id=None,is_active=True,is_primary=True), lambda:create_assignment(user=user,company=company,is_primary=True)) _service_record(UserRoleAssignment,{"user":user,"role":role},dict(organization_assignment_id=assignment.pk,is_active=True), lambda:create_role_assignment(user=user,role=role,organization_assignment=assignment)) brand=_master(Brand,{"code":CODE},name="Demo Brand",is_active=True) category=_master(ProductCategory,{"code":CODE},name="Demo Devices",is_active=True) model=_master(ProductModel,{"brand":brand,"code":CODE},category=category,name="Demo Model",is_active=True) policy=dict(imei1_requirement="NOT_APPLICABLE",imei2_requirement="NOT_APPLICABLE",serial_requirement="REQUIRED") _service_record(DeviceIdentificationPolicy,{"product_model":model},policy, lambda:set_identification_policy(product_model=model,**policy)) complaint=None for cls,name,operation,argument in [ (ComplaintSymptom,"Demo device will not power on",taxonomy.set_complaint_applicability,"complaint"), (FaultDiagnosis,"Demo power input fault",taxonomy.set_fault_diagnosis_applicability,"diagnosis"), (RepairAction,"Demo power connector replacement",taxonomy.set_repair_action_applicability,"repair_action")]: item=_master(cls,{"code":CODE},name=name,is_active=True,applies_to_all_product_categories=False) if initial: operation(**{argument:item},applies_to_all_product_categories=False,product_categories=[category]) elif set(item.category_mappings.values_list("product_category_id",flat=True))!={category.pk}: raise ValidationError("Demo applicability changed; it will not be replaced.") if cls is ComplaintSymptom: complaint=item part_category=_service_record(PartCategory,{"code":CODE},dict(name="Demo Parts",description=MARKER,is_active=True), lambda:parts.create_part_category(code=CODE,name="Demo Parts",description=MARKER),initial=initial) part=_service_record(SparePart,{"part_code":CODE},dict(name="Demo connector",category_id=part_category.pk, serialization_policy="NOT_SERIALIZED",description=MARKER,is_active=True),lambda:parts.create_spare_part( part_code=CODE,name="Demo connector",category=part_category,serialization_policy="NOT_SERIALIZED",description=MARKER),initial=initial) if initial: parts.set_spare_part_compatibility(spare_part=part,product_models=[model],product_variants=[]) else: from apps.parts.models import SparePartCompatibility mappings=list(SparePartCompatibility.objects.filter(spare_part=part).values_list("product_model_id","product_variant_id","is_active")) if mappings!=[(model.pk,None,True)]: raise ValidationError("Demo compatibility changed; it will not be replaced.") location=_service_record(InventoryLocation,{"company":company,"code":CODE}, dict(service_center_id=center.pk,name="Demo Store",location_type="STORE",is_active=True), lambda:create_location(actor=user,company=company,service_center=center,code=CODE,name="Demo Store",location_type="STORE")) customer=_service_record(Customer,{"company":company,"primary_email":"customer@ccare-demo.invalid"}, dict(full_name="Synthetic Demo Customer",customer_type="INDIVIDUAL",is_active=True), lambda:create_customer(company=company,customer_type="INDIVIDUAL",full_name="Synthetic Demo Customer",primary_email="customer@ccare-demo.invalid")) identity=DeviceIdentifier.objects.filter(identity_key="CCARE-DEMO-SERIAL-0001").select_related("device").first() if identity: _verify(identity,dict(is_active=True,identifier_type="SERIAL")) device=_verify(identity.device,dict(product_model_id=model.pk,product_variant_id=None,is_active=True)) else: device=register_device(product_model=model,serial="CCARE-DEMO-SERIAL-0001") assign_device_owner(device=device,customer=customer) case=_service_record(ServiceCase,{"company":company,"customer_reference":CODE}, dict(service_center_id=center.pk,customer_id=customer.pk,device_id=device.pk,created_by_id=user.pk), lambda:create_service_case(company=company,service_center=center,customer=customer,device=device,created_by=user, intake_channel="WALK_IN",received_at=datetime(2025,1,1,9,tzinfo=timezone.utc),customer_reference=CODE,intake_note=MARKER)) if initial: add_service_case_complaint(service_case=case,complaint_symptom=complaint) _service_record(NotificationTemplate,{"company":company,"code":"ccare-demo"}, dict(name="Demo status template",channel="EMAIL",body="Hello {customer_name}. Case {service_case_reference} has a recorded update.",is_active=True), lambda:save_template(actor=user,company=company,code="ccare-demo",name="Demo status template",channel="EMAIL", subject="Demo service update",body="Hello {customer_name}. Case {service_case_reference} has a recorded update.")) _service_record(SlaPolicy,{"company":company,"code":"ccare-demo"},dict(name="Demo acceptance to ready",target_minutes=1440, warning_minutes=60,is_active=True,service_center_id=center.pk),lambda:save_policy(actor=user,company=company, code="ccare-demo",name="Demo acceptance to ready",service_center=center,effective_from=date(2025,1,1),target_minutes=1440,warning_minutes=60)) enroll(actor=user,service_case=case) return dict(user=user,company=company,center=center,customer=customer,device=device,case=case,location=location,initial=initial)