214 lines
9.1 KiB
Python
214 lines
9.1 KiB
Python
"""Transactional WeCom outbox. Network calls never hold an order transaction.
|
|
|
|
Protocol: https://developer.work.weixin.qq.com/document/path/91770
|
|
The worker uses its own sessions; importing the app never sends a message.
|
|
"""
|
|
import ipaddress
|
|
import json
|
|
import re
|
|
import secrets
|
|
import threading
|
|
from datetime import datetime, timedelta, timezone
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlsplit
|
|
from urllib.request import HTTPRedirectHandler, Request, build_opener
|
|
|
|
from sqlalchemy import and_, or_, text
|
|
|
|
from models import Notification, NotificationSettings, Order
|
|
|
|
CHANNELS = {'approval': '审批通知群', 'result': '结果通知群'}
|
|
STATES = {'pending': '等待发送', 'sending': '发送中', 'retry': '等待重试',
|
|
'sent': '已发送', 'failed': '发送失败', 'cancelled': '已跳过'}
|
|
MAX_ATTEMPTS = 5
|
|
|
|
|
|
def validate_webhook(value):
|
|
value = value.strip()
|
|
if value and not re.fullmatch(
|
|
r'https://qyapi\.weixin\.qq\.com/cgi-bin/webhook/send\?key=[A-Za-z0-9_-]{1,200}', value):
|
|
raise ValueError('Webhook 须为企业微信群机器人的完整 HTTPS 地址')
|
|
return value
|
|
|
|
|
|
def validate_base_url(value):
|
|
value = value.strip().rstrip('/')
|
|
if not value:
|
|
return ''
|
|
try:
|
|
parsed = urlsplit(value)
|
|
host = parsed.hostname or ''
|
|
port = parsed.port
|
|
if (parsed.scheme not in ('http', 'https') or not host or parsed.username or parsed.password
|
|
or parsed.path or parsed.query or parsed.fragment or len(value) > 300
|
|
or re.search(r'[\s\\<>"()\[\]]', value.replace('[' + host + ']', host))
|
|
or port == 0):
|
|
raise ValueError()
|
|
try:
|
|
addr = ipaddress.ip_address(host)
|
|
except ValueError:
|
|
if (host.lower() == 'localhost' or host.lower().endswith('.localhost')
|
|
or not re.fullmatch(r'[a-zA-Z0-9](?:[a-zA-Z0-9.-]*[a-zA-Z0-9])?', host)
|
|
or re.fullmatch(r'[0-9.]+', host)):
|
|
raise ValueError()
|
|
else:
|
|
networks = ('10.0.0.0/8', '172.16.0.0/12', '192.168.0.0/16', 'fc00::/7')
|
|
if not any(addr in ipaddress.ip_network(net) for net in networks):
|
|
raise ValueError()
|
|
except ValueError:
|
|
raise ValueError('请填写手机可访问的内网根地址，例如 http://192.168.1.10:8080；不要使用本机回环地址或订单路径') from None
|
|
return value
|
|
|
|
|
|
def enqueue_transition(db, order, log):
|
|
config = db.get(NotificationSettings, 1)
|
|
if not config or not config.enabled:
|
|
return
|
|
channel = 'approval' if order.status == 'pending_approval' else 'result'
|
|
if not getattr(config, channel + '_webhook'):
|
|
return
|
|
db.add(Notification(log_id=log.id, order_id=order.id, revision=order.revision,
|
|
channel=channel, event=log.action, expected_status=order.status))
|
|
|
|
|
|
def display_time(value):
|
|
return value.replace(tzinfo=timezone.utc).astimezone().strftime('%Y-%m-%d %H:%M')
|
|
|
|
|
|
def payload_for(notice, order, base_url):
|
|
if order:
|
|
# Legacy order numbers are untrusted Markdown input. No prices, bank data,
|
|
# comments, supplier contacts or account tokens are sent to the group.
|
|
number = re.sub(r'[^\w\-]', ' ', order.order_no)[:50]
|
|
event = re.sub(r'[^\w\-]', ' ', notice.event)[:50]
|
|
content = ('**采购订单通知**\n订单：%s\n第 %s 次提交 · %s\n'
|
|
'记录时间：%s\n[打开订单](%s/orders/%s/view)\n'
|
|
'请连接公司 Wi-Fi 后登录；以订单页最新状态为准。') % (
|
|
number, notice.revision, event, display_time(notice.created_at),
|
|
base_url, order.id)
|
|
else:
|
|
content = '**采购订单测试通知**\n%s\n[打开工作台](%s/)\n请连接公司 Wi-Fi 后登录。' % (
|
|
CHANNELS[notice.channel], base_url)
|
|
return {'msgtype': 'markdown', 'markdown': {'content': content}}
|
|
|
|
|
|
class DeliveryError(Exception):
|
|
"""Only controlled, credential-free messages may cross this boundary."""
|
|
|
|
|
|
class NoRedirect(HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
|
|
def send_wecom(webhook, payload):
|
|
validate_webhook(webhook)
|
|
if not webhook:
|
|
raise DeliveryError('未配置该通知群')
|
|
data = json.dumps(payload, ensure_ascii=False).encode('utf-8')
|
|
req = Request(webhook, data=data, headers={'Content-Type': 'application/json; charset=utf-8'}, method='POST')
|
|
try:
|
|
with build_opener(NoRedirect()).open(req, timeout=5) as response:
|
|
raw = response.read(16385)
|
|
if response.status != 200 or len(raw) > 16384:
|
|
raise DeliveryError('企业微信响应异常')
|
|
result = json.loads(raw)
|
|
code = result.get('errcode') if isinstance(result, dict) else None
|
|
if type(code) is not int:
|
|
raise DeliveryError('企业微信响应格式错误')
|
|
if code != 0:
|
|
raise DeliveryError('企业微信拒绝消息，错误码 %s' % code)
|
|
except HTTPError as error:
|
|
raise DeliveryError('企业微信 HTTP 错误 %s' % error.code) from None
|
|
except (URLError, OSError, TimeoutError):
|
|
raise DeliveryError('无法连接企业微信，请检查网络后重试') from None
|
|
except (ValueError, UnicodeError):
|
|
raise DeliveryError('企业微信响应格式错误') from None
|
|
|
|
|
|
def dispatch_once(factory, sender=None, now=None):
|
|
"""Claim -> commit -> send -> acknowledge, with a recoverable lease.
|
|
|
|
At-least-once delivery: a timeout/crash after WeCom accepts a message can
|
|
cause a duplicate notice. Approvals still require authenticated POSTs.
|
|
"""
|
|
now = now or datetime.utcnow()
|
|
with factory() as db:
|
|
db.execute(text('BEGIN IMMEDIATE'))
|
|
config = db.get(NotificationSettings, 1)
|
|
if not config or not config.enabled:
|
|
return False
|
|
if config.next_dispatch_at and config.next_dispatch_at > now:
|
|
return False
|
|
notice = db.query(Notification).filter(or_(
|
|
and_(Notification.status.in_(('pending', 'retry')), Notification.next_attempt_at <= now),
|
|
and_(Notification.status == 'sending', Notification.lease_until <= now)
|
|
)).order_by(Notification.id).first()
|
|
if not notice:
|
|
return False
|
|
order = db.get(Order, notice.order_id) if notice.order_id else None
|
|
obsolete = notice.order_id and (not order or (notice.channel in ('review', 'approval') and (
|
|
order.revision != notice.revision or order.status != notice.expected_status)))
|
|
webhook = getattr(config, notice.channel + '_webhook')
|
|
if obsolete or not webhook:
|
|
notice.status = 'cancelled'
|
|
notice.last_error = '订单已进入其他状态或版本' if obsolete else '通知群配置已移除'
|
|
db.commit()
|
|
return True
|
|
if notice.attempts >= MAX_ATTEMPTS:
|
|
notice.status, notice.last_error = 'failed', '已达到最大尝试次数，请检查后手动重试'
|
|
db.commit()
|
|
return True
|
|
notice.attempts += 1
|
|
notice.status = 'sending'
|
|
notice.lease_token = secrets.token_hex(16)
|
|
notice.lease_until = now + timedelta(seconds=60)
|
|
# Shared across workers and channels, including when two channels use
|
|
# the same bot. Up to 15 attempts/minute from this installation.
|
|
config.next_dispatch_at = now + timedelta(seconds=4)
|
|
payload = payload_for(notice, order, config.base_url)
|
|
notice_id, lease = notice.id, notice.lease_token
|
|
db.commit()
|
|
|
|
error = ''
|
|
try:
|
|
(sender or send_wecom)(webhook, payload)
|
|
except DeliveryError as failure:
|
|
error = str(failure)
|
|
except Exception:
|
|
# Exceptions may embed the secret URL. Never persist/log their text.
|
|
error = '通知发送异常，请检查网络及配置'
|
|
with factory() as db:
|
|
db.execute(text('BEGIN IMMEDIATE'))
|
|
notice = db.get(Notification, notice_id)
|
|
if notice.status != 'sending' or notice.lease_token != lease:
|
|
return True
|
|
notice.lease_until = notice.lease_token = None
|
|
notice.last_error = error
|
|
if error:
|
|
notice.status = 'failed' if notice.attempts >= MAX_ATTEMPTS else 'retry'
|
|
notice.next_attempt_at = now + timedelta(seconds=min(900, 30 * 2 ** (notice.attempts - 1)))
|
|
else:
|
|
notice.status, notice.sent_at = 'sent', now
|
|
db.commit()
|
|
return True
|
|
|
|
|
|
def start_notification_worker(app, factory):
|
|
"""Call once from the server entry point; returns a stop event for shutdown."""
|
|
if 'po_notification_stop' in app.extensions:
|
|
return app.extensions['po_notification_stop']
|
|
stop = threading.Event()
|
|
app.extensions['po_notification_stop'] = stop
|
|
|
|
def work():
|
|
while not stop.is_set():
|
|
try:
|
|
dispatch_once(factory)
|
|
except Exception:
|
|
app.logger.error('Notification worker iteration failed; will retry (details suppressed)')
|
|
stop.wait(1)
|
|
|
|
threading.Thread(target=work, name='wecom-outbox', daemon=True).start()
|
|
return stop
|