zycandClaude Opus 4.8
1d5c2c0910
- core/bug: 各轮电商项目测试清单 xlsx + 验证报告/套图提示词梳理 md - core/qa: consumption_summary / diag_login / diag_sessions / probe_usage 诊断脚本 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
80 lines
3.4 KiB
Python
80 lines
3.4 KiB
Python
"""消费汇总:把 DB 里所有扣费/充值/退费流水按团队、按任务类型、按模型聚合。
|
|
跑法:cd core/backend && DJANGO_SETTINGS_MODULE=airshelf.settings.development .venv/bin/python ../qa/consumption_summary.py
|
|
"""
|
|
import os, sys
|
|
from collections import defaultdict
|
|
from decimal import Decimal
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "airshelf.settings.development")
|
|
import django # noqa: E402
|
|
|
|
django.setup()
|
|
|
|
from django.db.models import Sum, Count # noqa: E402
|
|
from apps.billing.models import CreditLedger # noqa: E402
|
|
from apps.ai.models import AITask # noqa: E402
|
|
|
|
Z = Decimal("0")
|
|
|
|
print("=" * 60)
|
|
print("一、全平台流水汇总(内部额度,单位 CNY)")
|
|
print("=" * 60)
|
|
for row in (
|
|
CreditLedger.objects.values("ledger_type")
|
|
.annotate(total=Sum("amount"), n=Count("id"))
|
|
.order_by("ledger_type")
|
|
):
|
|
print(f" {row['ledger_type']:12} 笔数={row['n']:6} 合计={row['total'] or Z}")
|
|
|
|
charge_total = CreditLedger.objects.filter(ledger_type="charge").aggregate(s=Sum("amount"))["s"] or Z
|
|
recharge_total = CreditLedger.objects.filter(ledger_type="recharge").aggregate(s=Sum("amount"))["s"] or Z
|
|
refund_total = CreditLedger.objects.filter(ledger_type="refund").aggregate(s=Sum("amount"))["s"] or Z
|
|
print(f"\n >>> 实际消费(charge)总额 = {charge_total}")
|
|
print(f" >>> 充值(recharge)总额 = {recharge_total}")
|
|
print(f" >>> 退费(refund)总额 = {refund_total}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("二、按团队拆分消费(charge)")
|
|
print("=" * 60)
|
|
for row in (
|
|
CreditLedger.objects.filter(ledger_type="charge")
|
|
.values("team__name", "team_id")
|
|
.annotate(total=Sum("amount"), n=Count("id"))
|
|
.order_by("-total")
|
|
):
|
|
print(f" team={row['team__name']!r:24} 任务数={row['n']:5} 消费={row['total']}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("三、按任务类型拆分消费(charge 关联的 AITask.task_type)")
|
|
print("=" * 60)
|
|
agg = defaultdict(lambda: [0, Z]) # task_type -> [count, sum]
|
|
qs = (
|
|
CreditLedger.objects.filter(ledger_type="charge", task__isnull=False)
|
|
.select_related("task")
|
|
.values_list("task__task_type", "amount")
|
|
)
|
|
for tt, amt in qs:
|
|
agg[tt][0] += 1
|
|
agg[tt][1] += amt or Z
|
|
for tt, (n, s) in sorted(agg.items(), key=lambda kv: -kv[1][1]):
|
|
print(f" {tt:16} 次数={n:6} 消费={s}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("四、单价表(ModelConfig.unit_price = 平台每次固定收费,非三方实际成本)")
|
|
print("=" * 60)
|
|
from apps.ai.models import ModelConfig # noqa: E402
|
|
for mc in ModelConfig.objects.all().order_by("capability", "name"):
|
|
print(f" {mc.capability:8} {mc.provider.name}:{mc.name:30} 单价={mc.unit_price} status={mc.status}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("五、AITask 成本字段抽样(estimated_cost vs actual_cost)")
|
|
print("=" * 60)
|
|
print(f" task 总数={AITask.objects.count()} 成功={AITask.objects.filter(status='succeeded').count()} 失败={AITask.objects.filter(status='failed').count()}")
|
|
print(f" actual_cost 总和={AITask.objects.aggregate(s=Sum('actual_cost'))['s'] or Z}")
|
|
# 检查 response_payload 里有没有三方返回的 usage/token/费用
|
|
sample = AITask.objects.filter(status="succeeded").exclude(response_payload={}).first()
|
|
if sample:
|
|
keys = list(sample.response_payload.keys()) if isinstance(sample.response_payload, dict) else type(sample.response_payload)
|
|
print(f" response_payload 样本 keys = {keys}")
|