128 lines
6.2 KiB
Python
128 lines
6.2 KiB
Python
#!/usr/bin/env python3
|
||
"""Send one signed daily DingTalk result from cloud, without an AI dependency."""
|
||
import argparse
|
||
import base64
|
||
import fcntl
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
from zoneinfo import ZoneInfo
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
from ding_daily_sync import STATE, atomic, now, read_json
|
||
|
||
|
||
def current(status, today):
|
||
try:
|
||
return status and datetime.fromisoformat(status['checked_at']).astimezone(
|
||
ZoneInfo('Asia/Shanghai')).date().isoformat() == today
|
||
except (KeyError, ValueError, TypeError):
|
||
return False
|
||
|
||
|
||
def message(cloud, local, today, test=False):
|
||
lines = [f"采购数据同步结果 | {today}"]
|
||
if test:
|
||
lines = ['钉钉机器人接入测试：通知通道及@配置验证。', '以下为最近记录，不代表今天09:00定时任务已执行。']
|
||
cloud_ok = bool(current(cloud, today) and cloud.get('state') in {'success', 'waiting_source'}
|
||
and cloud.get('version') and cloud.get('cache_refreshed'))
|
||
local_ok = bool(cloud_ok and current(local, today) and local.get('state') in {'success', 'waiting_source'}
|
||
and local.get('version') == cloud['version'] and local.get('cache_refreshed')
|
||
and local.get('target_inventory') == cloud.get('target_inventory'))
|
||
if cloud_ok:
|
||
coverage = cloud.get('coverage', {})
|
||
lines += [f"库存日期：{cloud.get('target_inventory')}，已同步{coverage.get('inventory', 0)}份仓库文件",
|
||
f"出库日期：{cloud.get('target_outbound')}，已同步{coverage.get('outbound', 0)}份来源文件",
|
||
'云端：已上传文件同步成功，缓存已刷新。']
|
||
elif current(cloud, today) and cloud.get('state') == 'error':
|
||
lines += ['云端：同步失败，请查看服务器任务日志。']
|
||
else:
|
||
lines += ['云端：未发现当天完成结果（任务未完成或未运行）。']
|
||
if local_ok:
|
||
lines += ['8.186：对应版本同步成功，缓存已刷新。']
|
||
elif current(local, today) and local.get('state') == 'error':
|
||
lines += ['8.186：同步失败，请查看本机任务日志。']
|
||
else:
|
||
lines += ['8.186：未收到当天对应版本完成回执，可能未开机、未联网或任务未完成。']
|
||
if cloud_ok:
|
||
missing = cloud.get('missing', [])
|
||
lines += [f'未上传文件：{len(missing)}份，已跳过，不等待补录。']
|
||
if missing:
|
||
labels = {'inventory': '库存', 'outbound': '出库', 'dymuto': '抖音M', 'dyqsimelr': '抖音Q'}
|
||
lines += ['跳过：' + '、'.join('/'.join(labels.get(part, part) for part in item.split(':')) for item in missing[:30])]
|
||
if len(missing) > 30:
|
||
lines += [f'其余{len(missing) - 30}份详见日志。']
|
||
lines += ['每日一次；同步和通知均不调用AI模型。']
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def send(config, content):
|
||
url = urllib.parse.urlsplit(config['webhook'])
|
||
if url.scheme != 'https' or url.netloc != 'oapi.dingtalk.com' or url.path != '/robot/send':
|
||
raise ValueError('Invalid DingTalk endpoint')
|
||
timestamp = str(int(time.time() * 1000))
|
||
signature = base64.b64encode(hmac.new(config['secret'].encode(),
|
||
(timestamp + '\n' + config['secret']).encode(), hashlib.sha256).digest()).decode()
|
||
query = urllib.parse.parse_qsl(url.query) + [('timestamp', timestamp), ('sign', signature)]
|
||
signed_url = urllib.parse.urlunsplit(url._replace(query=urllib.parse.urlencode(query)))
|
||
mobiles = config.get('at_mobiles', [])
|
||
content += '\n' + ' '.join('@' + mobile for mobile in mobiles)
|
||
payload = {'msgtype': 'text', 'text': {'content': content},
|
||
'at': {'atMobiles': mobiles, 'isAtAll': False}}
|
||
request = urllib.request.Request(signed_url, data=json.dumps(payload, ensure_ascii=False).encode(),
|
||
headers={'Content-Type': 'application/json; charset=utf-8'}, method='POST')
|
||
# Suppress exceptions and API messages that might contain the signed URL or credentials.
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=20) as response:
|
||
result = json.load(response)
|
||
except Exception:
|
||
raise RuntimeError('DingTalk request failed; no retry') from None
|
||
if result.get('errcode') != 0:
|
||
raise RuntimeError(f"DingTalk rejected notification (errcode={result.get('errcode')}); no retry")
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument('--test', action='store_true', help='Send one explicitly labelled setup test')
|
||
args = parser.parse_args()
|
||
os.umask(0o077)
|
||
with open(STATE / 'notify.lock', 'w') as lock:
|
||
try:
|
||
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except BlockingIOError:
|
||
print('Notifier already running; skipped')
|
||
return
|
||
today = now().date().isoformat()
|
||
receipt = STATE / 'notifications' / (('test-' if args.test else '') + today + '.json')
|
||
if receipt.exists():
|
||
print('Notification already attempted for this day; skipped')
|
||
return
|
||
config_path = STATE / 'robot.json'
|
||
if config_path.stat().st_mode & 0o077:
|
||
raise RuntimeError('Robot configuration permissions must be 0600')
|
||
config = read_json(config_path)
|
||
content = message(read_json(STATE / 'status.json', {}), read_json(STATE / 'local-ack.json', {}), today, args.test)
|
||
record = {'attempted_at': now().isoformat(), 'state': 'attempted', 'test': args.test}
|
||
# Record before sending: an ambiguous network failure must not send duplicate daily messages.
|
||
atomic(receipt, record)
|
||
try:
|
||
send(config, content)
|
||
except Exception as error:
|
||
atomic(receipt, {**record, 'state': 'error', 'error': str(error)})
|
||
raise
|
||
atomic(receipt, {**record, 'state': 'sent', 'sent_at': now().isoformat()})
|
||
print('DingTalk accepted notification; configured @ mention included')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
try:
|
||
main()
|
||
except Exception as error:
|
||
print(str(error), file=sys.stderr)
|
||
sys.exit(1)
|