1b4f3a7bbe
- 建立 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>
712 lines
20 KiB
Python
712 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Bot API 路由模組
|
|
提供給 Clawdbot/Telegram Bot 使用的 API 端點
|
|
使用 API Token 認證，不需要 session
|
|
"""
|
|
|
|
import os
|
|
from datetime import datetime, timezone, timedelta
|
|
from functools import wraps
|
|
from flask import Blueprint, request, jsonify
|
|
from sqlalchemy import func, desc, text
|
|
|
|
from config import BASE_DIR
|
|
from database.manager import DatabaseManager
|
|
from services.logger_manager import SystemLogger
|
|
|
|
# 時區設定
|
|
TAIPEI_TZ = timezone(timedelta(hours=8))
|
|
|
|
# Logger
|
|
sys_log = SystemLogger("BotAPI").get_logger()
|
|
|
|
# Blueprint 定義
|
|
bot_api_bp = Blueprint('bot_api', __name__)
|
|
|
|
# API Token (從環境變數讀取，預設值僅用於開發)
|
|
BOT_API_TOKEN = os.getenv('BOT_API_TOKEN', 'clawdbot_momo_2026')
|
|
|
|
|
|
def require_api_token(f):
|
|
"""API Token 認證裝飾器"""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
# 從 header 或 query string 取得 token
|
|
token = request.headers.get('X-API-Token') or request.args.get('token')
|
|
|
|
if not token:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Missing API token'
|
|
}), 401
|
|
|
|
if token != BOT_API_TOKEN:
|
|
sys_log.warning(f"[BotAPI] Invalid token attempt from {request.remote_addr}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Invalid API token'
|
|
}), 403
|
|
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/status')
|
|
def bot_api_status():
|
|
"""API 狀態檢查（不需要認證）"""
|
|
return jsonify({
|
|
'success': True,
|
|
'service': 'MOMO Pro Bot API',
|
|
'version': '1.0.0',
|
|
'timestamp': datetime.now(TAIPEI_TZ).isoformat()
|
|
})
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/daily_sales')
|
|
@require_api_token
|
|
def bot_daily_sales():
|
|
"""
|
|
查詢每日業績
|
|
|
|
Query Parameters:
|
|
- date: 日期 (YYYY-MM-DD)，預設今天
|
|
|
|
Returns:
|
|
- 業績總額、訂單數、商品數等統計
|
|
"""
|
|
try:
|
|
date_str = request.args.get('date')
|
|
|
|
if date_str:
|
|
try:
|
|
query_date = datetime.strptime(date_str, '%Y-%m-%d').date()
|
|
except ValueError:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'Invalid date format: {date_str}. Use YYYY-MM-DD'
|
|
}), 400
|
|
else:
|
|
query_date = datetime.now(TAIPEI_TZ).date()
|
|
|
|
db = DatabaseManager()
|
|
engine = db.engine
|
|
|
|
# 查詢當日業績
|
|
query = text("""
|
|
SELECT
|
|
COUNT(*) as total_records,
|
|
COALESCE(SUM(CAST("總業績" AS FLOAT)), 0) as total_revenue,
|
|
COUNT(DISTINCT "訂單編號") as order_count,
|
|
COUNT(DISTINCT "商品ID") as product_count
|
|
FROM daily_sales_snapshot
|
|
WHERE "日期" = :query_date
|
|
""")
|
|
|
|
with engine.connect() as conn:
|
|
result = conn.execute(query, {'query_date': str(query_date)})
|
|
row = result.fetchone()
|
|
|
|
if row and row[0] > 0:
|
|
return jsonify({
|
|
'success': True,
|
|
'date': str(query_date),
|
|
'data': {
|
|
'total_records': row[0],
|
|
'total_revenue': round(row[1], 2),
|
|
'order_count': row[2],
|
|
'product_count': row[3],
|
|
'formatted_revenue': f"${row[1]:,.0f}"
|
|
}
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'success': True,
|
|
'date': str(query_date),
|
|
'data': None,
|
|
'message': f'No sales data for {query_date}'
|
|
})
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] daily_sales error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/sales_summary')
|
|
@require_api_token
|
|
def bot_sales_summary():
|
|
"""
|
|
查詢業績摘要（今日 vs 昨日 vs 上週同日）
|
|
|
|
Returns:
|
|
- 今日、昨日、上週同日的業績比較
|
|
"""
|
|
try:
|
|
today = datetime.now(TAIPEI_TZ).date()
|
|
yesterday = today - timedelta(days=1)
|
|
last_week = today - timedelta(days=7)
|
|
|
|
db = DatabaseManager()
|
|
engine = db.engine
|
|
|
|
def get_daily_revenue(date):
|
|
query = text("""
|
|
SELECT COALESCE(SUM(CAST("總業績" AS FLOAT)), 0)
|
|
FROM daily_sales_snapshot
|
|
WHERE "日期" = :query_date
|
|
""")
|
|
with engine.connect() as conn:
|
|
result = conn.execute(query, {'query_date': str(date)})
|
|
row = result.fetchone()
|
|
return row[0] if row else 0
|
|
|
|
today_revenue = get_daily_revenue(today)
|
|
yesterday_revenue = get_daily_revenue(yesterday)
|
|
last_week_revenue = get_daily_revenue(last_week)
|
|
|
|
# 計算成長率
|
|
dod_growth = ((today_revenue - yesterday_revenue) / yesterday_revenue * 100) if yesterday_revenue > 0 else 0
|
|
wow_growth = ((today_revenue - last_week_revenue) / last_week_revenue * 100) if last_week_revenue > 0 else 0
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': {
|
|
'today': {
|
|
'date': str(today),
|
|
'revenue': round(today_revenue, 2),
|
|
'formatted': f"${today_revenue:,.0f}"
|
|
},
|
|
'yesterday': {
|
|
'date': str(yesterday),
|
|
'revenue': round(yesterday_revenue, 2),
|
|
'formatted': f"${yesterday_revenue:,.0f}"
|
|
},
|
|
'last_week': {
|
|
'date': str(last_week),
|
|
'revenue': round(last_week_revenue, 2),
|
|
'formatted': f"${last_week_revenue:,.0f}"
|
|
},
|
|
'growth': {
|
|
'dod': round(dod_growth, 2), # Day over Day
|
|
'wow': round(wow_growth, 2), # Week over Week
|
|
'dod_emoji': '📈' if dod_growth > 0 else ('📉' if dod_growth < 0 else '➡️'),
|
|
'wow_emoji': '📈' if wow_growth > 0 else ('📉' if wow_growth < 0 else '➡️')
|
|
}
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] sales_summary error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/stockout')
|
|
@require_api_token
|
|
def bot_stockout():
|
|
"""
|
|
查詢缺貨商品
|
|
|
|
Query Parameters:
|
|
- status: pending/sent/all (預設 pending)
|
|
- limit: 數量限制 (預設 10)
|
|
|
|
Returns:
|
|
- 缺貨商品清單
|
|
"""
|
|
try:
|
|
status = request.args.get('status', 'pending')
|
|
limit = min(int(request.args.get('limit', 10)), 50) # 最多 50 筆
|
|
|
|
db = DatabaseManager()
|
|
engine = db.engine
|
|
|
|
# 根據狀態查詢
|
|
if status == 'all':
|
|
status_condition = ""
|
|
elif status == 'sent':
|
|
status_condition = "WHERE sent_date IS NOT NULL"
|
|
else: # pending
|
|
status_condition = "WHERE sent_date IS NULL"
|
|
|
|
query = text(f"""
|
|
SELECT
|
|
id, vendor_name, product_code, product_name,
|
|
current_stock, stockout_days, created_at, sent_date
|
|
FROM vendor_stockout
|
|
{status_condition}
|
|
ORDER BY created_at DESC
|
|
LIMIT :limit
|
|
""")
|
|
|
|
with engine.connect() as conn:
|
|
result = conn.execute(query, {'limit': limit})
|
|
rows = result.fetchall()
|
|
|
|
items = []
|
|
for row in rows:
|
|
items.append({
|
|
'id': row[0],
|
|
'vendor_name': row[1],
|
|
'product_code': row[2],
|
|
'product_name': row[3],
|
|
'current_stock': row[4],
|
|
'stockout_days': row[5],
|
|
'created_at': str(row[6]) if row[6] else None,
|
|
'sent': row[7] is not None
|
|
})
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'status_filter': status,
|
|
'count': len(items),
|
|
'data': items
|
|
})
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] stockout error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/product_search')
|
|
@require_api_token
|
|
def bot_product_search():
|
|
"""
|
|
搜尋商品
|
|
|
|
Query Parameters:
|
|
- q: 搜尋關鍵字（商品名稱或貨號）
|
|
- limit: 數量限制 (預設 10)
|
|
|
|
Returns:
|
|
- 符合的商品清單
|
|
"""
|
|
try:
|
|
keyword = request.args.get('q', '').strip()
|
|
limit = min(int(request.args.get('limit', 10)), 20)
|
|
|
|
if not keyword:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Missing search keyword (q parameter)'
|
|
}), 400
|
|
|
|
db = DatabaseManager()
|
|
engine = db.engine
|
|
|
|
# 查詢商品並取得最新價格
|
|
query = text("""
|
|
SELECT
|
|
p.id, p.i_code, p.name, pr.price as current_price,
|
|
p.status, p.category, p.updated_at
|
|
FROM products p
|
|
LEFT JOIN LATERAL (
|
|
SELECT price FROM price_records
|
|
WHERE product_id = p.id
|
|
ORDER BY timestamp DESC
|
|
LIMIT 1
|
|
) pr ON true
|
|
WHERE p.name ILIKE :keyword OR p.i_code ILIKE :keyword
|
|
ORDER BY p.updated_at DESC
|
|
LIMIT :limit
|
|
""")
|
|
|
|
with engine.connect() as conn:
|
|
result = conn.execute(query, {
|
|
'keyword': f'%{keyword}%',
|
|
'limit': limit
|
|
})
|
|
rows = result.fetchall()
|
|
|
|
items = []
|
|
for row in rows:
|
|
items.append({
|
|
'id': row[0],
|
|
'i_code': row[1],
|
|
'name': row[2],
|
|
'current_price': row[3],
|
|
'status': row[4],
|
|
'category': row[5],
|
|
'updated_at': str(row[6]) if row[6] else None
|
|
})
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'keyword': keyword,
|
|
'count': len(items),
|
|
'data': items
|
|
})
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] product_search error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/top_products')
|
|
@require_api_token
|
|
def bot_top_products():
|
|
"""
|
|
查詢熱銷商品
|
|
|
|
Query Parameters:
|
|
- date: 日期 (YYYY-MM-DD)，預設今天
|
|
- limit: 數量限制 (預設 10)
|
|
|
|
Returns:
|
|
- 熱銷商品排行
|
|
"""
|
|
try:
|
|
date_str = request.args.get('date')
|
|
limit = min(int(request.args.get('limit', 10)), 20)
|
|
|
|
if date_str:
|
|
try:
|
|
query_date = datetime.strptime(date_str, '%Y-%m-%d').date()
|
|
except ValueError:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'Invalid date format: {date_str}. Use YYYY-MM-DD'
|
|
}), 400
|
|
else:
|
|
query_date = datetime.now(TAIPEI_TZ).date()
|
|
|
|
db = DatabaseManager()
|
|
engine = db.engine
|
|
|
|
query = text("""
|
|
SELECT
|
|
"商品ID",
|
|
"商品名稱",
|
|
SUM(CAST("總業績" AS FLOAT)) as total_revenue,
|
|
SUM(CAST("數量" AS INTEGER)) as total_quantity
|
|
FROM daily_sales_snapshot
|
|
WHERE "日期" = :query_date
|
|
GROUP BY "商品ID", "商品名稱"
|
|
ORDER BY total_revenue DESC
|
|
LIMIT :limit
|
|
""")
|
|
|
|
with engine.connect() as conn:
|
|
result = conn.execute(query, {
|
|
'query_date': str(query_date),
|
|
'limit': limit
|
|
})
|
|
rows = result.fetchall()
|
|
|
|
items = []
|
|
for i, row in enumerate(rows, 1):
|
|
items.append({
|
|
'rank': i,
|
|
'product_code': row[0],
|
|
'product_name': row[1],
|
|
'total_revenue': round(row[2], 2),
|
|
'total_quantity': row[3],
|
|
'formatted_revenue': f"${row[2]:,.0f}"
|
|
})
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'date': str(query_date),
|
|
'count': len(items),
|
|
'data': items
|
|
})
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] top_products error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
# ===== AI 助手 API =====
|
|
|
|
@bot_api_bp.route('/bot/api/ai/status')
|
|
@require_api_token
|
|
def bot_ai_status():
|
|
"""
|
|
查詢 AI 服務狀態
|
|
|
|
Returns:
|
|
- Ollama 和 Gemini 的連線狀態
|
|
"""
|
|
try:
|
|
from services.ai_provider import get_ai_status
|
|
status = get_ai_status(force_refresh=True)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': {
|
|
'default_provider': status.get('default_provider', 'ollama'),
|
|
'ollama': {
|
|
'connected': status.get('ollama', {}).get('connected', False),
|
|
'model': status.get('ollama', {}).get('model', 'unknown')
|
|
},
|
|
'gemini': {
|
|
'connected': status.get('gemini', {}).get('connected', False),
|
|
'model': status.get('gemini', {}).get('model', 'unknown')
|
|
}
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] ai_status error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/ai/generate_copy', methods=['POST'])
|
|
@require_api_token
|
|
def bot_generate_copy():
|
|
"""
|
|
生成銷售文案
|
|
|
|
Request JSON:
|
|
- product_name: 商品名稱 (必填)
|
|
- trend_keywords: 趨勢關鍵字 (選填, 陣列)
|
|
- style: 文案風格 (選填, 預設 '吸睛')
|
|
- provider: AI 提供者 (選填, 'ollama' 或 'gemini')
|
|
|
|
Returns:
|
|
- 生成的銷售文案
|
|
"""
|
|
try:
|
|
data = request.get_json() or {}
|
|
product_name = data.get('product_name', '')
|
|
trend_keywords = data.get('trend_keywords', [])
|
|
style = data.get('style', '吸睛')
|
|
provider = data.get('provider', None)
|
|
|
|
if not product_name:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Missing product_name'
|
|
}), 400
|
|
|
|
from services.ai_provider import ai_provider_service
|
|
|
|
result = ai_provider_service.generate_sales_copy(
|
|
product_name=product_name,
|
|
provider=provider,
|
|
trend_keywords=trend_keywords,
|
|
style=style
|
|
)
|
|
|
|
if result.success:
|
|
return jsonify({
|
|
'success': True,
|
|
'data': {
|
|
'copy': result.content,
|
|
'model': result.model,
|
|
'provider': result.provider,
|
|
'duration': round(result.total_duration, 2) if result.total_duration else None
|
|
}
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': result.error
|
|
}), 500
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] generate_copy error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/ai/trends')
|
|
@require_api_token
|
|
def bot_ai_trends():
|
|
"""
|
|
查詢趨勢資料
|
|
|
|
Query Parameters:
|
|
- categories: 分類 (選填, 可多個)
|
|
- time_range: 時間範圍 day/week/month (預設 week)
|
|
|
|
Returns:
|
|
- 新聞、社群熱門話題、關鍵字
|
|
"""
|
|
try:
|
|
from services.trend_crawler import TrendCrawler
|
|
|
|
categories = request.args.getlist('categories') or ['時尚美妝', '生活居家', '健康保健']
|
|
time_range = request.args.get('time_range', 'week')
|
|
|
|
if time_range not in ['day', 'week', 'month']:
|
|
time_range = 'week'
|
|
|
|
trend_crawler = TrendCrawler()
|
|
trend_data = trend_crawler.get_all_trends(
|
|
categories=categories,
|
|
time_range=time_range,
|
|
include_social=True
|
|
)
|
|
|
|
# 簡化輸出供 Bot 使用
|
|
result = {
|
|
'timestamp': trend_data.timestamp.isoformat(),
|
|
'time_range': time_range,
|
|
'keywords': trend_data.keywords[:10],
|
|
'news': [
|
|
{
|
|
'title': n.title,
|
|
'source': n.source,
|
|
'category': n.category
|
|
}
|
|
for n in trend_data.news_items[:10]
|
|
],
|
|
'social': [
|
|
{
|
|
'title': p.title,
|
|
'source': p.source,
|
|
'likes': p.likes
|
|
}
|
|
for p in trend_data.social_posts[:10]
|
|
]
|
|
}
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': result
|
|
})
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] ai_trends error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/ai/weather')
|
|
@require_api_token
|
|
def bot_ai_weather():
|
|
"""
|
|
查詢天氣資訊
|
|
|
|
Query Parameters:
|
|
- location: 地點 (預設 臺北市)
|
|
|
|
Returns:
|
|
- 天氣資訊和行銷建議
|
|
"""
|
|
try:
|
|
from services.trend_crawler import TrendCrawler
|
|
|
|
location = request.args.get('location', '臺北市')
|
|
|
|
trend_crawler = TrendCrawler()
|
|
weather = trend_crawler.fetch_weather(location)
|
|
|
|
if weather:
|
|
return jsonify({
|
|
'success': True,
|
|
'data': {
|
|
'location': weather.location,
|
|
'date': weather.date,
|
|
'description': weather.weather_description,
|
|
'temp_range': f"{weather.min_temp}°C ~ {weather.max_temp}°C",
|
|
'rain_probability': weather.rain_probability,
|
|
'humidity': weather.humidity,
|
|
'marketing_suggestions': weather.marketing_suggestions
|
|
}
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': '無法獲取天氣資訊'
|
|
}), 500
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] ai_weather error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@bot_api_bp.route('/bot/api/ai/suggest', methods=['POST'])
|
|
@require_api_token
|
|
def bot_ai_suggest():
|
|
"""
|
|
生成銷售策略建議
|
|
|
|
Request JSON:
|
|
- product_name: 商品名稱 (必填)
|
|
- trend_keywords: 趨勢關鍵字 (選填)
|
|
|
|
Returns:
|
|
- 銷售策略建議
|
|
"""
|
|
try:
|
|
data = request.get_json() or {}
|
|
product_name = data.get('product_name', '')
|
|
trend_keywords = data.get('trend_keywords', [])
|
|
|
|
if not product_name:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Missing product_name'
|
|
}), 400
|
|
|
|
from services.ollama_service import OllamaService
|
|
|
|
ollama_service = OllamaService()
|
|
|
|
system_prompt = """你是一位資深電商銷售策略顧問，專精於台灣市場。
|
|
請用繁體中文回答，簡潔明瞭。"""
|
|
|
|
context = f"熱門趨勢：{', '.join(trend_keywords)}" if trend_keywords else "無特定趨勢"
|
|
|
|
prompt = f"""請為「{product_name}」提供簡短銷售建議：
|
|
|
|
市場資訊：{context}
|
|
|
|
請提供：
|
|
1. 目標客群（一句話）
|
|
2. 主打賣點（一句話）
|
|
3. 促銷建議（一句話）
|
|
|
|
請簡潔回答，總共不超過 100 字。"""
|
|
|
|
result = ollama_service.generate(prompt, system_prompt=system_prompt, temperature=0.6)
|
|
|
|
if result.success:
|
|
return jsonify({
|
|
'success': True,
|
|
'data': {
|
|
'suggestion': result.content,
|
|
'model': result.model,
|
|
'duration': round(result.total_duration, 2) if result.total_duration else None
|
|
}
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': result.error
|
|
}), 500
|
|
|
|
except Exception as e:
|
|
sys_log.error(f"[BotAPI] ai_suggest error: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|