import calendar from datetime import date, timedelta from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from django.db import transaction from django.db.models import Sum from django.db.models.functions import TruncDate from django.utils import timezone from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from apps.ai.models import AITask from apps.common.api import can_manage_team, get_current_team from .models import CreditAccount, CreditLedger from .pricing import get_billing_config, team_price_multiplier from .serializers import CreditAccountSerializer, CreditLedgerSerializer # AITask.task_type → 账户页「按阶段分布」的 4 个聚合桶 _STAGE_BUCKET = { AITask.Type.SCRIPT_GENERATION: "script", AITask.Type.SCRIPT_OPTIMIZATION: "script", AITask.Type.ENTITY_EXTRACTION: "script", AITask.Type.PRODUCT_IMAGE: "base", AITask.Type.PERSON_IMAGE: "base", AITask.Type.MODEL_TRIVIEW: "base", AITask.Type.SCENE_IMAGE: "base", AITask.Type.STORYBOARD: "storyboard", AITask.Type.VIDEO_SEGMENT: "video", AITask.Type.VOICEOVER: "video", AITask.Type.FREE_VIDEO: "video", AITask.Type.EXPORT: "video", } @api_view(["GET"]) @permission_classes([IsAuthenticated]) def summary(request): team = get_current_team(request.user) account, _ = CreditAccount.objects.get_or_create(team=team) charged = CreditLedger.objects.filter(team=team, ledger_type=CreditLedger.Type.CHARGE).aggregate( total=Sum("amount") )["total"] or 0 return Response( { "account": CreditAccountSerializer(account).data, "charged_total": charged, } ) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def ledgers(request): team = get_current_team(request.user) # defer task 的两个巨型 payload 列:流水只展示金额/类型/关联,不需要 AI 请求/响应 payload。 # 与 assets/billing.summary 同一惯例,避免 select_related("task") 把大 JSON 拖出来。 queryset = ( CreditLedger.objects.filter(team=team) .select_related("user", "project", "task") .defer("task__request_payload", "task__response_payload") .order_by("-created_at") ) project_id = request.query_params.get("project") user_id = request.query_params.get("user") # 按类型过滤(账户页「全部类型/扣费/充值…」下拉):count/分页随过滤变化,未知值忽略 ledger_type = request.query_params.get("ledger_type") if project_id: queryset = queryset.filter(project_id=project_id) if user_id: queryset = queryset.filter(user_id=user_id) if ledger_type and ledger_type in CreditLedger.Type.values: queryset = queryset.filter(ledger_type=ledger_type) # 按自然月过滤(账户页流水月份下拉);count/分页随月份变化 month_param = (request.query_params.get("month") or "").strip() if len(month_param) == 7 and month_param[4] == "-": try: y = int(month_param[:4]) m = int(month_param[5:7]) if 1 <= m <= 12: month_start = date(y, m, 1) month_end = date(y, m, calendar.monthrange(y, m)[1]) queryset = queryset.filter(created_at__date__gte=month_start, created_at__date__lte=month_end) except ValueError: pass # 服务端分页:总数随流水增长(原先写死 [:100] 导致永远 100 条) try: page = max(1, int(request.query_params.get("page", 1))) except (TypeError, ValueError): page = 1 try: page_size = int(request.query_params.get("page_size", 10)) except (TypeError, ValueError): page_size = 10 page_size = max(1, min(page_size, 100)) total = queryset.count() start = (page - 1) * page_size rows = queryset[start:start + page_size] return Response( { "count": total, "page": page, "page_size": page_size, "results": CreditLedgerSerializer(rows, many=True).data, } ) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def config(request): """计费公共配置:前端预估(自由创作 token 预估 / 图片单价 / 充值到账换算)统一读这里, 与后端计价引擎同一来源,杜绝两端口径漂移。""" cfg = get_billing_config() # 团队价格系数(差异化调价):预估所见即所扣。平台超管无团队 → 回落 1。 try: multiplier = team_price_multiplier(get_current_team(request.user)) except Exception: # noqa: BLE001 — 无团队(平台超管)等情况一律按标准价 multiplier = Decimal("1") return Response( { "currency_unit": "points", "points_per_yuan": str(cfg.points_per_yuan), "video_margin_multiplier": str(cfg.video_margin_multiplier), "video_reserve_buffer": str(cfg.video_reserve_buffer), "team_price_multiplier": str(multiplier), } ) @api_view(["POST"]) @permission_classes([IsAuthenticated]) def recharge(request): """充值:amount 是真实支付人民币(¥),到账积分 = ¥ × points_per_yuan + bonus_points(赠送积分)。 metadata 快照 paid_amount(¥)与当时汇率——老流水的支付金额只能从这里读,不可拿 amount÷rate 反推。""" team = get_current_team(request.user) # 充值是团队资金操作:仅 owner/admin 可发起,普通成员/访客一律拒绝 if not can_manage_team(request.user, team): return Response({"detail": "permission denied"}, status=status.HTTP_403_FORBIDDEN) try: amount = Decimal(str(request.data.get("amount", "0"))) # 兼容旧键 bonus(历史前端按 ¥ 传):按汇率折成积分;新契约 bonus_points 直接是积分 if request.data.get("bonus_points") is not None: bonus_points = Decimal(str(request.data.get("bonus_points"))) else: bonus_yuan = Decimal(str(request.data.get("bonus", "0"))) bonus_points = bonus_yuan * get_billing_config().points_per_yuan except (InvalidOperation, TypeError): return Response({"detail": "invalid amount"}, status=status.HTTP_400_BAD_REQUEST) # Decimal("NaN")/("Infinity") 构造不抛,后续比较/入库才炸 500 → 显式拦(review 确认) if not amount.is_finite() or not bonus_points.is_finite(): return Response({"detail": "invalid amount"}, status=status.HTTP_400_BAD_REQUEST) # 最低 ¥1:分币级支付按 HALF_UP 会到账 0 积分(付了钱拿 0,投诉必至);上限防长尾脏数据 if amount < Decimal("1") or amount > Decimal("1000000"): return Response({"detail": "充值金额需在 ¥1 ~ ¥1,000,000 之间"}, status=status.HTTP_400_BAD_REQUEST) if bonus_points < 0: return Response({"detail": "bonus cannot be negative"}, status=status.HTTP_400_BAD_REQUEST) channel = str(request.data.get("channel") or "manual")[:32] rate = get_billing_config().points_per_yuan # 取整与计价引擎同口径 HALF_UP(quantize 默认是银行家舍入,¥0.25×10 会得 2 而非 3) credited = (amount * rate + bonus_points).quantize(Decimal("1"), rounding=ROUND_HALF_UP) with transaction.atomic(): account, _ = CreditAccount.objects.select_for_update().get_or_create(team=team) account.balance += credited account.save(update_fields=["balance", "updated_at"]) ledger = CreditLedger.objects.create( team=team, user=request.user, ledger_type=CreditLedger.Type.RECHARGE, amount=credited, balance_after=account.balance, reason="团队充值", metadata={ "channel": channel, "paid_amount": str(amount), "points_per_yuan": str(rate), "bonus_points": str(bonus_points), }, ) return Response( { "account": CreditAccountSerializer(account).data, "ledger": CreditLedgerSerializer(ledger).data, "credited_points": str(credited), "points_per_yuan": str(rate), }, status=status.HTTP_201_CREATED, ) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def trend(request): """账户页消费分析:消费趋势(日/周/月可切)+ 本月按阶段/按项目分布。全部来自真实 CHARGE 流水。""" team = get_current_team(request.user) today = timezone.localdate() rng = request.query_params.get("range", "day") charges = CreditLedger.objects.filter(team=team, ledger_type=CreditLedger.Type.CHARGE) def _daily_amounts(win_start): rows = ( charges.filter(created_at__date__gte=win_start) .annotate(day=TruncDate("created_at")) .values("day") .annotate(amount=Sum("amount")) ) return {row["day"]: row["amount"] or Decimal("0") for row in rows} # 按 range 选窗口与分桶:日=近 14 天 / 周=近 8 周 / 月=近 6 个自然月(缺口补 0) series = [] if rng == "week": monday = today - timedelta(days=today.weekday()) starts = [monday - timedelta(weeks=(7 - i)) for i in range(8)] amt_by_day = _daily_amounts(starts[0]) for s in starts: total = sum((amt_by_day.get(s + timedelta(days=k), Decimal("0")) for k in range(7)), Decimal("0")) series.append({"date": s.isoformat(), "label": s.strftime("%m/%d"), "amount": str(total)}) elif rng == "month": seq = [] y, m = today.year, today.month for _ in range(6): seq.append((y, m)) m -= 1 if m == 0: m, y = 12, y - 1 seq.reverse() amt_by_day = _daily_amounts(today.replace(year=seq[0][0], month=seq[0][1], day=1)) for yy, mm in seq: total = sum((v for d, v in amt_by_day.items() if d.year == yy and d.month == mm), Decimal("0")) series.append({"date": f"{yy}-{mm:02d}-01", "label": f"{mm}月", "amount": str(total)}) else: start = today - timedelta(days=13) amt_by_day = _daily_amounts(start) for i in range(14): d = start + timedelta(days=i) series.append({"date": d.isoformat(), "label": d.strftime("%m/%d"), "amount": str(amt_by_day.get(d, Decimal("0")))}) daily = series total_14d = sum((Decimal(s["amount"]) for s in series), Decimal("0")) peak = max((Decimal(s["amount"]) for s in series), default=Decimal("0")) avg = (total_14d / len(series)).quantize(Decimal("0.0001")) if series else Decimal("0") # 按任务类型分布(与后台「任务监控」类型列同源);?month=YYYY-MM 指定月份,默认当月 # 有扣费的类型全部返回,前端按金额降序全部展示,不截断 month_param = (request.query_params.get("month") or "").strip() dist_year, dist_month = today.year, today.month if len(month_param) == 7 and month_param[4] == "-": try: dist_year = int(month_param[:4]) dist_month = int(month_param[5:7]) if not (1 <= dist_month <= 12): raise ValueError("bad month") except ValueError: dist_year, dist_month = today.year, today.month month_param = f"{dist_year:04d}-{dist_month:02d}" else: month_param = f"{dist_year:04d}-{dist_month:02d}" month_start = date(dist_year, dist_month, 1) last_day = calendar.monthrange(dist_year, dist_month)[1] month_end = date(dist_year, dist_month, last_day) month_charges = ( charges.filter(created_at__date__gte=month_start, created_at__date__lte=month_end) .select_related("task") .defer("task__request_payload", "task__response_payload") ) by_stage = {"script": Decimal("0"), "base": Decimal("0"), "storyboard": Decimal("0"), "video": Decimal("0")} by_task_type: dict[str, Decimal] = {} project_amounts: dict[str, Decimal] = {} month_charged = Decimal("0") for row in month_charges: month_charged += row.amount task = row.task bucket = _STAGE_BUCKET.get(task.task_type) if task else None if bucket: by_stage[bucket] += row.amount if task and task.task_type: by_task_type[task.task_type] = by_task_type.get(task.task_type, Decimal("0")) + row.amount pid = str(row.project_id) if row.project_id else None if pid: project_amounts[pid] = project_amounts.get(pid, Decimal("0")) + row.amount return Response( { "daily": daily, "total_14d": str(total_14d), "avg": str(avg), "peak": str(peak), "month": month_param, "month_charged": str(month_charged), "by_stage": {k: str(v) for k, v in by_stage.items()}, "by_task_type": {k: str(v) for k, v in by_task_type.items()}, "by_project": {k: str(v) for k, v in project_amounts.items()}, } )