#!/usr/bin/env python3 import argparse import json import os import re import sys import time # 将 workspace 下的 site-packages 加入路径（离线安装的 SDK 位置） sys.path.insert(0, '/workspace/.local/lib/python3.11/site-packages') # ── 配置区：从 key.json 读取 ────────────────────────────────── _KEY_JSON_PATH = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))), 'key.json' ) def _load_sms_config(): """从 workspace 根目录的 key.json 读取 SMS 配置 结构: key.json -> "sms-notifier" -> "sms" """ try: with open(_KEY_JSON_PATH, 'r', encoding='utf-8') as f: data = json.load(f) cfg = data.get('sms-notifier', {}) return cfg except Exception as e: print(json.dumps({ 'status': 'error', 'message': f'无法读取 key.json 配置文件: {e}，路径: {_KEY_JSON_PATH}' }, ensure_ascii=False)) sys.exit(1) _cfg = _load_sms_config() SMS_ACCESS_KEY_ID = _cfg.get('SMS_ACCESS_KEY_ID', '') SMS_ACCESS_KEY_SECRET = _cfg.get('SMS_ACCESS_KEY_SECRET', '') SMS_SIGN_NAME = _cfg.get('SMS_SIGN_NAME', '') SMS_TEMPLATE_CODE = _cfg.get('SMS_TEMPLATE_CODE', '') SMS_ENDPOINT = _cfg.get('SMS_ENDPOINT', 'dysmsapi.aliyuncs.com') LOG_DIR = '/workspace/skills/sms-notifier/logs' CONTACTS_FILE = '/workspace/skills/sms-notifier/contacts_cache.json' os.makedirs(LOG_DIR, exist_ok=True) # ────────────────────────────────────────────────────────────── def looks_like_mobile(value: str) -> bool: return bool(re.fullmatch(r'1\d{10}', (value or '').strip())) def load_contacts(): if not os.path.exists(CONTACTS_FILE): return [] try: with open(CONTACTS_FILE, 'r', encoding='utf-8') as f: data = json.load(f) return data if isinstance(data, list) else [] except Exception: return [] def resolve_recipient(recipient: str): recipient = (recipient or '').strip() if looks_like_mobile(recipient): return {'mode': 'phone', 'name': recipient, 'phone': recipient} lowered = recipient.lower() for item in load_contacts(): name = str(item.get('name', '') or '').strip() phone = str(item.get('phone', '') or '').strip() if not name: continue if lowered == name.lower() or lowered in name.lower(): return {'mode': 'name', 'name': name, 'phone': phone} return None def extract_alarm_name(content: str, resolved_name: str) -> str: text = (content or '').strip() if text: text = re.sub(r'^(告警|报警|推送告警|发送告警|发送短信通知)[：:，,\s]*', '', text) text = text.strip('，,：: ') if text: return text return (resolved_name or '').strip() def send_via_sdk(phone: str, content: str, resolved_name: str, timestamp: int) -> dict: """使用阿里云官方 Python SDK 直接发送短信""" from alibabacloud_dysmsapi20170525.client import Client from alibabacloud_dysmsapi20170525 import models as sms_models from alibabacloud_tea_openapi import models as open_api_models alarm_name = extract_alarm_name(content, resolved_name) template_param = json.dumps({'message': alarm_name}, ensure_ascii=False) config = open_api_models.Config( access_key_id=SMS_ACCESS_KEY_ID, access_key_secret=SMS_ACCESS_KEY_SECRET, ) config.endpoint = SMS_ENDPOINT client = Client(config) send_req = sms_models.SendSmsRequest( phone_numbers=phone, sign_name=SMS_SIGN_NAME, template_code=SMS_TEMPLATE_CODE, template_param=template_param, ) try: resp = client.send_sms(send_req) body = resp.body if body.code == 'OK': result = { 'status': 'success', 'message': '短信发送成功', 'code': body.code, 'biz_id': body.biz_id or '', 'request_id': body.request_id or '', 'phone': phone, 'timestamp': timestamp, 'content_preview': (content[:50] + '...') if len(content) > 50 else content, 'template_code': SMS_TEMPLATE_CODE, 'sign_name': SMS_SIGN_NAME, 'template_param': {'message': alarm_name}, } else: result = { 'status': 'error', 'message': body.message or body.code, 'code': body.code, 'biz_id': body.biz_id or '', 'request_id': body.request_id or '', 'phone': phone, 'timestamp': timestamp, 'content_preview': (content[:50] + '...') if len(content) > 50 else content, 'template_code': SMS_TEMPLATE_CODE, 'sign_name': SMS_SIGN_NAME, 'template_param': {'message': alarm_name}, } except Exception as e: result = { 'status': 'error', 'message': 'SDK 调用失败: %s' % str(e), 'code': 'SDK_ERROR', 'phone': phone, 'timestamp': timestamp, 'content_preview': (content[:50] + '...') if len(content) > 50 else content, 'template_code': SMS_TEMPLATE_CODE, 'sign_name': SMS_SIGN_NAME, 'template_param': {'message': alarm_name}, } log_path = os.path.join(LOG_DIR, 'sms_log_%s.json' % timestamp) with open(log_path, 'w', encoding='utf-8') as f: json.dump(result, f, ensure_ascii=False, indent=2) return result def main(): parser = argparse.ArgumentParser() parser.add_argument('--recipient', required=True) parser.add_argument('--content', default='') args = parser.parse_args() recipient = (args.recipient or '').strip() content = (args.content or '').strip() if not recipient or recipient in {'相关负责人', '设备管理员', '发送短信通知相关人员'}: print(json.dumps({ 'status': 'need_contact_selection', 'message': '未提供明确联系人或手机号，禁止直接发送，请先查询联系人信息。' }, ensure_ascii=False)) return resolved = resolve_recipient(recipient) if not resolved: print(json.dumps({ 'status': 'need_contact_selection', 'message': '未匹配到联系人，请先查询联系人列表或提供正确手机号。', 'recipient': recipient }, ensure_ascii=False)) return if not looks_like_mobile(resolved['phone']): print(json.dumps({ 'status': 'need_contact_selection', 'message': '已匹配联系人，但手机号缺失或不可用。', 'name': resolved['name'], 'phone': resolved['phone'] }, ensure_ascii=False)) return timestamp = int(time.time()) try: result = send_via_sdk( phone=resolved['phone'], content=content, resolved_name=resolved['name'], timestamp=timestamp, ) result['name'] = resolved['name'] print(json.dumps(result, ensure_ascii=False)) except Exception as e: print(json.dumps({ 'status': 'error', 'message': str(e), 'name': resolved.get('name', recipient), 'phone': resolved.get('phone', ''), 'timestamp': timestamp, }, ensure_ascii=False)) if __name__ == '__main__': main()