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

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

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

4 

5from datetime import timedelta 

6 

7from celery.schedules import crontab 

8from django.conf import settings 

9from django.db import transaction 

10from django.db.models import Count, Q 

11from django.utils import timezone 

12from django.utils.translation import gettext 

13 

14from weblate.accounts.notifications import send_notification_email 

15from weblate.billing.models import Billing, BillingEvent 

16from weblate.trans.tasks import project_removal 

17from weblate.utils.celery import app 

18 

19 

20@app.task(trail=False) 

21def billing_check(billing_id: int | None = None) -> None: 

22 if billing_id is None: 

23 Billing.objects.check_limits() 

24 else: 

25 Billing.objects.get(pk=billing_id).check_limits() 

26 

27 

28@app.task(trail=False) 

29def billing_notify() -> None: 

30 billing_check() 

31 

32 limit = Billing.objects.get_out_of_limits() 

33 due = Billing.objects.get_unpaid() 

34 

35 with_project = Billing.objects.annotate(Count("projects")).filter( 

36 projects__count__gt=0 

37 ) 

38 toremove = with_project.exclude(removal=None).order_by("removal") 

39 trial = with_project.filter(removal=None, state=Billing.STATE_TRIAL).order_by( 

40 "expiry" 

41 ) 

42 

43 if limit or due or toremove or trial: 

44 send_notification_email( 

45 "en", 

46 [a[1] for a in settings.ADMINS] + settings.ADMINS_BILLING, 

47 "billing_check", 

48 context={ 

49 "limit": limit, 

50 "due": due, 

51 "toremove": toremove, 

52 "trial": trial, 

53 }, 

54 ) 

55 

56 

57@app.task(trail=False) 

58def notify_expired() -> None: 

59 # Notify about expired billings 

60 possible_billings = Billing.objects.filter( 

61 # Active without payment (checked later) 

62 Q(state=Billing.STATE_ACTIVE) 

63 # Scheduled removal 

64 | Q(removal__isnull=False) 

65 # Trials expiring soon 

66 | Q(state=Billing.STATE_TRIAL, expiry__lte=timezone.now() + timedelta(days=7)) 

67 ).exclude(projects__isnull=True) 

68 for bill in possible_billings: 

69 if bill.state == Billing.STATE_ACTIVE and bill.check_payment_status(now=True): 

70 continue 

71 if bill.plan.price: 

72 note = gettext( 

73 "You will stop receiving this notification once " 

74 "you pay the bills or the project is removed." 

75 ) 

76 else: 

77 note = gettext( 

78 "You will stop receiving this notification once " 

79 "you change to regular subscription or the project is removed." 

80 ) 

81 

82 for user in bill.get_notify_users(): 

83 bill.billinglog_set.create( 

84 event=BillingEvent.EMAIL, summary="Billing expired", user=user 

85 ) 

86 send_notification_email( 

87 user.profile.language, 

88 [user.email], 

89 "billing_expired", 

90 context={ 

91 "billing": bill, 

92 "payment_enabled": getattr(settings, "PAYMENT_ENABLED", False), 

93 "unsubscribe_note": note, 

94 }, 

95 info=bill, 

96 ) 

97 

98 

99@app.task(trail=False) 

100@transaction.atomic 

101def schedule_removal() -> None: 

102 removal = timezone.now() + timedelta(days=settings.BILLING_REMOVAL_PERIOD) 

103 for bill in Billing.objects.filter( 

104 state=Billing.STATE_ACTIVE, removal=None 

105 ).select_for_update(): 

106 if bill.check_payment_status(): 

107 continue 

108 bill.billinglog_set.create( 

109 event=BillingEvent.UNPAID, 

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

111 ) 

112 bill.removal = removal 

113 bill.save(update_fields=["removal"]) 

114 

115 

116@app.task(trail=False) 

117@transaction.atomic 

118def remove_single_billing(billing_id: int) -> None: 

119 bill = Billing.objects.select_for_update().get(pk=billing_id) 

120 for user in bill.get_notify_users(): 

121 bill.billinglog_set.create( 

122 event=BillingEvent.EMAIL, summary="Billing removed", user=user 

123 ) 

124 send_notification_email( 

125 user.profile.language, 

126 [user.email], 

127 "billing_expired", 

128 context={"billing": bill, "final_removal": True}, 

129 info=bill, 

130 ) 

131 for prj in bill.projects.iterator(): 

132 bill.billinglog_set.create( 

133 event=BillingEvent.REMOVED, summary=f"Removed project {prj}" 

134 ) 

135 prj.log_warning("removing due to unpaid billing") 

136 project_removal(prj.id, None) 

137 bill.removal = None 

138 bill.state = Billing.STATE_TERMINATED 

139 bill.save() 

140 bill.billinglog_set.create(event=BillingEvent.REMOVED, summary="Terminated billing") 

141 

142 

143@app.task(trail=False) 

144def perform_removal() -> None: 

145 for bill in Billing.objects.filter(removal__lte=timezone.now()).iterator(): 

146 remove_single_billing.delay(bill.pk) 

147 

148 

149@app.on_after_finalize.connect 

150def setup_periodic_tasks(sender, **kwargs) -> None: 

151 sender.add_periodic_task(3600, billing_check.s(), name="billing-check") 

152 sender.add_periodic_task( 

153 crontab(hour=3, minute=0, day_of_week="mon,thu"), 

154 billing_notify.s(), 

155 name="billing-notify", 

156 ) 

157 sender.add_periodic_task( 

158 crontab(hour=1, minute=0), 

159 perform_removal.s(), 

160 name="perform-removal", 

161 ) 

162 sender.add_periodic_task( 

163 crontab(hour=2, minute=0, day_of_week="mon,thu"), 

164 schedule_removal.s(), 

165 name="schedule-removal", 

166 ) 

167 sender.add_periodic_task( 

168 crontab(hour=2, minute=30, day_of_week="mon,thu"), 

169 notify_expired.s(), 

170 name="notify-expired", 

171 )