Evilom
64ac71a8ed
- Collected 9 new messages from dc group and innovation group - Fetched 6 new Feishu docs (黄静雯/陈楚真/韦译/夏莲/莫润麟/张家振) - Downloaded 狱国争霸分析拆解.html file attachment - Saved 6/12 game industry daily report from 韩丹 - Updated .gitignore to exclude temp scripts, images, downloaded files - Added utility scripts (group members extraction, invitation generation)
210 lines
6.9 KiB
Python
210 lines
6.9 KiB
Python
"""根据群成员从账号数据库匹配账号密码，逐个私聊发送"""
|
|
|
|
import csv
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__))))
|
|
from config import DWS
|
|
|
|
DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data")
|
|
ACCT_DB = os.path.join(DATA_DIR, "group_members_account.xlsx")
|
|
LOG_FILE = os.path.join(DATA_DIR, "send_account_log.json")
|
|
DEFAULT_GROUP_NAME = "账号发放测试"
|
|
|
|
|
|
def run_dws(args, timeout=30):
|
|
cmd = [DWS] + args + ["-f", "json"]
|
|
result = subprocess.run(cmd, capture_output=True, timeout=timeout)
|
|
out = result.stdout.decode("utf-8", errors="replace")
|
|
for i, ch in enumerate(out):
|
|
if ch == "{":
|
|
return json.loads(out[i:])
|
|
return None
|
|
|
|
|
|
def load_account_db():
|
|
import openpyxl
|
|
wb = openpyxl.load_workbook(ACCT_DB, read_only=True)
|
|
ws = wb["账号信息"]
|
|
acct_map = {}
|
|
for r in list(ws.iter_rows(values_only=True))[1:]:
|
|
t, acct, pwd = r[0], r[1], r[2]
|
|
if t is not None and acct is not None:
|
|
try:
|
|
acct_map[str(int(acct))] = str(int(pwd)) if pwd else ""
|
|
except:
|
|
acct_map[str(acct)] = str(pwd)
|
|
wb.close()
|
|
return acct_map
|
|
|
|
|
|
def fetch_all_members(group_id):
|
|
all_members = []
|
|
cursor = "0"
|
|
while True:
|
|
args = ["chat", "group", "members", "list", "--id", group_id, "--cursor", cursor]
|
|
data = run_dws(args, timeout=60)
|
|
if not data or not data.get("success"):
|
|
break
|
|
result = data.get("result", {})
|
|
all_members.extend(result.get("list", []))
|
|
if not result.get("hasMore"):
|
|
break
|
|
cursor = str(result.get("nextCursor", ""))
|
|
if not cursor:
|
|
break
|
|
time.sleep(0.3)
|
|
return all_members
|
|
|
|
|
|
def build_message(name, account, password):
|
|
return f"""## 隐秘之潮 - 试玩活动
|
|
|
|
{name}，你好！
|
|
|
|
项目试玩活动即将开启，以下是你的专属测试账号：
|
|
|
|
| 项目 | 信息 |
|
|
|------|------|
|
|
| **账号** | {account} |
|
|
| **密码** | {password} |
|
|
|
|
请妥善保管，勿转发给他人。
|
|
|
|
有问题欢迎在群里反馈。祝你玩得开心！"""
|
|
|
|
|
|
def send_message(open_dingtalk_id, title, text, dry_run=False):
|
|
if dry_run:
|
|
print(f" [DRY RUN] -> {open_dingtalk_id}")
|
|
return {"success": True, "dry_run": True}
|
|
args = [
|
|
"chat", "message", "send",
|
|
"--open-dingtalk-id", open_dingtalk_id,
|
|
"--title", title,
|
|
"--text", text,
|
|
]
|
|
return run_dws(args, timeout=15)
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description="群成员账号发放工具")
|
|
parser.add_argument("--group", type=str, help="群 openConversationId")
|
|
parser.add_argument("--group-name", type=str, default=DEFAULT_GROUP_NAME, help="群名 (默认: 账号发放测试)")
|
|
parser.add_argument("--dry-run", action="store_true", help="只预览不发送")
|
|
parser.add_argument("--limit", type=int, default=0, help="限制发送人数")
|
|
args = parser.parse_args()
|
|
|
|
group_id = args.group
|
|
if not group_id and args.group_name:
|
|
print(f"搜索群: {args.group_name}")
|
|
data = run_dws(["chat", "search", "--query", args.group_name], timeout=15)
|
|
if data and data.get("success"):
|
|
groups = data["result"].get("groups", [])
|
|
if groups:
|
|
group_id = groups[0]["openConversationId"]
|
|
print(f" 找到: {groups[0]['title']} ({group_id})")
|
|
else:
|
|
print(" 未找到群")
|
|
return
|
|
if not group_id:
|
|
print("请指定 --group <id> 或 --group-name <名称>")
|
|
return
|
|
|
|
print("加载账号数据库...")
|
|
acct_db = load_account_db()
|
|
print(f" {len(acct_db)} 个账号")
|
|
|
|
print("拉取群成员...")
|
|
members = fetch_all_members(group_id)
|
|
print(f" {len(members)} 人")
|
|
|
|
print("解析成员工号...")
|
|
member_accounts = []
|
|
no_match = []
|
|
for idx, m in enumerate(members, 1):
|
|
name = m.get("memberEmpName", "")
|
|
oid = m.get("openDingtalkId", "")
|
|
nick = m.get("memberNick", "")
|
|
if not name or not oid:
|
|
no_match.append({"name": name, "reason": "无姓名或ID"})
|
|
continue
|
|
|
|
args_search = ["contact", "user", "search", "--query", name]
|
|
data = run_dws(args_search, timeout=15)
|
|
user_id = ""
|
|
if data and data.get("success"):
|
|
for r in data.get("result", []):
|
|
if r.get("openDingTalkId") == oid:
|
|
user_id = r.get("userId", "")
|
|
break
|
|
|
|
job_number = ""
|
|
if user_id:
|
|
args_get = ["contact", "user", "get", "--ids", user_id]
|
|
data2 = run_dws(args_get, timeout=15)
|
|
if data2 and data2.get("success"):
|
|
for u in data2.get("result", []):
|
|
model = u.get("orgEmployeeModel", {})
|
|
job_number = str(model.get("jobNumber", ""))
|
|
break
|
|
|
|
if job_number and job_number in acct_db:
|
|
pwd = acct_db[job_number]
|
|
member_accounts.append({
|
|
"name": name, "nick": nick,
|
|
"openDingtalkId": oid, "userId": user_id,
|
|
"account": job_number, "password": pwd,
|
|
})
|
|
else:
|
|
no_match.append({"name": name, "job_number": job_number, "reason": "工号无匹配"})
|
|
|
|
if idx % 50 == 0:
|
|
print(f" {len(member_accounts)}/{idx} 已匹配")
|
|
time.sleep(0.15)
|
|
|
|
print(f"\n匹配结果: {len(member_accounts)} 人有账号, {len(no_match)} 人无匹配")
|
|
|
|
if args.limit > 0:
|
|
member_accounts = member_accounts[:args.limit]
|
|
print(f"限制发送: {args.limit} 人")
|
|
|
|
title = "隐秘之潮 - 试玩活动"
|
|
success_count = 0
|
|
fail_count = 0
|
|
log = []
|
|
|
|
print(f"\n{'[DRY RUN] ' if args.dry_run else ''}开始发送...")
|
|
for idx, ma in enumerate(member_accounts, 1):
|
|
text = build_message(ma["name"], ma["account"], ma["password"])
|
|
result = send_message(ma["openDingtalkId"], title, text, dry_run=args.dry_run)
|
|
status = "ok" if result and result.get("success") else "fail"
|
|
if status == "ok":
|
|
success_count += 1
|
|
else:
|
|
fail_count += 1
|
|
log.append({"name": ma["name"], "account": ma["account"], "status": status, "response": result})
|
|
print(f" [{idx}/{len(member_accounts)}] {ma['name']} ({ma['account']}) -> {status}")
|
|
if not args.dry_run:
|
|
time.sleep(0.5)
|
|
|
|
with open(LOG_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(log, f, ensure_ascii=False, indent=2)
|
|
|
|
print(f"\n完成! 成功: {success_count}, 失败: {fail_count}")
|
|
print(f"日志: {LOG_FILE}")
|
|
|
|
if no_match:
|
|
print(f"\n未匹配人员 ({len(no_match)}):")
|
|
for nm in no_match[:10]:
|
|
print(f" {nm['name']}: {nm.get('reason','')}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|