#!/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='正文按“已确认的进展、待办事项、遗留”分条填写。沿用项目记录中的负责人、原计划、当前计划、必须完成的阶段和影响；脚本不推断任务完成或阶段推进。') 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())