Coverage for app/venv/lib/python3.14/site-packages/weblate/billing/models.py: 0%

3# SPDX-License-Identifier: GPL-3.0-or-later

5from __future__ import annotations

7import os.path

8from contextlib import suppress

9from datetime import timedelta

10from functools import partial

11from pathlib import Path

12from typing import ClassVar

14from appconf import AppConf

15from django.conf import settings

16from django.contrib import admin

17from django.core.exceptions import ValidationError

18from django.core.serializers.json import DjangoJSONEncoder

19from django.db import models, transaction

20from django.db.models import Prefetch, Q

21from django.db.models.signals import m2m_changed, post_delete, post_save, pre_delete

22from django.dispatch import receiver

23from django.urls import reverse

24from django.utils import timezone

25from django.utils.functional import cached_property

26from django.utils.html import format_html

27from django.utils.translation import gettext, gettext_lazy, ngettext

29from weblate.auth.models import User

30from weblate.trans.models import Alert, Component, Project, Translation

31from weblate.utils.decorators import disable_for_loaddata

32from weblate.utils.html import format_html_join_comma, list_to_tuples

33from weblate.utils.stats import prefetch_stats

36class LibreCheck:

37 def __init__(self, result, message, component=None) -> None:

38 self.result = result

39 self.message = message

40 self.component = component

42 def __bool__(self) -> bool:

43 return self.result

45 def __str__(self) -> str:

46 return self.message

49class PlanQuerySet(models.QuerySet["Plan"]):

50 def public(self, user=None):

51 """List of public paid plans which are available."""

52 base = self.exclude(Q(price=0) & Q(yearly_price=0))

53 result = base.filter(public=True)

54 if user:

55 result |= base.filter(

56 public=False, billing__in=Billing.objects.for_user(user)

57 )

58 return result.distinct().order_by("price")

61class Plan(models.Model):

62 name = models.CharField(max_length=100, unique=True)

63 slug = models.SlugField(max_length=100, unique=True)

64 price = models.IntegerField(default=0)

65 yearly_price = models.IntegerField(default=0)

66 limit_strings = models.IntegerField(default=0)

67 display_limit_strings = models.IntegerField(default=0)

68 limit_languages = models.IntegerField(default=0)

69 display_limit_languages = models.IntegerField(default=0)

70 limit_projects = models.IntegerField(default=0)

71 display_limit_projects = models.IntegerField(default=0)

72 limit_hosted_strings = models.IntegerField(default=0)

73 display_limit_hosted_strings = models.IntegerField(default=0)

74 change_access_control = models.BooleanField(default=True)

75 public = models.BooleanField(default=False)

77 objects = PlanQuerySet.as_manager()

79 class Meta:

80 verbose_name = "Billing plan"

81 verbose_name_plural = "Billing plans"

83 def __str__(self) -> str:

84 return self.name

86 @property

87 def vat_price(self):

88 return round(self.price * settings.VAT_RATE, 2)

90 @property

91 def vat_yearly_price(self):

92 return round(self.yearly_price * settings.VAT_RATE, 2)

94 @property

95 def is_free(self):

96 return self.price == 0 and self.yearly_price == 0

99class BillingManager(models.Manager["Billing"]):

100 def check_limits(self) -> None:

101 for bill in self.iterator():

102 bill.check_limits()

103

104

105class BillingQuerySet(models.QuerySet["Billing"]):

106 def get_out_of_limits(self):

107 return self.filter(in_limits=False)

108

109 def get_unpaid(self):

110 return self.filter(paid=False, state=Billing.STATE_ACTIVE)

111

112 def get_valid(self):

113 return self.filter(

114 Q(in_limits=True)

115 & (

116 (Q(state=Billing.STATE_ACTIVE) & Q(paid=True))

117 | Q(state=Billing.STATE_TRIAL)

118 )

119 )

120

121 def for_user(self, user: User):

122 if user.has_perm("billing.manage"):

123 return self.all().order_by("state")

124 return (

125 self.filter(

126 Q(projects__in=user.projects_with_perm("billing.view")) | Q(owners=user)

127 )

128 .distinct()

129 .order_by("state")

130 )

131

132 def for_user_within_limits(self, user: User):

133 """Return billings for the given user which are valid and within project creation limits."""

134 billings = self.get_valid().for_user(user).prefetch()

135 pks = set()

136 for billing in billings:

137 limit = billing.plan.display_limit_projects

138 if limit == 0 or billing.count_projects < limit:

139 pks.add(billing.pk)

140 return Billing.objects.filter(pk__in=pks).prefetch()

141

142 def prefetch(self):

143 return self.prefetch_related(

144 "owners",

145 "owners__profile",

146 "plan",

147 Prefetch(

148 "projects",

149 queryset=Project.objects.order(),

150 to_attr="ordered_projects",

151 ),

152 )

153

154 def active(self):

155 return self.filter(state__in=Billing.ACTIVE_STATES)

156

157

158class Billing(models.Model):

159 STATE_ACTIVE = 0

160 STATE_TRIAL = 1

161 STATE_TERMINATED = 3

162

163 EXPIRING_STATES: ClassVar[set[int]] = {STATE_TRIAL}

164 ACTIVE_STATES: ClassVar[set[int]] = {STATE_ACTIVE, STATE_TRIAL}

165

166 plan = models.ForeignKey(

167 Plan,

168 on_delete=models.deletion.CASCADE,

169 verbose_name=gettext_lazy("Billing plan"),

170 )

171 projects = models.ManyToManyField(

172 Project, blank=True, verbose_name=gettext_lazy("Billed projects")

173 )

174 owners = models.ManyToManyField(

175 User, blank=True, verbose_name=gettext_lazy("Billing owners")

176 )

177 state = models.IntegerField(

178 choices=(

179 (STATE_ACTIVE, gettext_lazy("Active")),

180 (STATE_TRIAL, gettext_lazy("Trial")),

181 (STATE_TERMINATED, gettext_lazy("Terminated")),

182 ),

183 default=STATE_ACTIVE,

184 verbose_name=gettext_lazy("Billing state"),

185 )

186 expiry = models.DateTimeField(

187 blank=True,

188 null=True,

189 verbose_name=gettext_lazy("Trial expiry date"),

190 help_text="After expiry removal with 15 days grace period is scheduled.",

191 )

192 removal = models.DateTimeField(

193 blank=True,

194 null=True,

195 verbose_name=gettext_lazy("Scheduled removal"),

196 help_text="This is automatically set after trial expiry.",

197 )

198 paid = models.BooleanField(

199 default=True, verbose_name=gettext_lazy("Paid"), editable=False

200 )

201 # Translators: Whether the package is inside actual (hard) limits

202 in_limits = models.BooleanField(

203 default=True, verbose_name=gettext_lazy("In limits"), editable=False

204 )

205 # Payment detailed information, used for integration

206 # with payment processor

207 payment = models.JSONField(editable=False, default=dict, encoder=DjangoJSONEncoder)

208

209 objects = BillingManager.from_queryset(BillingQuerySet)()

210

211 class Meta:

212 verbose_name = "Customer billing"

213 verbose_name_plural = "Customer billings"

214

215 def __str__(self) -> str:

216 projects = self.projects_display

217 owners = self.owners.order()

218 if projects:

219 base = projects

220 elif owners:

221 base = format_html_join_comma(

222 "{}", list_to_tuples(x.get_visible_name() for x in owners)

223 )

224 else:

225 base = "Unassigned"

226 trial = ", trial" if self.is_trial else ""

227 return f"{base} ({self.plan}{trial})"

228

229 def save(

230 self,

231 force_insert=False,

232 force_update=False,

233 using=None,

234 update_fields=None,

235 skip_limits=False,

236 ) -> None:

237 if (

238 not skip_limits

239 and self.pk

240 and self.check_limits(save=False)

241 and update_fields

242 ):

243 update_fields = set(update_fields)

244 update_fields.update(("state", "expiry", "removal", "paid", "in_limits"))

245

246 super().save(

247 force_insert=force_insert,

248 force_update=force_update,

249 using=using,

250 update_fields=update_fields,

251 )

252

253 def get_absolute_url(self) -> str:

254 return reverse("billing-detail", kwargs={"pk": self.pk})

255

256 @cached_property

257 def ordered_projects(self):

258 return self.projects.order()

259

260 @cached_property

261 def all_projects(self):

262 return prefetch_stats(self.ordered_projects)

263

264 @cached_property

265 def projects_display(self):

266 return format_html_join_comma("{}", list_to_tuples(self.all_projects))

267

268 @property

269 def is_trial(self):

270 return self.state == Billing.STATE_TRIAL

271

272 @property

273 def is_terminated(self):

274 return self.state == Billing.STATE_TERMINATED

275

276 @property

277 def is_libre_trial(self) -> bool:

278 return self.is_trial and self.plan.price == 0

279

280 @cached_property

281 def can_be_paid(self) -> bool:

282 if self.state in Billing.ACTIVE_STATES:

283 return True

284 return self.count_projects > 0

285

286 @admin.display(description=gettext_lazy("Changes in last month"))

287 def monthly_changes(self) -> int:

288 return sum(project.stats.monthly_changes for project in self.all_projects)

289

290 @admin.display(description=gettext_lazy("Number of changes"))

291 def total_changes(self) -> int:

292 return sum(project.stats.total_changes for project in self.all_projects)

293

294 @cached_property

295 def count_projects(self) -> int:

296 return len(self.all_projects)

297

298 @admin.display(description=gettext_lazy("Projects"))

299 def display_projects(self) -> str:

300 return f"{self.count_projects} / {self.plan.display_limit_projects}"

301

302 @cached_property

303 def count_strings(self) -> int:

304 return sum(p.stats.source_strings for p in self.all_projects)

305

306 @admin.display(description=gettext_lazy("Source strings"))

307 def display_strings(self) -> str:

308 return f"{self.count_strings} / {self.plan.display_limit_strings}"

309

310 @cached_property

311 def count_hosted_strings(self) -> int:

312 return sum(p.stats.all for p in self.all_projects)

313

314 @admin.display(description=gettext_lazy("Hosted strings"))

315 def display_hosted_strings(self) -> str:

316 return f"{self.count_hosted_strings} / {self.plan.display_limit_hosted_strings}"

317

318 @cached_property

319 def count_words(self):

320 return sum(p.stats.source_words for p in self.all_projects)

321

322 @cached_property

323 def hosted_words(self):

324 return sum(p.stats.all_words for p in self.all_projects)

325

326 @admin.display(description=gettext_lazy("Source words"))

327 def display_words(self) -> str:

328 return f"{self.count_words}"

329

330 @cached_property

331 def count_languages(self):

332 return max((p.stats.languages for p in self.all_projects), default=0)

333

334 @admin.display(description=gettext_lazy("Languages"))

335 def display_languages(self) -> str:

336 return f"{self.count_languages} / {self.plan.display_limit_languages}"

337

338 def flush_cache(self) -> None:

339 keys = list(self.__dict__.keys())

340 for key in keys:

341 if key.startswith("count_"):

342 del self.__dict__[key]

343

344 def check_in_limits(self, plan=None):

345 if plan is None:

346 plan = self.plan

347 return (

348 (plan.limit_projects == 0 or self.count_projects <= plan.limit_projects)

349 and (

350 plan.limit_hosted_strings == 0

351 or self.count_hosted_strings <= plan.limit_hosted_strings

352 )

353 and (plan.limit_strings == 0 or self.count_strings <= plan.limit_strings)

354 and (

355 plan.limit_languages == 0

356 or self.count_languages <= plan.limit_languages

357 )

358 )

359

360 def check_expiry(self):

361 return (

362 self.state in Billing.EXPIRING_STATES

363 and self.expiry

364 and self.expiry < timezone.now()

365 )

366

367 @admin.display(description=gettext_lazy("Number of strings"))

368 def unit_count(self):

369 return sum(p.stats.all for p in self.all_projects)

370

371 def get_last_invoice_object(self):

372 return self.invoice_set.order_by("-start")[0]

373

374 @admin.display(description=gettext_lazy("Last invoice"))

375 def last_invoice(self):

376 try:

377 invoice = self.get_last_invoice_object()

378 except IndexError:

379 return gettext("N/A")

380 return f"{invoice.start} - {invoice.end}"

381

382 @admin.display(

383 # Translators: Whether the package is inside displayed (soft) limits

384 description=gettext_lazy("In display limits"),

385 boolean=True,

386 )

387 def in_display_limits(self, plan=None):

388 if plan is None:

389 plan = self.plan

390 return (

391 (

392 plan.display_limit_projects == 0

393 or self.count_projects <= plan.display_limit_projects

394 )

395 and (

396 plan.display_limit_hosted_strings == 0

397 or self.count_hosted_strings <= plan.display_limit_hosted_strings

398 )

399 and (

400 plan.display_limit_strings == 0

401 or self.count_strings <= plan.display_limit_strings

402 )

403 and (

404 plan.display_limit_languages == 0

405 or self.count_languages <= plan.display_limit_languages

406 )

407 )

408

409 def check_payment_status(self, now: bool = False) -> bool:

410 """

411 Check current payment status.

412

413 Compared to paid attribute, this does not include grace period.

414 """

415 end = timezone.now()

416 if not now:

417 end -= timedelta(days=settings.BILLING_GRACE_PERIOD)

418 return (

419 (self.plan.is_free and self.state == Billing.STATE_ACTIVE)

420 or self.invoice_set.filter(end__gte=end).exists()

421 or self.state == Billing.STATE_TRIAL

422 )

423

424 @transaction.atomic

425 def check_limits(self, save=True) -> bool:

426 self.flush_cache()

427 in_limits = self.check_in_limits()

428 paid = self.check_payment_status()

429 modified = False

430

431 if self.check_expiry():

432 self.expiry = None

433 self.removal = timezone.now() + timedelta(

434 days=settings.BILLING_REMOVAL_PERIOD

435 )

436 self.billinglog_set.create(

437 event=BillingEvent.EXPIRED,

438 summary=f"Scheduled removal at {self.removal.isoformat()}",

439 )

440 modified = True

441

442 if self.state not in Billing.EXPIRING_STATES and self.expiry:

443 self.expiry = None

444 modified = True

445

446 if self.in_limits != in_limits or self.paid != paid:

447 self.in_limits = in_limits

448 self.paid = paid

449 modified = True

450

451 if save:

452 if modified:

453 Billing.objects.select_for_update().get(pk=self.pk)

454 self.save(skip_limits=True)

455 self.update_alerts()

456

457 return modified

458

459 def update_alerts(self) -> None:

460 if self.in_limits:

461 Alert.objects.filter(

462 component__project__in=self.projects.all(), name="BillingLimit"

463 ).delete()

464 else:

465 for project in self.projects.iterator():

466 for component in project.component_set.iterator():

467 component.add_alert("BillingLimit")

468

469 def is_active(self):

470 return self.state in Billing.ACTIVE_STATES

471

472 def get_notify_users(self):

473 users = self.owners.distinct()

474 for project in self.projects.iterator():

475 users |= User.objects.having_perm("billing.view", project)

476 return users.exclude(is_superuser=True)

477

478 def _get_libre_checklist(self):

479 message = ngettext(

480 "Contains %d project", "Contains %d projects", self.count_projects

481 )

482 # Ignore when format string is not present

483 with suppress(TypeError):

484 message %= self.count_projects

485 yield LibreCheck(self.count_projects == 1, message)

486 for project in self.all_projects:

487 yield LibreCheck(

488 bool(project.web),

489 format_html(

490 '<a href="{0}">{1}</a>, <a href="{2}">{3}</a>',

491 project.get_absolute_url(),

492 project,

493 project.web

494 or reverse("settings", kwargs={"path": project.get_url_path()}),

495 project.web or gettext("Project website missing!"),

496 ),

497 )

498 if project.access_control:

499 yield LibreCheck(False, gettext("Only public projects are allowed"))

500 components = Component.objects.filter(

501 project__in=self.all_projects

502 ).prefetch_related("project")

503 yield LibreCheck(

504 len(components) > 0,

505 ngettext("Contains %d component", "Contains %d components", len(components))

506 % len(components),

507 )

508 for component in components:

509 license_name = component.get_license_display()

510 if not component.libre_license:

511 if not license_name:

512 license_name = format_html(

513 "<strong>{0}</strong>", gettext("Missing license")

514 )

515 else:

516 license_name = format_html(

517 "{0} (<strong>{1}</strong>)",

518 license_name,

519 gettext("Not a libre license"),

520 )

521 if component.license_url:

522 license_name = format_html(

523 '<a href="{0}">{1}</a>', component.license_url, license_name

524 )

525 repo_url = component.repo

526 if repo_url.startswith("https://"):

527 repo_url = format_html('<a href="{0}">{0}</a>', repo_url)

528 yield LibreCheck(

529 component.libre_license,

530 format_html(

531 """

532 <a href="{0}">{1}</a>,

533 {2},

534 {3},

535 {4}""",

536 component.get_absolute_url(),

537 component.name,

538 license_name,

539 repo_url,

540 component.get_file_format_display(),

541 ),

542 component=component,

543 )

544

545 @cached_property

546 def libre_checklist(self):

547 return list(self._get_libre_checklist())

548

549 @property

550 def valid_libre(self):

551 return all(self.libre_checklist)

552

553

554class InvoiceQuerySet(models.QuerySet["Invoice"]):

555 def order(self):

556 return self.order_by("-start")

557

558

559class Invoice(models.Model):

560 CURRENCY_EUR = 0

561 CURRENCY_BTC = 1

562 CURRENCY_USD = 2

563 CURRENCY_CZK = 3

564

565 billing = models.ForeignKey(Billing, on_delete=models.deletion.RESTRICT)

566 start = models.DateField()

567 end = models.DateField()

568 amount = models.FloatField()

569 currency = models.IntegerField(

570 choices=(

571 (CURRENCY_EUR, "EUR"),

572 (CURRENCY_BTC, "mBTC"),

573 (CURRENCY_USD, "USD"),

574 (CURRENCY_CZK, "CZK"),

575 ),

576 default=CURRENCY_EUR,

577 )

578 ref = models.CharField(blank=True, max_length=50)

579 note = models.TextField(blank=True)

580 # Payment detailed information, used for integration

581 # with payment processor

582 payment = models.JSONField(editable=False, default=dict)

583 created = models.DateTimeField(auto_now_add=True)

584

585 objects = InvoiceQuerySet.as_manager()

586

587 class Meta:

588 verbose_name = "Invoice"

589 verbose_name_plural = "Invoices"

590

591 def __str__(self) -> str:

592 return f"{self.start} - {self.end}: {self.billing if self.billing_id else None}"

593

594 @cached_property

595 def is_legacy(self):

596 return len(self.ref) <= 6

597

598 @cached_property

599 def filename(self) -> str | None:

600 if not self.ref:

601 return None

602 if self.is_legacy:

603 return f"{self.ref}.pdf"

604 return f"Weblate_Invoice_{self.ref}.pdf"

605

606 @cached_property

607 def full_filename(self) -> str | None:

608 if not self.ref:

609 return None

610 if self.is_legacy:

611 invoice_path = Path(settings.INVOICE_PATH_LEGACY)

612 else:

613 invoice_path = (

614 Path(settings.INVOICE_PATH)

615 / f"{self.created.year}"

616 / f"{self.created.month:02d}"

617 )

618 full_path = invoice_path / (self.filename or "")

619 return full_path.as_posix()

620

621 @cached_property

622 def filename_valid(self) -> bool:

623 return self.full_filename and os.path.exists(self.full_filename)

624

625 def clean(self) -> None:

626 if self.end is None or self.start is None:

627 return

628

629 if self.end <= self.start:

630 msg = "Start has be to before end!"

631 raise ValidationError(msg)

632

633 if not self.billing_id:

634 return

635

636 overlapping = Invoice.objects.filter(

637 (Q(start__lte=self.end) & Q(end__gte=self.end))

638 | (Q(start__lte=self.start) & Q(end__gte=self.start))

639 ).filter(billing=self.billing)

640

641 if self.pk:

642 overlapping = overlapping.exclude(pk=self.pk)

643

644 if overlapping.exists():

645 msg = "Overlapping invoices exist: {}".format(

646 format_html_join_comma("{}", list_to_tuples(overlapping))

647 )

648 raise ValidationError(msg)

649

650

651class BillingEvent(models.IntegerChoices):

652 EMAIL = 1, "Outbound e-mail"

653 EXPIRED = 2, "Trial expired"

654 REMOVED = 3, "Projects removed"

655 PAYMENT = 4, "Payment received"

656 UNPAID = 5, "Unpaid billing"

657 CREATED = 6, "Created billing"

658 DISABLED_RECURRING = 7, "Disabled recurring payment"

659 LIBRE_REQUEST = 8, "Requested Libre hosting"

660 LIBRE_APPROVED = 9, "Approved Libre hosting"

661 TERMINATED = 10, "Billing terminated"

662 EXTENDED_TRIAL = 11, "Trial extended"

663

664

665class BillingLogQuerySet(models.QuerySet["BillingLog"]):

666 def order(self) -> BillingLogQuerySet:

667 return self.order_by("-timestamp")

668

669 def recent(self) -> BillingLogQuerySet:

670 return self.order()[:20]

671

672

673class BillingLog(models.Model):

674 billing = models.ForeignKey(

675 Billing,

676 on_delete=models.deletion.RESTRICT,

677 verbose_name=gettext_lazy("Billing"),

678 )

679 timestamp = models.DateTimeField(default=timezone.now, verbose_name="Timestamp")

680 event = models.IntegerField(

681 choices=BillingEvent, verbose_name=gettext_lazy("Billing event")

682 )

683 summary = models.CharField(max_length=200, verbose_name="Summary")

684 user = models.ForeignKey(User, null=True, on_delete=models.RESTRICT)

685

686 objects = BillingLogQuerySet.as_manager()

687

688 def __str__(self) -> str:

689 return f"{self.timestamp.isoformat()}: {self.billing}: {self.get_event_display()} {self.summary}"

690

691

692@receiver(post_save, sender=Component)

693@receiver(post_save, sender=Project)

694@receiver(post_save, sender=Plan)

695@disable_for_loaddata

696def update_project_bill(sender, instance, **kwargs) -> None:

697 if isinstance(instance, Component):

698 instance = instance.project

699 for billing in instance.billing_set.all():

700 billing.check_limits()

701

702

703@receiver(pre_delete, sender=Project)

704@receiver(pre_delete, sender=Component)

705@receiver(post_delete, sender=Translation)

706@disable_for_loaddata

707def record_project_bill(

708 sender, instance: Project | Component | Translation, **kwargs

709) -> None:

710 if isinstance(instance, Translation):

711 try:

712 instance = instance.component

713 except Component.DoesNotExist:

714 # Happens during component removal

715 return

716 if isinstance(instance, Component):

717 instance = instance.project

718 # Collect billings to update for delete_project_bill

719 instance.billings_to_update = list(

720 instance.billing_set.values_list("pk", flat=True)

721 )

722

723

724@receiver(post_delete, sender=Project)

725@receiver(post_delete, sender=Component)

726@receiver(post_delete, sender=Translation)

727@disable_for_loaddata

728def delete_project_bill(

729 sender, instance: Project | Component | Translation, **kwargs

730) -> None:

731 from weblate.billing.tasks import billing_check

732

733 if isinstance(instance, Translation):

734 try:

735 instance = instance.component

736 except Component.DoesNotExist:

737 # Happens during component removal

738 return

739 if isinstance(instance, Component):

740 instance = instance.project

741 # This is collected in record_project_bill

742 for billing_id in instance.billings_to_update:

743 transaction.on_commit(partial(billing_check, billing_id))

744 # Clear the list to avoid repeated trigger

745 instance.billings_to_update.clear()

746

747

748@receiver(post_save, sender=Invoice)

749@disable_for_loaddata

750def update_invoice_bill(sender, instance, **kwargs) -> None:

751 instance.billing.check_limits()

752

753

754@receiver(m2m_changed, sender=Billing.projects.through)

755@disable_for_loaddata

756def change_billing_projects(sender, instance, action, **kwargs) -> None:

757 if not action.startswith("post_"):

758 return

759 instance.check_limits()

760

761

762class WeblateConf(AppConf):

763 GRACE_PERIOD = 15

764 REMOVAL_PERIOD = 15

765

766 class Meta:

767 prefix = "BILLING"