hz4th_coder
69edd1c63e
🎯 重构目标： - 将 3700+ 行单文件拆分为多个模块 - 清晰的职责分离，便于维护和扩展 📁 模块结构： - config.py: 配置常量和路径 - database.py: 数据库操作 - utils.py: 工具函数（日志、进程、cron等） - routes_cron.py: Cron管理API - routes_system.py: 系统监控API - routes_config.py: 配置管理API - routes_email.py: 邮件通知API - routes_project.py: 项目管理API - templates.py: HTML模板 - app.py: 主入口文件 ✨ 优势： - 易于维护：每个模块职责明确 - 易于扩展：新增功能只需添加对应模块 - 易于测试：模块独立，便于单元测试 - 易于理解：新人能快速上手 📝 更新 README.md，详细说明项目结构和API接口
303 lines
9.1 KiB
Python
303 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
工具函数模块
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import subprocess
|
||
import re
|
||
import signal
|
||
import shutil
|
||
from datetime import datetime
|
||
from config import LOG_FILE, PROJECTS_FILE, CRON_BACKUP_DIR, ALERT_CONFIG_FILE
|
||
|
||
|
||
# 进程状态缓存
|
||
process_cache = {}
|
||
|
||
|
||
def log_message(msg):
|
||
"""写入日志"""
|
||
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
log_line = f"[{timestamp}] {msg}\n"
|
||
print(log_line.strip())
|
||
try:
|
||
os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
|
||
with open(LOG_FILE, 'a') as f:
|
||
f.write(log_line)
|
||
except:
|
||
pass
|
||
|
||
|
||
def load_projects():
|
||
"""加载项目配置"""
|
||
try:
|
||
with open(PROJECTS_FILE, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
return data.get('projects', [])
|
||
except:
|
||
return []
|
||
|
||
|
||
def load_projects_full():
|
||
"""加载完整项目配置（包含server和external_services）"""
|
||
try:
|
||
with open(PROJECTS_FILE, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except:
|
||
return {'server': {}, 'projects': [], 'external_services': []}
|
||
|
||
|
||
def save_projects(projects):
|
||
"""保存项目配置"""
|
||
with open(PROJECTS_FILE, 'w', encoding='utf-8') as f:
|
||
json.dump({'projects': projects}, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def check_port(port):
|
||
"""检查端口是否有进程监听"""
|
||
try:
|
||
import socket
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(1)
|
||
result = sock.connect_ex(('localhost', port))
|
||
sock.close()
|
||
return result == 0
|
||
except:
|
||
return False
|
||
|
||
|
||
def stop_process(port):
|
||
"""停止占用指定端口的进程"""
|
||
try:
|
||
# 使用 ss 找到占用端口的进程
|
||
result = subprocess.run(f"ss -tlnp | grep :{port}", shell=True, capture_output=True, text=True)
|
||
if result.stdout:
|
||
# 解析 pid
|
||
import re
|
||
match = re.search(r'pid=(\d+)', result.stdout)
|
||
if match:
|
||
pid = int(match.group(1))
|
||
os.kill(pid, signal.SIGTERM)
|
||
log_message(f"已停止进程 {pid} (端口 {port})")
|
||
return 1
|
||
except Exception as e:
|
||
log_message(f"停止进程失败: {e}")
|
||
return 0
|
||
|
||
|
||
def run_command(cmd, cwd, project_id, action):
|
||
"""运行命令"""
|
||
log_message(f"[{project_id}] 执行 {action}: {cmd}")
|
||
try:
|
||
subprocess.Popen(cmd, shell=True, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
return True
|
||
except Exception as e:
|
||
log_message(f"命令执行失败: {e}")
|
||
return False
|
||
|
||
|
||
def backup_crontab():
|
||
"""备份当前 crontab"""
|
||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
backup_file = os.path.join(CRON_BACKUP_DIR, f'crontab_{timestamp}.txt')
|
||
|
||
try:
|
||
os.makedirs(CRON_BACKUP_DIR, exist_ok=True)
|
||
result = subprocess.run("crontab -l 2>/dev/null", shell=True, capture_output=True, text=True)
|
||
with open(backup_file, 'w') as f:
|
||
f.write(result.stdout)
|
||
log_message(f"crontab 已备份到 {backup_file}")
|
||
except Exception as e:
|
||
log_message(f"备份 crontab 失败: {e}")
|
||
|
||
|
||
def get_system_crons():
|
||
"""获取系统所有 cron 任务"""
|
||
crons = []
|
||
|
||
try:
|
||
result = subprocess.run("crontab -l 2>/dev/null", shell=True, capture_output=True, text=True, timeout=5)
|
||
for line in result.stdout.strip().split('\n'):
|
||
line = line.strip()
|
||
if line and not line.startswith('#'):
|
||
parts = line.split()
|
||
if len(parts) >= 6:
|
||
crons.append({'source': 'user', 'expression': ' '.join(parts[:5]), 'command': ' '.join(parts[5:]), 'line': line})
|
||
except:
|
||
pass
|
||
|
||
try:
|
||
if os.path.exists('/etc/crontab'):
|
||
with open('/etc/crontab', 'r') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line and not line.startswith('#'):
|
||
parts = line.split()
|
||
if len(parts) >= 7:
|
||
crons.append({'source': 'system', 'expression': ' '.join(parts[:5]), 'user': parts[5], 'command': ' '.join(parts[6:]), 'line': line})
|
||
except:
|
||
pass
|
||
|
||
try:
|
||
if os.path.exists('/etc/cron.d'):
|
||
for filename in os.listdir('/etc/cron.d'):
|
||
filepath = os.path.join('/etc/cron.d', filename)
|
||
if os.path.isfile(filepath):
|
||
with open(filepath, 'r') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line and not line.startswith('#'):
|
||
parts = line.split()
|
||
if len(parts) >= 7:
|
||
crons.append({'source': 'cron.d', 'file': filename, 'expression': ' '.join(parts[:5]), 'user': parts[5], 'command': ' '.join(parts[6:]), 'line': line})
|
||
except:
|
||
pass
|
||
|
||
return crons
|
||
|
||
|
||
def parse_cron_expression(expr):
|
||
"""解析 cron 表达式为人类可读格式"""
|
||
parts = expr.split()
|
||
if len(parts) != 5:
|
||
return expr
|
||
|
||
minute, hour, day, month, weekday = parts
|
||
desc = []
|
||
|
||
if minute == '*':
|
||
desc.append('每分钟')
|
||
elif minute.startswith('*/'):
|
||
desc.append(f'每{minute[2:]}分钟')
|
||
else:
|
||
desc.append(f'{minute}分')
|
||
|
||
if hour != '*':
|
||
if hour.startswith('*/'):
|
||
desc.append(f'每{hour[2:]}小时')
|
||
else:
|
||
desc.append(f'{hour}时')
|
||
|
||
if day != '*' and not day.startswith('*'):
|
||
desc.append(f'{day}日')
|
||
|
||
if month != '*' and not month.startswith('*'):
|
||
desc.append(f'{month}月')
|
||
|
||
if weekday != '*' and not weekday.startswith('*'):
|
||
weekdays = ['周日', '周一', '周二', '周三', '周四', '周五', '周六']
|
||
try:
|
||
desc.append(weekdays[int(weekday)])
|
||
except:
|
||
pass
|
||
|
||
return ' '.join(desc) if desc else expr
|
||
|
||
|
||
def guess_cron_name(command):
|
||
"""从命令推断任务名称"""
|
||
if 'openclaw' in command or 'agent' in command:
|
||
return 'OpenClaw Agent'
|
||
elif 'backup' in command:
|
||
return '备份任务'
|
||
elif 'sync' in command:
|
||
return '同步任务'
|
||
elif 'clean' in command or 'cleanup' in command:
|
||
return '清理任务'
|
||
elif 'report' in command:
|
||
return '报告任务'
|
||
elif 'monitor' in command or 'check' in command:
|
||
return '监控任务'
|
||
elif 'git' in command:
|
||
return 'Git 任务'
|
||
return '自定义任务'
|
||
|
||
|
||
def guess_cron_category(command):
|
||
"""从命令推断分类"""
|
||
if 'openclaw' in command:
|
||
return 'agent'
|
||
elif 'backup' in command:
|
||
return 'backup'
|
||
elif 'sync' in command:
|
||
return 'sync'
|
||
elif 'clean' in command or 'cleanup' in command:
|
||
return 'cleanup'
|
||
elif 'report' in command:
|
||
return 'report'
|
||
elif 'git' in command:
|
||
return 'git'
|
||
return 'other'
|
||
|
||
|
||
def sync_crontab_from_db(db):
|
||
"""从数据库同步到系统 crontab"""
|
||
c = db.cursor()
|
||
c.execute("SELECT expression, command, enabled FROM cron_tasks WHERE enabled = 1")
|
||
tasks = c.fetchall()
|
||
|
||
lines = []
|
||
for task in tasks:
|
||
lines.append(f"{task['expression']} {task['command']}")
|
||
|
||
# 写入临时文件
|
||
tmp_file = '/tmp/crontab_sync.tmp'
|
||
with open(tmp_file, 'w') as f:
|
||
f.write('\n'.join(lines) + '\n')
|
||
|
||
# 安装 crontab
|
||
subprocess.run(f"crontab {tmp_file}", shell=True)
|
||
os.remove(tmp_file)
|
||
|
||
# 备份
|
||
backup_crontab()
|
||
|
||
|
||
def get_next_run_time(expression):
|
||
"""获取下次执行时间"""
|
||
try:
|
||
from croniter import croniter
|
||
base = datetime.now()
|
||
cron = croniter(expression, base)
|
||
next_time = cron.get_next(datetime)
|
||
return next_time.strftime('%Y-%m-%d %H:%M:%S')
|
||
except:
|
||
return '无法计算'
|
||
|
||
|
||
def get_cpu_temperatures():
|
||
"""获取 CPU Package 温度"""
|
||
temps = {'package_0': None, 'package_1': None}
|
||
try:
|
||
result = subprocess.run(['sensors'], capture_output=True, text=True, timeout=5)
|
||
output = result.stdout
|
||
# 解析 Package id 0 和 Package id 1
|
||
for line in output.split('\n'):
|
||
if 'Package id 0:' in line:
|
||
match = re.search(r'Package id 0:\s*([+\d.]+)°C', line)
|
||
if match:
|
||
temps['package_0'] = float(match.group(1).replace('+', ''))
|
||
elif 'Package id 1:' in line:
|
||
match = re.search(r'Package id 1:\s*([+\d.]+)°C', line)
|
||
if match:
|
||
temps['package_1'] = float(match.group(1).replace('+', ''))
|
||
except Exception as e:
|
||
log_message(f"获取CPU温度失败: {e}")
|
||
return temps
|
||
|
||
|
||
def load_alert_config():
|
||
"""加载预警配置"""
|
||
try:
|
||
with open(ALERT_CONFIG_FILE, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except:
|
||
return {}
|
||
|
||
|
||
def save_alert_config(config):
|
||
"""保存预警配置"""
|
||
with open(ALERT_CONFIG_FILE, 'w', encoding='utf-8') as f:
|
||
json.dump(config, f, ensure_ascii=False, indent=2) |