81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
"""企业微信欢迎消息一次性群发:给可见范围内还没有应用会话的成员各推一条欢迎语。
|
|
|
|
背景(会话冷启动):应用配了「应用主页」后,点应用只进主页;应用会话要等应用发过
|
|
第一条消息才会出现在成员的消息列表里。本脚本给**存量成员**"种"会话入口;增量新
|
|
成员由免登入口(/v1/wecom/entry 回调)自动补推,不需要重复跑本脚本。
|
|
|
|
用法(服务器上,.env 同目录,需 WECOM_CORPID/AGENTID/SECRET + ZCBOT_DB_URL 可达):
|
|
.venv/Scripts/python.exe scripts/wecom_welcome_broadcast.py --dry-run # 只看名单不发送
|
|
.venv/Scripts/python.exe scripts/wecom_welcome_broadcast.py # 实际发送
|
|
|
|
去重(已推过 / 已有对话的跳过)由 service.maybe_send_wecom_welcome + wecom_welcomes
|
|
表保证,重复跑安全。ASCII 输出(Windows 控制台 GBK)。
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
# 仓库根加入 sys.path(脚本在 scripts/ 下,直跑时 core 在上一级)
|
|
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if _ROOT not in sys.path:
|
|
sys.path.insert(0, _ROOT)
|
|
|
|
|
|
def _load_env(path: str) -> None:
|
|
"""加载 .env:优先 python-dotenv,没装则手动解析(只填未设置的 key)。"""
|
|
try:
|
|
from dotenv import load_dotenv
|
|
load_dotenv(path)
|
|
return
|
|
except Exception:
|
|
pass
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
_load_env(os.path.join(_ROOT, ".env"))
|
|
|
|
from core.wechat import service, wecom # noqa: E402 —— 须在 _load_env 之后 import
|
|
|
|
|
|
def main() -> int:
|
|
dry_run = "--dry-run" in sys.argv[1:]
|
|
if not wecom.wecom_configured():
|
|
print("[FAIL] WECOM_CORPID/AGENTID/SECRET 没读到(确认 .env 在仓库根、值已填)")
|
|
return 1
|
|
|
|
print("[step1] list visible userids (agent/get + expand) ...")
|
|
try:
|
|
userids = wecom.list_visible_userids()
|
|
except Exception as e:
|
|
print(f"[step1] FAIL: {e}")
|
|
print(" -> 60011/48002 = 应用 secret 无通讯录读权限;可在企微后台把可见"
|
|
"范围改成按成员勾选(agent/get 直接给 userid,不用展开部门)后重跑")
|
|
return 2
|
|
print(f"[step1] OK: {len(userids)} member(s) in visible range")
|
|
|
|
counts: dict[str, int] = {}
|
|
for u in userids:
|
|
uid = service.get_user_by_wecom_userid(u)
|
|
reason = service.maybe_send_wecom_welcome(u, uid, dry_run=dry_run)
|
|
key = reason.split(":", 1)[0]
|
|
counts[key] = counts.get(key, 0) + 1
|
|
bound = "bound" if uid else "unbound"
|
|
print(f" [{reason}] userid={u} ({bound})")
|
|
|
|
print("[done]", " | ".join(f"{k}={v}" for k, v in sorted(counts.items())))
|
|
if dry_run:
|
|
print("[note] dry-run: nothing sent. Re-run without --dry-run to send.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|