76 lines
3.9 KiB
Python
76 lines
3.9 KiB
Python
"""Administrator-only notification setup and delivery diagnostics."""
|
|
from datetime import datetime
|
|
|
|
from flask import abort, flash, redirect, render_template, request, url_for
|
|
|
|
from models import Notification, NotificationSettings
|
|
from notifications import CHANNELS, STATES, display_time, validate_base_url, validate_webhook
|
|
from security import current_user, db, role_required
|
|
|
|
|
|
def register_notification_routes(app):
|
|
app.jinja_env.filters['notice_time'] = display_time
|
|
|
|
@app.route('/settings/notifications', methods=['GET', 'POST'])
|
|
@role_required('admin')
|
|
def notification_settings():
|
|
config = db.get(NotificationSettings, 1)
|
|
if request.method == 'POST':
|
|
if not config:
|
|
config = NotificationSettings(id=1)
|
|
db.add(config)
|
|
config.base_url = validate_base_url(request.form.get('base_url', ''))
|
|
for channel in CHANNELS:
|
|
field = channel + '_webhook'
|
|
value = request.form.get(field, '').strip()
|
|
if request.form.get('clear_' + channel) == '1':
|
|
setattr(config, field, '')
|
|
elif value:
|
|
setattr(config, field, validate_webhook(value))
|
|
elif not getattr(config, field):
|
|
setattr(config, field, '')
|
|
config.enabled = request.form.get('enabled') == '1'
|
|
if config.enabled and not (config.base_url and config.approval_webhook):
|
|
raise ValueError('启用通知前，请配置内网访问地址和审批通知群')
|
|
db.commit()
|
|
flash('通知配置已保存；启用后新发生的审批操作会自动通知', 'ok')
|
|
return redirect(url_for('notification_settings'))
|
|
# Never pass credentials to a template, even through hidden inputs.
|
|
settings = {'enabled': bool(config and config.enabled), 'base_url': config.base_url if config else '',
|
|
'configured': {c: bool(config and getattr(config, c + '_webhook')) for c in CHANNELS}}
|
|
rows = db.query(Notification).order_by(Notification.id.desc()).limit(100).all()
|
|
response = app.make_response(render_template('notifications.html', user=current_user(),
|
|
settings=settings, notices=rows, channels=CHANNELS, states=STATES))
|
|
response.headers['Cache-Control'] = 'no-store'
|
|
return response
|
|
|
|
@app.post('/settings/notifications/test')
|
|
@role_required('admin')
|
|
def notification_test():
|
|
config = db.get(NotificationSettings, 1)
|
|
channel = request.form.get('channel', '')
|
|
if channel not in CHANNELS or not config or not config.enabled or not getattr(config, channel + '_webhook'):
|
|
raise ValueError('请先启用并保存该群的通知配置')
|
|
# Double taps/reloads should not create a burst of test messages.
|
|
if not db.query(Notification).filter(Notification.order_id.is_(None), Notification.channel == channel,
|
|
Notification.status.in_(('pending', 'retry', 'sending'))).first():
|
|
db.add(Notification(channel=channel, event='测试通知'))
|
|
db.commit()
|
|
flash('测试通知已加入队列，请稍后刷新查看结果并在群内确认', 'ok')
|
|
return redirect(url_for('notification_settings'))
|
|
|
|
@app.post('/settings/notifications/<int:nid>/retry')
|
|
@role_required('admin')
|
|
def notification_retry(nid):
|
|
notice = db.get(Notification, nid)
|
|
if not notice:
|
|
abort(404)
|
|
if notice.status not in ('failed', 'retry'):
|
|
abort(409, description='该通知无需重试，请刷新查看最新状态')
|
|
notice.status, notice.attempts, notice.last_error = 'pending', 0, ''
|
|
notice.next_attempt_at = datetime.utcnow()
|
|
notice.lease_token = notice.lease_until = None
|
|
db.commit()
|
|
flash('已安排重试；发送前会再次核对订单状态', 'ok')
|
|
return redirect(url_for('notification_settings'))
|