60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
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))
|