fix: cron dspy移除import + load_path.py添加cron权限(any) + stat_total_users

- cron/*.dspy: 移除import,改为调用init.py注册的cron_etl_*函数
- init.py: 新增cron_etl_sync/aggregate/provider_cost函数(含IP白名单校验)
- load_path.py: 添加3个cron dspy路径为any角色,补充stat_total_users.ui
- dspy内权限: request['client_ip']不在127.0.0.1则拒绝
This commit is contained in:
yumoqing 2026-07-20 10:55:05 +08:00
parent 8a97ea4a86
commit 4f23e07bcd
5 changed files with 64 additions and 40 deletions

View File

@ -147,6 +147,44 @@ async def j2_avg_ttft(request):
return await get_avg_ttft(env, sor)
# ── Cron ETL 函数 ──
async def cron_etl_sync(request):
"""ETL: 增量同步 llmusage → dm_model_call_fact"""
ip = request['client_ip']
if ip not in ['127.0.0.1']:
return {'error': f'IP {ip} not allowed'}
env = request._run_ns
async with get_sor_context(env, MODULE_NAME) as sor:
await ensure_tables(sor)
count = await sync_call_fact(sor)
return {'status': 'ok', 'synced': count}
async def cron_etl_aggregate(request):
"""ETL: 聚合 dm_model_call_fact → dm_model_perf_daily"""
ip = request['client_ip']
if ip not in ['127.0.0.1']:
return {'error': f'IP {ip} not allowed'}
env = request._run_ns
async with get_sor_context(env, MODULE_NAME) as sor:
await ensure_tables(sor)
count = await aggregate_daily_perf(sor)
return {'status': 'ok', 'aggregated': count}
async def cron_etl_provider_cost(request):
"""ETL: 聚合供应商性价比 dm_provider_cost_daily"""
ip = request['client_ip']
if ip not in ['127.0.0.1']:
return {'error': f'IP {ip} not allowed'}
env = request._run_ns
async with get_sor_context(env, MODULE_NAME) as sor:
await ensure_tables(sor)
count = await aggregate_provider_cost(sor)
return {'status': 'ok', 'aggregated': count}
def load_sage_datamart():
env = ServerEnv()
env.api_model_perf = api_model_perf
@ -164,4 +202,7 @@ def load_sage_datamart():
env.j2_fail_count = j2_fail_count
env.j2_success_rate = j2_success_rate
env.j2_avg_ttft = j2_avg_ttft
debug("[sage_datamart] registered all API endpoints")
env.cron_etl_sync = cron_etl_sync
env.cron_etl_aggregate = cron_etl_aggregate
env.cron_etl_provider_cost = cron_etl_provider_cost
debug("[sage_datamart] registered all API endpoints and cron functions")

View File

@ -1,30 +1,40 @@
#!/usr/bin/env python3
"""sage_datamart 模块 load_path"""
import os, sys, subprocess
"""sage_datamart load_path — register RBAC permissions"""
import os, subprocess
MOD = "sage_datamart"
SAGE_ROOT = os.environ.get("SAGE_ROOT", os.path.expanduser("~/sage"))
SAGE_ROOT = os.path.expanduser("~/repos/sage")
PYTHON = os.path.join(SAGE_ROOT, "py3", "bin", "python")
SET_PERM = os.path.join(SAGE_ROOT, "set_role_perm.py")
# cron endpoints — called via curl from localhost crontab, IP-checked in dspy
PATHS_ANY = [
f"/{MOD}/cron/etl_sync.dspy",
f"/{MOD}/cron/etl_aggregate.dspy",
f"/{MOD}/cron/etl_provider_cost.dspy",
]
PATHS_LOGINED = [
f"/{MOD}",
f"/{MOD}/index.ui",
f"/{MOD}/provider_roi.ui",
f"/{MOD}/menu.ui",
f"/{MOD}/provider_roi.ui",
f"/{MOD}/api/%",
f"/{MOD}/stat_avg_ttft.ui",
f"/{MOD}/stat_success_rate.ui",
f"/{MOD}/stat_total_calls.ui",
f"/{MOD}/stat_total_users.ui",
f"/{MOD}/stat_fail_calls.ui",
f"/{MOD}/stat_today_usage.ui",
f"/{MOD}/stat_today_amount.ui",
f"/{MOD}/api/%",
]
def run(role, path):
subprocess.run([PYTHON, SET_PERM, role, path], capture_output=True)
for p in PATHS_ANY:
run("any", p)
for p in PATHS_LOGINED:
run("logined", p)
print(f"[{MOD}] {len(PATHS_LOGINED)} paths registered")
print(f"[{MOD}] {len(PATHS_ANY)} any + {len(PATHS_LOGINED)} logined paths registered")

View File

@ -1,12 +1,3 @@
# ETL: 每小时聚合 dm_model_call_fact → dm_model_perf_daily
ip = request['client_ip']
if ip not in ['127.0.0.1']:
return ip
import json
from sage_datamart.etl import ensure_tables, aggregate_daily_perf
env = request._run_ns
async with get_sor_context(env, 'sage_datamart') as sor:
await ensure_tables(sor)
count = await aggregate_daily_perf(sor)
return json.dumps({'status': 'ok', 'aggregated': count})
result = await request._run_ns.cron_etl_aggregate(request)
return str(result)

View File

@ -1,12 +1,3 @@
# ETL: 每天聚合供应商性价比 dm_provider_cost_daily
ip = request['client_ip']
if ip not in ['127.0.0.1']:
return ip
import json
from sage_datamart.etl import ensure_tables, aggregate_provider_cost
env = request._run_ns
async with get_sor_context(env, 'sage_datamart') as sor:
await ensure_tables(sor)
count = await aggregate_provider_cost(sor)
return json.dumps({'status': 'ok', 'aggregated': count})
result = await request._run_ns.cron_etl_provider_cost(request)
return str(result)

View File

@ -1,12 +1,3 @@
# ETL: 每5分钟增量同步 llmusage → dm_model_call_fact
ip = request['client_ip']
if ip not in ['127.0.0.1']:
return ip
import json
from sage_datamart.etl import ensure_tables, sync_call_fact
env = request._run_ns
async with get_sor_context(env, 'sage_datamart') as sor:
await ensure_tables(sor)
count = await sync_call_fact(sor)
return json.dumps({'status': 'ok', 'synced': count})
result = await request._run_ns.cron_etl_sync(request)
return str(result)