improve:导出接口护栏与两端控制台查询性能优化
- 导出统一行数上限:xlsx 超限在拉取阶段截停并报错,CSV 到限截断并在文件末尾写入可见标记 - 全部导出接口 OFFSET 循环翻页改为 keyset 游标(用量日志/计费记录/成员/渠道/钱包流水) - 用量日志导出强制时间窗(开始时间必填、跨度≤31天),校验下沉 common 供管理后台与租户端共用 - 管理后台仪表盘趋势/租户排行/模型分布/小时费用聚合加 5 分钟双层缓存 - 交易流水 Username 筛选改 EXISTS 子查询解除对 JOIN 别名的依赖,COUNT 与列表拆分免背联表 - 管理后台成员列表 COUNT 免联表(筛选条件只引用主表列) - 租户端用量日志列表/统计/导出共用筛选构造器,COUNT 与 SUM 全程剥离联表,钱包流水 COUNT 拆分
This commit is contained in:
1 parent
6370e87c8e
commit
62a23325c7
8 files changed
+405
-242
No files matched your search
@@ -1064,7 +1064,14 @@ func (s *sAdmin) ExportChannels(ctx context.Context, req *v1.ChannelExportReq) (
|
||||
}
|
||||
|
||||
return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) {
|
||||
offset := 0
|
||||
// keyset 翻页:导出按优先级倒序展示,游标取 (priority, id) 复合键 ——
|
||||
// priority 大量重复,必须带 id 作 tiebreaker,否则同优先级的行会被跳过或重复。
|
||||
// PostgreSQL 行值比较 (a, b) < (?, ?) 按字典序逐列比较,与排序方向一致。
|
||||
var (
|
||||
hasCursor bool
|
||||
cursorPriority int
|
||||
cursorID int64
|
||||
)
|
||||
for {
|
||||
query := dao.ChnChannels.Ctx(ctx).
|
||||
LeftJoin("chn_health_scores h ON chn_channels.id = h.channel_id")
|
||||
@@ -1087,6 +1094,9 @@ func (s *sAdmin) ExportChannels(ctx context.Context, req *v1.ChannelExportReq) (
|
||||
// 与 ListChannels 保持一致:EXISTS 相关子查询避免 JOIN 重复行
|
||||
query = query.Where("EXISTS (SELECT 1 FROM chn_abilities WHERE chn_abilities.channel_id = chn_channels.id AND (chn_abilities.model_name LIKE ? OR chn_abilities.upstream_model LIKE ?))", "%"+req.Model+"%", "%"+req.Model+"%")
|
||||
}
|
||||
if hasCursor {
|
||||
query = query.Where("(chn_channels.priority, chn_channels.id) < (?, ?)", cursorPriority, cursorID)
|
||||
}
|
||||
var batch []struct {
|
||||
ID int64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
@@ -1097,8 +1107,8 @@ func (s *sAdmin) ExportChannels(ctx context.Context, req *v1.ChannelExportReq) (
|
||||
CreatedAt *gtime.Time `json:"created_at"`
|
||||
HealthScore *float64 `json:"health_score"`
|
||||
}
|
||||
if err := query.Fields(channelFields).OrderDesc("chn_channels.priority").Limit(1000).Offset(offset).Scan(&batch); err != nil {
|
||||
g.Log().Errorf(ctx, "ExportChannels: query batch at offset %d failed: %v", offset, err)
|
||||
if err := query.Fields(channelFields).OrderDesc("chn_channels.priority").OrderDesc("chn_channels.id").Limit(1000).Scan(&batch); err != nil {
|
||||
g.Log().Errorf(ctx, "ExportChannels: query batch failed (cursor priority=%d id=%d): %v", cursorPriority, cursorID, err)
|
||||
return
|
||||
}
|
||||
for _, ch := range batch {
|
||||
@@ -1122,11 +1132,12 @@ func (s *sAdmin) ExportChannels(ctx context.Context, req *v1.ChannelExportReq) (
|
||||
}) {
|
||||
return
|
||||
}
|
||||
cursorPriority, cursorID = ch.Priority, ch.ID
|
||||
}
|
||||
if len(batch) < 1000 {
|
||||
break
|
||||
}
|
||||
offset += 1000
|
||||
hasCursor = true
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -20,6 +20,27 @@ import (
|
||||
"github.com/qianfree/team-api/internal/utility/export"
|
||||
)
|
||||
|
||||
// dashboardCache 仪表盘聚合接口的短缓存(L1 内存 + L2 Redis)。
|
||||
// 趋势 / 租户排行 / 模型分布 / 小时费用这几个接口都是对 bil_usage_logs
|
||||
// 的多天窗口聚合扫描,且数据本身不要求秒级新鲜 —— 5 分钟缓存能消除
|
||||
// 仪表盘反复打开 / 多管理员同时在线造成的绝大部分重复扫表。
|
||||
var dashboardCache = common.NewCache("admin:dashboard", 5*time.Minute)
|
||||
|
||||
// cachedDashboard 按 key 读仪表盘缓存,miss 时回源并写缓存。
|
||||
// key 必须编入全部查询参数(如天数 / TopN),避免不同参数共享一份结果。
|
||||
func cachedDashboard[T any](ctx context.Context, key string, load func() (*T, error)) (*T, error) {
|
||||
var cached T
|
||||
if dashboardCache.GetJSON(ctx, key, &cached) {
|
||||
return &cached, nil
|
||||
}
|
||||
res, err := load()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dashboardCache.Set(ctx, key, res)
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// GetDashboardStats 获取管理后台仪表盘统计
|
||||
func (s *sAdmin) GetDashboardStats(ctx context.Context, req *v1.AdminDashboardReq) (*v1.AdminDashboardRes, error) {
|
||||
today := time.Now().Format("2006-01-02")
|
||||
@@ -133,25 +154,27 @@ func (s *sAdmin) GetDashboardTrends(ctx context.Context, req *v1.AdminDashboardT
|
||||
days = 30
|
||||
}
|
||||
|
||||
startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02")
|
||||
return cachedDashboard(ctx, fmt.Sprintf("trends:%d", days), func() (*v1.AdminDashboardTrendsRes, error) {
|
||||
startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02")
|
||||
|
||||
result, err := g.DB().Ctx(ctx).Raw(`
|
||||
SELECT
|
||||
DATE(created_at) as date,
|
||||
COUNT(*) as requests,
|
||||
COUNT(DISTINCT tenant_id) as active_tenants,
|
||||
COALESCE(SUM(total_cost), 0) as revenue
|
||||
FROM bil_usage_logs
|
||||
WHERE created_at >= ?
|
||||
GROUP BY DATE(created_at)
|
||||
ORDER BY date ASC
|
||||
`, startDate+" 00:00:00").All()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result, err := g.DB().Ctx(ctx).Raw(`
|
||||
SELECT
|
||||
DATE(created_at) as date,
|
||||
COUNT(*) as requests,
|
||||
COUNT(DISTINCT tenant_id) as active_tenants,
|
||||
COALESCE(SUM(total_cost), 0) as revenue
|
||||
FROM bil_usage_logs
|
||||
WHERE created_at >= ?
|
||||
GROUP BY DATE(created_at)
|
||||
ORDER BY date ASC
|
||||
`, startDate+" 00:00:00").All()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
records := result.List()
|
||||
return &v1.AdminDashboardTrendsRes{List: records}, nil
|
||||
records := result.List()
|
||||
return &v1.AdminDashboardTrendsRes{List: records}, nil
|
||||
})
|
||||
}
|
||||
|
||||
// GetTopTenants returns the top 10 tenants by revenue.
|
||||
@@ -161,28 +184,30 @@ func (s *sAdmin) GetTopTenants(ctx context.Context, req *v1.AdminDashboardTopTen
|
||||
days = 30
|
||||
}
|
||||
|
||||
startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02")
|
||||
return cachedDashboard(ctx, fmt.Sprintf("top-tenants:%d", days), func() (*v1.AdminDashboardTopTenantsRes, error) {
|
||||
startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02")
|
||||
|
||||
result, err := g.DB().Ctx(ctx).Raw(`
|
||||
SELECT
|
||||
t.id as tenant_id,
|
||||
t.name as tenant_name,
|
||||
COALESCE(SUM(ul.total_cost), 0) as total_cost,
|
||||
COUNT(*) as requests,
|
||||
COUNT(DISTINCT ul.user_id) as active_members
|
||||
FROM bil_usage_logs ul
|
||||
JOIN tnt_tenants t ON t.id = ul.tenant_id
|
||||
WHERE ul.created_at >= ?
|
||||
GROUP BY t.id, t.name
|
||||
ORDER BY total_cost DESC
|
||||
LIMIT 10
|
||||
`, startDate+" 00:00:00").All()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result, err := g.DB().Ctx(ctx).Raw(`
|
||||
SELECT
|
||||
t.id as tenant_id,
|
||||
t.name as tenant_name,
|
||||
COALESCE(SUM(ul.total_cost), 0) as total_cost,
|
||||
COUNT(*) as requests,
|
||||
COUNT(DISTINCT ul.user_id) as active_members
|
||||
FROM bil_usage_logs ul
|
||||
JOIN tnt_tenants t ON t.id = ul.tenant_id
|
||||
WHERE ul.created_at >= ?
|
||||
GROUP BY t.id, t.name
|
||||
ORDER BY total_cost DESC
|
||||
LIMIT 10
|
||||
`, startDate+" 00:00:00").All()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
records := result.List()
|
||||
return &v1.AdminDashboardTopTenantsRes{List: records}, nil
|
||||
records := result.List()
|
||||
return &v1.AdminDashboardTopTenantsRes{List: records}, nil
|
||||
})
|
||||
}
|
||||
|
||||
// GetModelDistribution returns the model usage distribution.
|
||||
@@ -192,27 +217,29 @@ func (s *sAdmin) GetModelDistribution(ctx context.Context, req *v1.AdminDashboar
|
||||
days = 30
|
||||
}
|
||||
|
||||
startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02")
|
||||
return cachedDashboard(ctx, fmt.Sprintf("model-dist:%d", days), func() (*v1.AdminDashboardModelDistributionRes, error) {
|
||||
startDate := time.Now().AddDate(0, 0, -days).Format("2006-01-02")
|
||||
|
||||
result, err := g.DB().Ctx(ctx).Raw(`
|
||||
SELECT
|
||||
model_name,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(input_tokens), 0) as input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as output_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as total_cost
|
||||
FROM bil_usage_logs
|
||||
WHERE created_at >= ?
|
||||
GROUP BY model_name
|
||||
ORDER BY total_cost DESC
|
||||
LIMIT 20
|
||||
`, startDate+" 00:00:00").All()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result, err := g.DB().Ctx(ctx).Raw(`
|
||||
SELECT
|
||||
model_name,
|
||||
COUNT(*) as requests,
|
||||
COALESCE(SUM(input_tokens), 0) as input_tokens,
|
||||
COALESCE(SUM(output_tokens), 0) as output_tokens,
|
||||
COALESCE(SUM(total_cost), 0) as total_cost
|
||||
FROM bil_usage_logs
|
||||
WHERE created_at >= ?
|
||||
GROUP BY model_name
|
||||
ORDER BY total_cost DESC
|
||||
LIMIT 20
|
||||
`, startDate+" 00:00:00").All()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
records := result.List()
|
||||
return &v1.AdminDashboardModelDistributionRes{List: records}, nil
|
||||
records := result.List()
|
||||
return &v1.AdminDashboardModelDistributionRes{List: records}, nil
|
||||
})
|
||||
}
|
||||
|
||||
// GetModelHourlyCost 模型费用按小时堆叠统计,用于仪表盘堆叠柱状图:
|
||||
@@ -227,6 +254,13 @@ func (s *sAdmin) GetModelHourlyCost(ctx context.Context, req *v1.AdminDashboardM
|
||||
topN = 8
|
||||
}
|
||||
|
||||
return cachedDashboard(ctx, fmt.Sprintf("model-hourly:%d:%d", hours, topN), func() (*v1.AdminDashboardModelHourlyRes, error) {
|
||||
return s.queryModelHourlyCost(ctx, hours, topN)
|
||||
})
|
||||
}
|
||||
|
||||
// queryModelHourlyCost 实际执行小时费用聚合(GetModelHourlyCost 的缓存回源路径)。
|
||||
func (s *sAdmin) queryModelHourlyCost(ctx context.Context, hours, topN int) (*v1.AdminDashboardModelHourlyRes, error) {
|
||||
// 步骤1:取费用最高的 TopN 个模型名
|
||||
topResult, err := g.DB().Ctx(ctx).Raw(`
|
||||
SELECT model_name
|
||||
@@ -1179,6 +1213,15 @@ func (s *sAdmin) GetAllTransactions(ctx context.Context, req *v1.AdminTransactio
|
||||
EndDate: req.EndDate,
|
||||
})
|
||||
|
||||
// COUNT 与列表拆开执行:计数不需要租户/用户名称列,筛选条件也不依赖 JOIN
|
||||
//(Username 过滤在 BuildTransactionQuery 内是 EXISTS 子查询),大表
|
||||
// bil_transactions 的计数因此免背 2 个联表。不用 ScanAndCount —— 它的
|
||||
// COUNT 会携带模型上的全部 JOIN(gf v2.10.2 源码确认:Clone 保留 join)。
|
||||
total, err := query.Count()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
type transactionRow struct {
|
||||
Id int64 `json:"id"`
|
||||
TenantId int64 `json:"tenant_id"`
|
||||
@@ -1194,19 +1237,16 @@ func (s *sAdmin) GetAllTransactions(ctx context.Context, req *v1.AdminTransactio
|
||||
CreatedAt *gtime.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
var records []*transactionRow
|
||||
var total int
|
||||
err := query.Fields("bil_transactions.id, bil_transactions.tenant_id, COALESCE(tn.name, '') AS tenant_name, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.created_at").
|
||||
LeftJoin("tnt_users tu", "bil_transactions.user_id = tu.id AND bil_transactions.tenant_id = tu.tenant_id").
|
||||
LeftJoin("tnt_tenants tn", "bil_transactions.tenant_id = tn.id").
|
||||
OrderDesc("bil_transactions.created_at").
|
||||
Page(page, pageSize).
|
||||
ScanAndCount(&records, &total, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if records == nil {
|
||||
records = make([]*transactionRow, 0)
|
||||
records := make([]*transactionRow, 0)
|
||||
if total > 0 {
|
||||
if err := query.Fields("bil_transactions.id, bil_transactions.tenant_id, COALESCE(tn.name, '') AS tenant_name, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.created_at").
|
||||
LeftJoin("tnt_users tu", "bil_transactions.user_id = tu.id AND bil_transactions.tenant_id = tu.tenant_id").
|
||||
LeftJoin("tnt_tenants tn", "bil_transactions.tenant_id = tn.id").
|
||||
OrderDesc("bil_transactions.created_at").
|
||||
Page(page, pageSize).
|
||||
Scan(&records); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
list := make([]*v1.AdminTransactionItem, 0, len(records))
|
||||
@@ -1279,6 +1319,9 @@ func (s *sAdmin) ExportUsageLogs(ctx context.Context, req *v1.AdminUsageLogExpor
|
||||
if err := validateUsageLogDateRange(req.StartDate, req.EndDate); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := common.ValidateExportTimeWindow(req.StartDate, req.EndDate, common.UsageLogExportMaxWindowDays); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
columns := []export.Column{
|
||||
{Field: "id", Header: "ID"},
|
||||
@@ -1305,14 +1348,29 @@ func (s *sAdmin) ExportUsageLogs(ctx context.Context, req *v1.AdminUsageLogExpor
|
||||
selectFields := "u.id, COALESCE(tn.name, '') AS tenant_name, COALESCE(t.username, '') AS username, u.model_name, u.request_type, u.input_tokens, u.output_tokens, u.total_cost, u.status, u.created_at"
|
||||
|
||||
return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) {
|
||||
offset := 0
|
||||
// keyset(游标)翻页替代 OFFSET:OFFSET 每翻一批都要重新扫过并丢弃前面的
|
||||
// 所有行,深翻页代价线性上涨;游标条件让每批从上一批断点直接续读。
|
||||
// 游标取 id(bigserial 单调递增,PK 索引可直接定位):排序从 created_at DESC
|
||||
// 改为 id DESC,两者在追加写表上近似同序(同为「新的在前」),却避免了
|
||||
// 时间戳做游标时亚秒精度在驱动往返中丢失导致的跳行/重行问题。
|
||||
var cursorID int64
|
||||
for {
|
||||
sql := "SELECT " + selectFields + " FROM " + fromClause + where + " ORDER BY u.created_at DESC LIMIT ? OFFSET ?"
|
||||
// 复制一份再追加分页参数,避免污染循环外构建的共用筛选参数
|
||||
batchArgs := append(append([]any{}, filterArgs...), 1000, offset)
|
||||
sql := "SELECT " + selectFields + " FROM " + fromClause + where
|
||||
batchArgs := append([]any{}, filterArgs...)
|
||||
if cursorID > 0 {
|
||||
// 时间窗必填保证 where 非空,这里安全起见仍处理空 where 的情况
|
||||
if where == "" {
|
||||
sql += " WHERE u.id < ?"
|
||||
} else {
|
||||
sql += " AND u.id < ?"
|
||||
}
|
||||
batchArgs = append(batchArgs, cursorID)
|
||||
}
|
||||
sql += " ORDER BY u.id DESC LIMIT ?"
|
||||
batchArgs = append(batchArgs, 1000)
|
||||
result, err := g.DB().Ctx(ctx).Query(ctx, sql, batchArgs...)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "ExportUsageLogs: query batch at offset %d failed: %v", offset, err)
|
||||
g.Log().Errorf(ctx, "ExportUsageLogs: query batch failed (cursor id=%d): %v", cursorID, err)
|
||||
return
|
||||
}
|
||||
for _, row := range result {
|
||||
@@ -1334,11 +1392,11 @@ func (s *sAdmin) ExportUsageLogs(ctx context.Context, req *v1.AdminUsageLogExpor
|
||||
}) {
|
||||
return
|
||||
}
|
||||
cursorID = row["id"].Int64()
|
||||
}
|
||||
if len(result) < 1000 {
|
||||
break
|
||||
}
|
||||
offset += 1000
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -1364,11 +1422,17 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR
|
||||
Columns: columns,
|
||||
}
|
||||
|
||||
fetchRecords := func(offset, limit int) ([]map[string]any, error) {
|
||||
// fetchRecords 取一批记录并返回本批末行 id 作为下一批游标(0 表示无数据)
|
||||
fetchRecords := func(cursorID int64, limit int) ([]map[string]any, int64, error) {
|
||||
query := dao.BilRecords.Ctx(ctx)
|
||||
if req.TenantID > 0 {
|
||||
query = query.Where("tenant_id", req.TenantID)
|
||||
}
|
||||
// keyset 翻页:id 游标替代 OFFSET,避免深翻页时重复扫弃前面的行;
|
||||
// 排序取 id DESC(追加写表上与 created_at DESC 近似同序)
|
||||
if cursorID > 0 {
|
||||
query = query.Where("id < ?", cursorID)
|
||||
}
|
||||
var records []struct {
|
||||
Id int64 `json:"id"`
|
||||
TenantId int64 `json:"tenant_id"`
|
||||
@@ -1382,8 +1446,8 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR
|
||||
CreatedAt *gtime.Time `json:"created_at"`
|
||||
}
|
||||
if err := query.Fields("id, tenant_id, user_id, channel_id, model_name, input_tokens, output_tokens, total_cost, status, created_at").
|
||||
OrderDesc("created_at").Limit(limit).Offset(offset).Scan(&records); err != nil {
|
||||
return nil, err
|
||||
OrderDesc("id").Limit(limit).Scan(&records); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
// Batch resolve names
|
||||
@@ -1434,6 +1498,7 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR
|
||||
}
|
||||
|
||||
data := make([]map[string]any, len(records))
|
||||
var lastID int64
|
||||
for i, r := range records {
|
||||
data[i] = map[string]any{
|
||||
"id": r.Id,
|
||||
@@ -1447,16 +1512,17 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR
|
||||
"status": r.Status,
|
||||
"created_at": r.CreatedAt.String(),
|
||||
}
|
||||
lastID = r.Id
|
||||
}
|
||||
return data, nil
|
||||
return data, lastID, nil
|
||||
}
|
||||
|
||||
return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) {
|
||||
offset := 0
|
||||
var cursorID int64
|
||||
for {
|
||||
batch, err := fetchRecords(offset, 1000)
|
||||
batch, lastID, err := fetchRecords(cursorID, 1000)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "ExportBillingRecords: query batch at offset %d failed: %v", offset, err)
|
||||
g.Log().Errorf(ctx, "ExportBillingRecords: query batch failed (cursor id=%d): %v", cursorID, err)
|
||||
return
|
||||
}
|
||||
for _, row := range batch {
|
||||
@@ -1467,7 +1533,7 @@ func (s *sAdmin) ExportBillingRecords(ctx context.Context, req *v1.AdminBillingR
|
||||
if len(batch) < 1000 {
|
||||
break
|
||||
}
|
||||
offset += 1000
|
||||
cursorID = lastID
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -130,17 +130,15 @@ func (s *sAdmin) CreateMember(ctx context.Context, req *v1.AdminMemberCreateReq)
|
||||
func (s *sAdmin) ListAllMembers(ctx context.Context, req *v1.AdminMemberListReq) (*v1.AdminMemberListRes, error) {
|
||||
page, pageSize := common.NormalizePagination(req.Page, req.PageSize)
|
||||
|
||||
m := dao.TntUsers.Ctx(ctx).
|
||||
LeftJoin("tnt_tenants t ON tnt_users.tenant_id = t.id")
|
||||
m = buildMemberFilters(m, req.Keyword, req.Status, req.Role, req.TenantID)
|
||||
|
||||
total, err := m.Count()
|
||||
// COUNT 不需要租户名称列,且 buildMemberFilters 的条件只引用 tnt_users.* ——
|
||||
// 计数免联表(联表目标是 tnt_tenants 主键,1:1 不影响行数)
|
||||
total, err := buildMemberFilters(dao.TntUsers.Ctx(ctx), req.Keyword, req.Status, req.Role, req.TenantID).Count()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Rebuild model for data query (Count() modifies internal state)
|
||||
m = dao.TntUsers.Ctx(ctx).
|
||||
// 列表查询才需要 JOIN 补租户名称/代码
|
||||
m := dao.TntUsers.Ctx(ctx).
|
||||
LeftJoin("tnt_tenants t ON tnt_users.tenant_id = t.id")
|
||||
m = buildMemberFilters(m, req.Keyword, req.Status, req.Role, req.TenantID)
|
||||
|
||||
@@ -393,12 +391,16 @@ func (s *sAdmin) ExportMembers(ctx context.Context, req *v1.AdminMemberExportReq
|
||||
}
|
||||
|
||||
return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) {
|
||||
offset := 0
|
||||
// keyset 翻页:id 游标替代 OFFSET,避免深翻页时重复扫弃前面的行(排序本就是 id DESC)
|
||||
var cursorID int64
|
||||
for {
|
||||
m := buildMemberFilters(
|
||||
dao.TntUsers.Ctx(ctx).LeftJoin("tnt_tenants t ON tnt_users.tenant_id = t.id"),
|
||||
req.Keyword, req.Status, req.Role, req.TenantID,
|
||||
)
|
||||
if cursorID > 0 {
|
||||
m = m.Where("tnt_users.id < ?", cursorID)
|
||||
}
|
||||
var batch []struct {
|
||||
Id int64 `json:"id"`
|
||||
TenantName string `json:"tenant_name"`
|
||||
@@ -410,7 +412,7 @@ func (s *sAdmin) ExportMembers(ctx context.Context, req *v1.AdminMemberExportReq
|
||||
LastLoginAt *gtime.Time `json:"last_login_at"`
|
||||
CreatedAt *gtime.Time `json:"created_at"`
|
||||
}
|
||||
if err := m.Fields(memberFields).OrderDesc("tnt_users.id").Limit(1000).Offset(offset).Scan(&batch); err != nil {
|
||||
if err := m.Fields(memberFields).OrderDesc("tnt_users.id").Limit(1000).Scan(&batch); err != nil {
|
||||
return
|
||||
}
|
||||
for _, m := range batch {
|
||||
@@ -427,11 +429,11 @@ func (s *sAdmin) ExportMembers(ctx context.Context, req *v1.AdminMemberExportReq
|
||||
}) {
|
||||
return
|
||||
}
|
||||
cursorID = m.Id
|
||||
}
|
||||
if len(batch) < 1000 {
|
||||
break
|
||||
}
|
||||
offset += 1000
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -36,7 +36,11 @@ func BuildTransactionQuery(ctx context.Context, params TransactionQueryParams) *
|
||||
query = query.Where("bil_transactions.created_at <= ?", params.EndDate+" 23:59:59")
|
||||
}
|
||||
if params.Username != "" {
|
||||
query = query.Where("tu.username LIKE ?", "%"+params.Username+"%")
|
||||
// 用 EXISTS 子查询而非引用调用方 JOIN 的别名(tu.username):
|
||||
// ① 筛选条件因此不依赖任何 JOIN,调用方的 COUNT / 聚合路径可以完全不联表;
|
||||
// ② 避免把 tnt_users 拖成驱动表(EXISTS 内是主键等值探测,代价低)。
|
||||
// 语义与 LEFT JOIN 后过滤一致:user_id 无对应用户的行同样被过滤掉。
|
||||
query = query.Where("EXISTS (SELECT 1 FROM tnt_users ut WHERE ut.id = bil_transactions.user_id AND ut.tenant_id = bil_transactions.tenant_id AND ut.username LIKE ?)", "%"+params.Username+"%")
|
||||
}
|
||||
if params.ModelName != "" {
|
||||
query = query.Where("bil_transactions.model_name LIKE ?", "%"+params.ModelName+"%")
|
||||
|
||||
@@ -2,12 +2,14 @@ package common
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode"
|
||||
|
||||
"github.com/gogf/gf/v2/errors/gcode"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/os/gtime"
|
||||
|
||||
"github.com/qianfree/team-api/internal/consts"
|
||||
)
|
||||
@@ -111,6 +113,37 @@ func EndOfRange(value string) string {
|
||||
return value
|
||||
}
|
||||
|
||||
// UsageLogExportMaxWindowDays 用量日志类导出的时间窗上限(天),管理后台与租户控制台共用。
|
||||
const UsageLogExportMaxWindowDays = 31
|
||||
|
||||
// ValidateExportTimeWindow 校验导出必须限定时间窗:开始时间必填,结束时间缺省按
|
||||
// 「现在」计(与前端「截止留空 = 到现在」的口径一致),窗口不得超过 maxDays 天。
|
||||
//
|
||||
// 列表接口天然有分页 LIMIT 兜底,导出没有 —— 不限窗口等于允许一次请求扫全量大表。
|
||||
// 更长跨度的取数应按窗口分多次导出。
|
||||
func ValidateExportTimeWindow(start, end string, maxDays int) error {
|
||||
if start == "" {
|
||||
return NewBadRequestError(fmt.Sprintf("导出必须指定开始时间,且时间跨度不超过 %d 天", maxDays))
|
||||
}
|
||||
startAt := gtime.NewFromStr(StartOfRange(start))
|
||||
if startAt == nil {
|
||||
return NewBadRequestError("开始时间格式不正确")
|
||||
}
|
||||
endAt := gtime.Now()
|
||||
if end != "" {
|
||||
if endAt = gtime.NewFromStr(EndOfRange(end)); endAt == nil {
|
||||
return NewBadRequestError("结束时间格式不正确")
|
||||
}
|
||||
}
|
||||
if endAt.Before(startAt) {
|
||||
return NewBadRequestError("结束时间不能早于开始时间")
|
||||
}
|
||||
if endAt.Sub(startAt) > time.Duration(maxDays)*24*time.Hour {
|
||||
return NewBadRequestError(fmt.Sprintf("导出时间跨度不能超过 %d 天,请缩小范围后分批导出", maxDays))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ValidateForbiddenWords 校验名称是否包含系统禁用词
|
||||
// 从系统配置 register_forbidden_words 读取逗号分隔的禁用词列表,对 value 做大小写不敏感的包含检查
|
||||
func ValidateForbiddenWords(ctx context.Context, value, fieldName string) error {
|
||||
|
||||
+141
-137
@@ -116,19 +116,24 @@ func (s *sTenant) WalletTransactions(ctx context.Context, req *v1.TenantWalletTr
|
||||
CreatedAt *gtime.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
var records []transactionRow
|
||||
var err error
|
||||
var total int
|
||||
err = query.Fields("bil_transactions.id, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.frozen_after, bil_transactions.related_id, bil_transactions.related_type, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.project_id, bil_transactions.api_key_id, bil_transactions.task_id, bil_transactions.created_at").
|
||||
LeftJoin("tnt_users tu", "bil_transactions.user_id = tu.id AND bil_transactions.tenant_id = tu.tenant_id").
|
||||
OrderDesc("bil_transactions.created_at").
|
||||
Page(page, pageSize).
|
||||
ScanAndCount(&records, &total, false)
|
||||
// COUNT 与列表拆开执行:计数不需要用户名称列,筛选条件也不依赖 JOIN
|
||||
//(Username 过滤在 BuildTransactionQuery 内是 EXISTS 子查询),大表
|
||||
// bil_transactions 的计数免背联表。不用 ScanAndCount —— 它的 COUNT
|
||||
// 会携带模型上的全部 JOIN。
|
||||
total, err := query.Count()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if records == nil {
|
||||
records = []transactionRow{}
|
||||
|
||||
records := []transactionRow{}
|
||||
if total > 0 {
|
||||
if err := query.Fields("bil_transactions.id, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.frozen_after, bil_transactions.related_id, bil_transactions.related_type, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.project_id, bil_transactions.api_key_id, bil_transactions.task_id, bil_transactions.created_at").
|
||||
LeftJoin("tnt_users tu", "bil_transactions.user_id = tu.id AND bil_transactions.tenant_id = tu.tenant_id").
|
||||
OrderDesc("bil_transactions.created_at").
|
||||
Page(page, pageSize).
|
||||
Scan(&records); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
list := make([]map[string]any, 0, len(records))
|
||||
@@ -161,6 +166,63 @@ func (s *sTenant) WalletTransactions(ctx context.Context, req *v1.TenantWalletTr
|
||||
}, nil
|
||||
}
|
||||
|
||||
// tenantUsageLogFilter 用量日志查询的筛选参数(列表 / 统计 / 导出三个入口共用)。
|
||||
type tenantUsageLogFilter struct {
|
||||
TenantID int64
|
||||
UserID int64
|
||||
Role string
|
||||
Username string
|
||||
Model string
|
||||
Status string
|
||||
RequestType int
|
||||
StartDate string
|
||||
EndDate string
|
||||
}
|
||||
|
||||
// buildTenantUsageLogFilter 构建用量日志查询的 WHERE 条件与绑定参数(不含 WHERE 前缀;
|
||||
// tenant_id 恒在条件里,结果必非空)。三个入口共用,保证筛选口径一致。
|
||||
//
|
||||
// 条件只引用主表别名 u,不依赖任何 JOIN —— 计数与统计路径因此可以完全不联表,
|
||||
// 只有需要展示名称的列表 / 导出才挂 LEFT JOIN。租户隔离(u.tenant_id)与
|
||||
// member 角色只见自己(u.user_id)在此统一强制,调用方不得绕过。
|
||||
func buildTenantUsageLogFilter(f tenantUsageLogFilter) (where string, args []any) {
|
||||
conditions := []string{"u.tenant_id = ?"}
|
||||
args = append(args, f.TenantID)
|
||||
|
||||
// member 角色只能查看自己的用量日志
|
||||
if f.Role == "member" {
|
||||
conditions = append(conditions, "u.user_id = ?")
|
||||
args = append(args, f.UserID)
|
||||
} else if f.Username != "" {
|
||||
// EXISTS 子查询替代对 JOIN 别名 t.username 的引用:条件不再依赖联表,
|
||||
// 且避免把 tnt_users 拖成驱动表(EXISTS 内是主键等值探测,代价低)。
|
||||
// 语义与 LEFT JOIN 后过滤一致:用户不存在的行同样被过滤掉。
|
||||
conditions = append(conditions, "EXISTS (SELECT 1 FROM tnt_users ut WHERE ut.id = u.user_id AND ut.tenant_id = u.tenant_id AND ut.username LIKE ?)")
|
||||
args = append(args, "%"+f.Username+"%")
|
||||
}
|
||||
if f.Model != "" {
|
||||
conditions = append(conditions, "u.model_name = ?")
|
||||
args = append(args, f.Model)
|
||||
}
|
||||
if f.Status != "" {
|
||||
conditions = append(conditions, "u.status = ?")
|
||||
args = append(args, f.Status)
|
||||
}
|
||||
if f.RequestType > 0 {
|
||||
conditions = append(conditions, "u.request_type = ?")
|
||||
args = append(args, f.RequestType)
|
||||
}
|
||||
if f.StartDate != "" {
|
||||
conditions = append(conditions, "u.created_at >= ?")
|
||||
args = append(args, common.StartOfRange(f.StartDate))
|
||||
}
|
||||
if f.EndDate != "" {
|
||||
conditions = append(conditions, "u.created_at <= ?")
|
||||
args = append(args, common.EndOfRange(f.EndDate))
|
||||
}
|
||||
return strings.Join(conditions, " AND "), args
|
||||
}
|
||||
|
||||
// UsageLogs 获取租户用量日志
|
||||
func (s *sTenant) UsageLogs(ctx context.Context, req *v1.TenantUsageLogsReq) (*v1.TenantUsageLogsRes, error) {
|
||||
if err := common.ValidateDateTimeParam(req.StartDate, "开始时间"); err != nil {
|
||||
@@ -175,46 +237,25 @@ func (s *sTenant) UsageLogs(ctx context.Context, req *v1.TenantUsageLogsReq) (*v
|
||||
role := middleware.GetUserRole(ctx)
|
||||
page, pageSize := common.NormalizePagination(req.Page, req.PageSize)
|
||||
|
||||
var conditions []string
|
||||
var args []any
|
||||
where, args := buildTenantUsageLogFilter(tenantUsageLogFilter{
|
||||
TenantID: tenantID,
|
||||
UserID: userID,
|
||||
Role: role,
|
||||
Username: req.Username,
|
||||
Model: req.Model,
|
||||
Status: req.Status,
|
||||
RequestType: req.RequestType,
|
||||
StartDate: req.StartDate,
|
||||
EndDate: req.EndDate,
|
||||
})
|
||||
|
||||
conditions = append(conditions, "u.tenant_id = ?")
|
||||
args = append(args, tenantID)
|
||||
|
||||
// member 角色只能查看自己的用量日志
|
||||
if role == "member" {
|
||||
conditions = append(conditions, "u.user_id = ?")
|
||||
args = append(args, userID)
|
||||
} else if req.Username != "" {
|
||||
conditions = append(conditions, "t.username LIKE ?")
|
||||
args = append(args, "%"+req.Username+"%")
|
||||
}
|
||||
if req.Model != "" {
|
||||
conditions = append(conditions, "u.model_name = ?")
|
||||
args = append(args, req.Model)
|
||||
}
|
||||
if req.Status != "" {
|
||||
conditions = append(conditions, "u.status = ?")
|
||||
args = append(args, req.Status)
|
||||
}
|
||||
if req.RequestType > 0 {
|
||||
conditions = append(conditions, "u.request_type = ?")
|
||||
args = append(args, req.RequestType)
|
||||
}
|
||||
if req.StartDate != "" {
|
||||
conditions = append(conditions, "u.created_at >= ?")
|
||||
args = append(args, common.StartOfRange(req.StartDate))
|
||||
}
|
||||
if req.EndDate != "" {
|
||||
conditions = append(conditions, "u.created_at <= ?")
|
||||
args = append(args, common.EndOfRange(req.EndDate))
|
||||
}
|
||||
|
||||
where := strings.Join(conditions, " AND ")
|
||||
// 列表查询不 JOIN mdl_models(model_display_name 仅导出使用)
|
||||
fromClause := "bil_usage_logs u LEFT JOIN tnt_users t ON u.user_id = t.id AND u.tenant_id = t.tenant_id LEFT JOIN tnt_projects p ON u.project_id = p.id LEFT JOIN api_keys ak ON u.api_key_id = ak.id"
|
||||
|
||||
countSQL := "SELECT COUNT(*) AS total FROM " + fromClause + " WHERE " + where
|
||||
// 计数不展示任何名称,且筛选条件只引用 u.*(Username 过滤是 EXISTS 子查询),
|
||||
// 3 个展示用 LEFT JOIN 整体剥离 —— bil_usage_logs 是分区大表,列表每翻一页
|
||||
// 都要重算一次计数,这里是本接口最重的一次查询。
|
||||
countSQL := "SELECT COUNT(*) AS total FROM bil_usage_logs u WHERE " + where
|
||||
countResult, err := g.DB().Ctx(ctx).Query(ctx, countSQL, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -288,53 +329,26 @@ func (s *sTenant) UsageLogsSummary(ctx context.Context, req *v1.TenantUsageLogsS
|
||||
userID := middleware.GetUserID(ctx)
|
||||
role := middleware.GetUserRole(ctx)
|
||||
|
||||
var conditions []string
|
||||
var args []any
|
||||
|
||||
conditions = append(conditions, "u.tenant_id = ?")
|
||||
args = append(args, tenantID)
|
||||
|
||||
// member 角色只能查看自己的用量日志
|
||||
if role == "member" {
|
||||
conditions = append(conditions, "u.user_id = ?")
|
||||
args = append(args, userID)
|
||||
} else if req.Username != "" {
|
||||
conditions = append(conditions, "t.username LIKE ?")
|
||||
args = append(args, "%"+req.Username+"%")
|
||||
}
|
||||
if req.Model != "" {
|
||||
conditions = append(conditions, "u.model_name = ?")
|
||||
args = append(args, req.Model)
|
||||
}
|
||||
if req.Status != "" {
|
||||
conditions = append(conditions, "u.status = ?")
|
||||
args = append(args, req.Status)
|
||||
}
|
||||
if req.RequestType > 0 {
|
||||
conditions = append(conditions, "u.request_type = ?")
|
||||
args = append(args, req.RequestType)
|
||||
}
|
||||
if req.StartDate != "" {
|
||||
conditions = append(conditions, "u.created_at >= ?")
|
||||
args = append(args, common.StartOfRange(req.StartDate))
|
||||
}
|
||||
if req.EndDate != "" {
|
||||
conditions = append(conditions, "u.created_at <= ?")
|
||||
args = append(args, common.EndOfRange(req.EndDate))
|
||||
}
|
||||
|
||||
// 统计聚合不展示名称,仅在按用户名筛选时才 JOIN 用户表,避免大表 SUM 背负无关 JOIN
|
||||
fromClause := "bil_usage_logs u"
|
||||
if role != "member" && req.Username != "" {
|
||||
fromClause += " LEFT JOIN tnt_users t ON u.user_id = t.id AND u.tenant_id = t.tenant_id"
|
||||
}
|
||||
where, args := buildTenantUsageLogFilter(tenantUsageLogFilter{
|
||||
TenantID: tenantID,
|
||||
UserID: userID,
|
||||
Role: role,
|
||||
Username: req.Username,
|
||||
Model: req.Model,
|
||||
Status: req.Status,
|
||||
RequestType: req.RequestType,
|
||||
StartDate: req.StartDate,
|
||||
EndDate: req.EndDate,
|
||||
})
|
||||
|
||||
// 统计聚合不展示名称,Username 过滤已由 EXISTS 子查询承担,全程无需联表:
|
||||
// 大表 SUM 不再背负任何 join。
|
||||
summarySQL := `SELECT
|
||||
COALESCE(SUM(COALESCE(NULLIF(u.actual_cost, 0), u.total_cost)), 0) AS total_cost,
|
||||
COALESCE(SUM(u.output_tokens), 0) AS total_output_tokens,
|
||||
COALESCE(SUM(u.input_tokens), 0) AS total_input_tokens,
|
||||
COALESCE(SUM(u.cache_read_tokens), 0) AS total_cache_read
|
||||
FROM ` + fromClause + ` WHERE ` + strings.Join(conditions, " AND ")
|
||||
FROM bil_usage_logs u WHERE ` + where
|
||||
|
||||
summaryResult, err := g.DB().Ctx(ctx).Query(ctx, summarySQL, args...)
|
||||
if err != nil {
|
||||
@@ -379,6 +393,10 @@ func (s *sTenant) ExportUsageLogs(ctx context.Context, req *v1.TenantUsageLogsEx
|
||||
if err := common.ValidateDateTimeParam(req.EndDate, "结束时间"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// 导出护栏:时间窗必填且不超上限(与管理后台导出同一规则)
|
||||
if err := common.ValidateExportTimeWindow(req.StartDate, req.EndDate, common.UsageLogExportMaxWindowDays); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
tenantID := middleware.GetTenantID(ctx)
|
||||
userID := middleware.GetUserID(ctx)
|
||||
@@ -403,58 +421,39 @@ func (s *sTenant) ExportUsageLogs(ctx context.Context, req *v1.TenantUsageLogsEx
|
||||
Columns: columns,
|
||||
}
|
||||
|
||||
var conditions []string
|
||||
var args []any
|
||||
where, args := buildTenantUsageLogFilter(tenantUsageLogFilter{
|
||||
TenantID: tenantID,
|
||||
UserID: userID,
|
||||
Role: role,
|
||||
Username: req.Username,
|
||||
Model: req.Model,
|
||||
Status: req.Status,
|
||||
RequestType: req.RequestType,
|
||||
StartDate: req.StartDate,
|
||||
EndDate: req.EndDate,
|
||||
})
|
||||
|
||||
conditions = append(conditions, "u.tenant_id = ?")
|
||||
args = append(args, tenantID)
|
||||
|
||||
// member 角色只能导出自己的用量日志
|
||||
if role == "member" {
|
||||
conditions = append(conditions, "u.user_id = ?")
|
||||
args = append(args, userID)
|
||||
} else if req.Username != "" {
|
||||
conditions = append(conditions, "t.username LIKE ?")
|
||||
args = append(args, "%"+req.Username+"%")
|
||||
}
|
||||
if req.Model != "" {
|
||||
conditions = append(conditions, "u.model_name = ?")
|
||||
args = append(args, req.Model)
|
||||
}
|
||||
if req.Status != "" {
|
||||
conditions = append(conditions, "u.status = ?")
|
||||
args = append(args, req.Status)
|
||||
}
|
||||
if req.RequestType > 0 {
|
||||
conditions = append(conditions, "u.request_type = ?")
|
||||
args = append(args, req.RequestType)
|
||||
}
|
||||
if req.StartDate != "" {
|
||||
conditions = append(conditions, "u.created_at >= ?")
|
||||
args = append(args, common.StartOfRange(req.StartDate))
|
||||
}
|
||||
if req.EndDate != "" {
|
||||
conditions = append(conditions, "u.created_at <= ?")
|
||||
args = append(args, common.EndOfRange(req.EndDate))
|
||||
}
|
||||
|
||||
where := strings.Join(conditions, " AND ")
|
||||
fromClause := "bil_usage_logs u LEFT JOIN tnt_users t ON u.user_id = t.id AND u.tenant_id = t.tenant_id LEFT JOIN tnt_projects p ON u.project_id = p.id LEFT JOIN api_keys ak ON u.api_key_id = ak.id LEFT JOIN mdl_models mdl ON u.model_name = mdl.model_id"
|
||||
|
||||
return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) {
|
||||
offset := 0
|
||||
// keyset(游标)翻页替代 OFFSET:OFFSET 每翻一批都要重新扫过并丢弃前面的
|
||||
// 所有行,深翻页代价线性上涨;id 游标(bigserial 单调递增,走 PK 索引)
|
||||
// 让每批从上一批断点续读。排序 created_at DESC → id DESC,追加写表近似同序,
|
||||
// 且避免时间戳做游标时亚秒精度在驱动往返中丢失导致的跳行/重行。
|
||||
var cursorID int64
|
||||
for {
|
||||
dataSQL := fmt.Sprintf(
|
||||
`SELECT u.id, COALESCE(t.username, '') AS username, COALESCE(mdl.model_name, '') AS model_display_name, u.model_name, u.request_type,
|
||||
u.input_tokens, u.output_tokens, u.total_cost, u.status, u.created_at
|
||||
FROM %s WHERE %s ORDER BY u.created_at DESC LIMIT 1000 OFFSET ?`,
|
||||
fromClause, where,
|
||||
)
|
||||
exportArgs := make([]any, len(args)+1)
|
||||
copy(exportArgs, args)
|
||||
exportArgs[len(args)] = offset
|
||||
dataSQL := `SELECT u.id, COALESCE(t.username, '') AS username, COALESCE(mdl.model_name, '') AS model_display_name, u.model_name, u.request_type,
|
||||
u.input_tokens, u.output_tokens, u.total_cost, u.status, u.created_at
|
||||
FROM ` + fromClause + ` WHERE ` + where
|
||||
exportArgs := append([]any{}, args...)
|
||||
if cursorID > 0 {
|
||||
dataSQL += " AND u.id < ?"
|
||||
exportArgs = append(exportArgs, cursorID)
|
||||
}
|
||||
dataSQL += " ORDER BY u.id DESC LIMIT 1000"
|
||||
result, err := g.DB().Ctx(ctx).Query(ctx, dataSQL, exportArgs...)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "ExportUsageLogs: query batch failed (cursor id=%d): %v", cursorID, err)
|
||||
return
|
||||
}
|
||||
for _, row := range result {
|
||||
@@ -475,11 +474,11 @@ func (s *sTenant) ExportUsageLogs(ctx context.Context, req *v1.TenantUsageLogsEx
|
||||
if !yield(m) {
|
||||
return
|
||||
}
|
||||
cursorID = row["id"].Int64()
|
||||
}
|
||||
if len(result) < 1000 {
|
||||
break
|
||||
}
|
||||
offset += 1000
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -509,7 +508,9 @@ func (s *sTenant) ExportWalletTransactions(ctx context.Context, req *v1.TenantWa
|
||||
}
|
||||
|
||||
return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) {
|
||||
offset := 0
|
||||
// keyset 翻页:id 游标替代 OFFSET,避免深翻页时重复扫弃前面的行;
|
||||
// 排序取 id DESC(追加写表上与 created_at DESC 近似同序)
|
||||
var cursorID int64
|
||||
for {
|
||||
type transactionRow struct {
|
||||
Id int64 `json:"id"`
|
||||
@@ -544,11 +545,14 @@ func (s *sTenant) ExportWalletTransactions(ctx context.Context, req *v1.TenantWa
|
||||
if req.ModelName != "" {
|
||||
q = q.Where("model_name LIKE ?", "%"+req.ModelName+"%")
|
||||
}
|
||||
if cursorID > 0 {
|
||||
q = q.Where("id < ?", cursorID)
|
||||
}
|
||||
|
||||
var records []transactionRow
|
||||
err := q.Fields("id, type, amount, balance_after, user_id, request_id, model_name, description, created_at").
|
||||
OrderDesc("created_at").
|
||||
Limit(1000).Offset(offset).
|
||||
OrderDesc("id").
|
||||
Limit(1000).
|
||||
Scan(&records)
|
||||
if err != nil {
|
||||
return
|
||||
@@ -564,11 +568,11 @@ func (s *sTenant) ExportWalletTransactions(ctx context.Context, req *v1.TenantWa
|
||||
}) {
|
||||
return
|
||||
}
|
||||
cursorID = r.Id
|
||||
}
|
||||
if len(records) < 1000 {
|
||||
break
|
||||
}
|
||||
offset += 1000
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1163,7 +1163,8 @@ func (s *sTenant) ExportMembers(ctx context.Context, req *v1.TenantMemberExportR
|
||||
}
|
||||
|
||||
return nil, export.GenericExport(ctx, config, func(yield func(map[string]any) bool) {
|
||||
offset := 0
|
||||
// keyset 翻页:id 游标替代 OFFSET,避免深翻页时重复扫弃前面的行(排序本就是 id DESC)
|
||||
var cursorID int64
|
||||
for {
|
||||
model := dao.TntUsers.Ctx(ctx).
|
||||
Where("tenant_id", tenantID)
|
||||
@@ -1174,6 +1175,9 @@ func (s *sTenant) ExportMembers(ctx context.Context, req *v1.TenantMemberExportR
|
||||
if req.Role != "" {
|
||||
model = model.Where("role", req.Role)
|
||||
}
|
||||
if cursorID > 0 {
|
||||
model = model.Where("id < ?", cursorID)
|
||||
}
|
||||
|
||||
var users []struct {
|
||||
Id int64 `json:"id" orm:"id"`
|
||||
@@ -1188,7 +1192,7 @@ func (s *sTenant) ExportMembers(ctx context.Context, req *v1.TenantMemberExportR
|
||||
QuotaPeriod string `json:"quota_period" orm:"quota_period"`
|
||||
UpdatedAt string `json:"updated_at" orm:"updated_at"`
|
||||
}
|
||||
err := model.OrderDesc("id").Limit(1000).Offset(offset).Scan(&users)
|
||||
err := model.OrderDesc("id").Limit(1000).Scan(&users)
|
||||
if err = common.IgnoreScanNoRows(err); err != nil {
|
||||
return
|
||||
}
|
||||
@@ -1265,11 +1269,11 @@ func (s *sTenant) ExportMembers(ctx context.Context, req *v1.TenantMemberExportR
|
||||
}) {
|
||||
return
|
||||
}
|
||||
cursorID = u.Id
|
||||
}
|
||||
if len(users) < 1000 {
|
||||
break
|
||||
}
|
||||
offset += 1000
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -2,7 +2,9 @@ package export
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
)
|
||||
|
||||
@@ -31,20 +33,57 @@ func (c Config) GetMaxRows() int {
|
||||
|
||||
// GenericExport handles format normalization and xlsx/csv branching.
|
||||
// queryFn yields data rows — same callback as StreamCSV's queryFn.
|
||||
// For xlsx: collects all rows from queryFn, then calls WriteExcel.
|
||||
// For csv: passes queryFn directly to StreamCSV for streaming.
|
||||
// For xlsx: collects rows from queryFn (capped), then calls WriteExcel.
|
||||
// For csv: passes queryFn to StreamCSV for streaming (capped).
|
||||
//
|
||||
// 行数上限在这里统一强制执行(对拉取过程截停,而不是拉完再检查):
|
||||
// 导出的数据源是「无总量保证」的分批循环查询,上限是防止一次导出请求
|
||||
// 拖着数据库扫全量大表的最后一道护栏。
|
||||
// - xlsx:达到上限直接返回错误(此时还未写出任何响应,能干净地报错),
|
||||
// 并且不再继续拉取 —— 旧实现是全部拉进内存后才检查,护栏形同虚设;
|
||||
// - csv:边查边流式写出,表头和数据已经发给客户端,超限时无法再改报错误,
|
||||
// 只能停止拉取并在文件末尾追加一行可见的截断标记,避免使用者
|
||||
// 误把部分数据当作全量。
|
||||
func GenericExport(ctx context.Context, config Config, queryFn func(yield func(map[string]any) bool)) error {
|
||||
r := g.RequestFromCtx(ctx)
|
||||
config.Format = detectFormat(config.Format)
|
||||
maxRows := config.GetMaxRows()
|
||||
|
||||
if config.Format == "xlsx" {
|
||||
var data []map[string]any
|
||||
var (
|
||||
data []map[string]any
|
||||
overflow bool
|
||||
)
|
||||
queryFn(func(row map[string]any) bool {
|
||||
if len(data) >= maxRows {
|
||||
overflow = true
|
||||
return false
|
||||
}
|
||||
data = append(data, row)
|
||||
return true
|
||||
})
|
||||
if overflow {
|
||||
return gerror.Newf("数据量超过 %d 行,请缩小筛选范围后重试,或改用 CSV 格式导出", maxRows)
|
||||
}
|
||||
return WriteExcel(r, config, data)
|
||||
}
|
||||
|
||||
return StreamCSV(r, config, queryFn)
|
||||
truncated := false
|
||||
capped := func(yield func(map[string]any) bool) {
|
||||
rowCount := 0
|
||||
queryFn(func(row map[string]any) bool {
|
||||
if rowCount >= maxRows {
|
||||
truncated = true
|
||||
return false
|
||||
}
|
||||
rowCount++
|
||||
return yield(row)
|
||||
})
|
||||
if truncated && len(config.Columns) > 0 {
|
||||
yield(map[string]any{
|
||||
config.Columns[0].Field: fmt.Sprintf("注意:已达到导出上限 %d 行,结果被截断,请缩小时间范围或筛选条件后分批导出", maxRows),
|
||||
})
|
||||
}
|
||||
}
|
||||
return StreamCSV(r, config, capped)
|
||||
}
|
||||
Reference in new issue
Block a user