341 lines
11 KiB
Python
341 lines
11 KiB
Python
from flask import Blueprint, request, jsonify, current_app
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
import os
|
|
import time
|
|
import psutil
|
|
import sys
|
|
import platform
|
|
from sqlalchemy import func
|
|
from ..models import get_session, Domain, Mailbox, Email
|
|
from ..services import get_smtp_server, get_email_processor
|
|
|
|
api_bp = Blueprint('api', __name__, url_prefix='/api')
|
|
|
|
|
|
@api_bp.route('/domains', methods=['GET'])
|
|
def get_domains():
|
|
"""获取所有可用域名"""
|
|
db = get_session()
|
|
try:
|
|
domains = db.query(Domain).filter_by(active=True).all()
|
|
return jsonify({
|
|
'success': True,
|
|
'domains': [domain.to_dict() for domain in domains]
|
|
})
|
|
except Exception as e:
|
|
current_app.logger.exception(f"获取域名失败: {str(e)}")
|
|
return jsonify({'success': False, 'error': '获取域名失败'}), 500
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@api_bp.route('/domains', methods=['POST'])
|
|
def create_domain():
|
|
"""创建新域名"""
|
|
data = request.json
|
|
if not data or 'name' not in data:
|
|
return jsonify({'success': False, 'error': '缺少必要字段'}), 400
|
|
|
|
db = get_session()
|
|
try:
|
|
# 检查域名是否已存在
|
|
domain_exists = db.query(Domain).filter_by(name=data['name']).first()
|
|
if domain_exists:
|
|
return jsonify({'success': False, 'error': '域名已存在'}), 400
|
|
|
|
# 创建新域名
|
|
domain = Domain(
|
|
name=data['name'],
|
|
description=data.get('description', ''),
|
|
active=data.get('active', True)
|
|
)
|
|
db.add(domain)
|
|
db.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': '域名创建成功',
|
|
'domain': domain.to_dict()
|
|
})
|
|
except Exception as e:
|
|
db.rollback()
|
|
current_app.logger.exception(f"创建域名失败: {str(e)}")
|
|
return jsonify({'success': False, 'error': '创建域名失败'}), 500
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@api_bp.route('/mailboxes', methods=['GET'])
|
|
def get_mailboxes():
|
|
"""获取所有邮箱"""
|
|
db = get_session()
|
|
try:
|
|
mailboxes = db.query(Mailbox).all()
|
|
return jsonify({
|
|
'success': True,
|
|
'mailboxes': [mailbox.to_dict() for mailbox in mailboxes]
|
|
})
|
|
except Exception as e:
|
|
current_app.logger.exception(f"获取邮箱失败: {str(e)}")
|
|
return jsonify({'success': False, 'error': '获取邮箱失败'}), 500
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@api_bp.route('/mailboxes', methods=['POST'])
|
|
def create_mailbox():
|
|
"""创建新邮箱"""
|
|
data = request.json
|
|
if not data or 'address' not in data or 'domain_id' not in data:
|
|
return jsonify({'success': False, 'error': '缺少必要字段'}), 400
|
|
|
|
db = get_session()
|
|
try:
|
|
# 检查域名是否存在
|
|
domain = db.query(Domain).filter_by(id=data['domain_id'], active=True).first()
|
|
if not domain:
|
|
return jsonify({'success': False, 'error': '域名不存在或未激活'}), 400
|
|
|
|
# 检查邮箱是否已存在
|
|
mailbox_exists = db.query(Mailbox).filter_by(
|
|
address=data['address'], domain_id=data['domain_id']).first()
|
|
if mailbox_exists:
|
|
return jsonify({'success': False, 'error': '邮箱已存在'}), 400
|
|
|
|
# 创建新邮箱
|
|
mailbox = Mailbox(
|
|
address=data['address'],
|
|
domain_id=data['domain_id'],
|
|
description=data.get('description', ''),
|
|
active=data.get('active', True)
|
|
)
|
|
db.add(mailbox)
|
|
db.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': '邮箱创建成功',
|
|
'mailbox': mailbox.to_dict()
|
|
})
|
|
except Exception as e:
|
|
db.rollback()
|
|
current_app.logger.exception(f"创建邮箱失败: {str(e)}")
|
|
return jsonify({'success': False, 'error': '创建邮箱失败'}), 500
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@api_bp.route('/mailboxes/batch', methods=['POST'])
|
|
def batch_create_mailboxes():
|
|
"""批量创建邮箱"""
|
|
data = request.json
|
|
if not data or 'domain_id' not in data or 'usernames' not in data or not isinstance(data['usernames'], list):
|
|
return jsonify({'success': False, 'error': '缺少必要字段或格式不正确'}), 400
|
|
|
|
domain_id = data['domain_id']
|
|
usernames = data['usernames']
|
|
description = data.get('description', '')
|
|
|
|
db = get_session()
|
|
try:
|
|
# 检查域名是否存在
|
|
domain = db.query(Domain).filter_by(id=domain_id, active=True).first()
|
|
if not domain:
|
|
return jsonify({'success': False, 'error': '域名不存在或未激活'}), 400
|
|
|
|
created_mailboxes = []
|
|
existed_mailboxes = []
|
|
|
|
for username in usernames:
|
|
# 检查邮箱是否已存在
|
|
mailbox_exists = db.query(Mailbox).filter_by(
|
|
username=username, domain_id=domain_id).first()
|
|
if mailbox_exists:
|
|
existed_mailboxes.append(username)
|
|
continue
|
|
|
|
# 创建新邮箱
|
|
mailbox = Mailbox(
|
|
username=username,
|
|
domain_id=domain_id,
|
|
description=description,
|
|
active=True
|
|
)
|
|
db.add(mailbox)
|
|
created_mailboxes.append(username)
|
|
|
|
db.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'成功创建 {len(created_mailboxes)} 个邮箱，{len(existed_mailboxes)} 个已存在',
|
|
'created': created_mailboxes,
|
|
'existed': existed_mailboxes
|
|
})
|
|
except Exception as e:
|
|
db.rollback()
|
|
current_app.logger.exception(f"批量创建邮箱失败: {str(e)}")
|
|
return jsonify({'success': False, 'error': '批量创建邮箱失败'}), 500
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@api_bp.route('/mailboxes/<int:mailbox_id>', methods=['GET'])
|
|
def get_mailbox(mailbox_id):
|
|
"""获取特定邮箱的信息"""
|
|
db = get_session()
|
|
try:
|
|
mailbox = db.query(Mailbox).filter_by(id=mailbox_id).first()
|
|
if not mailbox:
|
|
return jsonify({'success': False, 'error': '邮箱不存在'}), 404
|
|
|
|
# 更新最后访问时间
|
|
mailbox.last_accessed = datetime.utcnow()
|
|
db.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'mailbox': mailbox.to_dict()
|
|
})
|
|
except Exception as e:
|
|
current_app.logger.exception(f"获取邮箱信息失败: {str(e)}")
|
|
return jsonify({'success': False, 'error': '获取邮箱信息失败'}), 500
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@api_bp.route('/mailboxes/<int:mailbox_id>/emails', methods=['GET'])
|
|
def get_emails(mailbox_id):
|
|
"""获取邮箱中的所有邮件"""
|
|
db = get_session()
|
|
try:
|
|
mailbox = db.query(Mailbox).filter_by(id=mailbox_id).first()
|
|
if not mailbox:
|
|
return jsonify({'success': False, 'error': '邮箱不存在'}), 404
|
|
|
|
# 更新最后访问时间
|
|
mailbox.last_accessed = datetime.utcnow()
|
|
db.commit()
|
|
|
|
emails = db.query(Email).filter_by(mailbox_id=mailbox_id).order_by(Email.received_at.desc()).all()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'emails': [email.to_dict() for email in emails]
|
|
})
|
|
except Exception as e:
|
|
current_app.logger.exception(f"获取邮件失败: {str(e)}")
|
|
return jsonify({'success': False, 'error': '获取邮件失败'}), 500
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@api_bp.route('/emails/<int:email_id>', methods=['GET'])
|
|
def get_email(email_id):
|
|
"""获取特定邮件的详细内容"""
|
|
db = get_session()
|
|
try:
|
|
email = db.query(Email).filter_by(id=email_id).first()
|
|
if not email:
|
|
return jsonify({'success': False, 'error': '邮件不存在'}), 404
|
|
|
|
# 标记为已读
|
|
if not email.read:
|
|
email.read = True
|
|
db.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'email': email.to_dict()
|
|
})
|
|
except Exception as e:
|
|
current_app.logger.exception(f"获取邮件详情失败: {str(e)}")
|
|
return jsonify({'success': False, 'error': '获取邮件详情失败'}), 500
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@api_bp.route('/emails/<int:email_id>/verification', methods=['GET'])
|
|
def get_verification_info(email_id):
|
|
"""获取邮件中的验证信息（链接和验证码）"""
|
|
db = get_session()
|
|
try:
|
|
email = db.query(Email).filter_by(id=email_id).first()
|
|
if not email:
|
|
return jsonify({'success': False, 'error': '邮件不存在'}), 404
|
|
|
|
verification_links = json.loads(email.verification_links) if email.verification_links else []
|
|
verification_codes = json.loads(email.verification_codes) if email.verification_codes else []
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'email_id': email_id,
|
|
'verification_links': verification_links,
|
|
'verification_codes': verification_codes
|
|
})
|
|
except Exception as e:
|
|
current_app.logger.exception(f"获取验证信息失败: {str(e)}")
|
|
return jsonify({'success': False, 'error': '获取验证信息失败'}), 500
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@api_bp.route('/status', methods=['GET'])
|
|
def system_status():
|
|
"""获取系统状态"""
|
|
session = get_session()
|
|
|
|
# 获取基本统计信息
|
|
domain_count = session.query(func.count(Domain.id)).scalar()
|
|
mailbox_count = session.query(func.count(Mailbox.id)).scalar()
|
|
email_count = session.query(func.count(Email.id)).scalar()
|
|
|
|
# 获取最近24小时的邮件数量
|
|
recent_emails = session.query(func.count(Email.id)).filter(
|
|
Email.received_at > datetime.now() - timedelta(hours=24)
|
|
).scalar()
|
|
|
|
# 获取系统资源信息
|
|
cpu_percent = psutil.cpu_percent(interval=0.5)
|
|
memory = psutil.virtual_memory()
|
|
disk = psutil.disk_usage('/')
|
|
|
|
# 获取服务状态
|
|
smtp_server = get_smtp_server()
|
|
email_processor = get_email_processor()
|
|
|
|
smtp_status = "running" if smtp_server and smtp_server.controller else "stopped"
|
|
processor_status = "running" if email_processor and email_processor.is_running else "stopped"
|
|
|
|
# 构建响应
|
|
status = {
|
|
"system": {
|
|
"uptime": round(time.time() - psutil.boot_time()),
|
|
"time": datetime.now().isoformat(),
|
|
"platform": platform.platform(),
|
|
"python_version": sys.version
|
|
},
|
|
"resources": {
|
|
"cpu_percent": cpu_percent,
|
|
"memory_percent": memory.percent,
|
|
"memory_used": memory.used,
|
|
"memory_total": memory.total,
|
|
"disk_percent": disk.percent,
|
|
"disk_used": disk.used,
|
|
"disk_total": disk.total
|
|
},
|
|
"application": {
|
|
"domain_count": domain_count,
|
|
"mailbox_count": mailbox_count,
|
|
"email_count": email_count,
|
|
"recent_emails_24h": recent_emails,
|
|
"storage_path": os.path.abspath("email_data"),
|
|
"services": {
|
|
"smtp_server": smtp_status,
|
|
"email_processor": processor_status
|
|
}
|
|
}
|
|
}
|
|
|
|
return jsonify(status) |