1b4f3a7bbe
Some checks failed
CD Pipeline / deploy (push) Failing after 59s
- 建立 Gitea Actions CD pipeline (.gitea/workflows/cd.yaml) - 部署模式: rsync Python 檔案至 188 → docker restart (volume mount) - Dockerfile/requirements 變動時自動重建 Docker image - 部署通知: Telegram (開始/成功/失敗) - 健康檢查: https://mo.wooo.work/health (最多 5 次重試) - 同步最新 CLAUDE.md / ADR-008 / memory (2026-04-19) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
124 lines
4.2 KiB
Python
124 lines
4.2 KiB
Python
import os
|
||
from flask import Flask, render_template, session, redirect, url_for, send_from_directory
|
||
from database.manager import DatabaseManager
|
||
from auth import init_auth_routes, login_required
|
||
from config import SECRET_KEY, DATA_DIR, LOG_DIR, EXCEL_EXPORT_DIR
|
||
|
||
# --- 1. 初始化路徑與 Flask ---
|
||
BASE_DIR = os.path.abspath(os.path.dirname(__file__))
|
||
TEMPLATE_DIR = os.path.join(BASE_DIR, 'web', 'templates')
|
||
STATIC_DIR = os.path.join(BASE_DIR, 'web', 'static')
|
||
|
||
app = Flask(__name__,
|
||
template_folder=TEMPLATE_DIR,
|
||
static_folder=STATIC_DIR)
|
||
|
||
app.secret_key = SECRET_KEY
|
||
|
||
# 初始化權限路由
|
||
init_auth_routes(app)
|
||
|
||
# 初始化資料庫管理員
|
||
db_mgr = DatabaseManager()
|
||
|
||
def get_db_size():
|
||
"""獲取資料庫實體檔案大小"""
|
||
db_path = os.path.join(DATA_DIR, 'momo_database.db')
|
||
if os.path.exists(db_path):
|
||
size_bytes = os.path.getsize(db_path)
|
||
return f"{size_bytes / (1024 * 1024):.2f} MB"
|
||
return "0.00 MB"
|
||
|
||
# --- 2. 路由定義 ---
|
||
|
||
@app.route('/')
|
||
@login_required
|
||
def dashboard():
|
||
"""主看板：包含統計數據與商品明細清單"""
|
||
stats = {
|
||
"total_products": 0,
|
||
"price_changes": 0,
|
||
"db_size": get_db_size()
|
||
}
|
||
recent_products = []
|
||
|
||
session_db = db_mgr.Session()
|
||
try:
|
||
from database.models import Product, PriceRecord
|
||
from datetime import datetime
|
||
from sqlalchemy import desc
|
||
|
||
# A. 計算統計數據
|
||
stats["total_products"] = session_db.query(Product).count()
|
||
|
||
today_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
||
stats["price_changes"] = session_db.query(PriceRecord).filter(
|
||
PriceRecord.timestamp >= today_start
|
||
).count()
|
||
|
||
# B. 抓取最新 50 筆商品明細
|
||
# 邏輯：抓取 Product，並針對每個 Product 找出最後一筆價格
|
||
products_query = session_db.query(Product).order_by(Product.id.desc()).limit(50).all()
|
||
|
||
for p in products_query:
|
||
# 取得該商品在 PriceRecord 表中最新的一筆紀錄
|
||
latest_price_rec = session_db.query(PriceRecord)\
|
||
.filter_by(product_id=p.id)\
|
||
.order_by(desc(PriceRecord.timestamp))\
|
||
.first()
|
||
|
||
recent_products.append({
|
||
"category": p.category,
|
||
"name": p.name,
|
||
"price": f"{latest_price_rec.price:,.0f}" if latest_price_rec else "N/A",
|
||
"url": p.url,
|
||
"time": latest_price_rec.timestamp.strftime('%m/%d %H:%M') if latest_price_rec else "-"
|
||
})
|
||
|
||
except Exception as e:
|
||
app.logger.error(f"❌ 看板數據讀取失敗: {e}")
|
||
finally:
|
||
session_db.close()
|
||
|
||
return render_template('dashboard.html', stats=stats, products=recent_products)
|
||
|
||
@app.route('/logs')
|
||
@login_required
|
||
def view_logs():
|
||
"""日誌查看頁面"""
|
||
log_path = os.path.join(LOG_DIR, 'system.log')
|
||
content = "⚠️ 尚無日誌紀錄。"
|
||
if os.path.exists(log_path):
|
||
try:
|
||
with open(log_path, 'r', encoding='utf-8') as f:
|
||
f.seek(0, os.SEEK_END)
|
||
size = f.tell()
|
||
f.seek(max(0, size - 15000)) # 讀取最後 1.5 萬字元
|
||
content = f.read()
|
||
except Exception as e:
|
||
content = f"讀取日誌出錯: {e}"
|
||
return render_template('logs.html', log_content=content)
|
||
|
||
@app.route('/download/<path:filename>')
|
||
@login_required
|
||
def download_excel(filename):
|
||
"""Excel 報表下載"""
|
||
return send_from_directory(EXCEL_EXPORT_DIR, filename)
|
||
|
||
# --- 3. 啟動自我檢查 ---
|
||
|
||
if __name__ == '__main__':
|
||
print("\n" + "="*50)
|
||
print("🚀 MOMO 伺服器啟動中...")
|
||
print(f"📂 模板目錄: {TEMPLATE_DIR}")
|
||
|
||
# 檢查核心檔案
|
||
for f in ['dashboard.html', 'login.html']:
|
||
path = os.path.join(TEMPLATE_DIR, f)
|
||
status = "✅ 存在" if os.path.exists(path) else "❌ 缺失"
|
||
size = f"({os.path.getsize(path)} bytes)" if os.path.exists(path) else ""
|
||
print(f" - {f}: {status} {size}")
|
||
|
||
print("="*50 + "\n")
|
||
|
||
app.run(host='0.0.0.0', port=5888, debug=True) |