"""Replay synthetic AuditLog rows across the demo dataset. Runs LAST in the seed_all chain — by then every other app has populated its Application / Award / Disbursement / Course / Lesson rows, so the audit trail can backreference real objects via target_app + target_model + object_id. Uses ``bulk_create`` with explicit ``timestamp=`` to backdate rows across the last 90 days. Idempotent: re-running fills in any deficit up to the target count. """ from __future__ import annotations import random from datetime import timedelta from django.contrib.auth import get_user_model from django.core.management.base import BaseCommand from django.utils import timezone from apps.core.audit.models import AuditLog from apps.core.seeders.volumes import RECORD_COUNTS User = get_user_model() _ACTIONS = ["create", "update", "delete", "access", "publish", "approve", "submit", "reject"] _USER_AGENTS = [ "Mozilla/5.0 (X11; Linux x86_64) Gecko/20100101 Firefox/123.0", "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_3) AppleWebKit/605.1.15", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/124.0", "Mozilla/5.0 (iPad; CPU OS 17_4) Safari/604.1", "RUFORUM-IILMP-Mobile/1.4 (iOS 17)", ] _PATHS = [ "/admin/", "/rims/grants/", "/rims/applications/", "/rims/finance/disbursements/", "/rep/courses/", "/rep/learner/", "/repository/", "/alumni/directory/", "/smehub/", "/mel/dashboard/", "/notifications/", ] def _candidate_targets() -> list[tuple[str, str, str]]: """Snapshot (target_app, target_model, object_id) tuples from already-seeded rows.""" targets: list[tuple[str, str, str]] = [] pull_specs = [ ("apps.rims.grants.models", "GrantCall", "grants"), ("apps.rims.grants.models", "Application", "grants"), ("apps.rims.grants.models", "Award", "grants"), ("apps.rims.finance.models", "DisbursementRequest", "finance"), ("apps.repository.documents.models", "Document", "documents"), ("apps.alumni.profiles.models", "AlumniProfile", "profiles"), ] for module_path, model_name, app_label in pull_specs: try: mod = __import__(module_path, fromlist=[model_name]) model = getattr(mod, model_name) for pk in model.objects.values_list("pk", flat=True)[:200]: targets.append((app_label, model_name, str(pk))) except Exception: # noqa: BLE001 — seeder must keep going if any module is missing continue return targets class Command(BaseCommand): help = "Seed synthetic AuditLog rows that backreference real seeded objects." def add_arguments(self, parser): parser.add_argument( "--volume", choices=("minimal", "demo", "heavy"), default="demo", ) parser.add_argument( "--reset", action="store_true", help="Delete all existing AuditLog rows before seeding.", ) def handle(self, *args, **opts): tier = opts["volume"] target = RECORD_COUNTS[tier]["audit_logs"] random.seed(2026) if opts["reset"]: removed = AuditLog.objects.count() AuditLog.objects.all().delete() self.stdout.write(self.style.WARNING(f"Removed {removed} AuditLog rows.")) existing = AuditLog.objects.count() if existing >= target: self.stdout.write(self.style.SUCCESS( f"AuditLog: already have {existing} rows (target {target}). Skipping." )) return actors = list(User.objects.filter(is_active=True).values_list("pk", flat=True)[:200]) if not actors: self.stdout.write(self.style.WARNING("No users — cannot seed AuditLog.")) return targets = _candidate_targets() # Fall back to placeholder targets if no models seeded yet. if not targets: targets = [("core", "system", str(n)) for n in range(1, 50)] now = timezone.now() to_create = target - existing # We can't rely on bulk_create() to preserve our explicit `timestamp=` # because the field has auto_now_add=True (Django's pre_save overrides). # Strategy: bulk_create with default timestamp, capture returned PKs, # then issue one UPDATE per row to set the backdated timestamp. This is # the cleanest way to populate a realistic audit history. created_pks: list[tuple[int, "timezone.timedelta"]] = [] batch: list[AuditLog] = [] def _flush(batch_rows: list[AuditLog]) -> None: saved = AuditLog.objects.bulk_create(batch_rows, batch_size=500) for obj in saved: created_pks.append((obj.pk, obj._seed_backdate)) for i in range(to_create): actor_pk = random.choice(actors) target_app, target_model, object_id = random.choice(targets) action = random.choice(_ACTIONS) backdate = timedelta( days=random.randint(0, 90), hours=random.randint(0, 23), minutes=random.randint(0, 59), ) row = AuditLog( actor_id=actor_pk, action=action, target_app=target_app, target_model=target_model, object_id=object_id, object_repr=f"{target_model}#{object_id}", changes={"summary": f"Seeded {action} on {target_model}"} if action == "update" else None, ip_address=f"10.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 254)}", user_agent=random.choice(_USER_AGENTS), path=random.choice(_PATHS), method=random.choice(["GET", "POST", "PUT", "PATCH", "DELETE"]), ) # Attach the desired backdate so the post-create UPDATE can use it. row._seed_backdate = backdate batch.append(row) if len(batch) >= 500: _flush(batch) batch = [] if batch: _flush(batch) # Now backdate. We do this in chunks of 200 with a CASE WHEN expression # to keep round-trips low (≈25 queries for 5000 rows). from django.db.models import Case, When, Value, DateTimeField for offset in range(0, len(created_pks), 200): chunk = created_pks[offset:offset + 200] cases = Case( *[When(pk=pk, then=Value(now - delta)) for pk, delta in chunk], output_field=DateTimeField(), ) AuditLog.objects.filter(pk__in=[pk for pk, _ in chunk]).update(timestamp=cases) total = AuditLog.objects.count() self.stdout.write(self.style.SUCCESS( f"AuditLog seed: created={to_create} total={total} (tier={tier})" ))