Merge remote-tracking branch 'origin/master' into dev

# Conflicts:
#	src/api/handlers/base/request_builder.py
#	src/api/handlers/openai_cli/adapter.py
#	src/services/system/maintenance_scheduler.py
This commit is contained in:
fawney19
2026-02-05 00:07:29 +08:00
22 changed files with 979 additions and 164 deletions

View File

@@ -15,7 +15,7 @@ export interface CandidateRecord {
key_preview?: string // 密钥脱敏预览(如 sk-***abc
key_capabilities?: Record<string, boolean> | null // Key 支持的能力
required_capabilities?: Record<string, boolean> | null // 请求实际需要的能力标签
status: 'pending' | 'streaming' | 'success' | 'failed' | 'skipped' | 'cancelled'
status: 'pending' | 'streaming' | 'success' | 'failed' | 'skipped' | 'cancelled' | 'available' | 'unused' | 'stream_interrupted'
skip_reason?: string
is_cached: boolean
// 执行结果字段

View File

@@ -433,7 +433,17 @@ const formatLatency = (ms: number | undefined | null): string => {
const timeline = computed<CandidateRecord[]>(() => {
if (!trace.value) return []
return [...trace.value.candidates]
.filter(c => ['success', 'failed', 'skipped', 'available', 'pending', 'streaming'].includes(c.status))
.filter(c => [
'success',
'failed',
'skipped',
'cancelled',
'pending',
'streaming',
'available',
'unused',
'stream_interrupted'
].includes(c.status))
.sort((a, b) => {
const startedA = a.started_at ? new Date(a.started_at).getTime() : Infinity
const startedB = b.started_at ? new Date(b.started_at).getTime() : Infinity
@@ -730,8 +740,10 @@ const formatDuration = (startStr: string, endStr: string): string => {
const getStatusLabel = (status: string) => {
const labels: Record<string, string> = {
available: '未执行',
unused: '未执行',
pending: '等待中',
streaming: '传输中',
stream_interrupted: '流中断',
success: '成功',
failed: '失败',
cancelled: '已取消',
@@ -744,8 +756,10 @@ const getStatusLabel = (status: string) => {
const getStatusColorClass = (status: string) => {
const classes: Record<string, string> = {
available: 'status-available',
unused: 'status-available',
pending: 'status-pending',
streaming: 'status-pending',
stream_interrupted: 'status-failed',
success: 'status-success',
failed: 'status-failed',
cancelled: 'status-cancelled',

View File

@@ -868,7 +868,7 @@ export const MOCK_API_FORMATS = {
{ value: 'claude:chat', label: 'Claude Chat', default_path: '/v1/messages', aliases: [] },
{ value: 'claude:cli', label: 'Claude CLI', default_path: '/v1/messages', aliases: [] },
{ value: 'openai:chat', label: 'OpenAI Chat', default_path: '/v1/chat/completions', aliases: [] },
{ value: 'openai:cli', label: 'OpenAI CLI', default_path: '/responses', aliases: [] },
{ value: 'openai:cli', label: 'OpenAI CLI', default_path: '/v1/responses', aliases: [] },
{ value: 'openai:video', label: 'OpenAI Video', default_path: '/v1/videos', aliases: [] },
{ value: 'gemini:chat', label: 'Gemini Chat', default_path: '/v1beta/models/{model}:{action}', aliases: [] },
{ value: 'gemini:cli', label: 'Gemini CLI', default_path: '/v1beta/models/{model}:{action}', aliases: [] },

View File

@@ -217,7 +217,7 @@ const MOCK_ENDPOINT_STATUS = {
},
{
api_format: 'openai:cli',
api_path: '/responses',
api_path: '/v1/responses',
total_attempts: 2340,
success_count: 2200,
failed_count: 100,

View File

@@ -487,82 +487,141 @@
title="定时任务"
description="配置系统后台定时任务"
>
<template #actions>
<Button
size="sm"
:disabled="checkinTimeLoading || !hasCheckinTimeChanged"
@click="handleCheckinTimeSave"
<div class="space-y-3">
<template
v-for="task in scheduledTasks"
:key="task.id"
>
{{ checkinTimeLoading ? '保存中...' : '保存' }}
</Button>
</template>
<div class="grid grid-cols-1 md:grid-cols-2 gap-6">
<div class="flex items-center space-x-2">
<Switch
id="enable-provider-checkin"
:model-value="systemConfig.enable_provider_checkin"
@update:model-value="handleProviderCheckinToggle"
/>
<div>
<Label
for="enable-provider-checkin"
class="cursor-pointer"
>
启用 Provider 自动签到
</Label>
<p class="text-xs text-muted-foreground">
自动执行已配置 Provider 的签到任务
</p>
</div>
</div>
<div
class="group relative rounded-xl border transition-all duration-300"
:class="task.enabled
? 'border-primary/30 bg-primary/[0.02] shadow-sm shadow-primary/5'
: 'border-border bg-card hover:border-border/80'"
>
<!-- 主行 -->
<div class="flex items-center gap-4 p-4">
<!-- 左侧开关 -->
<div class="shrink-0">
<Switch
:id="`enable-${task.id}`"
:model-value="task.enabled"
@update:model-value="task.onToggle"
/>
</div>
<div v-if="systemConfig.enable_provider_checkin">
<Label class="block text-sm font-medium">
执行时间
</Label>
<div class="mt-1 flex items-center gap-2">
<Select
v-model:open="checkinHourSelectOpen"
:model-value="checkinHour"
@update:model-value="(val: string) => updateCheckinTime(val, checkinMinute)"
>
<SelectTrigger class="w-20">
<SelectValue placeholder="" />
</SelectTrigger>
<SelectContent>
<SelectItem
v-for="h in 24"
:key="h - 1"
:value="String(h - 1).padStart(2, '0')"
<!-- 中间图标标题描述 -->
<div class="flex items-center gap-3 flex-1 min-w-0">
<div
class="w-9 h-9 rounded-lg flex items-center justify-center shrink-0 transition-colors duration-300"
:class="task.enabled
? 'bg-primary/10 text-primary'
: 'text-muted-foreground'"
>
{{ String(h - 1).padStart(2, '0') }}
</SelectItem>
</SelectContent>
</Select>
<span class="text-muted-foreground">:</span>
<Select
v-model:open="checkinMinuteSelectOpen"
:model-value="checkinMinute"
@update:model-value="(val: string) => updateCheckinTime(checkinHour, val)"
>
<SelectTrigger class="w-20">
<SelectValue placeholder="" />
</SelectTrigger>
<SelectContent>
<SelectItem
v-for="m in 60"
:key="m - 1"
:value="String(m - 1).padStart(2, '0')"
<component
:is="task.icon"
class="w-4.5 h-4.5"
/>
</div>
<div class="flex-1 min-w-0">
<h4 class="font-medium text-sm">
{{ task.title }}
</h4>
<p class="text-xs text-muted-foreground mt-0.5 truncate">
{{ task.description }}
</p>
</div>
</div>
<!-- 右侧时间选择器 + 保存按钮 -->
<div
v-if="task.enabled"
class="flex items-center gap-2 shrink-0"
>
<Clock class="w-4 h-4 text-muted-foreground" />
<Select
v-model:open="task.hourSelectOpen.value"
:model-value="task.hour"
@update:model-value="(val: string) => task.updateTime(val, task.minute)"
>
{{ String(m - 1).padStart(2, '0') }}
</SelectItem>
</SelectContent>
</Select>
<SelectTrigger class="w-14 h-8 text-xs">
<SelectValue placeholder="" />
</SelectTrigger>
<SelectContent>
<SelectItem
v-for="h in 24"
:key="h - 1"
:value="String(h - 1).padStart(2, '0')"
>
{{ String(h - 1).padStart(2, '0') }}
</SelectItem>
</SelectContent>
</Select>
<span class="text-sm text-muted-foreground">:</span>
<Select
v-model:open="task.minuteSelectOpen.value"
:model-value="task.minute"
@update:model-value="(val: string) => task.updateTime(task.hour, val)"
>
<SelectTrigger class="w-14 h-8 text-xs">
<SelectValue placeholder="" />
</SelectTrigger>
<SelectContent>
<SelectItem
v-for="m in 60"
:key="m - 1"
:value="String(m - 1).padStart(2, '0')"
>
{{ String(m - 1).padStart(2, '0') }}
</SelectItem>
</SelectContent>
</Select>
<Button
v-if="task.hasChanges"
variant="default"
size="sm"
class="h-8 px-2.5 text-xs"
:disabled="task.loading"
@click="task.onSave"
>
<Check
v-if="!task.loading"
class="w-3.5 h-3.5"
/>
<Loader2
v-else
class="w-3.5 h-3.5 animate-spin"
/>
</Button>
</div>
</div>
<!-- 额外配置区域(仅用户配额重置任务有) -->
<div
v-if="task.id === 'user-quota-reset' && task.enabled"
class="px-4 pb-4 pt-0"
>
<div class="flex items-center gap-3 p-3 rounded-lg bg-muted/30 border border-border/50">
<div class="flex items-center gap-2 text-sm">
<span class="text-muted-foreground">重置周期</span>
<div class="flex items-center gap-1.5">
<span class="text-muted-foreground">每</span>
<Input
v-model.number="systemConfig.user_quota_reset_interval_days"
type="number"
min="1"
step="1"
class="w-14 h-7 text-xs text-center px-2"
/>
<span class="text-muted-foreground">天</span>
</div>
</div>
</div>
<p class="text-[11px] text-muted-foreground mt-2 ml-1">
滚动计算:距离上次成功执行满 N 天后再次执行
</p>
</div>
</div>
<p class="mt-1 text-xs text-muted-foreground">
每天定时执行24小时制
</p>
</div>
</template>
</div>
</CardSection>
@@ -936,7 +995,7 @@
<script setup lang="ts">
import { ref, computed, onMounted } from 'vue'
import { Download, Upload } from 'lucide-vue-next'
import { Download, Upload, CalendarCheck, RotateCcw, Clock, Check, Loader2 } from 'lucide-vue-next'
import Button from '@/components/ui/button.vue'
import Input from '@/components/ui/input.vue'
import Label from '@/components/ui/label.vue'
@@ -982,6 +1041,9 @@ interface SystemConfig {
// 定时任务
enable_provider_checkin: boolean
provider_checkin_time: string
enable_user_quota_reset: boolean
user_quota_reset_time: string
user_quota_reset_interval_days: number
}
const basicConfigLoading = ref(false)
@@ -1039,6 +1101,9 @@ const systemConfig = ref<SystemConfig>({
// 定时任务
enable_provider_checkin: true,
provider_checkin_time: '01:05',
enable_user_quota_reset: false,
user_quota_reset_time: '05:00',
user_quota_reset_interval_days: 1,
})
// 原始配置值(用于检测变动)
@@ -1147,6 +1212,9 @@ async function loadSystemConfig() {
// 定时任务
'enable_provider_checkin',
'provider_checkin_time',
'enable_user_quota_reset',
'user_quota_reset_time',
'user_quota_reset_interval_days',
]
for (const key of configs) {
@@ -1163,6 +1231,9 @@ async function loadSystemConfig() {
originalConfig.value = JSON.parse(JSON.stringify(systemConfig.value))
// 初始化签到时间的原始值(用于回滚)
previousCheckinTime.value = systemConfig.value.provider_checkin_time
// 初始化配额重置时间的原始值(用于回滚)
previousUserQuotaResetTime.value = systemConfig.value.user_quota_reset_time
previousUserQuotaResetIntervalDays.value = systemConfig.value.user_quota_reset_interval_days
} catch (err) {
error('加载系统配置失败')
log.error('加载系统配置失败:', err)
@@ -1305,9 +1376,9 @@ async function handleProviderCheckinToggle(enabled: boolean) {
}
}
// 签到时间相关
const previousCheckinTime = ref('')
const checkinTimeLoading = ref(false)
const checkinConfigLoading = ref(false)
const quotaResetConfigLoading = ref(false)
const checkinHourSelectOpen = ref(false)
const checkinMinuteSelectOpen = ref(false)
@@ -1334,16 +1405,103 @@ const hasCheckinTimeChanged = computed(() => {
return systemConfig.value.provider_checkin_time !== previousCheckinTime.value
})
async function handleUserQuotaResetToggle(enabled: boolean) {
const previousValue = systemConfig.value.enable_user_quota_reset
systemConfig.value.enable_user_quota_reset = enabled
try {
await adminApi.updateSystemConfig(
'enable_user_quota_reset',
enabled,
'是否启用用户配额自动重置任务'
)
success(enabled ? '已启用用户配额自动重置' : '已禁用用户配额自动重置')
} catch (err) {
error('保存配置失败')
log.error('保存用户配额自动重置配置失败:', err)
// 回滚状态
systemConfig.value.enable_user_quota_reset = previousValue
}
}
// 用户配额重置时间相关
const previousUserQuotaResetTime = ref('')
const userQuotaResetHourSelectOpen = ref(false)
const userQuotaResetMinuteSelectOpen = ref(false)
const previousUserQuotaResetIntervalDays = ref(1)
const userQuotaResetHour = computed(() => {
const time = systemConfig.value.user_quota_reset_time
if (!time || !time.includes(':')) return '05'
return time.split(':')[0]
})
const userQuotaResetMinute = computed(() => {
const time = systemConfig.value.user_quota_reset_time
if (!time || !time.includes(':')) return '00'
return time.split(':')[1]
})
function updateUserQuotaResetTime(hour: string, minute: string) {
systemConfig.value.user_quota_reset_time = `${hour}:${minute}`
}
const hasUserQuotaResetTimeChanged = computed(() => {
return systemConfig.value.user_quota_reset_time !== previousUserQuotaResetTime.value
})
const hasUserQuotaResetIntervalChanged = computed(() => {
return systemConfig.value.user_quota_reset_interval_days !== previousUserQuotaResetIntervalDays.value
})
const hasQuotaResetConfigChanged = computed(() => {
return hasUserQuotaResetTimeChanged.value || hasUserQuotaResetIntervalChanged.value
})
// 定时任务配置列表
const scheduledTasks = computed(() => [
{
id: 'provider-checkin',
icon: CalendarCheck,
title: 'Provider 自动签到',
description: '自动执行已配置 Provider 的签到任务',
enabled: systemConfig.value.enable_provider_checkin,
hour: checkinHour.value,
minute: checkinMinute.value,
hourSelectOpen: checkinHourSelectOpen,
minuteSelectOpen: checkinMinuteSelectOpen,
updateTime: updateCheckinTime,
hasChanges: hasCheckinTimeChanged.value,
loading: checkinConfigLoading.value,
onToggle: handleProviderCheckinToggle,
onSave: handleCheckinTimeSave,
},
{
id: 'user-quota-reset',
icon: RotateCcw,
title: '用户配额自动重置',
description: '定时将用户已使用配额重置为零',
enabled: systemConfig.value.enable_user_quota_reset,
hour: userQuotaResetHour.value,
minute: userQuotaResetMinute.value,
hourSelectOpen: userQuotaResetHourSelectOpen,
minuteSelectOpen: userQuotaResetMinuteSelectOpen,
updateTime: updateUserQuotaResetTime,
hasChanges: hasQuotaResetConfigChanged.value,
loading: quotaResetConfigLoading.value,
onToggle: handleUserQuotaResetToggle,
onSave: handleQuotaResetConfigSave,
},
])
// Provider 签到时间保存
async function handleCheckinTimeSave() {
const newTime = systemConfig.value.provider_checkin_time
// 验证时间格式
if (!newTime || !/^\d{2}:\d{2}$/.test(newTime)) {
error('请输入有效的时间格式 (HH:MM)')
return
}
checkinTimeLoading.value = true
checkinConfigLoading.value = true
try {
await adminApi.updateSystemConfig(
'provider_checkin_time',
@@ -1356,7 +1514,71 @@ async function handleCheckinTimeSave() {
error('保存签到时间失败')
log.error('保存签到时间失败:', err)
} finally {
checkinTimeLoading.value = false
checkinConfigLoading.value = false
}
}
// 用户配额重置配置保存
async function handleQuotaResetConfigSave() {
const configItems: Array<{ key: string, value: any, description: string, onSuccess: () => void }> = []
if (hasUserQuotaResetTimeChanged.value) {
const newTime = systemConfig.value.user_quota_reset_time
if (!newTime || !/^\d{2}:\d{2}$/.test(newTime)) {
error('请输入有效的时间格式 (HH:MM)')
return
}
configItems.push({
key: 'user_quota_reset_time',
value: newTime,
description: '用户配额自动重置执行时间HH:MM 格式)',
onSuccess: () => {
previousUserQuotaResetTime.value = newTime
}
})
}
if (hasUserQuotaResetIntervalChanged.value) {
let intervalDays = Number(systemConfig.value.user_quota_reset_interval_days)
if (!Number.isFinite(intervalDays) || intervalDays < 1) intervalDays = 1
intervalDays = Math.trunc(intervalDays)
systemConfig.value.user_quota_reset_interval_days = intervalDays
configItems.push({
key: 'user_quota_reset_interval_days',
value: intervalDays,
description: '用户配额重置周期(天数),滚动计算',
onSuccess: () => {
previousUserQuotaResetIntervalDays.value = intervalDays
}
})
}
if (configItems.length === 0) return
quotaResetConfigLoading.value = true
const failedKeys: string[] = []
try {
for (const item of configItems) {
try {
await adminApi.updateSystemConfig(item.key, item.value, item.description)
item.onSuccess()
} catch (err) {
failedKeys.push(item.key)
log.error(`保存配额重置配置失败: ${item.key}`, err)
}
}
if (failedKeys.length > 0) {
error(`部分配置保存失败: ${failedKeys.join(', ')}`)
return
}
success('配额重置配置已保存')
} finally {
quotaResetConfigLoading.value = false
}
}

View File

@@ -490,6 +490,16 @@ async def test_model(
"temperature": 0.7,
}
# 获取端点规则(不在此处应用,传递给 check_endpoint 在格式转换后应用)
body_rules = getattr(endpoint, "body_rules", None)
header_rules = getattr(endpoint, "header_rules", None)
extra_headers = endpoint_config.get("extra_headers") or {}
if body_rules:
logger.debug(f"[test-model] 将传递 body_rules 给 check_endpoint: {body_rules}")
if header_rules:
logger.debug(f"[test-model] 将传递 header_rules 给 check_endpoint: {header_rules}")
# 发送测试请求
async with httpx.AsyncClient(
timeout=endpoint_config["timeout"], verify=get_ssl_context()
@@ -501,7 +511,10 @@ async def test_model(
endpoint_config["base_url"],
endpoint_config["api_key"],
check_request,
endpoint_config.get("extra_headers"),
extra_headers if extra_headers else None,
# 端点规则(在 check_endpoint 内部格式转换后应用)
body_rules=body_rules,
header_rules=header_rules,
# 用量计算参数(现在强制记录)
db=db,
user=current_user,

View File

@@ -339,29 +339,29 @@ async def import_models_from_upstream(
"""
从上游提供商导入模型
从上游提供商导入模型列表。导入的模型作为独立的 ProviderModel 存储,
不会自动创建 GlobalModel。后续需要手动关联 GlobalModel 才能参与路由。
从上游提供商导入模型列表。自动匹配已有的 GlobalModel如果不存在则自动创建。
**流程说明**:
1. 检查模型是否已存在于当前 Provider按 provider_model_name 匹配)
2. 创建新的 ProviderModelglobal_model_id = NULL
3. 支持设置价格覆盖tiered_pricing, price_per_request
2. 尝试按名称精确匹配已有的 GlobalModel
3. 如果没有匹配到,自动创建新的 GlobalModel
4. 创建 Model 记录并关联到 GlobalModel
**路径参数**:
- `provider_id`: 提供商 ID
**请求体字段**:
- `model_ids`: 模型 ID 数组(必填,每个 ID 长度 1-100 字符)
- `tiered_pricing`: 可选的阶梯计费配置(应用于所有导入的模型)
- `price_per_request`: 可选的按次计费价格(应用于所有导入的模型)
- `tiered_pricing`: 可选的阶梯计费配置(应用于所有导入的模型和新创建的 GlobalModel
- `price_per_request`: 可选的按次计费价格(应用于所有导入的模型和新创建的 GlobalModel
**返回字段**:
- `success`: 成功导入的模型数组,每项包含:
- `model_id`: 模型 ID
- `provider_model_id`: 提供商模型 ID
- `global_model_id`: 全局模型 ID(如果已关联)
- `global_model_name`: 全局模型名称(如果已关联)
- `created_global_model`: 是否新创建了全局模型(始终为 false
- `global_model_id`: 全局模型 ID
- `global_model_name`: 全局模型名称
- `created_global_model`: 是否新创建了全局模型
- `errors`: 失败的模型数组,每项包含:
- `model_id`: 模型 ID
- `error`: 错误信息
@@ -657,7 +657,7 @@ class AdminBatchAssignModelsToProviderAdapter(AdminApiAdapter):
@dataclass
class AdminImportFromUpstreamAdapter(AdminApiAdapter):
"""从上游提供商导入模型(创建 GlobalModel,作为独立 ProviderModel"""
"""从上游提供商导入模型(自动匹配或创建 GlobalModel"""
provider_id: str
payload: ImportFromUpstreamRequest
@@ -682,6 +682,17 @@ class AdminImportFromUpstreamAdapter(AdminApiAdapter):
):
price_per_request = self.payload.price_per_request
# 默认价格配置(用于自动创建的 GlobalModel
default_pricing = {
"tiers": [
{
"up_to": None,
"input_price_per_1m": 0.0,
"output_price_per_1m": 0.0,
}
]
}
for model_id in self.payload.model_ids:
# 输入验证:检查 model_id 长度
if not model_id or len(model_id) > 100:
@@ -726,10 +737,33 @@ class AdminImportFromUpstreamAdapter(AdminApiAdapter):
)
continue
# 2. 创建新的 Model 记录(不关联 GlobalModel
# 2. 尝试匹配已有的 GlobalModel(按名称精确匹配
global_model = (
db.query(GlobalModel).filter(GlobalModel.name == model_id).first()
)
created_global_model = False
# 3. 如果没有匹配到,自动创建新的 GlobalModel
if not global_model:
global_model = GlobalModel(
name=model_id,
display_name=model_id,
default_tiered_pricing=tiered_pricing or default_pricing,
default_price_per_request=price_per_request,
is_active=True,
)
db.add(global_model)
db.flush()
created_global_model = True
logger.info(
f"Auto-created GlobalModel: {model_id} for provider {provider.name} "
f"by {context.user.username}"
)
# 4. 创建新的 Model 记录(关联到 GlobalModel
new_model = Model(
provider_id=self.provider_id,
global_model_id=None, # 独立模型,不关联 GlobalModel
global_model_id=global_model.id,
provider_model_name=model_id,
is_active=True,
tiered_pricing=tiered_pricing,
@@ -743,14 +777,15 @@ class AdminImportFromUpstreamAdapter(AdminApiAdapter):
success.append(
ImportFromUpstreamSuccessItem(
model_id=model_id,
global_model_id="", # 未关联
global_model_name="", # 未关联
global_model_id=global_model.id,
global_model_name=global_model.name,
provider_model_id=new_model.id,
created_global_model=False,
created_global_model=created_global_model,
)
)
logger.info(
f"Created independent ProviderModel: {model_id} for provider {provider.name}"
f"Imported model: {model_id} -> GlobalModel: {global_model.name} "
f"(created={created_global_model}) for provider {provider.name}"
)
except Exception as e:
# 回滚到 savepoint
@@ -762,9 +797,11 @@ class AdminImportFromUpstreamAdapter(AdminApiAdapter):
db.commit()
logger.info(
f"Imported {len(success)} independent models to provider {provider.name} by {context.user.username}"
f"Imported {len(success)} models to provider {provider.name} by {context.user.username}"
)
# 不需要清除 /v1/models 缓存,因为独立模型不参与路由
# 清除 /v1/models 列表缓存(导入的模型现在参与路由
if success:
await invalidate_models_list_cache()
return ImportFromUpstreamResponse(success=success, errors=errors)

View File

@@ -624,6 +624,16 @@ class AdminSetSystemConfigAdapter(AdminApiAdapter):
except Exception as e:
logger.warning(f"更新签到任务时间失败: {e}")
# 如果更新的是用户配额重置任务时间,动态更新调度器
if self.key == "user_quota_reset_time" and value:
try:
from src.services.system.maintenance_scheduler import get_maintenance_scheduler
scheduler = get_maintenance_scheduler()
scheduler.update_user_quota_reset_time(value)
except Exception as e:
logger.warning(f"更新用户配额重置任务时间失败: {e}")
# 返回时不暴露加密后的值
display_value = "********" if self.key in self.ENCRYPTED_KEYS else config.value

View File

@@ -126,16 +126,7 @@ class MessageTelemetry:
# Provider 响应元数据(如 Gemini 的 modelVersion
response_metadata: dict[str, Any] | None = None,
) -> float:
total_cost = await self.calculate_cost(
provider,
model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cache_creation_tokens=cache_creation_tokens,
cache_read_tokens=cache_read_tokens,
)
await UsageService.record_usage(
usage = await UsageService.record_usage(
db=self.db,
user=self.user,
api_key=self.api_key,
@@ -170,6 +161,8 @@ class MessageTelemetry:
metadata=response_metadata,
)
total_cost = float(getattr(usage, "total_cost_usd", 0.0) or 0.0)
if self.user and self.api_key:
audit_service.log_api_request(
db=self.db,
@@ -181,8 +174,8 @@ class MessageTelemetry:
success=True,
ip_address=self.client_ip,
status_code=status_code,
input_tokens=input_tokens,
output_tokens=output_tokens,
input_tokens=getattr(usage, "input_tokens", input_tokens),
output_tokens=getattr(usage, "output_tokens", output_tokens),
cost_usd=total_cost,
)

View File

@@ -630,6 +630,9 @@ class ChatAdapterBase(ApiAdapter):
api_key: str,
request_data: dict[str, Any],
extra_headers: dict[str, str] | None = None,
# 端点规则参数
body_rules: list[dict[str, Any]] | None = None,
header_rules: list[dict[str, Any]] | None = None,
# 用量计算参数(现在强制记录)
db: Any | None = None,
user: Any | None = None,
@@ -647,6 +650,8 @@ class ChatAdapterBase(ApiAdapter):
api_key: API 密钥(已解密)
request_data: 请求数据
extra_headers: 端点配置的额外请求头
body_rules: 请求体规则(在格式转换后应用)
header_rules: 请求头规则(在请求头构建后应用)
db: 数据库会话
user: 用户对象
provider_name: 提供商名称
@@ -658,12 +663,30 @@ class ChatAdapterBase(ApiAdapter):
测试响应数据
"""
from src.api.handlers.base.endpoint_checker import run_endpoint_check
from src.api.handlers.base.request_builder import apply_body_rules
from src.core.api_format.headers import HeaderBuilder
# 使用子类配置方法构建请求组件
url = cls.build_endpoint_url(base_url)
headers = cls.build_headers_with_extra(api_key, extra_headers)
body = cls.build_request_body(request_data)
# 应用请求体规则(在格式转换后应用,确保规则效果不被覆盖)
if body_rules:
body = apply_body_rules(body, body_rules)
# 应用请求头规则(在请求头构建后应用)
if header_rules:
# 获取认证头名称,防止被规则覆盖
from src.core.api_format import get_auth_config_for_endpoint
auth_header, _ = get_auth_config_for_endpoint(cls.FORMAT_ID)
protected_keys = {auth_header.lower(), "content-type"}
header_builder = HeaderBuilder()
header_builder.add_many(headers)
header_builder.apply_rules(header_rules, protected_keys)
headers = header_builder.build()
# 使用通用的endpoint checker执行请求
return await run_endpoint_check(
client=client,

View File

@@ -598,6 +598,9 @@ class CliAdapterBase(ApiAdapter):
api_key: str,
request_data: dict[str, Any],
extra_headers: dict[str, str] | None = None,
# 端点规则参数
body_rules: list[dict[str, Any]] | None = None,
header_rules: list[dict[str, Any]] | None = None,
# 用量计算参数
db: Any | None = None,
user: Any | None = None,
@@ -622,6 +625,8 @@ class CliAdapterBase(ApiAdapter):
api_key: API 密钥(已解密)
request_data: 请求数据
extra_headers: 端点配置的额外请求头
body_rules: 请求体规则(在格式转换后应用)
header_rules: 请求头规则(在请求头构建后应用)
db: 数据库会话
user: 用户对象
provider_name: 提供商名称
@@ -633,6 +638,8 @@ class CliAdapterBase(ApiAdapter):
测试响应数据
"""
from src.api.handlers.base.endpoint_checker import run_endpoint_check
from src.api.handlers.base.request_builder import apply_body_rules
from src.core.api_format.headers import HeaderBuilder
# 构建请求组件
url = cls.build_endpoint_url(base_url, request_data, model_name)
@@ -646,6 +653,22 @@ class CliAdapterBase(ApiAdapter):
headers = cls.build_headers_with_extra(api_key, merged_extra if merged_extra else None)
body = cls.build_request_body(request_data, base_url=base_url)
# 应用请求体规则(在格式转换后应用,确保规则效果不被覆盖)
if body_rules:
body = apply_body_rules(body, body_rules)
# 应用请求头规则(在请求头构建后应用)
if header_rules:
# 获取认证头名称,防止被规则覆盖
from src.core.api_format import get_auth_config_for_endpoint
auth_header, _ = get_auth_config_for_endpoint(cls.FORMAT_ID)
protected_keys = {auth_header.lower(), "content-type"}
header_builder = HeaderBuilder()
header_builder.add_many(headers)
header_builder.apply_rules(header_rules, protected_keys)
headers = header_builder.build()
# 获取有效的模型名称
effective_model_name = model_name or request_data.get("model")

View File

@@ -1922,10 +1922,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
logger.warning(f"[{ctx.request_id}] 流式请求失败,未选中提供商")
return
# Claude API 的 input_tokens 已经是非缓存部分,不需要再减去 cached_tokens
# 实际计费的输入 tokens = input_tokens + cache_creation_tokens缓存读取免费或折扣
actual_input_tokens = ctx.input_tokens
# 获取新的 DB session
db_gen = get_db()
bg_db = next(db_gen)
@@ -1987,7 +1983,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
is_stream=True,
api_format=ctx.api_format,
provider_request_headers=ctx.provider_request_headers,
input_tokens=actual_input_tokens,
input_tokens=ctx.input_tokens,
output_tokens=ctx.output_tokens,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
@@ -2001,7 +1997,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
logger.debug(f"{self.FORMAT_ID} 流式响应被客户端取消")
logger.info(
f"[CANCEL] {self.request_id[:8]} | {ctx.model} | {ctx.provider_name} | {response_time_ms}ms | "
f"{ctx.status_code} | in:{actual_input_tokens} out:{ctx.output_tokens} cache:{ctx.cached_tokens}"
f"{ctx.status_code} | in:{ctx.input_tokens} out:{ctx.output_tokens} cache:{ctx.cached_tokens}"
)
else:
# 服务端/上游异常:记录为失败
@@ -2017,7 +2013,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
api_format=ctx.api_format,
provider_request_headers=ctx.provider_request_headers,
# 预估 token 信息(来自 message_start 事件)
input_tokens=actual_input_tokens,
input_tokens=ctx.input_tokens,
output_tokens=ctx.output_tokens,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
@@ -2033,7 +2029,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
logger.debug(f"{self.FORMAT_ID} 流式响应中断")
logger.info(
f"[FAIL] {self.request_id[:8]} | {ctx.model} | {ctx.provider_name} | {response_time_ms}ms | "
f"{ctx.status_code} | in:{actual_input_tokens} out:{ctx.output_tokens} cache:{ctx.cached_tokens}"
f"{ctx.status_code} | in:{ctx.input_tokens} out:{ctx.output_tokens} cache:{ctx.cached_tokens}"
)
else:
# 在记录统计前,允许子类从 parsed_chunks 中提取额外的元数据
@@ -2052,12 +2048,12 @@ class CliMessageHandlerBase(BaseMessageHandler):
logger.debug(
f"[{ctx.request_id}] 开始记录 Usage: "
f"provider={ctx.provider_name}, model={ctx.model}, "
f"in={actual_input_tokens}, out={ctx.output_tokens}"
f"in={ctx.input_tokens}, out={ctx.output_tokens}"
)
total_cost = await bg_telemetry.record_success(
provider=ctx.provider_name,
model=ctx.model,
input_tokens=actual_input_tokens,
input_tokens=ctx.input_tokens,
output_tokens=ctx.output_tokens,
response_time_ms=response_time_ms,
first_byte_time_ms=ctx.first_byte_time_ms, # 传递首字时间
@@ -2545,8 +2541,6 @@ class CliMessageHandlerBase(BaseMessageHandler):
output_tokens = usage.get("output_tokens", 0)
cached_tokens = usage.get("cache_read_tokens", 0)
cache_creation_tokens = usage.get("cache_creation_tokens", 0)
# Claude API 的 input_tokens 已经是非缓存部分,不需要再减去 cached_tokens
actual_input_tokens = input_tokens
output_text = self.parser.extract_text_content(response_json)[:200]
@@ -2560,7 +2554,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
total_cost = await self.telemetry.record_success(
provider=provider_name,
model=model,
input_tokens=actual_input_tokens,
input_tokens=input_tokens,
output_tokens=output_tokens,
response_time_ms=response_time_ms,
status_code=status_code,

View File

@@ -252,6 +252,9 @@ class GeminiChatAdapter(ChatAdapterBase):
api_key: str,
request_data: dict[str, Any],
extra_headers: dict[str, str] | None = None,
# 端点规则参数
body_rules: list[dict[str, Any]] | None = None,
header_rules: list[dict[str, Any]] | None = None,
# 用量计算参数
db: Any | None = None,
user: Any | None = None,
@@ -261,6 +264,10 @@ class GeminiChatAdapter(ChatAdapterBase):
model_name: str | None = None,
) -> dict[str, Any]:
"""测试 Gemini API 模型连接性(非流式)"""
from src.api.handlers.base.endpoint_checker import run_endpoint_check
from src.api.handlers.base.request_builder import apply_body_rules
from src.core.api_format.headers import HeaderBuilder
# Gemini需要从request_data或model_name参数获取model名称
effective_model_name = model_name or request_data.get("model", "")
if not effective_model_name:
@@ -277,8 +284,21 @@ class GeminiChatAdapter(ChatAdapterBase):
headers = cls.build_headers_with_extra(api_key, extra_headers)
body = cls.build_request_body(request_data)
# 使用基类的通用endpoint checker
from src.api.handlers.base.endpoint_checker import run_endpoint_check
# 应用请求体规则(在格式转换后应用,确保规则效果不被覆盖)
if body_rules:
body = apply_body_rules(body, body_rules)
# 应用请求头规则(在请求头构建后应用)
if header_rules:
# 获取认证头名称,防止被规则覆盖
from src.core.api_format import get_auth_config_for_endpoint
auth_header, _ = get_auth_config_for_endpoint(cls.FORMAT_ID)
protected_keys = {auth_header.lower(), "content-type"}
header_builder = HeaderBuilder()
header_builder.add_many(headers)
header_builder.apply_rules(header_rules, protected_keys)
headers = header_builder.build()
return await run_endpoint_check(
client=client,

View File

@@ -116,7 +116,7 @@ _ENDPOINT_DEFINITIONS: dict[tuple[ApiFamily, EndpointKind], EndpointDefinition]
api_family=ApiFamily.OPENAI,
endpoint_kind=EndpointKind.CLI,
aliases=("openai_cli", "responses"),
default_path="/responses",
default_path="/v1/responses",
auth_method=AuthMethod.BEARER,
auth_header="Authorization",
auth_type="bearer",

View File

@@ -0,0 +1,47 @@
"""
计费相关 token 归一化工具。
"""
from __future__ import annotations
from src.core.api_format.enums import ApiFamily
from src.core.api_format.signature import parse_signature_key
def _get_api_family(api_format: str | None) -> ApiFamily | None:
"""解析 api_format 字符串,返回对应的 ApiFamily 枚举。"""
if not api_format:
return None
text = str(api_format).strip()
if not text:
return None
sig = parse_signature_key(text)
return sig.api_family
def normalize_input_tokens_for_billing(
api_format: str | None,
input_tokens: int,
cache_read_tokens: int,
) -> int:
"""
归一化 `input_tokens`,使其在计费中表示"非缓存输入 token"
计费口径:`input_tokens`=非缓存输入 token`cache_read_tokens`=缓存命中 token折扣/免费维度)。
- Claude 系:保持上游口径(不扣除),因为 Claude API 的 input_tokens 本身就不包含缓存部分。
- OpenAI 系:`input_tokens` 包含缓存命中部分,需要扣除 `cache_read_tokens`。
- Gemini 系:`promptTokenCount` 包含 `cachedContentTokenCount`,需要扣除。
"""
if input_tokens <= 0:
return 0 if input_tokens == 0 else input_tokens
if cache_read_tokens <= 0:
return input_tokens
api_family = _get_api_family(api_format)
if api_family == ApiFamily.CLAUDE:
return input_tokens
if api_family in (ApiFamily.OPENAI, ApiFamily.GEMINI):
return max(input_tokens - cache_read_tokens, 0)
# 未知格式,保守处理,不扣除
return input_tokens

View File

@@ -626,13 +626,15 @@ class CacheAwareScheduler:
target_format,
)
# 0. 解析 model_name 到 GlobalModel支持直接匹配和映射名匹配,使用 ModelCacheService
global_model = await ModelCacheService.resolve_global_model_by_name_or_mapping(
db, model_name
)
# 0. 解析 model_name 到 GlobalModel仅接受 GlobalModel.name
normalized_name = model_name.strip() if isinstance(model_name, str) else ""
if not normalized_name:
logger.warning("GlobalModel not found: <empty model name>")
raise ModelNotSupportedException(model=model_name)
if not global_model:
logger.warning(f"GlobalModel not found: {model_name}")
global_model = await ModelCacheService.get_global_model_by_name(db, normalized_name)
if not global_model or not global_model.is_active:
logger.warning(f"GlobalModel not found or inactive: {normalized_name}")
raise ModelNotSupportedException(model=model_name)
logger.debug(
@@ -828,14 +830,12 @@ class CacheAwareScheduler:
- 模型支持的能力是全局的,与具体的 Key 无关
- 如果模型不支持某能力,整个 Provider 的所有 Key 都应该被跳过
支持两种匹配方式:
1. 直接匹配 GlobalModel.name
2. 通过 ModelCacheService 匹配映射名(全局查找)
支持直接匹配 GlobalModel.name外部请求不接受映射名
Args:
db: 数据库会话
provider: Provider 对象
model_name: 模型名称(可以是 GlobalModel.name 或映射名
model_name: 模型名称(必须是 GlobalModel.name
is_stream: 是否是流式请求,如果为 True 则同时检查流式支持
capability_requirements: 能力需求(可选),用于检查模型是否支持所需能力
@@ -849,14 +849,14 @@ class CacheAwareScheduler:
# Avoid holding a DB connection while awaiting cache/Redis inside ModelCacheService.
self._release_db_connection_before_await(db)
# 使用 ModelCacheService 解析模型名称(支持映射名)
global_model = await ModelCacheService.resolve_global_model_by_name_or_mapping(
db, model_name
)
# 仅接受 GlobalModel.name不允许映射名)
normalized_name = model_name.strip() if isinstance(model_name, str) else ""
if not normalized_name:
return False, "模型不存在或名称无效", None, None
if not global_model:
# 完全未找到匹配
return False, "模型不存在或 Provider 未配置此模型", None, None
global_model = await ModelCacheService.get_global_model_by_name(db, normalized_name)
if not global_model or not global_model.is_active:
return False, "模型不存在或已停用", None, None
# 找到 GlobalModel 后,检查当前 Provider 是否支持
is_supported, skip_reason, caps, provider_model_names = (

View File

@@ -86,30 +86,33 @@ class ModelMapperMiddleware:
获取模型映射
简化后的逻辑:
1. 通过 GlobalModel.name 或映射名解析 GlobalModel
1. 通过 GlobalModel.name 解析 GlobalModel
2. 找到 GlobalModel 后,查找该 Provider 的 Model 实现
Args:
source_model: 用户请求的模型名(可以是 GlobalModel.name 或映射名
source_model: 用户请求的模型名(必须是 GlobalModel.name
provider_id: 提供商ID (UUID)
Returns:
模型映射对象(包含 model 字段如果没有找到返回None
"""
# 检查缓存
cache_key = f"{provider_id}:{source_model}"
# 步骤 1: 规范化模型名称
normalized_name = source_model.strip() if isinstance(source_model, str) else ""
if not normalized_name:
logger.debug("GlobalModel not found: <empty model name>")
return None
# 检查缓存(使用规范化后的名称)
cache_key = f"{provider_id}:{normalized_name}"
if cache_key in self._cache:
return self._cache[cache_key]
mapping = None
# 步骤 1: 解析 GlobalModel(支持映射名)
global_model = await ModelCacheService.resolve_global_model_by_name_or_mapping(
self.db, source_model
)
global_model = await ModelCacheService.get_global_model_by_name(self.db, normalized_name)
if not global_model:
logger.debug(f"GlobalModel not found: {source_model}")
if not global_model or not global_model.is_active:
logger.debug(f"GlobalModel not found or inactive: {normalized_name}")
self._cache[cache_key] = None
return None
@@ -132,7 +135,7 @@ class ModelMapperMiddleware:
)()
logger.debug(
f"Found model mapping: {source_model} -> {model.provider_model_name} "
f"Found model mapping: {normalized_name} -> {model.provider_model_name} "
f"(provider={provider_id[:8]}...)"
)

View File

@@ -123,6 +123,18 @@ class SystemConfigService:
"value": "01:05",
"description": "Provider 自动签到执行时间HH:MM 格式24小时制",
},
"enable_user_quota_reset": {
"value": False,
"description": "是否启用用户配额自动重置任务(按配置时间触发,按周期执行)",
},
"user_quota_reset_time": {
"value": "05:00",
"description": "用户配额自动重置执行时间HH:MM 格式24小时制",
},
"user_quota_reset_interval_days": {
"value": 1,
"description": "用户配额重置周期(天数)",
},
"provider_priority_mode": {
"value": "provider",
"description": "优先级策略provider(提供商优先模式) 或 global_key(全局Key优先模式)",

View File

@@ -41,6 +41,8 @@ class MaintenanceScheduler:
CHECKIN_JOB_ID = "provider_checkin"
# OAuth 刷新任务的 job_id
OAUTH_REFRESH_JOB_ID = "oauth_token_refresh"
# 用户配额重置任务的 job_id
USER_QUOTA_RESET_JOB_ID = "user_quota_reset"
def __init__(self) -> None:
self.running = False
@@ -72,6 +74,19 @@ class MaintenanceScheduler:
finally:
db.close()
def _get_user_quota_reset_time(self) -> tuple[int, int]:
"""获取用户配额重置任务的执行时间
Returns:
(hour, minute) 元组
"""
db = create_session()
try:
time_str = SystemConfigService.get_config(db, "user_quota_reset_time", "05:00")
return self._parse_user_quota_reset_time_string(time_str)
finally:
db.close()
@staticmethod
def _parse_time_string(time_str: str) -> tuple[int, int]:
"""解析时间字符串为 (hour, minute) 元组
@@ -95,6 +110,26 @@ class MaintenanceScheduler:
except (ValueError, IndexError):
return (1, 5)
@staticmethod
def _parse_user_quota_reset_time_string(time_str: str) -> tuple[int, int]:
"""解析用户配额重置时间字符串为 (hour, minute) 元组
Returns:
(hour, minute) 元组,解析失败返回默认值 (5, 0)
"""
try:
if not time_str or ":" not in time_str:
return (5, 0)
parts = time_str.split(":")
hour = int(parts[0])
minute = int(parts[1])
# 验证范围
if 0 <= hour <= 23 and 0 <= minute <= 59:
return (hour, minute)
return (5, 0)
except (ValueError, IndexError):
return (5, 0)
def update_checkin_time(self, time_str: str) -> bool:
"""更新签到任务的执行时间
@@ -118,6 +153,29 @@ class MaintenanceScheduler:
return success
def update_user_quota_reset_time(self, time_str: str) -> bool:
"""更新用户配额重置任务的执行时间
Args:
time_str: HH:MM 格式的时间字符串
Returns:
是否成功更新
"""
hour, minute = self._parse_user_quota_reset_time_string(time_str)
scheduler = get_scheduler()
success = scheduler.reschedule_cron_job(
self.USER_QUOTA_RESET_JOB_ID,
hour=hour,
minute=minute,
)
if success:
logger.info(f"用户配额重置任务时间已更新为: {hour:02d}:{minute:02d}")
return success
def get_checkin_job_info(self) -> dict | None:
"""获取签到任务的信息
@@ -223,6 +281,16 @@ class MaintenanceScheduler:
# 启动时先执行一次,计算下次执行时间
asyncio.create_task(self._schedule_next_oauth_refresh())
# 用户配额重置任务 - 根据配置时间执行(按周期配置决定是否执行)
quota_reset_hour, quota_reset_minute = self._get_user_quota_reset_time()
scheduler.add_cron_job(
self._scheduled_user_quota_reset,
hour=quota_reset_hour,
minute=quota_reset_minute,
job_id=self.USER_QUOTA_RESET_JOB_ID,
name="用户配额自动重置",
)
# 启动时执行一次初始化任务
asyncio.create_task(self._run_startup_tasks())
@@ -440,6 +508,10 @@ class MaintenanceScheduler:
except Exception:
pass
async def _scheduled_user_quota_reset(self) -> None:
"""用户配额重置任务(定时调用)"""
await self._perform_user_quota_reset()
# ========== 实际任务实现 ==========
async def _perform_stats_aggregation(self, backfill: bool = False) -> None:
@@ -843,6 +915,108 @@ class MaintenanceScheduler:
if db is not None:
db.close()
async def _perform_user_quota_reset(self) -> None:
"""执行用户配额自动重置任务
适用范围:
- 未删除is_deleted=false
- 仅对 quota_usd != NULL 的用户生效
"""
db = create_session()
try:
# 检查是否启用用户配额重置
if not SystemConfigService.get_config(db, "enable_user_quota_reset", False):
logger.info("用户配额自动重置已禁用,跳过任务")
return
# 重置周期(天数),不限制上限
interval_value = SystemConfigService.get_config(db, "user_quota_reset_interval_days", 1)
try:
interval_days = int(interval_value)
except Exception:
interval_days = 1
if interval_days < 1:
interval_days = 1
# 滚动计算根据上次执行日APP_TIMEZONE判断是否到期
last_reset_at = SystemConfigService.get_config(db, "user_quota_last_reset_at")
should_run = True
if last_reset_at:
last_dt: datetime | None = None
try:
if isinstance(last_reset_at, str):
last_dt = datetime.fromisoformat(last_reset_at)
except Exception:
last_dt = None
if last_dt is None:
logger.warning("user_quota_last_reset_at 格式无效,视为需要执行一次")
else:
if last_dt.tzinfo is None:
last_dt = last_dt.replace(tzinfo=timezone.utc)
from zoneinfo import ZoneInfo
from src.services.system.scheduler import APP_TIMEZONE
tz = ZoneInfo(APP_TIMEZONE)
now_local = datetime.now(tz)
last_local_date = last_dt.astimezone(tz).date()
days_since_reset = (now_local.date() - last_local_date).days
if days_since_reset < 0:
logger.warning(
"user_quota_last_reset_at 在未来,跳过本次用户配额自动重置"
)
should_run = False
elif days_since_reset < interval_days:
logger.info(
f"用户配额自动重置未到周期,跳过任务({days_since_reset}/{interval_days}天)"
)
should_run = False
if not should_run:
return
from src.models.database import User as DBUser
now_utc = datetime.now(timezone.utc)
reset_count = (
db.query(DBUser)
.filter(
DBUser.is_deleted.is_(False),
DBUser.quota_usd.isnot(None),
)
.update(
{
DBUser.used_usd: 0.0,
DBUser.updated_at: now_utc,
},
synchronize_session=False,
)
)
db.commit()
# 记录 last_reset_at成功执行后更新滚动计算用
SystemConfigService.set_config(
db,
"user_quota_last_reset_at",
now_utc.isoformat(),
"用户配额自动重置的上次执行时间UTC内部使用",
)
logger.info(f"用户配额自动重置完成: interval_days={interval_days}, 重置用户数={reset_count}")
except Exception as e:
logger.exception(f"用户配额自动重置任务执行失败: {e}")
try:
db.rollback()
except Exception:
pass
finally:
db.close()
async def _perform_cleanup(self) -> None:
"""执行清理任务"""
db = create_session()

View File

@@ -25,6 +25,7 @@ from src.models.database import (
User,
UserRole,
)
from src.services.billing.token_normalization import normalize_input_tokens_for_billing
from src.services.model.cost import ModelCostService
from src.services.system.config import SystemConfigService
from src.services.usage.error_classifier import classify_error
@@ -843,9 +844,28 @@ class UsageService:
Returns:
(usage_params 字典, total_cost 总成本)
"""
# 计费口径以 Provider 为准(优先 endpoint_api_format
billing_api_format: str | None = None
if params.endpoint_api_format:
try:
billing_api_format = normalize_signature_key(str(params.endpoint_api_format))
except Exception:
billing_api_format = None
if billing_api_format is None and params.api_format:
try:
billing_api_format = normalize_signature_key(str(params.api_format))
except Exception:
billing_api_format = None
input_tokens_for_billing = normalize_input_tokens_for_billing(
billing_api_format,
params.input_tokens,
params.cache_read_input_tokens,
)
# 获取费率倍数和是否免费套餐(传递 api_format 支持按格式配置的倍率)
actual_rate_multiplier, is_free_tier = await cls._get_rate_multiplier_and_free_tier(
params.db, params.provider_api_key_id, params.provider_id, params.api_format
params.db, params.provider_api_key_id, params.provider_id, billing_api_format
)
metadata = dict(params.metadata or {})
@@ -884,7 +904,7 @@ class UsageService:
request_count = 0 if is_failed_request else 1
dims: dict[str, Any] = {
"input_tokens": params.input_tokens,
"input_tokens": input_tokens_for_billing,
"output_tokens": params.output_tokens,
"cache_creation_input_tokens": params.cache_creation_input_tokens,
"cache_read_input_tokens": params.cache_read_input_tokens,
@@ -956,11 +976,11 @@ class UsageService:
db=params.db,
provider=params.provider,
model=params.model,
input_tokens=params.input_tokens,
input_tokens=input_tokens_for_billing,
output_tokens=params.output_tokens,
cache_creation_input_tokens=params.cache_creation_input_tokens,
cache_read_input_tokens=params.cache_read_input_tokens,
api_format=params.api_format,
api_format=billing_api_format,
cache_ttl_minutes=params.cache_ttl_minutes,
use_tiered_pricing=params.use_tiered_pricing,
is_failed_request=is_failed_request,
@@ -989,8 +1009,8 @@ class UsageService:
provider_id=params.provider_id,
model=params.model,
task_type=billing_task_type,
api_format=params.api_format,
input_tokens=params.input_tokens,
api_format=billing_api_format,
input_tokens=input_tokens_for_billing,
output_tokens=params.output_tokens,
cache_creation_input_tokens=params.cache_creation_input_tokens,
cache_read_input_tokens=params.cache_read_input_tokens,
@@ -1019,7 +1039,7 @@ class UsageService:
api_key=params.api_key,
provider=params.provider,
model=params.model,
input_tokens=params.input_tokens,
input_tokens=input_tokens_for_billing,
output_tokens=params.output_tokens,
cache_creation_input_tokens=params.cache_creation_input_tokens,
cache_read_input_tokens=params.cache_read_input_tokens,

View File

@@ -0,0 +1,44 @@
from src.services.billing.token_normalization import normalize_input_tokens_for_billing
from src.services.billing.usage_mapper import UsageMapper
class TestNormalizeInputTokensForBilling:
def test_openai_family_subtracts_cached_tokens(self) -> None:
assert normalize_input_tokens_for_billing("openai:cli", 160_070, 81_664) == 78_406
def test_claude_family_does_not_change(self) -> None:
assert normalize_input_tokens_for_billing("claude:cli", 160_070, 81_664) == 160_070
def test_gemini_family_subtracts_cached_tokens(self) -> None:
# Gemini 的 promptTokenCount 包含 cachedContentTokenCount需要扣除
assert normalize_input_tokens_for_billing("gemini:chat", 323_392, 323_384) == 8
assert normalize_input_tokens_for_billing("gemini:cli", 100, 20) == 80
def test_missing_format_does_not_change(self) -> None:
assert normalize_input_tokens_for_billing(None, 100, 20) == 100
assert normalize_input_tokens_for_billing("", 100, 20) == 100
def test_clamps_when_cached_tokens_exceed_input(self) -> None:
assert normalize_input_tokens_for_billing("openai:cli", 10, 20) == 0
assert normalize_input_tokens_for_billing("gemini:chat", 10, 20) == 0
class TestUsageMapperOpenAICacheTokens:
def test_openai_mapping_maps_cached_tokens_details(self) -> None:
raw_usage = {
"prompt_tokens": 100,
"completion_tokens": 50,
"prompt_tokens_details": {"cached_tokens": 20},
}
usage = UsageMapper.map(raw_usage, api_format="openai:chat")
assert usage.input_tokens == 100
assert usage.output_tokens == 50
assert usage.cache_read_tokens == 20
def test_openai_mapping_without_cached_tokens_is_unchanged(self) -> None:
raw_usage = {"prompt_tokens": 100, "completion_tokens": 50}
usage = UsageMapper.map(raw_usage, api_format="openai:chat")
assert usage.input_tokens == 100
assert usage.output_tokens == 50
assert usage.cache_read_tokens == 0

View File

@@ -0,0 +1,166 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock
import pytest
from src.services.system.maintenance_scheduler import MaintenanceScheduler
@pytest.mark.asyncio
async def test_user_quota_reset_disabled(monkeypatch):
scheduler = MaintenanceScheduler()
mock_db = MagicMock()
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.create_session",
lambda: mock_db,
)
def fake_get_config(cls, db, key, default=None):
if key == "enable_user_quota_reset":
return False
return default
mock_set_config = MagicMock()
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.SystemConfigService.get_config",
classmethod(fake_get_config),
)
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.SystemConfigService.set_config",
mock_set_config,
)
await scheduler._perform_user_quota_reset()
assert not mock_db.query.called
assert not mock_db.commit.called
assert not mock_set_config.called
@pytest.mark.asyncio
async def test_user_quota_reset_not_due_skips(monkeypatch):
scheduler = MaintenanceScheduler()
mock_db = MagicMock()
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.create_session",
lambda: mock_db,
)
last_reset_at = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
def fake_get_config(cls, db, key, default=None):
if key == "enable_user_quota_reset":
return True
if key == "user_quota_reset_interval_days":
return 2
if key == "user_quota_last_reset_at":
return last_reset_at
return default
mock_set_config = MagicMock()
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.SystemConfigService.get_config",
classmethod(fake_get_config),
)
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.SystemConfigService.set_config",
mock_set_config,
)
await scheduler._perform_user_quota_reset()
assert not mock_db.query.called
assert not mock_db.commit.called
assert not mock_set_config.called
@pytest.mark.asyncio
async def test_user_quota_reset_due_runs(monkeypatch):
scheduler = MaintenanceScheduler()
mock_db = MagicMock()
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.create_session",
lambda: mock_db,
)
last_reset_at = (datetime.now(timezone.utc) - timedelta(days=2)).isoformat()
def fake_get_config(cls, db, key, default=None):
if key == "enable_user_quota_reset":
return True
if key == "user_quota_reset_interval_days":
return 2
if key == "user_quota_last_reset_at":
return last_reset_at
return default
mock_set_config = MagicMock()
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.SystemConfigService.get_config",
classmethod(fake_get_config),
)
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.SystemConfigService.set_config",
mock_set_config,
)
mock_query = MagicMock()
mock_filter = MagicMock()
mock_filter.update.return_value = 7
mock_query.filter.return_value = mock_filter
mock_db.query.return_value = mock_query
await scheduler._perform_user_quota_reset()
mock_db.query.assert_called_once()
mock_filter.update.assert_called_once()
_, update_kwargs = mock_filter.update.call_args
assert update_kwargs["synchronize_session"] is False
mock_db.commit.assert_called_once()
mock_set_config.assert_called_once()
@pytest.mark.asyncio
async def test_user_quota_reset_invalid_interval_defaults_to_1(monkeypatch):
scheduler = MaintenanceScheduler()
mock_db = MagicMock()
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.create_session",
lambda: mock_db,
)
def fake_get_config(cls, db, key, default=None):
if key == "enable_user_quota_reset":
return True
if key == "user_quota_reset_interval_days":
return "abc"
if key == "user_quota_last_reset_at":
return None
return default
mock_set_config = MagicMock()
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.SystemConfigService.get_config",
classmethod(fake_get_config),
)
monkeypatch.setattr(
"src.services.system.maintenance_scheduler.SystemConfigService.set_config",
mock_set_config,
)
mock_query = MagicMock()
mock_filter = MagicMock()
mock_filter.update.return_value = 1
mock_query.filter.return_value = mock_filter
mock_db.query.return_value = mock_query
await scheduler._perform_user_quota_reset()
mock_db.commit.assert_called_once()
mock_set_config.assert_called_once()