308 lines
16 KiB
Python
308 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""Hub notifications via a confirmed personal DWS account. Python 3.9+, stdlib."""
|
||
import argparse
|
||
import base64
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
PREFIX = 'DINGTALK_PROGRESS_'
|
||
KEYS = {'ENABLED', 'PROFILE', 'TARGET', 'PROJECT'}
|
||
EVENTS = ('start', 'milestone', 'blocked', 'complete', 'failed')
|
||
INLINE_MENTION = re.compile(r'@\{([^{}\r\n]+)\}')
|
||
DNS_ERROR = re.compile(r'lookup [\w.-]+(?: on [^:\s]+(?::\d+)?)?: [^\n]*'
|
||
r'(?:i/o timeout|no such host|server misbehaving|temporary failure in name resolution)', re.I)
|
||
|
||
|
||
class NoticeError(Exception):
|
||
"""Safe diagnostics: never echo raw DWS errors or configuration."""
|
||
|
||
|
||
def emit(status, message, **extra):
|
||
print(json.dumps(dict(status=status, message=message, **extra), ensure_ascii=False))
|
||
|
||
|
||
def json_object(text):
|
||
try:
|
||
data = json.loads(text)
|
||
return data if isinstance(data, dict) else {}
|
||
except (ValueError, TypeError):
|
||
return {}
|
||
|
||
|
||
def read_config(root):
|
||
path = root / '.dingtalk' / 'config.env'
|
||
if not path.exists():
|
||
return {}
|
||
result = {}
|
||
for line in path.read_text(encoding='utf-8-sig').splitlines():
|
||
if not line.strip() or line.lstrip().startswith('#'):
|
||
continue
|
||
key, sep, value = line.partition('=')
|
||
key = key.strip()
|
||
if not sep or key not in {PREFIX + k for k in KEYS} or key in result:
|
||
raise NoticeError('通知配置不是当前个人账号格式，请由 Hub 确认账号和群后重新 init；不沿用机器人配置。')
|
||
result[key] = value.strip()
|
||
config = {k[len(PREFIX):]: v for k, v in result.items()}
|
||
if config.get('ENABLED', '0') not in ('0', '1'):
|
||
raise NoticeError('通知开关只能是 0 或 1。')
|
||
return config
|
||
|
||
|
||
def clean_env():
|
||
return {k: v for k, v in os.environ.items()
|
||
if k not in ('DING_SEC', 'DWS_CLIENT_ID', 'DWS_CLIENT_SECRET') and not k.startswith(PREFIX)}
|
||
|
||
|
||
def dws_path():
|
||
executable = sys.executable.replace('\\', '/').lower()
|
||
if sys.platform == 'win32' and ('/windowsapps/' in executable or 'pythonsoftwarefoundation.python.' in executable):
|
||
raise NoticeError('请使用项目 PowerShell 入口或独立 Python 3.9+；商店版 Python 可能无法识别 DWS 登录。')
|
||
path = shutil.which('dws')
|
||
if not path or (os.name == 'nt' and Path(path).suffix.lower() != '.exe'):
|
||
raise NoticeError('未找到原生 dws，请安装并加入当前 Hub 运行环境的 PATH。')
|
||
return path
|
||
|
||
|
||
def accepted_task(data):
|
||
if (data.get('ok') is not True or data.get('identity') != 'user'
|
||
or data.get('tool') != 'send_personal_message' or data.get('partial') or data.get('failures')):
|
||
return None
|
||
result = data.get('result')
|
||
if not isinstance(result, dict) or result.get('success') is not True or result.get('errorCode') not in (None, 0, ''):
|
||
return None
|
||
payload = result.get('result')
|
||
task = payload.get('openTaskId') if isinstance(payload, dict) else None
|
||
return task.strip() if isinstance(task, str) and task.strip() else None
|
||
|
||
|
||
def invoke(command, run=subprocess.run, sleep=time.sleep, sending=False):
|
||
for attempt in range(3):
|
||
try:
|
||
result = run(command, env=clean_env(), capture_output=True, text=True,
|
||
encoding='utf-8', errors='replace', timeout=60)
|
||
except subprocess.TimeoutExpired:
|
||
raise NoticeError('钉钉调用超时，结果未知，不自动重发；主任务继续。') from None
|
||
except OSError:
|
||
raise NoticeError('无法启动 DWS，请检查安装；本次停止。') from None
|
||
data = json_object(result.stdout)
|
||
if sending and accepted_task(data):
|
||
return data
|
||
if result.returncode == 0:
|
||
if sending:
|
||
raise NoticeError('未取得有效的个人发送回执，结果未知，不自动重发。')
|
||
if not data or data.get('success') is False or data.get('ok') is False:
|
||
raise NoticeError('DWS 查询未返回有效结果，本次停止。')
|
||
return data
|
||
error = json_object(result.stderr).get('error', {})
|
||
if isinstance(error, dict) and error.get('category') == 'auth':
|
||
raise NoticeError('个人 DWS 登录失效，请在 Hub 的运行账号下恢复登录；不换身份发送。')
|
||
if isinstance(error, dict) and error.get('retryable') is False:
|
||
raise NoticeError('DWS 拒绝本次调用，请检查权限或参数；不自动重试。')
|
||
if DNS_ERROR.search(result.stdout + '\n' + result.stderr):
|
||
if attempt < 2:
|
||
sleep(10)
|
||
continue
|
||
raise NoticeError('域名解析连续失败 3 次，通知停止，主任务继续。')
|
||
raise NoticeError('钉钉调用失败或结果未知；请检查登录、权限或网络，不自动重发。')
|
||
|
||
|
||
def check_login(cli, profile, run, sleep):
|
||
if not profile or ':' not in profile or any(c.isspace() for c in profile):
|
||
raise NoticeError('缺少已确认的发送账号，请先由 Hub 初始化通知。')
|
||
data = invoke([cli, 'auth', 'status', '--profile', profile, '--format', 'json'], run, sleep)
|
||
if data.get('authenticated') is not True or data.get('token_valid') is not True:
|
||
raise NoticeError('个人 DWS 未登录或授权已失效，请负责人完成登录；本次未发送。')
|
||
|
||
|
||
def mention_names(names):
|
||
names = list(dict.fromkeys(n.strip() for value in names for n in value.replace('，', ',').split(',')))
|
||
if any(not name for name in names):
|
||
raise NoticeError('提醒对象不能为空。')
|
||
return names
|
||
|
||
|
||
def format_mentions(text, names, people=None):
|
||
names = mention_names(names)
|
||
placed = {match.group(1).strip() for match in INLINE_MENTION.finditer(text)}
|
||
if '@{' in INLINE_MENTION.sub('', text) or not placed.issubset(names):
|
||
raise NoticeError('正文中的 @{完整姓名} 必须完整填写，并在 --at 中列出同一姓名。')
|
||
by_name = {name: person for person, name in people.items()} if people is not None else None
|
||
if by_name is not None and not set(names).issubset(by_name):
|
||
raise NoticeError('提醒对象未全部唯一匹配，本次停止，不删掉 @ 后发送。')
|
||
|
||
def marker(name):
|
||
return f'<@{by_name[name]}>' if by_name is not None else f'@{name}'
|
||
|
||
text = INLINE_MENTION.sub(lambda match: marker(match.group(1).strip()), text)
|
||
remaining = [marker(name) for name in names if name not in placed]
|
||
if remaining:
|
||
text += '\n\n' + ' '.join(remaining)
|
||
return text
|
||
|
||
|
||
def mentions(names, cli, profile, run, sleep):
|
||
people = {}
|
||
names = mention_names(names)
|
||
for name in names:
|
||
data = invoke([cli, 'aisearch', 'person', '--query', name, '--dimension', 'name',
|
||
'--profile', profile, '--format', 'json'], run, sleep)
|
||
candidates = {}
|
||
for row in data.get('result', []):
|
||
if not isinstance(row, dict):
|
||
continue
|
||
meta = row.get('meta') if isinstance(row.get('meta'), dict) else {}
|
||
person = row.get('openDingTalkId')
|
||
if (meta.get('name') or row.get('title') or row.get('author')) == name and isinstance(person, str) and person.strip():
|
||
candidates[person] = name
|
||
if len(candidates) != 1:
|
||
raise NoticeError('提醒对象未唯一匹配，请确认完整姓名；不猜编号、不删掉 @ 后发送。')
|
||
people.update(candidates)
|
||
return people
|
||
|
||
|
||
def initialize(args, root, run, sleep):
|
||
if not args.yes:
|
||
raise NoticeError('先向负责人确认发送账号、项目群及自动通知，再使用 init --yes。')
|
||
values = {'ENABLED': '1', 'PROFILE': args.profile.strip(), 'TARGET': args.group.strip(), 'PROJECT': args.project.strip()}
|
||
if any(not v or '\n' in v or '\r' in v or '\0' in v for v in values.values()):
|
||
raise NoticeError('初始化内容不能为空或含换行。')
|
||
path = root / '.dingtalk' / 'config.env'
|
||
if path.is_symlink() or path.parent.is_symlink():
|
||
raise NoticeError('配置路径不能是符号链接。')
|
||
if path.exists() and not args.replace:
|
||
raise NoticeError('配置已存在；变更账号或群须重新确认后使用 --replace。')
|
||
cli = dws_path()
|
||
profiles = invoke([cli, 'profile', 'list', '--format', 'json'], run, sleep).get('profiles', [])
|
||
if len([p for p in profiles if isinstance(p, dict) and p.get('profile') == values['PROFILE']]) != 1:
|
||
raise NoticeError('初始化账号不在 DWS 实际账号列表中。')
|
||
check_login(cli, values['PROFILE'], run, sleep)
|
||
# Hub supplies a group ID from same-profile discovery after human confirmation.
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
content = '# Hub 经负责人确认后生成；仅个人 DWS，不保存登录凭据。\n' + ''.join(
|
||
f'{PREFIX}{k}={v}\n' for k, v in values.items())
|
||
descriptor, temporary = tempfile.mkstemp(prefix='.notify-', suffix='.tmp', dir=path.parent)
|
||
try:
|
||
with os.fdopen(descriptor, 'w', encoding='utf-8', newline='\n') as stream:
|
||
stream.write(content)
|
||
os.replace(temporary, path)
|
||
finally:
|
||
if os.path.exists(temporary):
|
||
os.unlink(temporary)
|
||
emit('initialized', '已保存负责人确认的个人账号和项目群，通知已启用；未发送测试消息。')
|
||
|
||
|
||
def parse_args(argv):
|
||
parser = argparse.ArgumentParser(description='Hub 个人 DWS 通知：初始化、进度、提醒、只读检查。')
|
||
commands = parser.add_subparsers(dest='command', required=True)
|
||
init = commands.add_parser('init', help='负责人确认后创建配置，不发送消息')
|
||
init.add_argument('--profile', required=True)
|
||
init.add_argument('--group', required=True)
|
||
init.add_argument('--project', required=True)
|
||
init.add_argument('--yes', action='store_true')
|
||
init.add_argument('--replace', action='store_true')
|
||
progress = commands.add_parser(
|
||
'progress', help='根据已留档的项目事实汇报当前阶段',
|
||
description='正文按“已确认的进展、待办事项、遗留”分条填写。阶段交接用一条 milestone 合并上一阶段总结与下一阶段已确认安排；项目 complete 用于明确关闭后的总结。沿用负责人、原计划、当前计划、必须完成的阶段和影响；脚本不判断授权、任务完成或阶段推进。')
|
||
progress.add_argument('event', choices=EVENTS)
|
||
progress.add_argument('--stage', required=True, help='项目当前阶段编号及中文名称；每次从有效项目记录取得')
|
||
progress.add_argument('summary', nargs='*')
|
||
progress.add_argument('--file', help='UTF-8 正文文件，仅含三个栏目；标题和时间由脚本生成')
|
||
message = commands.add_parser('message')
|
||
message.add_argument('--title', required=True)
|
||
body = message.add_mutually_exclusive_group(required=True)
|
||
body.add_argument('--text')
|
||
body.add_argument('--file')
|
||
check = commands.add_parser('check', help='检查固定账号登录及可选姓名，不发送')
|
||
for command in (progress, message, check):
|
||
command.add_argument('--at', action='append', default=[], help='完整姓名；正文可用 @{完整姓名} 定位，可重复或用逗号分隔')
|
||
for command in (progress, message):
|
||
command.add_argument('--dry-run', action='store_true', help='完全离线，只预览中文正文')
|
||
return parser.parse_args(argv)
|
||
|
||
|
||
def render(args, config, root):
|
||
if args.command == 'progress' and bool(args.summary) == bool(args.file):
|
||
raise NoticeError('进度摘要和 --file 必须且只能提供一项。')
|
||
if args.file:
|
||
path = Path(args.file).resolve()
|
||
config_path = root / '.dingtalk' / 'config.env'
|
||
if path == config_path.resolve() or (path.exists() and config_path.exists() and path.samefile(config_path)):
|
||
raise NoticeError('不能把配置文件当作消息正文。')
|
||
text = path.read_text(encoding='utf-8')
|
||
else:
|
||
text = ' '.join(args.summary) if args.command == 'progress' else args.text
|
||
if not text.strip() or len(text.encode('utf-8')) > 12000:
|
||
raise NoticeError('请提供不超过 12000 字节的简短中文摘要。')
|
||
raw_title = args.stage if args.command == 'progress' else args.title
|
||
title = raw_title.strip()
|
||
if not title or len(title) > 200 or any(ord(c) < 32 or ord(c) == 127 for c in raw_title):
|
||
raise NoticeError('请使用不超过 200 字的单行标题。')
|
||
project = config.get('PROJECT') or root.name
|
||
timestamp = dt.datetime.now().astimezone().strftime('%Y-%m-%d %H:%M %z')
|
||
return f'**{project}｜{title}**\n\n项目进度自动汇报 · {timestamp}\n\n{text}'
|
||
|
||
|
||
def main(argv=None, root=ROOT, run=subprocess.run, sleep=time.sleep):
|
||
try:
|
||
encoded = os.environ.pop(PREFIX + 'ARGV_B64', '')
|
||
if argv is None and encoded:
|
||
argv = json.loads(base64.b64decode(encoded, validate=True).decode('utf-8'))
|
||
if not isinstance(argv, list) or not all(isinstance(v, str) for v in argv):
|
||
raise NoticeError('PowerShell 参数无效。')
|
||
args = parse_args(argv)
|
||
if args.command == 'init':
|
||
initialize(args, root, run, sleep)
|
||
return 0
|
||
config = read_config(root)
|
||
if args.command != 'check' and args.dry_run:
|
||
text = format_mentions(render(args, config, root), args.at)
|
||
emit('preview', '仅离线预览，账号、群与 @ 效果未验证。', text=text, mentions=mention_names(args.at))
|
||
return 0
|
||
if args.command != 'check' and config.get('ENABLED') != '1':
|
||
emit('skipped', '通知尚未初始化或未启用；本次未发送，主任务继续。')
|
||
return 0
|
||
if not config.get('PROFILE') or not config.get('TARGET'):
|
||
raise NoticeError('请由 Hub 先确认个人账号和项目群并执行 init。')
|
||
text = render(args, config, root) if args.command != 'check' else None
|
||
if text is not None:
|
||
format_mentions(text, args.at) # Validate inline names before any DWS call.
|
||
cli = dws_path()
|
||
profile = config['PROFILE']
|
||
if args.command == 'check':
|
||
check_login(cli, profile, run, sleep)
|
||
people = mentions(args.at, cli, profile, run, sleep)
|
||
if args.command == 'check':
|
||
emit('checked', '登录及所请求的姓名查询通过；未发送，未验证实际提醒效果。', names=list(people.values()))
|
||
return 0
|
||
text = format_mentions(text, args.at, people)
|
||
command = [cli, 'chat', '+messages-send', '--as', 'user', '--group', config['TARGET'],
|
||
'--markdown', text, '--profile', profile, '--yes', '--format', 'json']
|
||
if people:
|
||
command += ['--at-open-dingtalk-ids', ','.join(people)]
|
||
data = invoke(command, run, sleep, sending=True)
|
||
emit('accepted', '个人 DWS 已接受发送请求，不代表已投递、已读或已触发接收端提醒。',
|
||
openTaskId=accepted_task(data), mentions=list(people.values()))
|
||
return 0
|
||
except NoticeError as error:
|
||
emit('error', str(error))
|
||
return 1
|
||
except (OSError, ValueError, UnicodeError, TypeError, subprocess.SubprocessError):
|
||
emit('error', '配置、正文或调用结果异常；本次停止，不自动重发，主任务继续。')
|
||
return 1
|
||
|
||
|
||
if __name__ == '__main__':
|
||
for stream in (sys.stdout, sys.stderr):
|
||
if hasattr(stream, 'reconfigure'):
|
||
stream.reconfigure(encoding='utf-8')
|
||
sys.exit(main())
|