feat: grant transition plan to existing users

apps/licensing/management/commands/grant_existing_users_plan.py

@@ -0,0 +1,59 @@
+from django.core.management.base import BaseCommand, CommandError
+
+from apps.licensing.models import SoftwarePlan
+from apps.licensing.services import LicensingError, grant_plan_to_existing_users
+
+
+class Command(BaseCommand):
+    help = "Preview or grant one active software plan to existing active non-staff users."
+
+    def add_arguments(self, parser):
+        parser.add_argument("--plan-id", type=int, required=True, help="Exact SoftwarePlan ID.")
+        parser.add_argument(
+            "--reason",
+            required=True,
+            help="Non-empty audit reason written to every grant event.",
+        )
+        parser.add_argument(
+            "--execute",
+            action="store_true",
+            help="Apply grants. Without this flag the command is read-only.",
+        )
+        parser.add_argument(
+            "--expected-grant-count",
+            type=int,
+            help="Required with --execute and must match the current preview count.",
+        )
+
+    def handle(self, *args, **options):
+        execute = options["execute"]
+        expected_grant_count = options["expected_grant_count"]
+        if execute and expected_grant_count is None:
+            raise CommandError("--execute 必须同时提供 --expected-grant-count")
+        if not execute and expected_grant_count is not None:
+            raise CommandError("预演模式不能提供 --expected-grant-count")
+
+        try:
+            plan = SoftwarePlan.objects.get(pk=options["plan_id"])
+        except SoftwarePlan.DoesNotExist as exc:
+            raise CommandError("指定套餐不存在") from exc
+
+        try:
+            result = grant_plan_to_existing_users(
+                plan=plan,
+                reason=options["reason"],
+                execute=execute,
+                expected_grant_count=expected_grant_count,
+            )
+        except LicensingError as exc:
+            raise CommandError(exc.message) from exc
+
+        mode = "EXECUTED" if result.executed else "DRY-RUN"
+        summary = (
+            f"mode={mode} plan_id={plan.pk} product={plan.product_code} "
+            f"eligible={result.eligible_count} "
+            f"skipped_existing={result.skipped_existing_count} "
+            f"grant_count={result.grant_count}"
+        )
+        style = self.style.SUCCESS if result.executed else self.style.WARNING
+        self.stdout.write(style(summary))

apps/licensing/services.py

@@ -7,6 +7,7 @@ from decimal import Decimal
 from decimal import InvalidOperation
 
 from django.conf import settings
+from django.contrib.auth import get_user_model
 from django.db import IntegrityError, transaction
 from django.utils import timezone
 
@@ -74,6 +75,14 @@ class SoftwarePaymentResult:
     applied: bool
 
 
+@dataclass(frozen=True)
+class BulkEntitlementGrantResult:
+    eligible_count: int
+    skipped_existing_count: int
+    grant_count: int
+    executed: bool
+
+
 @dataclass(frozen=True)
 class AuthorizationDecision:
     product_code: str
@@ -344,6 +353,76 @@ def grant_software_entitlement(*, user, plan: SoftwarePlan, reason: str, actor=N
     return entitlement
 
 
+@transaction.atomic
+def grant_plan_to_existing_users(
+    *,
+    plan: SoftwarePlan,
+    reason: str,
+    execute: bool = False,
+    expected_grant_count: int | None = None,
+    actor=None,
+    now=None,
+) -> BulkEntitlementGrantResult:
+    reason = _required_reason(reason)
+    if execute and expected_grant_count is None:
+        raise LicensingError("expected_grant_count_required", "实际执行必须提供预期授予人数")
+    if expected_grant_count is not None and expected_grant_count < 0:
+        raise LicensingError("invalid_expected_grant_count", "预期授予人数不能为负数")
+
+    plan_query = SoftwarePlan.objects
+    if execute:
+        plan_query = plan_query.select_for_update()
+    locked_plan = plan_query.get(pk=plan.pk)
+    if locked_plan.status != SoftwarePlan.Status.ACTIVE:
+        raise LicensingError("plan_inactive", "套餐已停用，不能批量授予权益")
+
+    users_query = get_user_model().objects.filter(
+        is_active=True,
+        is_staff=False,
+        is_superuser=False,
+    ).order_by("pk")
+    if execute:
+        users_query = users_query.select_for_update()
+    eligible_users = list(users_query)
+    eligible_user_ids = [user.pk for user in eligible_users]
+    effective_now = now or timezone.now()
+    existing_user_ids = set(
+        SoftwareEntitlement.objects.filter(
+            user_id__in=eligible_user_ids,
+            product_code=locked_plan.product_code,
+            status=SoftwareEntitlement.Status.ACTIVE,
+            grace_expires_at__gt=effective_now,
+        ).values_list("user_id", flat=True)
+    )
+    users_to_grant = [
+        user for user in eligible_users if user.pk not in existing_user_ids
+    ]
+    grant_count = len(users_to_grant)
+
+    if execute and expected_grant_count != grant_count:
+        raise LicensingError(
+            "grant_count_mismatch",
+            f"预期授予 {expected_grant_count} 人，当前实际应授予 {grant_count} 人",
+        )
+
+    if execute:
+        for user in users_to_grant:
+            grant_software_entitlement(
+                user=user,
+                plan=locked_plan,
+                reason=reason,
+                actor=actor,
+                starts_at=effective_now,
+            )
+
+    return BulkEntitlementGrantResult(
+        eligible_count=len(eligible_users),
+        skipped_existing_count=len(existing_user_ids),
+        grant_count=grant_count,
+        executed=execute,
+    )
+
+
 @transaction.atomic
 def renew_software_entitlement(
     *,

apps/licensing/tests.py

@@ -1,10 +1,13 @@
 from datetime import timedelta
 from decimal import Decimal
 from concurrent.futures import ThreadPoolExecutor
+from io import StringIO
 from types import SimpleNamespace
 from unittest.mock import patch
 
 from django.contrib import admin
+from django.core.management import call_command
+from django.core.management.base import CommandError
 from django.db import close_old_connections
 from django.test import SimpleTestCase, TestCase, TransactionTestCase, override_settings
 from django.urls import reverse
@@ -40,6 +43,7 @@ from apps.licensing.services import (
     evaluate_device_authorization,
     evaluate_subscription_access,
     get_subscription_mode,
+    grant_plan_to_existing_users,
     grant_software_entitlement,
     record_device_heartbeat,
     release_license_seat,
@@ -51,6 +55,171 @@ from apps.licensing.services import (
 from apps.users.models import ApiKey, User, UserWallet
 
 
+class ExistingUserPlanGrantCommandTests(TestCase):
+    def setUp(self):
+        self.plan = SoftwarePlan.objects.create(
+            product_code=ClientDevice.ProductCode.CMSHOPEE,
+            name="测试",
+            duration_days=30,
+            price=Decimal("99.00"),
+            device_limit=1,
+            grace_days=2,
+        )
+        self.user_one = User.objects.create_user(
+            username="existing-one",
+            email="existing-one@example.com",
+            password="password",
+        )
+        self.user_two = User.objects.create_user(
+            username="existing-two",
+            email="existing-two@example.com",
+            password="password",
+        )
+        self.staff_user = User.objects.create_user(
+            username="staff-user",
+            email="staff-user@example.com",
+            password="password",
+            is_staff=True,
+        )
+        self.inactive_user = User.objects.create_user(
+            username="inactive-user",
+            email="inactive-user@example.com",
+            password="password",
+            is_active=False,
+        )
+
+    def command_options(self, **overrides):
+        options = {
+            "plan_id": self.plan.pk,
+            "reason": "T-633 存量用户测试套餐过渡",
+        }
+        options.update(overrides)
+        return options
+
+    def test_dry_run_is_read_only_and_reports_scope(self):
+        stdout = StringIO()
+
+        call_command("grant_existing_users_plan", stdout=stdout, **self.command_options())
+
+        self.assertIn("mode=DRY-RUN", stdout.getvalue())
+        self.assertIn("eligible=2", stdout.getvalue())
+        self.assertIn("skipped_existing=0", stdout.getvalue())
+        self.assertIn("grant_count=2", stdout.getvalue())
+        self.assertFalse(SoftwareEntitlement.objects.exists())
+        self.assertFalse(LicenseSeat.objects.exists())
+        self.assertFalse(LicenseEvent.objects.exists())
+
+    def test_execute_grants_only_active_nonstaff_users_with_audit_events(self):
+        before = timezone.now()
+
+        call_command(
+            "grant_existing_users_plan",
+            execute=True,
+            expected_grant_count=2,
+            **self.command_options(),
+        )
+
+        entitlements = SoftwareEntitlement.objects.order_by("user_id")
+        self.assertEqual(entitlements.count(), 2)
+        self.assertSetEqual(
+            set(entitlements.values_list("user_id", flat=True)),
+            {self.user_one.pk, self.user_two.pk},
+        )
+        entitlement = entitlements.first()
+        self.assertEqual(entitlement.source_plan, self.plan)
+        self.assertEqual(entitlement.plan_name, "测试")
+        self.assertEqual(entitlement.plan_duration_days, 30)
+        self.assertEqual(entitlement.plan_grace_days, 2)
+        self.assertGreaterEqual(entitlement.starts_at, before)
+        self.assertEqual(
+            entitlement.expires_at,
+            entitlement.starts_at + timedelta(days=30),
+        )
+        self.assertEqual(
+            entitlement.grace_expires_at,
+            entitlement.expires_at + timedelta(days=2),
+        )
+        self.assertEqual(LicenseSeat.objects.count(), 2)
+        self.assertEqual(
+            LicenseEvent.objects.filter(
+                action=LicenseEvent.Action.GRANTED,
+                reason="T-633 存量用户测试套餐过渡",
+            ).count(),
+            2,
+        )
+
+    def test_existing_usable_product_entitlement_is_skipped_and_rerun_is_idempotent(self):
+        formal_plan = SoftwarePlan.objects.create(
+            product_code=ClientDevice.ProductCode.CMSHOPEE,
+            name="正式会员",
+            duration_days=365,
+            price=Decimal("999.00"),
+            device_limit=1,
+            grace_days=7,
+        )
+        grant_software_entitlement(
+            user=self.user_one,
+            plan=formal_plan,
+            reason="已有正式权益",
+        )
+
+        call_command(
+            "grant_existing_users_plan",
+            execute=True,
+            expected_grant_count=1,
+            **self.command_options(),
+        )
+        result = grant_plan_to_existing_users(
+            plan=self.plan,
+            reason="重复预演",
+        )
+
+        self.assertEqual(SoftwareEntitlement.objects.count(), 2)
+        self.assertEqual(result.eligible_count, 2)
+        self.assertEqual(result.skipped_existing_count, 2)
+        self.assertEqual(result.grant_count, 0)
+        self.assertFalse(result.executed)
+
+    def test_execute_requires_matching_preview_count(self):
+        with self.assertRaisesRegex(CommandError, "当前实际应授予 2 人"):
+            call_command(
+                "grant_existing_users_plan",
+                execute=True,
+                expected_grant_count=1,
+                **self.command_options(),
+            )
+
+        self.assertFalse(SoftwareEntitlement.objects.exists())
+        self.assertFalse(LicenseEvent.objects.exists())
+
+    def test_failure_rolls_back_entire_batch(self):
+        original_grant = grant_software_entitlement
+        call_count = 0
+
+        def fail_second_grant(**kwargs):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 2:
+                raise RuntimeError("simulated grant failure")
+            return original_grant(**kwargs)
+
+        with patch(
+            "apps.licensing.services.grant_software_entitlement",
+            side_effect=fail_second_grant,
+        ):
+            with self.assertRaisesRegex(RuntimeError, "simulated grant failure"):
+                call_command(
+                    "grant_existing_users_plan",
+                    execute=True,
+                    expected_grant_count=2,
+                    **self.command_options(),
+                )
+
+        self.assertFalse(SoftwareEntitlement.objects.exists())
+        self.assertFalse(LicenseSeat.objects.exists())
+        self.assertFalse(LicenseEvent.objects.exists())
+
+
 class SubscriptionModeUnitTests(SimpleTestCase):
     @override_settings(
         CMSHOPEE_SUBSCRIPTION_MODE="",