Files
ewoooc/config.py
OoO 97e7e2843b
All checks were successful
CD Pipeline / deploy (push) Successful in 1m6s
V10.568 優化價格決策信封通知
2026-06-02 10:47:09 +08:00

421 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import os
import json
import sys
from urllib.parse import urlparse
from dotenv import load_dotenv
# 載入 .env 環境變數
load_dotenv()
# 基本路徑設定
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, 'data')
LOG_DIR = os.path.join(BASE_DIR, 'logs')
UPLOAD_FOLDER = os.path.join(BASE_DIR, 'web/static/uploads')
# 建立必要目錄
for d in [DATA_DIR, LOG_DIR, UPLOAD_FOLDER]:
os.makedirs(d, exist_ok=True)
# ==========================================
# 資料庫設定
# ==========================================
# 支援 SQLite 和 PostgreSQL 兩種資料庫
# 預設使用 SQLite (本地開發),可透過環境變數切換到 PostgreSQL
USE_POSTGRESQL = os.getenv('USE_POSTGRESQL', 'false').lower() == 'true'
if USE_POSTGRESQL:
# PostgreSQL 連線設定
POSTGRES_HOST = os.getenv('POSTGRES_HOST', 'momo-postgres')
POSTGRES_PORT = os.getenv('POSTGRES_PORT', '5432')
POSTGRES_USER = os.getenv('POSTGRES_USER', 'momo')
POSTGRES_PASSWORD = os.getenv('POSTGRES_PASSWORD')
POSTGRES_DB = os.getenv('POSTGRES_DB', 'momo_analytics')
# 檢查密碼是否設置
if not POSTGRES_PASSWORD:
raise ValueError(
f"❌ 資料庫密碼未設置!請檢查:\n"
f"1. .env 檔案中是否設置了 POSTGRES_PASSWORD\n"
f"2. Docker Compose 環境變數是否正確傳遞\n"
f"3. 容器重啟後環境變數是否生效\n"
f"當前環境變數USE_POSTGRESQL=true, POSTGRES_HOST={POSTGRES_HOST}, POSTGRES_USER={POSTGRES_USER}"
)
DATABASE_PATH = f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"
DATABASE_TYPE = 'postgresql'
else:
# SQLite 連線設定 (開發環境或備用)
DATABASE_PATH = f"sqlite:///{os.path.join(DATA_DIR, 'momo_database.db')}"
DATABASE_TYPE = 'sqlite'
# ==========================================
# 安全設定(從環境變數讀取)
# ==========================================
import sys as _sys
def _is_test_context() -> bool:
return (
os.getenv("PYTEST_CURRENT_TEST") is not None
or "pytest" in sys.modules
or os.getenv("MOMO_ALLOW_INSECURE_CONFIG_FOR_TESTS", "").lower() == "true"
)
def _require_env(var_name: str, *, insecure_values: tuple[str, ...] = ()) -> str:
value = os.getenv(var_name)
if value and value not in insecure_values:
return value
if _is_test_context():
fallback = f"test-{var_name.lower()}"
print(f"[WARN] {var_name} 未設定,測試環境使用暫時值。", file=_sys.stderr)
return fallback
if var_name == "LOGIN_PASSWORD":
print("[FATAL] LOGIN_PASSWORD 環境變數未設定,拒絕啟動。請在 .env 或 Docker 環境設定此值。", file=_sys.stderr)
else:
print(f"[FATAL] {var_name} 環境變數未設定或仍為不安全預設值,拒絕啟動。請改用安全值。", file=_sys.stderr)
_sys.exit(1)
LOGIN_PASSWORD = _require_env("LOGIN_PASSWORD")
SECRET_KEY = _require_env("SECRET_KEY", insecure_values=('your_flask_secret_key', 'change_me', ''))
# ==========================================
# 通訊模組設定(從環境變數讀取)
# ==========================================
# --- Telegram Bot ---
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN', '')
try:
TELEGRAM_CHAT_IDS = json.loads(os.getenv('TELEGRAM_CHAT_IDS', '[]'))
except json.JSONDecodeError:
TELEGRAM_CHAT_IDS = []
# --- Line Notify ---
LINE_ENABLED = os.getenv('LINE_ENABLED', 'false').lower() == 'true' # 預設關閉
LINE_CHANNEL_ACCESS_TOKEN = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', '')
LINE_GROUP_ID = os.getenv('LINE_GROUP_ID', '')
# --- Email (SMTP) ---
EMAIL_HOST = os.getenv('EMAIL_HOST', 'smtp.gmail.com')
EMAIL_PORT = int(os.getenv('EMAIL_PORT', '587'))
EMAIL_HOST_USER = os.getenv('EMAIL_HOST_USER', '')
EMAIL_HOST_PASSWORD = os.getenv('EMAIL_HOST_PASSWORD', '') # 注意:若使用 Gmail需設定「應用程式密碼」
EMAIL_SENDER = os.getenv('EMAIL_SENDER', '')
EMAIL_RECEIVER = os.getenv('EMAIL_RECEIVER', '')
# ==========================================
# 網路設定(從環境變數讀取)
# ==========================================
PUBLIC_URL = os.getenv('PUBLIC_URL', 'https://mo.wooo.work')
def _env_bool(name: str, default: str = 'false') -> bool:
return os.getenv(name, default).strip().lower() in {'1', 'true', 'yes', 'on'}
def _safe_public_http_url(value: str) -> str:
"""只接受 http(s) URL避免模板把 javascript/data URI 輸出成 script src。"""
candidate = (value or '').strip().rstrip('/')
if not candidate:
return ''
parsed = urlparse(candidate)
if parsed.scheme not in {'http', 'https'} or not parsed.netloc:
return ''
return candidate
def _safe_webcrumbs_reference(value: str) -> str:
"""允許 http(s) URL 或 momo-pro 同源絕對路徑,供 script/plugin URL 使用。"""
candidate = (value or '').strip().rstrip('/')
if not candidate:
return ''
if candidate.startswith('/') and not candidate.startswith('//') and '\\' not in candidate:
return candidate
return _safe_public_http_url(candidate)
# Webcrumbs 共用 microfrontend runtime。
# 預設透過 momo-pro 同源 proxy 讀 188 Shared UI Hub避免正式頁面被跨域 TLS 狀態卡住。
WEBCRUMBS_ENABLED = _env_bool('WEBCRUMBS_ENABLED', 'true')
WEBCRUMBS_BASE_URL = _safe_public_http_url(
os.getenv('WEBCRUMBS_BASE_URL', 'https://webcrumbs.wooo.work')
)
WEBCRUMBS_RUNTIME_VERSION = os.getenv('WEBCRUMBS_RUNTIME_VERSION', 'shared-ui-poc-0.1.0').strip().strip('/')
WEBCRUMBS_RUNTIME_PATH = os.getenv(
'WEBCRUMBS_RUNTIME_PATH',
'/webcrumbs-assets/loader/webcrumbs-compatible-loader.js',
).strip()
_webcrumbs_runtime_url = os.getenv('WEBCRUMBS_RUNTIME_URL', '').strip()
WEBCRUMBS_RUNTIME_URL = _safe_webcrumbs_reference(_webcrumbs_runtime_url)
if not WEBCRUMBS_RUNTIME_URL:
WEBCRUMBS_RUNTIME_URL = _safe_webcrumbs_reference(WEBCRUMBS_RUNTIME_PATH)
WEBCRUMBS_PLUGIN_BASE_URL = _safe_webcrumbs_reference(
os.getenv(
'WEBCRUMBS_PLUGIN_BASE_URL',
'/webcrumbs-assets/plugins',
)
)
WEBCRUMBS_ASSET_UPSTREAM_URL = _safe_public_http_url(
os.getenv('WEBCRUMBS_ASSET_UPSTREAM_URL', 'http://192.168.0.188:18088')
)
# ==========================================
# 市場情報模組設定(預設全部關閉)
# ==========================================
MARKET_INTEL_ENABLED = os.getenv('MARKET_INTEL_ENABLED', 'false').lower() == 'true'
MARKET_INTEL_CRAWLER_ENABLED = os.getenv('MARKET_INTEL_CRAWLER_ENABLED', 'false').lower() == 'true'
MARKET_INTEL_WRITE_ENABLED = os.getenv('MARKET_INTEL_WRITE_ENABLED', 'false').lower() == 'true'
# 補上 EXCEL_EXPORT_DIR 定義
EXCEL_EXPORT_DIR = os.path.join(DATA_DIR, 'excel_exports')
# 更新建立目錄清單 (確保系統會自動建立這個資料夾)
for d in [DATA_DIR, LOG_DIR, UPLOAD_FOLDER, EXCEL_EXPORT_DIR]:
os.makedirs(d, exist_ok=True)
# MOMO 分類清單 (從 JSON 檔案動態讀取)
def load_momo_categories():
import json
import time
categories_path = os.path.join(DATA_DIR, 'categories.json')
# 預設分類,用於首次啟動或檔案遺失時
default_categories = [
{"name": "保養超值組", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700012"},
{"name": "化妝水", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700001"},
{"name": "精華液", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700002&p_orderType=4&showType=chessboardType"},
{"name": "乳液", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700003&p_orderType=4&showType=chessboardType"},
{"name": "乳霜", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700004&p_orderType=4&showType=chessboardType"},
{"name": "凝膠", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700005&p_orderType=4&showType=chessboardType"},
{"name": "面膜", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700006&p_orderType=4&showType=chessboardType"},
{"name": "眼霜", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700007&p_orderType=4&showType=chessboardType"},
{"name": "護唇膏", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700008&p_orderType=4&showType=chessboardType"},
{"name": "防曬", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700009&p_orderType=4&showType=chessboardType"},
{"name": "素顏霜", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700010&p_orderType=4&showType=chessboardType"},
{"name": "美顏霜", "url": "https://www.momoshop.com.tw/category/DgrpCategory.jsp?d_code=1111700011&p_orderType=4&showType=chessboardType"},
{"name": "身體護理", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1100901499&sourcePageType=4"},
{"name": "手部保養", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1100901655&sourcePageType=4"},
{"name": "足部保養", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1100901656&sourcePageType=4"},
{"name": "局部護理", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1100901505&sourcePageType=4"},
{"name": "止汗體香", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1100901503&sourcePageType=4"},
{"name": "嬰幼身體保養品牌旗艦", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1100901724&sourcePageType=4"},
{"name": "嬰幼本月主打", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=2705200081&sourcePageType=4"},
{"name": "嬰幼清潔用品", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=2705200349&p_orderType=4&showType=chessboardType"},
{"name": "嬰幼保養護膚", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=2705200350&p_orderType=4&showType=chessboardType"},
{"name": "媽咪孕期保養", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=2705200352&p_orderType=4&showType=chessboardType"},
{"name": "送禮超值組", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=2705200348&p_orderType=4&showType=chessboardType"},
{"name": "嬰幼品牌總覽", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=2705200348&p_orderType=4&showType=chessboardType"},
{"name": "私密保養本月主打", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1105900060&sourcePageType=4"},
{"name": "私密保養", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1105900002&sourcePageType=4"},
{"name": "私密清潔", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1105900003&sourcePageType=4"},
{"name": "除毛", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1105900154&sourcePageType=4"},
{"name": "私密保養推薦品牌", "url": "https://www.momoshop.com.tw/category/MgrpCategory.jsp?m_code=1105900023&sourcePageType=4"}
]
if not os.path.exists(categories_path):
# 如果檔案不存在,使用預設值並為每項加上 ID 後建立新檔案
categories_with_id = []
for i, cat in enumerate(default_categories):
cat['id'] = int(time.time() * 1000) + i
categories_with_id.append(cat)
try:
with open(categories_path, 'w', encoding='utf-8') as f:
json.dump(categories_with_id, f, ensure_ascii=False, indent=4)
return categories_with_id
except Exception as e:
print(f"Error creating categories.json: {e}")
return default_categories
try:
# V-Fix: 檢查檔案大小,如果過大 (例如超過 1MB) 則視為異常,直接使用預設值,避免讀取卡死
if os.path.exists(categories_path):
try:
if os.path.getsize(categories_path) > 1024 * 1024:
print(f"⚠️ Warning: categories.json is too large ({os.path.getsize(categories_path)} bytes). Using defaults.")
return default_categories
except OSError:
return default_categories
with open(categories_path, 'r', encoding='utf-8') as f:
try:
content = f.read().strip()
except Exception:
return default_categories
if not content:
return default_categories
data = json.loads(content)
# 若讀取到的列表為空,則回傳預設值 (修正空檔案導致爬蟲不工作的問題)
if not data:
return default_categories
return data
except (json.JSONDecodeError, FileNotFoundError, OSError):
# 如果檔案損毀或讀取失敗,回傳預設值
return default_categories
MOMO_CATEGORIES = load_momo_categories()
# ==========================================
# 密碼安全設定
# ==========================================
PASSWORD_MIN_LENGTH = int(os.getenv('PASSWORD_MIN_LENGTH', '8'))
PASSWORD_REQUIRE_UPPERCASE = os.getenv('PASSWORD_REQUIRE_UPPERCASE', 'true').lower() == 'true'
PASSWORD_REQUIRE_LOWERCASE = os.getenv('PASSWORD_REQUIRE_LOWERCASE', 'true').lower() == 'true'
PASSWORD_REQUIRE_DIGIT = os.getenv('PASSWORD_REQUIRE_DIGIT', 'true').lower() == 'true'
PASSWORD_REQUIRE_SPECIAL = os.getenv('PASSWORD_REQUIRE_SPECIAL', 'false').lower() == 'true'
PASSWORD_SPECIAL_CHARS = os.getenv('PASSWORD_SPECIAL_CHARS', '!@#$%^&*()_+-=[]{}|;:,.<>?')
PASSWORD_EXPIRY_DAYS = int(os.getenv('PASSWORD_EXPIRY_DAYS', '90'))
# ==========================================
# 外部服務連結 (分析報表選單)
# ==========================================
METABASE_URL = os.getenv('METABASE_URL', '') # Metabase BI 連結
GRIST_URL = os.getenv('GRIST_URL', '') # Grist 資料協作連結
# ==========================================
# AI 服務設定
# ==========================================
_APPROVED_OLLAMA_HOST_SUBSTRINGS = (
'34.143.170.20:11434',
'34.21.145.224:11434',
'192.168.0.111:11434',
'192.168.0.110:11435',
'192.168.0.110:11436',
)
def _static_approved_ollama_env(name: str, default: str = '') -> str:
"""Import-time safe Ollama host env reader; never probes network."""
value = os.getenv(name, '').strip()
if value and any(approved in value for approved in _APPROVED_OLLAMA_HOST_SUBSTRINGS):
return value
return default
_STATIC_OLLAMA_PRIMARY = _static_approved_ollama_env(
'OLLAMA_HOST_PRIMARY',
'http://34.143.170.20:11434',
)
# Hermes AI Service (競價情報分析)
# V-New (ADR-027 Phase 2):所有 host 解析必須 lazy禁止 import-time freeze。
# 理由import 時 GCP 還沒探測resolve_ollama_host 內部 cache 120s
# 若 freeze 到 module attr主機 GCP 後來掛掉、cache 失效仍取不到新值。
def get_hermes_url():
"""Lazy 取得 Hermes 主機(每次呼叫都嘗試 resolve內部有 120s cache
優先序:
1. env HERMES_URL僅接受 GCP-A/GCP-B/111
2. resolve_ollama_host()GCP 優先 / 111 備援)
3. 兜底 http://192.168.0.111:11434防 ImportError
"""
try:
from services.ollama_service import approved_ollama_env
env_val = approved_ollama_env('HERMES_URL')
except Exception:
env_val = os.getenv('HERMES_URL', '').strip()
if env_val:
return env_val
try:
from services.ollama_service import resolve_ollama_host
return resolve_ollama_host()
except Exception:
return 'http://192.168.0.111:11434'
def get_embedding_host():
"""Lazy 取得 Embedding 主機。
優先序env EMBEDDING_HOST僅接受 GCP-A/GCP-B/111> get_hermes_url()。
"""
try:
from services.ollama_service import approved_ollama_env
env_val = approved_ollama_env('EMBEDDING_HOST')
except Exception:
env_val = os.getenv('EMBEDDING_HOST', '').strip()
if env_val:
return env_val
return get_hermes_url()
def get_ollama_host():
"""Lazy 取得 Ollama 主機(一般 LLM 推理用)。
優先序env OLLAMA_HOST僅接受 GCP-A/GCP-B/111> resolve_ollama_host()。
與舊 module-level OLLAMA_HOST 不同:本函數不會 freeze 結果,每次呼叫都重新 resolve。
"""
try:
from services.ollama_service import approved_ollama_env
env_val = approved_ollama_env('OLLAMA_HOST')
except Exception:
env_val = os.getenv('OLLAMA_HOST', '').strip()
if env_val:
return env_val
try:
from services.ollama_service import resolve_ollama_host
return resolve_ollama_host()
except Exception:
return 'http://192.168.0.111:11434'
# 向下相容:舊 caller 仍可 `from config import HERMES_URL`,但此常數不得
# import-time probe也不得在 GCP 短暫不可用時 freeze 到 111新 caller 應
# 改用 `get_hermes_url()` 或 `OllamaService` 取得動態三主機級聯結果。
HERMES_URL = _static_approved_ollama_env('HERMES_URL', _STATIC_OLLAMA_PRIMARY)
HERMES_TIMEOUT = int(os.getenv('HERMES_TIMEOUT', '120')) # 秒;批量 300 筆預估 ~90s
# Embedding 服務ADR-003 對齊embedding 走 Hermes 主機,內網免認證)
# 向下相容;新 caller 應改用 `get_embedding_host()` 或
# `OllamaService.generate_embedding()`,不可依賴 import-time 探測結果。
EMBEDDING_HOST = _static_approved_ollama_env('EMBEDDING_HOST', HERMES_URL)
EMBEDDING_TIMEOUT = int(os.getenv('EMBEDDING_TIMEOUT', os.getenv('OLLAMA_EMBED_TIMEOUT', '45')))
# Ollama 本地 AI 服務
# ADR-027 Phase 2OLLAMA_HOST 改為 lazy resolve禁止寫死 nginx URL 繞過 GCP 優先策略。
# 舊行為:寫死 'https://ollama.wooo.work/ollama' → 任何 import 都跳過 GCP 探測。
# 新行為env OLLAMA_HOST 優先;否則動態 caller 走 resolve_ollama_host()GCP 優先 / 111 備援)。
# 向下相容:保留 OLLAMA_HOST module attribute但不再於 import 時呼叫 get_ollama_host()。
# 新 caller 應改用 `from config import get_ollama_host` + `host = get_ollama_host()`。
OLLAMA_HOST = _static_approved_ollama_env('OLLAMA_HOST', _STATIC_OLLAMA_PRIMARY)
OLLAMA_MODEL = os.getenv('OLLAMA_MODEL', 'gemma3:4b')
# Google Gemini AI 雲端服務
GEMINI_API_KEY = os.getenv('GEMINI_API_KEY', '')
GEMINI_MODEL = os.getenv('GEMINI_MODEL', 'gemini-1.5-flash')
# Gemini is emergency fallback only. The hard kill switch defaults to true, so
# GEMINI_API_KEY/GEMINI_FALLBACK_ENABLED cannot create paid egress by accident.
GEMINI_API_HARD_DISABLED = os.getenv('GEMINI_API_HARD_DISABLED', 'true')
GEMINI_FALLBACK_ENABLED = os.getenv('GEMINI_FALLBACK_ENABLED', 'false')
GEMINI_ALLOWED_CONTEXTS = os.getenv('GEMINI_ALLOWED_CONTEXTS', '')
# 預設 AI 提供者: 'ollama' (本地免費) 或 'gemini' (雲端付費)
AI_PROVIDER = os.getenv('AI_PROVIDER', 'ollama')
# YouTube API Key (用於趨勢爬蟲)
YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
# ==========================================
# 系統版本與路徑
# ==========================================
SYSTEM_VERSION = "V10.568"
LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
public_url = PUBLIC_URL # 用於模板顯示
def validate_critical_config():
"""啟動時驗證選用配置,缺少則回傳 warning 清單(非 fatal"""
warnings = []
optional_vars = ['HERMES_URL', 'GOOGLE_DRIVE_CREDENTIALS']
for var in optional_vars:
if not os.getenv(var):
warnings.append(f"[Config] 選用設定 {var} 未設,部分功能可能停用")
if WEBCRUMBS_ENABLED and not WEBCRUMBS_RUNTIME_URL:
warnings.append("[Config] WEBCRUMBS_ENABLED=true 但 WEBCRUMBS_RUNTIME_URL 無效Webcrumbs runtime 將不會載入")
if WEBCRUMBS_ENABLED and WEBCRUMBS_RUNTIME_URL.startswith('/webcrumbs-assets/') and not WEBCRUMBS_ASSET_UPSTREAM_URL:
warnings.append("[Config] Webcrumbs 使用同源 asset proxy但 WEBCRUMBS_ASSET_UPSTREAM_URL 無效")
return warnings