"""task_runner.py - asyncio + asyncssh rewrite for I/O parallelism. Runs deploy in a background thread that executes asyncio event loop. Uses asyncssh for non-blocking SSH + stdout streaming. Falls back to paramiko if asyncssh not available.""" import asyncio, json, time, os, re, random, string, threading, base64, queue as queue_mod from store import get_servers, save_servers, update_server, get_options, save_options from dkim_modes import resolve_dkim from cf_api import CloudflareAPI try: import asyncssh HAS_ASYNCSSH = True except ImportError: HAS_ASYNCSSH = False import paramiko SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "install_scripts") TASK_LABELS = { "deploy": "并发部署", "policy": "策略更新", "retry_failed": "重试失败邮局", "monitor": "投递监控", "dns": "DNS/Zone 检测", "reload_zone": "重新加载 Zone", "download_logs": "下载邮局发送日志", "clear_dns": "清空 DNS 解析", } class TaskRunner: def __init__(self): self.running = False self.exit_code = None self.task_name = "" self._stop_flag = False self._thread = None self.ws_clients = set() self._loop = None self.byte_offset = 0 self.offset = 0 self.download_path = "" self.download_name = "" self.summary = "" def is_running(self): return self.running def add_client(self, ws): try: ws = ws._get_current_object() except: pass self.ws_clients.add(ws) if self._loop is None: try: self._loop = asyncio.get_event_loop() except: self._loop = None def remove_client(self, ws): self.ws_clients.discard(ws) def push(self, text, running=None, exit_code=None, task_name=None, download_path=None, download_name=None, summary=None): if running is not None: self.running = running if exit_code is not None: self.exit_code = exit_code if task_name: self.task_name = task_name if download_path: self.download_path = download_path if download_name: self.download_name = download_name if summary: self.summary = summary if text: self.byte_offset += len(text.encode("utf-8")) self.offset += text.count("\n") msg = { "type": "task", "lines": [text], "running": self.running, "byte_offset": self.byte_offset, "offset": self.offset, "task_name": self.task_name, } if exit_code is not None: msg["exit_code"] = exit_code msg["done"] = True if download_path: msg["download_path"] = download_path msg["download_name"] = download_name if summary: msg["summary"] = summary data = json.dumps(msg, ensure_ascii=False) dead = set() for ws in self.ws_clients: try: if self._loop and not self._loop.is_closed(): # Quart async WS: schedule send on main event loop asyncio.run_coroutine_threadsafe(ws.send(data), self._loop) elif hasattr(ws, 'send') and not asyncio.iscoroutinefunction(ws.send): # flask-sock sync WS ws.send(data) else: dead.add(ws) except: dead.add(ws) self.ws_clients -= dead def start(self, action, options, confirmed=False): if self.running: return {"error": "任务正在运行，请先等待完成或点击停止"} self._stop_flag = False self.running = True self.exit_code = None self.task_name = TASK_LABELS.get(action, action) self.byte_offset = 0 self.offset = 0 self._thread = threading.Thread(target=self._run_wrapper, args=(action, options, confirmed), daemon=True) self._thread.start() return {"message": f"已启动：{self.task_name}"} def stop(self): self._stop_flag = True return {"message": "已请求停止", "running": self.running} def clear_output(self): self.byte_offset = 0 self.offset = 0 return {"offset": 0} def _run_wrapper(self, action, options, confirmed): """Run async task in new event loop.""" try: asyncio.run(self._async_run(action, options, confirmed)) except Exception as e: self.push(f"\n[error] {e}\n") self.push("", running=False, exit_code=-1) finally: if self.exit_code is None: self.push("", running=False, exit_code=0) async def _async_run(self, action, options, confirmed): if action == "deploy": await self._run_deploy(options) elif action == "policy": await self._run_policy(options) elif action == "monitor": await self._run_monitor(options) elif action == "dns": await self._run_dns_check(options) elif action == "reload_zone": await self._run_reload_zone(options) elif action == "download_logs": await self._run_download_logs(options) elif action == "verify": from verify_check import run_verify await run_verify(self, options) elif action == "clear_dns": await self._run_clear_dns(options) elif action == "retry_failed": await self._run_deploy(options, retry_only=True) def _read_script(self, stack): stack_map = { "postfix": "install-postfix.sh", "postfix_outbound": "install-postfix_outbound.sh", "haraka": "install-haraka.sh", "maddy": "install-maddy.sh", "opensmtpd": "install-opensmtpd.sh", "exim4": "install-exim4.sh", "exim4_outbound": "install-exim4_outbound.sh", "courier_mta": "install-courier_mta.sh", "poste_io": "install-poste_io.sh", "stalwart": "install-stalwart.sh", } fname = stack_map.get(stack, f"install-{stack}.sh") path = os.path.join(SCRIPTS_DIR, fname) with open(path, "r", encoding="utf-8") as f: return f.read() def _inject_params(self, script_body, server, options): hostname = server.get("hostname", "") helo = options.get("helo_name", "") or hostname selector, _ = resolve_dkim( options.get("dkim_sign_mode", "mkt_lu_news"), options.get("dkim_selector", ""), options.get("dkim_custom_headers", ""), ) ssh_pass = server.get("cred", server.get("password", "")) cf_tok = server.get("token", "") import re as _re def _replace_var(text, var, new_val): return _re.sub(rf'^({var})=.*$', rf'\g<1>={new_val}', text, flags=_re.MULTILINE) script_body = _replace_var(script_body, "MAIL_DOMAIN", hostname) script_body = _replace_var(script_body, "MAIL_HOSTNAME", hostname) script_body = _replace_var(script_body, "MAIL_HELO", helo) script_body = _replace_var(script_body, "MSGID_DOMAIN", hostname) script_body = _replace_var(script_body, "DKIM_SELECTOR", selector) script_body = _replace_var(script_body, "SMTP_AUTH_USER", f"admin@{hostname}") script_body = _replace_var(script_body, "SMTP_PASS", ssh_pass) script_body = _replace_var(script_body, "ADMIN_EMAIL", f"admin@{hostname}") script_body = _replace_var(script_body, "CF_TOKEN", cf_tok) script_body = _replace_var(script_body, "ACME_CA", options.get("acme_ca", "selfsigned")) script_body = _replace_var(script_body, "TLS_KEY_TYPE", options.get("tls_key_type", "rsa")) script_body = _replace_var(script_body, "RECEIVED_MODE", options.get("received_mode", "single")) script_body = _replace_var(script_body, "OUTBOUND_CLIENT_MODE", options.get("outbound_client_mode", "mobile")) script_body = _replace_var(script_body, "DKIM_SIGN_COUNT", str(options.get("dkim_sign_count", 1))) script_body = _replace_var(script_body, "DKIM_SIGN_MODE", options.get("dkim_sign_mode", "mkt_lu_news")) script_body = _replace_var(script_body, "DMARC_POLICY", options.get("dmarc_policy", "none")) script_body = _replace_var(script_body, "SPF_QUALIFIER", options.get("spf_qualifier", "-all")) script_body = _replace_var(script_body, "STRIP_HEADERS", options.get("strip_headers", "")) script_body = _replace_var(script_body, "IPV6_ENABLED", str(options.get("ipv6_enabled", False)).lower()) script_body = _re.sub(r'(const\s+EHLO_DOMAIN\s*=\s*)"[^"]*"', rf'\1"{hostname}"', script_body) placeholder_host = "mail1.weicai168.com" if hostname and hostname != placeholder_host: script_body = script_body.replace(placeholder_host, hostname) placeholder_pass = "qwffHPPQ4966" if ssh_pass and ssh_pass != placeholder_pass: script_body = script_body.replace(placeholder_pass, ssh_pass) placeholder_token = "cfut_oea3LitKRIKQl0ursISIpPUD2E24BoNopC5l4oI009ced023" if cf_tok and cf_tok != placeholder_token: script_body = script_body.replace(placeholder_token, cf_tok) return script_body async def _read_proc_output(self, proc, server): """Read all stdout lines from an asyncssh process, push to WS, return kumomta_result if found.""" kumomta_result = None try: while True: if self._stop_flag: proc.terminate() break line = await proc.stdout.readline() if not line: break line = line.rstrip("\n\r") self.push(f"[{server['hostname']}] {line}\n", running=True) if "__KUMOMTA_RESULT__" in line: self.push("\n", running=True) json_text = "" while True: if self._stop_flag: break jline = await proc.stdout.readline() if not jline: break json_text += jline if "}" in jline: break try: kumomta_result = json.loads(json_text.strip()) except: pass except: pass return kumomta_result async def _async_ssh_readline(self, proc): """Read one line from asyncssh process stdout. Uses proc.stdout with timeout.""" try: line = await asyncio.wait_for(proc.stdout.readline(), timeout=920) return line.rstrip('\n\r') if line else None except asyncio.TimeoutError: return None except: return None async def _deploy_one_async(self, server, script_body, options, stack, sem): """Deploy to one server using asyncssh (non-blocking I/O).""" async with sem: if self._stop_flag: return False delay = random.uniform(0, 10) self.push(f"[{server['hostname']}] 随机启动延迟 {delay:.1f} 秒\n\n", running=True) await asyncio.sleep(delay) self.push(f"==== {server['hostname']} / {server['ip']}（{stack}） ====\n\n", running=True) self.push(f"[{server['hostname']}] 阶段 2/3：远程安装（{stack}，超时 900 秒）\n", running=True) self.push(f"[{server['hostname']}] 正在上传 安装脚本 到\n", running=True) try: injected = self._inject_params(script_body, server, options) if HAS_ASYNCSSH: conn = await asyncio.wait_for( asyncssh.connect(server["ip"], port=int(server.get("port", 22)), username=server.get("user", "root"), password=server.get("cred", server.get("password", "")), known_hosts=None, login_timeout=30), timeout=30) remote_path = f"/tmp/install-{stack}-{int(time.time())}-{random.randint(1000,9999)}.sh" sftp = await conn.start_sftp_client() async with sftp.open(remote_path, "w") as f: await f.write(injected) sftp.exit() proc = await conn.create_process(f"bash {remote_path}") kumomta_result = None # Read stdout concurrently while waiting for process to finish stdout_task = asyncio.create_task(self._read_proc_output(proc, server)) try: exit_status = await asyncio.wait_for(proc.wait(), timeout=900) except asyncio.TimeoutError: proc.terminate() # Get any remaining output kumomta_result = await stdout_task if stdout_task.done() else None if stdout_task and not stdout_task.done(): stdout_task.cancel() conn.close() else: # Fallback: paramiko (sync, runs in executor) loop = asyncio.get_event_loop() conn = await loop.run_in_executor(None, lambda: self._ssh_connect_sync(server)) remote_path = f"/tmp/install-{stack}-{int(time.time())}-{random.randint(1000,9999)}.sh" sftp = conn.open_sftp() with sftp.open(remote_path, "w") as f: f.write(injected) sftp.close() stdin, stdout, stderr = conn.exec_command(f"bash {remote_path}", timeout=900) kumomta_result = None while True: if self._stop_flag: stdin.channel.close() break line = await loop.run_in_executor(None, stdout.readline) if not line: break line = line.rstrip("\n\r") self.push(f"[{server['hostname']}] {line}\n", running=True) if "__KUMOMTA_RESULT__" in line: self.push("\n", running=True) json_text = "" while True: if self._stop_flag: break jline = await loop.run_in_executor(None, stdout.readline) if not jline: break json_text += jline if "}" in jline: break try: kumomta_result = json.loads(json_text.strip()) except: pass exit_status = stdout.channel.recv_exit_status() conn.close() if kumomta_result: if not kumomta_result.get("public_ip"): kumomta_result["public_ip"] = server.get("ip", "") self.push(f"\n成功：{server['hostname']}\n", running=True) update_server(server["ip"], status="成功", dkim_public=kumomta_result.get("dkim_public", ""), dkim_selector=kumomta_result.get("selector", "")) return True else: self.push(f"\n失败：{server['hostname']} / {server['ip']}\n", running=True) self.push(f"原因分类：未知失败\n\n", running=True) update_server(server["ip"], status="失败") return False except Exception as e: self.push(f"\n失败：{server['hostname']} / {server['ip']}\n原因：{e}\n\n", running=True) update_server(server["ip"], status="失败") return False def _ssh_connect_sync(self, server): ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(server["ip"], port=int(server.get("port", 22)), username=server.get("user", "root"), password=server.get("cred", server.get("password", "")), timeout=30, banner_timeout=30) return ssh async def _run_deploy(self, options, retry_only=False): servers = get_servers() if retry_only: servers = [s for s in servers if s.get("status") == "失败"] if not servers: self.push("没有服务器可部署\n", running=False, exit_code=0) return self.push(f"\n$ {TASK_LABELS['deploy']}\n\n正在连接部署引擎…\n\n正在加载部署引擎…\n部署脚本已加载，正在初始化…\n\n", running=True, task_name=TASK_LABELS["deploy"]) self.push(f"strip headers: {options.get('strip_headers','')}\n\n", running=True) if options.get("dkim_sign_count", 1) > 1: self.push(f"DKIM 多签已启用：每封邮件将带 {options['dkim_sign_count']} 个 DKIM-Signature（同域多 selector）\n\n", running=True) self.push(f"已读取 {len(servers)} 台服务器配置\n\n", running=True) stack = options.get("mail_stack", "haraka") workers = max(1, min(200, options.get("workers", 40))) self.push(f"开始并发部署：{len(servers)} 台服务器，线程={workers}，随机延迟=10 秒，证书CA={options.get('acme_ca','selfsigned')}，TLS密钥类型={options.get('tls_key_type','rsa')}，默认邮件栈={stack}\n\n", running=True) self.push(f"将执行全新远程安装：Haraka 邮局（Haraka 首次 npm 约 5～15 分钟/台）\n\n", running=True) # Phase 1: CF Zone self.push("阶段 1/3：预加载 Cloudflare Zone（仅查询账户域名，尚未修改 DNS）\n\n", running=True) zones_cache = {} tokens = list(set(s.get("token", "") for s in servers if s.get("token"))) for token in tokens[:4]: try: cf = CloudflareAPI(token) zones = cf.get_zones() zones_cache[token] = zones self.push(f" Token {token[:8]}...{token[-4:]}：已缓存 {len(zones)} 个 Zone\n", running=True) except Exception as e: self.push(f" Token 加载失败：{e}\n", running=True) self.push(f"\nZone 获取完成：{len(tokens)} 个 Token，共 {sum(len(z) for z in zones_cache.values())} 个 Zone\n\n", running=True) # Phase 2: SSH install (asyncio concurrent) self.push("阶段 2/3：SSH 远程安装邮件栈（完成后才会进入 DNS 写入）\n\n", running=True) script_body = self._read_script(stack) sem = asyncio.Semaphore(workers) tasks = [self._deploy_one_async(server, script_body, options, stack, sem) for server in servers] results = await asyncio.gather(*tasks, return_exceptions=True) ok_count = sum(1 for r in results if r is True) fail_count = sum(1 for r in results if r is False or isinstance(r, Exception)) # Phase 3: DNS write dns_fail_count = 0 self.push(f"\n阶段 3/3：安装完成，开始写入 Cloudflare DNS\n\n", running=True) servers = get_servers() for server in servers: if server.get("status") != "成功": continue try: token = server.get("token", "") if not token: self.push(f" {server['hostname']}: 无 CF Token，跳过 DNS\n", running=True) continue cf = CloudflareAPI(token) zone = cf.find_zone_for_domain(server.get("domain", ""), zones_cache.get(token, [])) if not zone: self.push(f" Zone 查询失败：{server['hostname']} -> cloudflare zone not found\n", running=True) continue zone_id = zone["id"] dkim_sels = [] dkim_pub = server.get("dkim_public", "") dkim_sel = server.get("dkim_selector", "default") if dkim_pub: dkim_sels.append({"selector": dkim_sel, "public_key": dkim_pub}) for extra_sel in ["default2", "default3"]: pub_key = server.get(f"dkim_public_{extra_sel}", "") if pub_key: dkim_sels.append({"selector": extra_sel, "public_key": pub_key}) public_ip = server.get("dkim_public_ip", "") or server.get("ip", "") ipv6_addr = server.get("ipv6", "") or options.get("ipv6_address", "") records = cf.build_dns_records( server["hostname"], public_ip, dkim_sels, options.get("dmarc_policy", "none"), options.get("spf_qualifier", "-all"), server.get("domain", ""), ) results_dns = cf.write_dns_records(zone_id, records) ok_recs = sum(1 for r in results_dns if r["success"]) if ok_recs == 0: dns_fail_count += 1 self.push(f" {server['hostname']}: DNS 写入 {ok_recs}/{len(records)} 条成功\n", running=True) except Exception as e: self.push(f" DNS 写入失败：{server['hostname']} -> {e}\n", running=True) total_fail = fail_count + dns_fail_count self.push(f"\n部署完成统计：成功 {ok_count}，失败 {fail_count}\n\n", running=True) if dns_fail_count > 0: self.push(f" ⚠️ DNS 写入失败：{dns_fail_count} 台，请更新 CF Token 后点「重试失败」\n", running=True) summary = f"搭建完成：成功 {ok_count} 台" if fail_count == 0 and ok_count > 0 else f"搭建结束：成功 {ok_count} 台，失败 {fail_count} 台" self.push(f"\n【任务结束】并发部署 exit={0 if fail_count==0 else 1}\n", running=False, exit_code=0 if fail_count==0 else 1, summary=summary) async def _run_policy(self, options): servers = get_servers() if not servers: self.push("没有服务器\n", running=False, exit_code=0) return self.push(f"\n$ 策略更新\n\n正在连接部署引擎…\n\n", running=True, task_name="策略更新") stack = options.get("mail_stack", "haraka") script_body = self._read_script(stack) workers = max(1, min(200, options.get("workers", 40))) self.push(f"开始仅策略更新：{len(servers)} 台服务器，线程={workers}\n\n", running=True) sem = asyncio.Semaphore(workers) async def policy_one(server): async with sem: if self._stop_flag: return False self.push(f"\n==== 策略更新 {server['hostname']} / {server['ip']} ====\n\n", running=True) try: injected = self._inject_params(script_body, server, options) if HAS_ASYNCSSH: conn = await asyncio.wait_for( asyncssh.connect(server["ip"], port=int(server.get("port", 22)), username=server.get("user", "root"), password=server.get("cred", server.get("password", "")), known_hosts=None, login_timeout=30), timeout=30) remote_path = f"/tmp/policy-{stack}-{int(time.time())}-{random.randint(1000,9999)}.sh" sftp = await conn.start_sftp_client() async with sftp.open(remote_path, "w") as f: await f.write(injected) sftp.exit() proc = await conn.create_process(f"bash {remote_path}") stdout_task = asyncio.create_task(self._read_proc_output(proc, server)) try: await asyncio.wait_for(proc.wait(), timeout=600) except asyncio.TimeoutError: proc.terminate() if not stdout_task.done(): stdout_task.cancel() conn.close() else: loop = asyncio.get_event_loop() conn = await loop.run_in_executor(None, lambda: self._ssh_connect_sync(server)) remote_path = f"/tmp/policy-{stack}-{int(time.time())}-{random.randint(1000,9999)}.sh" sftp = conn.open_sftp() with sftp.open(remote_path, "w") as f: f.write(injected) sftp.close() stdin, stdout, stderr = conn.exec_command(f"bash {remote_path}", timeout=600) while True: if self._stop_flag: stdin.channel.close() break line = await loop.run_in_executor(None, stdout.readline) if not line: break self.push(f"[{server['hostname']}] {line.rstrip()}\n", running=True) exit_status = stdout.channel.recv_exit_status() conn.close() self.push(f"\n成功：{server['hostname']}\n", running=True) return True except Exception as e: self.push(f"\n失败：{server['hostname']} -> {e}\n", running=True) return False tasks = [policy_one(server) for server in servers] results = await asyncio.gather(*tasks, return_exceptions=True) ok_count = sum(1 for r in results if r is True) fail_count = sum(1 for r in results if r is False or isinstance(r, Exception)) self.push(f"\n策略更新完成：成功 {ok_count}，失败 {fail_count}\n", running=False, exit_code=0 if fail_count==0 else 1, summary=f"策略更新完成：成功 {ok_count} 台，失败 {fail_count} 台") async def _run_monitor(self, options): servers = get_servers() workers = max(1, min(200, options.get("workers", 40))) self.push(f"\n$ 投递监控\n\n已读取 {len(servers)} 台服务器配置\n\n投递监控：{len(servers)} 台服务器，线程={workers}\n\n", running=True, task_name="投递监控") stats = {"total": 0, "delivered": 0, "hard_bounce": 0, "deferred": 0, "rejected": 0, "soft_bounce": 0, "pending": 0} sem = asyncio.Semaphore(workers) async def monitor_one(server): async with sem: self.push(f"投递拉取：{server['hostname']}（{server.get('stack','')}）\n", running=True) try: loop = asyncio.get_event_loop() def _read_logs(): conn = self._ssh_connect_sync(server) i,so,se = conn.exec_command("cat /var/log/haraka/delivery.log 2>/dev/null | tail -5000; echo '---'; journalctl -u haraka --no-pager -n 5000 2>/dev/null | grep -E 'delivered|deferred|bounce' | tail -5000", timeout=60) log_text = so.read().decode("utf-8", "replace") conn.close() return log_text log_text = await loop.run_in_executor(None, _read_logs) self.push(f" 日志读取：{server['hostname']}：{len(log_text)} bytes\n", running=True) local = {"delivered":0, "hard_bounce":0, "deferred":0, "total":0} for line in log_text.split("\n"): if "haraka_delivery delivered" in line: local["delivered"] += 1; local["total"] += 1 elif "haraka_delivery bounce" in line: local["hard_bounce"] += 1; local["total"] += 1 elif "haraka_delivery deferred" in line: local["deferred"] += 1; local["total"] += 1 for k in ("delivered", "hard_bounce", "deferred", "total"): stats[k] += local[k] except Exception as e: self.push(f" 日志读取失败：{server['hostname']} -> {e}\n", running=True) tasks = [monitor_one(server) for server in servers] await asyncio.gather(*tasks, return_exceptions=True) stats["success_rate"] = round(stats["delivered"] / stats["total"] * 100, 1) if stats["total"] > 0 else 0.0 self.push(f"\n投递监控统计\n\n", running=True) self.push(f" 总发送： {stats['total']}\n", running=True) self.push(f" 送达： {stats['delivered']}\n", running=True) self.push(f" 成功率： {stats['success_rate']}%\n", running=True) self.push(f" 硬退信： {stats['hard_bounce']}\n", running=True) self.push(f" 重试中： {stats['deferred']}\n", running=True) self.push(f" 拒收： {stats['rejected']}\n", running=True) self.push(f" 软退信： {stats['soft_bounce']}\n", running=True) self.push(f"\n[任务结束] 投递监控 exit=0\n", running=False, exit_code=0, summary=f"投递监控完成：成功率 {stats['success_rate']}%") async def _run_dns_check(self, options): servers = get_servers() self.push(f"\n$ DNS/Zone 检测\n\n已读取 {len(servers)} 台服务器配置\n\n", running=True, task_name="DNS/Zone 检测") for server in servers: token = server.get("token", "") if not token: self.push(f" {server['hostname']}: 无 CF Token，跳过\n", running=True) continue try: cf = CloudflareAPI(token) zones = cf.get_zones() zone = cf.find_zone_for_domain(server.get("domain", ""), zones) if zone: self.push(f" {server['hostname']}: Zone={zone['name']}\n", running=True) else: self.push(f" Zone 查询失败：{server['hostname']} -> cloudflare zone not found\n", running=True) except Exception as e: self.push(f" DNS 检测失败：{server['hostname']} -> {e}\n", running=True) self.push(f"\n[任务结束] DNS/Zone 检测 exit=0\n", running=False, exit_code=0) async def _run_reload_zone(self, options): servers = get_servers() self.push(f"\n$ 重新加载 Zone\n\n", running=True, task_name="重新加载 Zone") tokens = list(set(s.get("token","") for s in servers if s.get("token"))) total = 0 for token in tokens: try: cf = CloudflareAPI(token) zones = cf.get_zones() total += len(zones) self.push(f" Token {token[:8]}...{token[-4:]}：已缓存 {len(zones)} 个 Zone\n", running=True) except Exception as e: self.push(f" Token 加载失败：{e}\n", running=True) self.push(f"\nZone 获取完成：{len(tokens)} 个 Token，共 {total} 个 Zone\n\n", running=False, exit_code=0) async def _run_download_logs(self, options): servers = get_servers() self.push(f"\n$ 下载邮局发送日志\n\n", running=True, task_name="下载邮局发送日志") sem = asyncio.Semaphore(max(1, min(200, options.get("workers", 40)))) async def dl_one(server): async with sem: try: loop = asyncio.get_event_loop() def _read(): conn = self._ssh_connect_sync(server) i,so,se = conn.exec_command("cat /var/log/haraka/delivery.log 2>/dev/null | tail -10000", timeout=60) log = so.read().decode("utf-8","replace") conn.close() return log log = await loop.run_in_executor(None, _read) self.push(f" {server['hostname']}: ({len(log)} bytes)\n", running=True) except Exception as e: self.push(f" {server['hostname']}: 失败 -> {e}\n", running=True) tasks = [dl_one(server) for server in servers] await asyncio.gather(*tasks, return_exceptions=True) self.push(f"\n[任务结束] 下载邮局发送日志 exit=0\n", running=False, exit_code=0) async def _run_clear_dns(self, options): servers = get_servers() self.push(f"\n$ 清空 DNS 解析\n\n", running=True, task_name="清空 DNS 解析") for server in servers: token = server.get("token", "") if not token: continue try: cf = CloudflareAPI(token) zone = cf.find_zone_for_domain(server.get("domain","")) if zone: deleted = cf.clear_dns_records(zone["id"], server.get("hostname","")) self.push(f" {server['hostname']}: 已删除 {deleted} 条记录\n", running=True) except Exception as e: self.push(f" {server['hostname']}: 失败 -> {e}\n", running=True) self.push(f"\n[任务结束] 清空 DNS 解析 exit=0\n", running=False, exit_code=0) runner = TaskRunner()