429 lines
18 KiB
Python
429 lines
18 KiB
Python
"""
|
||
Subscription Invoice Processing Job
|
||
Processes active subscriptions when next_invoice_date is reached
|
||
Creates ordre drafts and advances subscription periods
|
||
Runs daily at 04:00
|
||
"""
|
||
|
||
import logging
|
||
from datetime import datetime, date
|
||
import json
|
||
from dateutil.relativedelta import relativedelta
|
||
|
||
from app.core.database import execute_query, get_db_connection
|
||
from app.services.subscription_billing_calendar import (
|
||
advance_billing_periods,
|
||
billing_date_for_period,
|
||
prorated_30_day_factor,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def process_subscriptions():
|
||
"""
|
||
Main job: Process subscriptions due for invoicing.
|
||
- Find active subscriptions where next_invoice_date <= today
|
||
- Skip subscriptions blocked for invoicing (missing asset/serial)
|
||
- Aggregate eligible subscriptions into one ordre_draft per customer + merge key + due date + billing direction
|
||
- Advance period_start and next_invoice_date for processed subscriptions
|
||
"""
|
||
|
||
try:
|
||
logger.info("💰 Processing subscription invoices...")
|
||
from app.subscriptions.backend.router import apply_due_subscription_changes, expire_ended_subscriptions
|
||
applied_changes = apply_due_subscription_changes()
|
||
expired_subscriptions = expire_ended_subscriptions()
|
||
if applied_changes:
|
||
logger.info("✅ Applied %s scheduled subscription change request(s)", applied_changes)
|
||
if expired_subscriptions:
|
||
logger.info("✅ Expired %s ended subscription(s)", expired_subscriptions)
|
||
|
||
# Find subscriptions due for invoicing
|
||
query = """
|
||
SELECT
|
||
s.id,
|
||
s.sag_id,
|
||
sg.titel AS sag_name,
|
||
s.customer_id,
|
||
c.name AS customer_name,
|
||
s.product_name,
|
||
s.billing_interval,
|
||
s.billing_schedule_type,
|
||
s.billing_day,
|
||
s.billing_direction,
|
||
s.advance_months,
|
||
s.billing_lead_months,
|
||
s.first_full_period_start,
|
||
s.proration_basis,
|
||
s.price,
|
||
s.next_invoice_date,
|
||
s.period_start,
|
||
s.invoice_merge_key,
|
||
s.billing_blocked,
|
||
s.billing_block_reason,
|
||
COALESCE(
|
||
(
|
||
SELECT json_agg(
|
||
json_build_object(
|
||
'id', si.id,
|
||
'description', si.description,
|
||
'quantity', si.quantity,
|
||
'unit_price', si.unit_price,
|
||
'line_total', si.line_total,
|
||
'product_id', si.product_id,
|
||
'asset_id', si.asset_id,
|
||
'billing_blocked', si.billing_blocked,
|
||
'billing_block_reason', si.billing_block_reason,
|
||
'period_from', si.period_from,
|
||
'period_to', si.period_to
|
||
) ORDER BY si.id
|
||
)
|
||
FROM sag_subscription_items si
|
||
WHERE si.subscription_id = s.id
|
||
),
|
||
'[]'::json
|
||
) as line_items
|
||
,COALESCE(
|
||
(
|
||
SELECT json_agg(json_build_object(
|
||
'id', fi.id,
|
||
'description', fi.description,
|
||
'quantity', fi.quantity,
|
||
'unit_price', fi.unit_price,
|
||
'line_total', fi.line_total,
|
||
'product_id', fi.product_id
|
||
) ORDER BY fi.line_no, fi.id)
|
||
FROM sag_subscription_first_invoice_items fi
|
||
WHERE fi.subscription_id = s.id AND fi.billed_at IS NULL
|
||
),
|
||
'[]'::json
|
||
) AS first_invoice_items
|
||
FROM sag_subscriptions s
|
||
LEFT JOIN sag_sager sg ON sg.id = s.sag_id
|
||
LEFT JOIN customers c ON c.id = s.customer_id
|
||
WHERE s.status = 'active'
|
||
AND s.next_invoice_date <= CURRENT_DATE
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM subscription_billing_runs br
|
||
WHERE br.subscription_id = s.id AND br.period_start = s.period_start
|
||
)
|
||
ORDER BY s.next_invoice_date, s.id
|
||
"""
|
||
|
||
subscriptions = execute_query(query)
|
||
|
||
if not subscriptions:
|
||
logger.info("✅ No subscriptions due for invoicing")
|
||
return
|
||
|
||
logger.info(f"📋 Found {len(subscriptions)} subscription(s) to process")
|
||
|
||
blocked_count = 0
|
||
processed_count = 0
|
||
error_count = 0
|
||
|
||
grouped_subscriptions = {}
|
||
for sub in subscriptions:
|
||
if sub.get('billing_blocked'):
|
||
blocked_count += 1
|
||
logger.warning(
|
||
"⚠️ Subscription %s skipped due to billing block: %s",
|
||
sub.get('id'),
|
||
sub.get('billing_block_reason') or 'unknown reason'
|
||
)
|
||
continue
|
||
|
||
group_key = (
|
||
int(sub['customer_id']),
|
||
str(sub.get('invoice_merge_key') or f"cust-{sub['customer_id']}"),
|
||
str(sub.get('next_invoice_date')),
|
||
str(sub.get('billing_direction') or 'forward'),
|
||
)
|
||
grouped_subscriptions.setdefault(group_key, []).append(sub)
|
||
|
||
for group in grouped_subscriptions.values():
|
||
try:
|
||
count = await _process_subscription_group(group)
|
||
processed_count += count
|
||
except Exception as e:
|
||
logger.error("❌ Failed processing subscription group: %s", e, exc_info=True)
|
||
error_count += 1
|
||
|
||
logger.info(
|
||
"✅ Subscription processing complete: %s processed, %s blocked, %s errors",
|
||
processed_count,
|
||
blocked_count,
|
||
error_count,
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Subscription processing job failed: {e}", exc_info=True)
|
||
|
||
|
||
async def _process_subscription_group(subscriptions: list[dict]) -> int:
|
||
"""Create one aggregated ordre draft for a group of subscriptions and advance all periods."""
|
||
|
||
if not subscriptions:
|
||
return 0
|
||
|
||
first = subscriptions[0]
|
||
customer_id = first['customer_id']
|
||
customer_name = first.get('customer_name') or f"Customer #{customer_id}"
|
||
billing_direction = first.get('billing_direction') or 'forward'
|
||
invoice_aggregate_key = first.get('invoice_merge_key') or f"cust-{customer_id}"
|
||
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
|
||
try:
|
||
ordre_lines = []
|
||
source_subscription_ids = []
|
||
coverage_start = None
|
||
coverage_end = None
|
||
|
||
# Claim every subscription period. A competing worker will make this
|
||
# transaction roll back before an order draft can be duplicated.
|
||
claimed_run_ids = []
|
||
for sub in subscriptions:
|
||
cursor.execute(
|
||
"""INSERT INTO subscription_billing_runs (subscription_id, period_start)
|
||
VALUES (%s, %s) ON CONFLICT DO NOTHING RETURNING id""",
|
||
(int(sub['id']), sub.get('period_start') or sub.get('next_invoice_date')),
|
||
)
|
||
claimed = cursor.fetchone()
|
||
if not claimed:
|
||
conn.rollback()
|
||
logger.info("Subscription period already claimed by another worker")
|
||
return 0
|
||
claimed_run_ids.append(int(claimed[0]))
|
||
|
||
for sub in subscriptions:
|
||
subscription_id = int(sub['id'])
|
||
source_subscription_ids.append(subscription_id)
|
||
|
||
line_items = sub.get('line_items', [])
|
||
if isinstance(line_items, str):
|
||
line_items = json.loads(line_items)
|
||
|
||
period_start = sub.get('period_start') or sub.get('next_invoice_date')
|
||
first_full_period_start = sub.get('first_full_period_start')
|
||
if isinstance(first_full_period_start, str):
|
||
first_full_period_start = datetime.strptime(first_full_period_start, '%Y-%m-%d').date()
|
||
advance_periods = max(1, int(sub.get('advance_months') or 1))
|
||
has_short_opening_period = bool(first_full_period_start and period_start < first_full_period_start)
|
||
full_period_start = first_full_period_start if has_short_opening_period else period_start
|
||
period_end = advance_billing_periods(full_period_start, sub['billing_interval'], advance_periods)
|
||
if coverage_start is None or period_start < coverage_start:
|
||
coverage_start = period_start
|
||
if coverage_end is None or period_end > coverage_end:
|
||
coverage_end = period_end
|
||
|
||
for item in line_items:
|
||
if item.get('billing_blocked'):
|
||
logger.warning(
|
||
"⚠️ Skipping blocked subscription item %s on subscription %s",
|
||
item.get('id'),
|
||
subscription_id,
|
||
)
|
||
continue
|
||
|
||
product_number = str(item.get('product_id', 'SUB'))
|
||
if has_short_opening_period:
|
||
factor = prorated_30_day_factor(period_start, first_full_period_start)
|
||
if factor > 0:
|
||
prorated_unit_price = round(float(item.get('unit_price', 0)) * factor, 2)
|
||
ordre_lines.append({
|
||
"product": {
|
||
"productNumber": product_number,
|
||
"description": f"{item.get('description', '')} – skæv periode {period_start} til {first_full_period_start} (30 dage)"
|
||
},
|
||
"quantity": float(item.get('quantity', 1)),
|
||
"unitNetPrice": prorated_unit_price,
|
||
"totalNetAmount": round(float(item.get('quantity', 1)) * prorated_unit_price, 2),
|
||
"discountPercentage": 0,
|
||
"metadata": {
|
||
"subscription_id": subscription_id,
|
||
"proration_basis": "30_day",
|
||
"proration_factor": factor,
|
||
"period_from": str(period_start),
|
||
"period_to": str(first_full_period_start),
|
||
}
|
||
})
|
||
full_unit_price = float(item.get('unit_price', 0)) * advance_periods
|
||
ordre_lines.append({
|
||
"product": {
|
||
"productNumber": product_number,
|
||
"description": item.get('description', '') + (f" – {advance_periods} perioder" if advance_periods > 1 else '')
|
||
},
|
||
"quantity": float(item.get('quantity', 1)),
|
||
"unitNetPrice": full_unit_price,
|
||
"totalNetAmount": float(item.get('quantity', 1)) * full_unit_price,
|
||
"discountPercentage": 0,
|
||
"metadata": {
|
||
"subscription_id": subscription_id,
|
||
"asset_id": item.get('asset_id'),
|
||
"period_from": str(full_period_start),
|
||
"period_to": str(item.get('period_to') or period_end),
|
||
"advance_periods": advance_periods,
|
||
}
|
||
})
|
||
|
||
first_invoice_items = sub.get('first_invoice_items', [])
|
||
if isinstance(first_invoice_items, str):
|
||
first_invoice_items = json.loads(first_invoice_items)
|
||
for item in first_invoice_items:
|
||
ordre_lines.append({
|
||
"product": {
|
||
"productNumber": str(item.get('product_id') or 'ENGANG'),
|
||
"description": item.get('description', '')
|
||
},
|
||
"quantity": float(item.get('quantity', 1)),
|
||
"unitNetPrice": float(item.get('unit_price', 0)),
|
||
"totalNetAmount": float(item.get('line_total', 0)),
|
||
"discountPercentage": 0,
|
||
"metadata": {
|
||
"subscription_id": subscription_id,
|
||
"first_invoice_item_id": item.get('id'),
|
||
"one_time": True,
|
||
}
|
||
})
|
||
|
||
if not ordre_lines:
|
||
logger.warning("⚠️ No invoiceable lines in subscription group for customer %s", customer_id)
|
||
return 0
|
||
|
||
title = f"Abonnementer: {customer_name}"
|
||
notes = (
|
||
f"Aggregated abonnement faktura\n"
|
||
f"Kunde: {customer_name}\n"
|
||
f"Coverage: {coverage_start} til {coverage_end}\n"
|
||
f"Subscription IDs: {', '.join(str(sid) for sid in source_subscription_ids)}"
|
||
)
|
||
|
||
insert_query = """
|
||
INSERT INTO ordre_drafts (
|
||
title,
|
||
customer_id,
|
||
lines_json,
|
||
notes,
|
||
coverage_start,
|
||
coverage_end,
|
||
billing_direction,
|
||
source_subscription_ids,
|
||
invoice_aggregate_key,
|
||
layout_number,
|
||
created_by_user_id,
|
||
sync_status,
|
||
export_status_json,
|
||
updated_at
|
||
) VALUES (%s, %s, %s::jsonb, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, CURRENT_TIMESTAMP)
|
||
RETURNING id
|
||
"""
|
||
|
||
cursor.execute(insert_query, (
|
||
title,
|
||
customer_id,
|
||
json.dumps(ordre_lines, ensure_ascii=False),
|
||
notes,
|
||
coverage_start,
|
||
coverage_end,
|
||
billing_direction,
|
||
source_subscription_ids,
|
||
invoice_aggregate_key,
|
||
1, # Default layout
|
||
None, # System-created
|
||
'pending',
|
||
json.dumps({"source": "subscription", "subscription_ids": source_subscription_ids}, ensure_ascii=False)
|
||
))
|
||
|
||
ordre_id = cursor.fetchone()[0]
|
||
cursor.execute(
|
||
"UPDATE subscription_billing_runs SET ordre_draft_id = %s WHERE id = ANY(%s)",
|
||
(ordre_id, claimed_run_ids),
|
||
)
|
||
for sub, run_id in zip(subscriptions, claimed_run_ids):
|
||
first_items = sub.get('first_invoice_items', [])
|
||
if isinstance(first_items, str):
|
||
first_items = json.loads(first_items)
|
||
first_item_ids = [int(item['id']) for item in first_items if item.get('id')]
|
||
if first_item_ids:
|
||
cursor.execute(
|
||
"""UPDATE sag_subscription_first_invoice_items
|
||
SET billed_at = CURRENT_TIMESTAMP, billing_run_id = %s, updated_at = CURRENT_TIMESTAMP
|
||
WHERE subscription_id = %s AND id = ANY(%s) AND billed_at IS NULL""",
|
||
(run_id, int(sub['id']), first_item_ids),
|
||
)
|
||
logger.info(
|
||
"✅ Created aggregated ordre draft #%s for %s subscription(s)",
|
||
ordre_id,
|
||
len(source_subscription_ids),
|
||
)
|
||
|
||
for sub in subscriptions:
|
||
subscription_id = int(sub['id'])
|
||
current_period_start = sub.get('period_start') or sub.get('next_invoice_date')
|
||
first_full_period_start = sub.get('first_full_period_start')
|
||
if isinstance(first_full_period_start, str):
|
||
first_full_period_start = datetime.strptime(first_full_period_start, '%Y-%m-%d').date()
|
||
full_period_start = first_full_period_start if first_full_period_start and current_period_start < first_full_period_start else current_period_start
|
||
new_period_start = advance_billing_periods(
|
||
full_period_start, sub['billing_interval'], max(1, int(sub.get('advance_months') or 1))
|
||
)
|
||
new_next_invoice_date = billing_date_for_period(
|
||
new_period_start,
|
||
int(sub.get('billing_lead_months') or 0),
|
||
sub.get('billing_schedule_type') or 'fixed_day',
|
||
int(sub.get('billing_day') or 1),
|
||
)
|
||
|
||
cursor.execute(
|
||
"""
|
||
UPDATE sag_subscriptions
|
||
SET period_start = %s,
|
||
next_invoice_date = %s,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s
|
||
""",
|
||
(new_period_start, new_next_invoice_date, subscription_id)
|
||
)
|
||
|
||
conn.commit()
|
||
return len(source_subscription_ids)
|
||
|
||
except Exception as e:
|
||
conn.rollback()
|
||
raise e
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
def _calculate_next_period_start(current_date, billing_interval: str) -> date:
|
||
"""Calculate next period start date based on billing interval"""
|
||
|
||
# Parse current_date if it's a string
|
||
if isinstance(current_date, str):
|
||
current_date = datetime.strptime(current_date, '%Y-%m-%d').date()
|
||
elif isinstance(current_date, datetime):
|
||
current_date = current_date.date()
|
||
|
||
# Calculate delta based on interval
|
||
if billing_interval == 'daily':
|
||
delta = relativedelta(days=1)
|
||
elif billing_interval == 'biweekly':
|
||
delta = relativedelta(weeks=2)
|
||
elif billing_interval == 'monthly':
|
||
delta = relativedelta(months=1)
|
||
elif billing_interval == 'quarterly':
|
||
delta = relativedelta(months=3)
|
||
elif billing_interval == 'yearly':
|
||
delta = relativedelta(years=1)
|
||
else:
|
||
# Default to monthly if unknown
|
||
logger.warning(f"Unknown billing interval '{billing_interval}', defaulting to monthly")
|
||
delta = relativedelta(months=1)
|
||
|
||
next_date = current_date + delta
|
||
return next_date
|