mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-17 21:25:47 +02:00
72 lines
3.2 KiB
Python
72 lines
3.2 KiB
Python
#coding: utf-8
|
|
# -------------------------------------------------------------------
|
|
# aapanel - agent notification CLI
|
|
# Standalone entry that sends a notification via mod/base/push_mod channels.
|
|
# Call directly for an immediate notify, or from a cron task for scheduled ones:
|
|
# btpython /www/server/panel/script/notify_cli.py --title "..." --message "..." --channels mail,discord
|
|
#
|
|
# send_notify() is the single source of truth for the sending logic - it is also
|
|
# imported by the agent Notify tool (panel_tools.Notify), so the CLI and the tool
|
|
# never diverge.
|
|
# -------------------------------------------------------------------
|
|
import sys
|
|
import os
|
|
import argparse
|
|
|
|
|
|
def send_notify(title, message, channels=None):
|
|
"""Send a notification to all enabled non-sms push channels.
|
|
channels: optional list of sender_type (e.g. ['mail','discord']); None = all enabled.
|
|
Returns a list of per-channel dicts {channel, ok, error}.
|
|
Caller must ensure panel root is on sys.path (so mod.base.* is importable)."""
|
|
from mod.base.push_mod.mods import SenderConfig
|
|
from mod.base.push_mod.system import PushSystem
|
|
ps = PushSystem()
|
|
wanted = set(channels) if channels else None
|
|
results = []
|
|
for s in SenderConfig().config:
|
|
if not s.get("used"):
|
|
continue
|
|
st = s.get("sender_type")
|
|
if st == "sms": # sms uses a template mechanism, not free text
|
|
continue
|
|
if wanted is not None and st not in wanted:
|
|
continue
|
|
try:
|
|
res = ps.sender_cls(st)(s).send_msg(msg=message, title=title)
|
|
results.append({"channel": st, "ok": not isinstance(res, str),
|
|
"error": res if isinstance(res, str) else None})
|
|
except Exception as e:
|
|
results.append({"channel": st, "ok": False, "error": str(e)})
|
|
return results
|
|
|
|
|
|
def main():
|
|
# Panel runtime environment - only needed when run as a standalone script.
|
|
os.chdir('/www/server/panel')
|
|
sys.path.insert(0, '/www/server/panel')
|
|
sys.path.insert(0, 'class/')
|
|
sys.path.insert(0, 'class_v2/')
|
|
|
|
parser = argparse.ArgumentParser(description="Send a notification via configured push channels.")
|
|
parser.add_argument("--title", required=True, help="Notification title")
|
|
parser.add_argument("--message", required=True, help="Notification body (markdown supported by most channels)")
|
|
parser.add_argument("--channels", default="",
|
|
help="Comma-separated sender_type list, e.g. 'mail,discord'. Empty = all enabled non-sms channels")
|
|
args = parser.parse_args()
|
|
|
|
channels = [c.strip() for c in args.channels.split(",") if c.strip()] or None
|
|
# Shell double-quotes don't parse \n/\t; convert literal escapes so a cron sBody like
|
|
# --message "line1\nline2" renders as two lines (Notify tool passes real newlines, unaffected).
|
|
message = args.message.replace('\\n', '\n').replace('\\t', '\t')
|
|
results = send_notify(args.title, message, channels)
|
|
sent = sum(1 for r in results if r["ok"])
|
|
print("notify sent %d/%d channels" % (sent, len(results)))
|
|
for r in results:
|
|
print(" - %s: %s" % (r["channel"], "ok" if r["ok"] else "fail: %s" % r.get("error")))
|
|
sys.exit(0 if sent > 0 else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|