d51d8031f5
ADR-017 Phase 3f-1 system sprint；新增無 prefix system_public_bp，保留公開 URL 與 backup CSRF；ABC detail 併入 sales_bp。
205 lines
7.6 KiB
Python
205 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""公開系統頁面與健康檢查路由。
|
||
|
||
此 blueprint 無 url_prefix，保留外部監控與既有前端使用的公開 URL。
|
||
"""
|
||
|
||
import os
|
||
import zipfile
|
||
from datetime import datetime, timezone, timedelta
|
||
|
||
from flask import Blueprint, Response, jsonify, render_template, send_from_directory, url_for
|
||
from sqlalchemy import text
|
||
|
||
from auth import login_required
|
||
from config import BASE_DIR, DATABASE_TYPE, SYSTEM_VERSION
|
||
from database.manager import DatabaseManager
|
||
from database.models import Product, PriceRecord
|
||
from services.json_storage import load_categories
|
||
from services.logger_manager import SystemLogger
|
||
from utils.security import safe_join
|
||
|
||
|
||
system_public_bp = Blueprint('system_public', __name__)
|
||
|
||
sys_log = SystemLogger("SystemPublicRoutes").get_logger()
|
||
TAIPEI_TZ = timezone(timedelta(hours=8))
|
||
LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
|
||
public_url = os.getenv('PUBLIC_URL', '服務啟動中...')
|
||
|
||
|
||
@system_public_bp.route('/health')
|
||
def health_check():
|
||
"""健康檢查端點 - 供 Nginx 和 Docker healthcheck 使用"""
|
||
try:
|
||
return jsonify({
|
||
'status': 'healthy',
|
||
'database': DATABASE_TYPE,
|
||
'version': SYSTEM_VERSION
|
||
}), 200
|
||
except Exception as e:
|
||
return jsonify({
|
||
'status': 'unhealthy',
|
||
'error': str(e)
|
||
}), 500
|
||
|
||
|
||
@system_public_bp.route('/metrics')
|
||
def prometheus_metrics():
|
||
"""Prometheus 指標端點 - 供 Prometheus 抓取監控資料"""
|
||
try:
|
||
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST, Gauge, CollectorRegistry
|
||
|
||
registry = CollectorRegistry()
|
||
|
||
app_info = Gauge('momo_app_info', '應用程式資訊', ['version', 'database_type'], registry=registry)
|
||
app_info.labels(version=SYSTEM_VERSION, database_type=DATABASE_TYPE).set(1)
|
||
|
||
app_health = Gauge('momo_app_health', '應用程式健康狀態', registry=registry)
|
||
db_status = Gauge('momo_database_up', '資料庫連線狀態', registry=registry)
|
||
|
||
try:
|
||
db = DatabaseManager()
|
||
with db.engine.connect() as conn:
|
||
conn.execute(text("SELECT 1"))
|
||
db_status.set(1)
|
||
app_health.set(1)
|
||
except Exception:
|
||
db_status.set(0)
|
||
app_health.set(0)
|
||
|
||
try:
|
||
db = DatabaseManager()
|
||
session = db.get_session()
|
||
|
||
product_count = Gauge('momo_products_total', '商品總數', registry=registry)
|
||
product_count.set(session.query(Product).count())
|
||
|
||
price_record_count = Gauge('momo_price_records_total', '價格記錄總數', registry=registry)
|
||
price_record_count.set(session.query(PriceRecord).count())
|
||
|
||
from database.realtime_sales_models import RealtimeSalesMonthly
|
||
sales_count = Gauge('momo_sales_records_total', '業績資料總數', registry=registry)
|
||
sales_count.set(session.query(RealtimeSalesMonthly).count())
|
||
|
||
session.close()
|
||
except Exception as e:
|
||
sys_log.warning(f"[Metrics] 無法取得資料庫統計: {e}")
|
||
|
||
return Response(generate_latest(registry), mimetype=CONTENT_TYPE_LATEST)
|
||
|
||
except ImportError:
|
||
metrics_text = f"""# HELP momo_app_health 應用程式健康狀態
|
||
# TYPE momo_app_health gauge
|
||
momo_app_health 1
|
||
# HELP momo_app_info 應用程式資訊
|
||
# TYPE momo_app_info gauge
|
||
momo_app_info{{version="{SYSTEM_VERSION}",database_type="{DATABASE_TYPE}"}} 1
|
||
"""
|
||
return Response(metrics_text, mimetype='text/plain; charset=utf-8')
|
||
except Exception as e:
|
||
sys_log.error(f"[Metrics] 指標生成錯誤: {e}")
|
||
return Response(f"# Error: {e}\n", mimetype='text/plain; charset=utf-8'), 500
|
||
|
||
|
||
@system_public_bp.route('/settings')
|
||
def settings():
|
||
"""分類設定頁面"""
|
||
categories = load_categories()
|
||
return render_template('settings.html',
|
||
categories=categories,
|
||
public_url=public_url,
|
||
system_version=SYSTEM_VERSION)
|
||
|
||
|
||
@system_public_bp.route('/system_settings')
|
||
def system_settings_page():
|
||
"""系統設定與匯入頁面"""
|
||
return render_template('system_settings.html', system_version=SYSTEM_VERSION)
|
||
|
||
|
||
@system_public_bp.route('/logs')
|
||
def show_logs():
|
||
return render_template('logs.html')
|
||
|
||
|
||
@system_public_bp.route('/api/logs')
|
||
def get_logs_api():
|
||
if os.path.exists(LOG_FILE_PATH):
|
||
try:
|
||
with open(LOG_FILE_PATH, 'r', encoding='utf-8') as f:
|
||
return jsonify({"logs": "".join(f.readlines()[-60:])})
|
||
except Exception as e:
|
||
sys_log.error(f"[Web] [Logs] ❌ 日誌 API 讀取異常 | Error: {e}")
|
||
return jsonify({"logs": "讀取日誌異常"})
|
||
return jsonify({"logs": "等待系統啟動中..."})
|
||
|
||
|
||
@system_public_bp.route('/api/backup', methods=['POST'])
|
||
@login_required
|
||
def trigger_backup():
|
||
"""API: 觸發系統完整備份"""
|
||
try:
|
||
sys_log.info("[System] [Backup] 💾 開始執行系統完整備份...")
|
||
backup_dir = os.path.join(BASE_DIR, 'backups')
|
||
if not os.path.exists(backup_dir):
|
||
os.makedirs(backup_dir)
|
||
|
||
timestamp = datetime.now(TAIPEI_TZ).strftime('%Y%m%d_%H%M')
|
||
zip_filename = f"momo_system_backup_{SYSTEM_VERSION}_{timestamp}.zip"
|
||
zip_filepath = os.path.join(backup_dir, zip_filename)
|
||
|
||
with zipfile.ZipFile(zip_filepath, 'w', zipfile.ZIP_DEFLATED) as zipf:
|
||
for root, dirs, files in os.walk(BASE_DIR):
|
||
dirs[:] = [d for d in dirs if d not in ['backups', '__pycache__', 'venv', '.git', '.idea', '.vscode', 'node_modules']]
|
||
|
||
for file in files:
|
||
if file == zip_filename:
|
||
continue
|
||
if file.endswith('.pyc') or file.endswith('.DS_Store'):
|
||
continue
|
||
|
||
file_path = os.path.join(root, file)
|
||
arcname = os.path.relpath(file_path, BASE_DIR)
|
||
zipf.write(file_path, arcname)
|
||
|
||
sys_log.info(f"[System] [Backup] ✅ 系統備份完成 | File: {zip_filename}")
|
||
|
||
download_url = url_for('system_public.download_backup', filename=zip_filename)
|
||
|
||
return jsonify({
|
||
"status": "success",
|
||
"message": f"備份成功！\n檔案已儲存為: {zip_filename}\n即將開始下載...",
|
||
"download_url": download_url
|
||
})
|
||
except Exception as e:
|
||
sys_log.error(f"[System] [Backup] ❌ 備份失敗 | Error: {e}")
|
||
return jsonify({"status": "error", "message": str(e)}), 500
|
||
|
||
|
||
@system_public_bp.route('/api/backup/download/<path:filename>')
|
||
@login_required
|
||
def download_backup(filename):
|
||
"""API: 下載備份檔案（已加入路徑遍歷防護）"""
|
||
try:
|
||
backup_dir = os.path.join(BASE_DIR, 'backups')
|
||
safe_path = safe_join(backup_dir, filename)
|
||
|
||
if not safe_path.exists():
|
||
sys_log.warning(f"[Security] 備份檔案不存在 | File: {filename}")
|
||
return jsonify({'error': '檔案不存在'}), 404
|
||
|
||
if not safe_path.is_file():
|
||
sys_log.warning(f"[Security] 嘗試下載非檔案路徑 | Path: {filename}")
|
||
return jsonify({'error': '非法路徑'}), 400
|
||
|
||
return send_from_directory(backup_dir, safe_path.name, as_attachment=True)
|
||
|
||
except ValueError as e:
|
||
sys_log.error(f"[Security] 路徑遍歷攻擊嘗試被阻擋 | Filename: {filename} | Error: {e}")
|
||
return jsonify({'error': '非法路徑'}), 400
|
||
except Exception as e:
|
||
sys_log.error(f"[System] 下載備份失敗 | Error: {e}")
|
||
return jsonify({'error': '下載失敗'}), 500
|