Files
stock-info-crawler/app/api/server.py

122 lines
4.6 KiB
Python

from __future__ import annotations
from datetime import datetime
from flask import Flask, jsonify, request
from app.services import notifications as notif
def create_app(crawler) -> Flask:
app = Flask(__name__)
# Support single crawler or a list of crawlers
crawlers = None
if isinstance(crawler, (list, tuple)):
crawlers = list(crawler)
@app.get('/health')
def health():
return jsonify({"status": "healthy", "timestamp": datetime.now().isoformat()})
@app.get('/stats')
def stats():
if crawlers is not None:
return jsonify({
(getattr(c, 'symbol', getattr(c, 'name', f"crawler_{i}")) or f"crawler_{i}"):
c.stats for i, c in enumerate(crawlers)
})
if crawler:
return jsonify(crawler.stats)
return jsonify({"error": "Crawler not initialized"}), 500
@app.get('/info')
def info():
if crawlers is not None:
out = []
for c in crawlers:
out.append({
"name": getattr(c, 'name', 'unknown'),
"type": c.__class__.__name__,
"symbol": getattr(c, 'symbol', None),
"schedule": getattr(c.config, 'run_daily_at', None) or f"every {c.config.check_interval}s",
})
return jsonify(out)
if not crawler:
return jsonify({"error": "Crawler not initialized"}), 500
return jsonify({
"name": getattr(crawler, 'name', 'unknown'),
"type": crawler.__class__.__name__,
"symbol": getattr(crawler, 'symbol', None),
"schedule": getattr(crawler.config, 'run_daily_at', None) or f"every {crawler.config.check_interval}s",
})
@app.get('/check')
def manual_check():
if crawlers is not None:
results = []
for c in crawlers:
r = c.run_check() or []
results.append({
"symbol": getattr(c, 'symbol', None),
"new": len(r)
})
return jsonify({"results": results})
if not crawler:
return jsonify({"error": "Crawler not initialized"}), 500
result = crawler.run_check() or []
return jsonify({"result": f"Found {len(result)} new picks"})
@app.get('/notify_test')
def notify_test():
channel = (request.args.get('channel') or 'email').lower()
target = request.args.get('target')
test_pick = [notif.build_test_pick()]
def _send_for(c):
if channel == 'email':
if not c.config.email:
return {"error": "Email config not set"}
# Build subject/body using crawler's formatting hook for consistency
if hasattr(c, '_build_email'):
subject, body = c._build_email(test_pick)
else:
subject = f"{getattr(c, 'name', '通知測試')}(測試)"
body = notif.format_email_body(test_pick)
notif.send_custom_email(subject, body, c.config.email)
elif channel == 'webhook':
if not c.config.webhook_url:
return {"error": "Webhook URL not set"}
notif.send_webhook(test_pick, c.config.webhook_url)
elif channel == 'discord':
if not c.config.discord_webhook:
return {"error": "Discord webhook not set"}
notif.send_discord(test_pick, c.config.discord_webhook)
else:
return {"error": f"Unsupported channel: {channel}"}
return {"result": f"Test notification sent via {channel}"}
if crawlers is not None:
results = {}
for c in crawlers:
key = getattr(c, 'symbol', getattr(c, 'name', 'unknown'))
if target and key != target:
continue
try:
results[key] = _send_for(c)
except Exception as e:
c.logger.error(f"測試通知發送失敗({key}): {e}")
results[key] = {"error": str(e)}
return jsonify(results)
if not crawler:
return jsonify({"error": "Crawler not initialized"}), 500
try:
res = _send_for(crawler)
if 'error' in res:
return jsonify(res), 400
return jsonify(res)
except Exception as e:
crawler.logger.error(f"測試通知發送失敗: {e}")
return jsonify({"error": str(e)}), 500
return app