#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ routes/admin_observability_routes.py Operation Ollama-First v5.0 / Phase 27 — Admin Observability Dashboard 提供 admin 介面看戰役累積的觀測資料: /observability/ai_calls — ai_calls 即時查詢(含篩選 / 圖表) /observability/promotion_review — Phase 28 PromotionGate 待審核列表 /observability/quality_trend — Phase 25 caller 反饋趨勢 /observability/host_health — 三主機 Ollama + MCP 健康度 設計原則: - 純讀(除了 promotion approve/reject 是 mutation) - 失敗安全:DB 失敗回空清單 + 警告 banner - 每頁 100 筆分頁,無限捲動 - 不暴露 secret / prompt 原文 """ from datetime import datetime, timedelta from flask import Blueprint, render_template, request, jsonify from sqlalchemy import text as sa_text from database.manager import get_session admin_observability_bp = Blueprint( 'admin_observability', __name__, url_prefix='/observability', ) # ───────────────────────────────────────────────────────────────────────────── # /observability/ai_calls — Phase 27 主入口 # ───────────────────────────────────────────────────────────────────────────── @admin_observability_bp.route('/ai_calls') def ai_calls_dashboard(): """ai_calls 表觀測 dashboard(24h 預設視窗)""" hours = int(request.args.get('hours', '24')) caller_filter = request.args.get('caller', '').strip() provider_filter = request.args.get('provider', '').strip() since = datetime.now() - timedelta(hours=hours) session = get_session() try: # 1. 總覽 summary = session.execute( sa_text(""" SELECT COUNT(*) AS total_calls, COALESCE(SUM(input_tokens + output_tokens), 0) AS total_tokens, COALESCE(SUM(cost_usd), 0) AS total_cost, COALESCE(AVG(duration_ms), 0) AS avg_duration, COUNT(*) FILTER (WHERE status = 'ok') AS ok_calls, COUNT(*) FILTER (WHERE status NOT IN ('ok','cache_only')) AS error_calls, COUNT(*) FILTER (WHERE rag_hit) AS rag_hits, COUNT(*) FILTER (WHERE cache_hit) AS cache_hits FROM ai_calls WHERE called_at >= :since """), {'since': since}, ).fetchone() # 2. by provider by_provider = session.execute( sa_text(""" SELECT provider, COUNT(*) AS calls, COALESCE(SUM(input_tokens + output_tokens), 0) AS tokens, COALESCE(SUM(cost_usd), 0) AS cost FROM ai_calls WHERE called_at >= :since GROUP BY provider ORDER BY tokens DESC """), {'since': since}, ).fetchall() # 3. TOP 20 calls(最近)— 動態 WHERE where_parts = ["called_at >= :since"] params = {'since': since} if caller_filter: where_parts.append("caller = :caller") params['caller'] = caller_filter if provider_filter: where_parts.append("provider = :provider") params['provider'] = provider_filter recent = session.execute( sa_text(f""" SELECT id, called_at, caller, provider, model, input_tokens, output_tokens, duration_ms, status, cost_usd, cache_hit, rag_hit FROM ai_calls WHERE {' AND '.join(where_parts)} ORDER BY called_at DESC LIMIT 100 """), params, ).fetchall() # 4. caller 列表(給篩選 dropdown) callers = session.execute( sa_text(""" SELECT DISTINCT caller FROM ai_calls WHERE called_at >= :since ORDER BY caller """), {'since': since}, ).fetchall() return render_template( 'admin/ai_calls_dashboard.html', hours=hours, caller_filter=caller_filter, provider_filter=provider_filter, summary={ 'total_calls': int(summary[0] or 0), 'total_tokens': int(summary[1] or 0), 'total_cost': float(summary[2] or 0), 'avg_duration': int(summary[3] or 0), 'ok_calls': int(summary[4] or 0), 'error_calls': int(summary[5] or 0), 'rag_hits': int(summary[6] or 0), 'cache_hits': int(summary[7] or 0), }, by_provider=[ {'provider': r[0], 'calls': int(r[1] or 0), 'tokens': int(r[2] or 0), 'cost': float(r[3] or 0)} for r in by_provider ], recent=[ {'id': r[0], 'called_at': r[1].strftime('%H:%M:%S'), 'caller': r[2], 'provider': r[3], 'model': r[4], 'in_tokens': int(r[5] or 0), 'out_tokens': int(r[6] or 0), 'duration_ms': int(r[7] or 0), 'status': r[8], 'cost': float(r[9] or 0), 'cache_hit': bool(r[10]), 'rag_hit': bool(r[11])} for r in recent ], callers=[r[0] for r in callers], error=None, ) except Exception as e: return render_template( 'admin/ai_calls_dashboard.html', hours=hours, caller_filter=caller_filter, provider_filter=provider_filter, summary={}, by_provider=[], recent=[], callers=[], error=f'查詢失敗: {type(e).__name__}: {str(e)[:200]}', ) finally: session.close() # ───────────────────────────────────────────────────────────────────────────── # /observability/promotion_review — Phase 28 PromotionGate 待審核列表 # ───────────────────────────────────────────────────────────────────────────── @admin_observability_bp.route('/promotion_review') def promotion_review_list(): """awaiting_review episodes 列表(24h 內 reviewed_at IS NULL)""" session = get_session() try: rows = session.execute( sa_text(""" SELECT id, created_at, episode_type, source_table, source_id, distilled_text, quality_score, weight, promotion_status FROM learning_episodes WHERE promotion_status = 'awaiting_review' AND reviewed_at IS NULL ORDER BY weight DESC, created_at ASC LIMIT 50 """), ).fetchall() episodes = [ {'id': r[0], 'created_at': r[1].strftime('%Y-%m-%d %H:%M'), 'episode_type': r[2], 'source_table': r[3], 'source_id': r[4], 'distilled_text': (r[5] or '')[:600], 'quality_score': float(r[6] or 0), 'weight': float(r[7] or 0), 'status': r[8]} for r in rows ] return render_template( 'admin/promotion_review.html', episodes=episodes, error=None, ) except Exception as e: return render_template( 'admin/promotion_review.html', episodes=[], error=f'查詢失敗: {type(e).__name__}: {str(e)[:200]}', ) finally: session.close() @admin_observability_bp.route('/promotion_review/approve/', methods=['POST']) def promotion_review_approve(episode_id: int): """Web 介面「通過」按鈕 — 等同於 Telegram pg_ok callback""" try: from services.learning_pipeline import promotion_gate, hash_human_approver username = request.headers.get('X-Forwarded-User', 'web_admin') approver_hash = hash_human_approver(username) insight_id = promotion_gate.promote(episode_id) if insight_id: return jsonify({'ok': True, 'insight_id': insight_id, 'approver': approver_hash}) return jsonify({'ok': False, 'error': 'promote failed'}), 500 except Exception as e: return jsonify({'ok': False, 'error': str(e)[:200]}), 500 @admin_observability_bp.route('/promotion_review/reject/', methods=['POST']) def promotion_review_reject(episode_id: int): """Web 介面「拒絕」按鈕""" try: from services.learning_pipeline import promotion_gate ok = promotion_gate.reject(episode_id, 'rejected_human', detail='web admin reject') return jsonify({'ok': ok}) except Exception as e: return jsonify({'ok': False, 'error': str(e)[:200]}), 500 # ───────────────────────────────────────────────────────────────────────────── # /observability/quality_trend — Phase 25 caller 反饋趨勢視覺化 # ───────────────────────────────────────────────────────────────────────────── @admin_observability_bp.route('/quality_trend') def quality_trend_dashboard(): """caller × feedback 趨勢(30 日窗格)""" days = int(request.args.get('days', '30')) try: from services.feedback_quality_tracker import ( compute_caller_quality_trend, get_caller_recommendations, ) trends = compute_caller_quality_trend(days=days) recommendations = get_caller_recommendations(days=days) # 排序:avg_score 升序(最差先看) sorted_trends = sorted( trends.items(), key=lambda kv: kv[1].get('avg_score', 5), ) return render_template( 'admin/quality_trend.html', days=days, trends=[(c, info) for c, info in sorted_trends], recommendations=recommendations, error=None, ) except Exception as e: return render_template( 'admin/quality_trend.html', days=days, trends=[], recommendations=[], error=f'查詢失敗: {type(e).__name__}: {str(e)[:200]}', ) # ───────────────────────────────────────────────────────────────────────────── # /observability/budget — Phase 29 預算管理 + 手動 throttle # ───────────────────────────────────────────────────────────────────────────── @admin_observability_bp.route('/budget') def budget_dashboard(): """ai_call_budgets 編輯 + 當月 spent 即時對比""" from datetime import datetime as _dt today = _dt.now() month_start = _dt(today.year, today.month, 1) session = get_session() try: budgets = session.execute( sa_text(""" SELECT id, period, provider, budget_usd, alert_pct, updated_at FROM ai_call_budgets ORDER BY period, provider NULLS FIRST """), ).fetchall() spent_rows = session.execute( sa_text(""" SELECT provider, COALESCE(SUM(cost_usd), 0) AS spent FROM ai_calls WHERE called_at >= :ms GROUP BY provider """), {'ms': month_start}, ).fetchall() spent_map = {r[0]: float(r[1] or 0) for r in spent_rows} # throttle 狀態 throttle_state = {} try: from services.cost_throttle_service import get_throttle_state throttle_state = get_throttle_state() except Exception: pass rows = [] for b in budgets: provider = b[2] # 可能 None(全供應商總額) spent = spent_map.get(provider, 0.0) if provider else sum(spent_map.values()) budget_usd = float(b[3] or 0) ratio = (spent / budget_usd) if budget_usd > 0 else 0 rows.append({ 'id': b[0], 'period': b[1], 'provider': provider or '(all)', 'budget_usd': budget_usd, 'alert_pct': int(b[4] or 80), 'spent': spent, 'ratio': ratio, 'throttled': throttle_state.get(provider, {}).get('throttled', False) if provider else False, 'updated_at': b[5].strftime('%Y-%m-%d %H:%M') if b[5] else '-', }) return render_template('admin/budget.html', rows=rows, error=None) except Exception as e: return render_template('admin/budget.html', rows=[], error=f'查詢失敗: {type(e).__name__}: {str(e)[:200]}') finally: session.close() @admin_observability_bp.route('/budget/update/', methods=['POST']) def budget_update(budget_id: int): """更新 budget_usd / alert_pct""" try: new_budget = float(request.json.get('budget_usd')) new_alert = int(request.json.get('alert_pct', 80)) if new_budget <= 0 or not (1 <= new_alert <= 100): return jsonify({'ok': False, 'error': 'invalid range'}), 400 session = get_session() try: session.execute( sa_text(""" UPDATE ai_call_budgets SET budget_usd = :b, alert_pct = :a, updated_at = NOW() WHERE id = :id """), {'b': new_budget, 'a': new_alert, 'id': budget_id}, ) session.commit() return jsonify({'ok': True}) finally: session.close() except Exception as e: return jsonify({'ok': False, 'error': str(e)[:200]}), 500 # ───────────────────────────────────────────────────────────────────────────── # /observability/ppt_audit_history — Phase 29 PPT 視覺審核歷史 # ───────────────────────────────────────────────────────────────────────────── @admin_observability_bp.route('/ppt_audit_history') def ppt_audit_history(): """掃 reports/ 目錄列近 7 日 .pptx 檔 + 即時跑 audit(如已啟用)""" import os reports_dir = 'reports' files = [] error = None try: if not os.path.isdir(reports_dir): error = f'{reports_dir} 目錄不存在' else: cutoff = __import__('time').time() - 7 * 86400 for f in os.listdir(reports_dir): if not f.lower().endswith('.pptx'): continue full = os.path.join(reports_dir, f) try: mtime = os.path.getmtime(full) if mtime >= cutoff: files.append({ 'name': f, 'size_kb': round(os.path.getsize(full) / 1024, 1), 'mtime': __import__('datetime').datetime.fromtimestamp(mtime).strftime('%Y-%m-%d %H:%M'), 'mtime_ts': mtime, }) except OSError: continue files.sort(key=lambda x: x['mtime_ts'], reverse=True) except Exception as e: error = f'{type(e).__name__}: {str(e)[:200]}' # PPT vision 啟用狀態 try: from services.ppt_vision_service import is_ppt_vision_enabled vision_enabled = is_ppt_vision_enabled() except Exception: vision_enabled = False return render_template( 'admin/ppt_audit_history.html', files=files, vision_enabled=vision_enabled, error=error, ) # ───────────────────────────────────────────────────────────────────────────── # /observability/host_health — 三主機 + MCP 健康度 # ───────────────────────────────────────────────────────────────────────────── @admin_observability_bp.route('/host_health') def host_health_dashboard(): """三主機 Ollama + 4 個 MCP server 即時健康""" ollama_hosts = [] try: from services.ollama_service import ( OLLAMA_HOST_PRIMARY, OLLAMA_HOST_SECONDARY, OLLAMA_HOST_FALLBACK, _is_unhealthy, _unhealthy_marks, ) import requests as _r for label, host in [ ('Primary (GCP)', OLLAMA_HOST_PRIMARY), ('Secondary (GCP)', OLLAMA_HOST_SECONDARY), ('Fallback (111)', OLLAMA_HOST_FALLBACK), ]: entry = {'label': label, 'host': host, 'healthy': False, 'unhealthy_mark': _is_unhealthy(host), 'models': []} try: resp = _r.get(f"{host.rstrip('/')}/api/tags", timeout=3) if resp.status_code == 200: entry['healthy'] = True entry['models'] = [ m.get('name', '') for m in resp.json().get('models', []) ][:15] except Exception: pass ollama_hosts.append(entry) except Exception: pass # MCP server 健康 mcp_status = {} try: from services.mcp_router import mcp_router mcp_status = mcp_router.health_check() except Exception: pass # cost throttle 狀態 throttle_state = {} try: from services.cost_throttle_service import get_throttle_state throttle_state = get_throttle_state() except Exception: pass return render_template( 'admin/host_health.html', ollama_hosts=ollama_hosts, mcp_status=mcp_status, throttle_state=throttle_state, )