620 lines
24 KiB
Python
620 lines
24 KiB
Python
"""
|
||
Nginx域名转发管理 - Python API 主入口
|
||
部署在 Nginx 服务器 (10.1.1.160:5000) 上
|
||
直接操作本机 Nginx 配置和证书
|
||
"""
|
||
import os
|
||
import logging
|
||
from flask import Flask, request, g
|
||
from services.nginx_service import NginxService
|
||
from services.ssl_service import SSLService
|
||
from services.dns_service import DnsService
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
app = Flask(__name__)
|
||
app.config['JSON_AS_ASCII'] = False # 中文不转 Unicode
|
||
|
||
# ════════════════════════════════════════════
|
||
# 网关用户信息中间件
|
||
# ════════════════════════════════════════════
|
||
USER_HEADERS = [
|
||
'X-User-Id', 'X-User-Name', 'X-User-Account',
|
||
'X-User-Email', 'X-User-Post', 'X-User-Roles',
|
||
'X-System-Id', 'X-System-Code', 'X-System-Name',
|
||
]
|
||
|
||
|
||
@app.before_request
|
||
def user_context_middleware():
|
||
"""从网关注入的请求头提取用户信息，存到 Flask g 对象"""
|
||
for h in USER_HEADERS:
|
||
val = request.headers.get(h)
|
||
if val:
|
||
setattr(g, h.replace('-', '_').lower(), val)
|
||
|
||
# 初始化服务
|
||
nginx_conf_dir = os.environ.get('NGINX_CONF_DIR', '/etc/nginx/conf.d')
|
||
nginx_ssl_dir = os.environ.get('NGINX_SSL_DIR', '/etc/nginx/ssl')
|
||
ca_dir = os.environ.get('CA_DIR', '/etc/nginx/ssl/ca')
|
||
|
||
nginx_service = NginxService(nginx_conf_dir, nginx_ssl_dir)
|
||
ssl_service = SSLService(ca_dir, nginx_ssl_dir)
|
||
dns_service = DnsService()
|
||
|
||
# 启动时初始化内部 CA（gunicorn 模式下也必须执行）
|
||
ssl_service.init_ca()
|
||
|
||
# ──────────────────────────────────────────────
|
||
# API 路由
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
@app.route('/api/nginx/health', methods=['GET'])
|
||
def health_check():
|
||
"""健康检查"""
|
||
return {
|
||
'code': 0,
|
||
'message': 'Nginx Manager API is running',
|
||
'data': {
|
||
'nginx_installed': os.path.exists('/usr/sbin/nginx') or os.path.exists('/usr/bin/nginx'),
|
||
'conf_dir': nginx_conf_dir,
|
||
'ssl_dir': nginx_ssl_dir
|
||
}
|
||
}
|
||
|
||
|
||
@app.route('/api/nginx/rules', methods=['GET'])
|
||
def list_rules():
|
||
"""获取所有转发规则列表"""
|
||
try:
|
||
rules = nginx_service.list_rules()
|
||
return {'code': 0, 'message': 'success', 'data': rules}
|
||
except Exception as e:
|
||
logger.error(f"获取规则列表失败: {e}")
|
||
return {'code': -1, 'message': str(e), 'data': []}
|
||
|
||
|
||
@app.route('/api/nginx/rules', methods=['POST'])
|
||
def add_rule():
|
||
"""新增转发规则"""
|
||
try:
|
||
body = request.get_json(force=True)
|
||
domain = body.get('domain', '').strip()
|
||
protocol = body.get('protocol', '').strip().lower()
|
||
target = body.get('target', '').strip()
|
||
|
||
# 参数校验
|
||
if not domain:
|
||
return {'code': -1, 'message': '域名不能为空'}, 400
|
||
if protocol not in ('http', 'https'):
|
||
return {'code': -1, 'message': '协议类型必须为 http 或 https'}, 400
|
||
if not target.startswith('http://') and not target.startswith('https://'):
|
||
return {'code': -1, 'message': '转发目标必须以 http:// 或 https:// 开头'}, 400
|
||
|
||
# 系统保护域名不允许手动创建（zgitm.com / www.zgitm.com）
|
||
if ssl_service.is_system_domain(domain):
|
||
return {'code': -1, 'message': f'{domain} 是系统保护域名，不能手动创建'}, 400
|
||
|
||
logger.info(f"新增规则: domain={domain}, protocol={protocol}, target={target}")
|
||
|
||
# 检查是否已存在
|
||
existing = nginx_service.find_rule(domain)
|
||
if existing:
|
||
return {'code': -1, 'message': f'域名 {domain} 的规则已存在'}, 409
|
||
|
||
# HTTPS证书签发
|
||
ssl_cert = None
|
||
ssl_key = None
|
||
if protocol == 'https':
|
||
if ssl_service.is_zgitm_domain(domain):
|
||
# zgitm.com 子域名 → Let's Encrypt DNS-01
|
||
ssl_cert, ssl_key = ssl_service.ensure_le_certificate(domain)
|
||
if not ssl_cert or not ssl_key:
|
||
return {'code': -1, 'message': f'{domain} Let\'s Encrypt 证书签发失败，请查看服务器日志'}, 500
|
||
else:
|
||
# 其他域名 → 内部 CA
|
||
ssl_cert, ssl_key = ssl_service.ensure_certificate(domain)
|
||
|
||
# 创建 Nginx 配置文件
|
||
config_file = nginx_service.create_config(domain, protocol, target, ssl_cert, ssl_key)
|
||
|
||
# 检查 Nginx 配置语法
|
||
valid, error_msg = nginx_service.check_config()
|
||
if not valid:
|
||
# 回滚：删除刚创建的配置文件
|
||
nginx_service.delete_config(domain)
|
||
logger.error(f"Nginx 配置语法错误: {error_msg}")
|
||
return {'code': -1, 'message': f'Nginx 配置语法检查失败: {error_msg}'}, 400
|
||
|
||
return {
|
||
'code': 0,
|
||
'message': '规则创建成功',
|
||
'data': {
|
||
'domain': domain,
|
||
'protocol': protocol,
|
||
'target': target,
|
||
'config_file': config_file,
|
||
'ssl_cert': ssl_cert,
|
||
'ssl_key': ssl_key
|
||
}
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"新增规则失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/rules/<domain>', methods=['GET'])
|
||
def get_rule(domain):
|
||
"""获取单个规则详情"""
|
||
try:
|
||
rule = nginx_service.find_rule(domain)
|
||
if rule:
|
||
return {'code': 0, 'message': 'success', 'data': rule}
|
||
return {'code': -1, 'message': f'域名 {domain} 的规则不存在'}, 404
|
||
except Exception as e:
|
||
logger.error(f"获取规则详情失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/rules/<domain>', methods=['PUT'])
|
||
def update_rule(domain):
|
||
"""修改转发规则"""
|
||
try:
|
||
body = request.get_json(force=True)
|
||
new_domain = body.get('domain', '').strip()
|
||
new_protocol = body.get('protocol', '').strip().lower()
|
||
new_target = body.get('target', '').strip()
|
||
|
||
if not new_domain:
|
||
return {'code': -1, 'message': '域名不能为空'}, 400
|
||
if new_protocol not in ('http', 'https'):
|
||
return {'code': -1, 'message': '协议类型必须为 http 或 https'}, 400
|
||
if not new_target.startswith('http://') and not new_target.startswith('https://'):
|
||
return {'code': -1, 'message': '转发目标必须以 http:// 或 https:// 开头'}, 400
|
||
|
||
logger.info(f"修改规则: domain={domain} -> new_domain={new_domain}")
|
||
|
||
# 检查原规则是否存在
|
||
old_rule = nginx_service.find_rule(domain)
|
||
if not old_rule:
|
||
return {'code': -1, 'message': f'域名 {domain} 的规则不存在'}, 404
|
||
|
||
# 删除旧配置
|
||
nginx_service.delete_config(domain)
|
||
|
||
# HTTPS证书签发
|
||
ssl_cert = None
|
||
ssl_key = None
|
||
if new_protocol == 'https':
|
||
if ssl_service.is_zgitm_domain(new_domain):
|
||
ssl_cert, ssl_key = ssl_service.ensure_le_certificate(new_domain)
|
||
if not ssl_cert or not ssl_key:
|
||
return {'code': -1, 'message': f'{new_domain} Let\'s Encrypt 证书签发失败'}, 500
|
||
else:
|
||
ssl_cert, ssl_key = ssl_service.ensure_certificate(new_domain)
|
||
|
||
# 创建新配置
|
||
config_file = nginx_service.create_config(new_domain, new_protocol, new_target, ssl_cert, ssl_key)
|
||
|
||
# 语法检查
|
||
valid, error_msg = nginx_service.check_config()
|
||
if not valid:
|
||
nginx_service.delete_config(new_domain)
|
||
# 恢复旧配置
|
||
old_proto = old_rule.get('protocol', 'http')
|
||
old_cert = old_rule.get('ssl_cert')
|
||
old_key = old_rule.get('ssl_key')
|
||
nginx_service.create_config(domain, old_proto, old_rule.get('target', ''), old_cert, old_key)
|
||
logger.error(f"Nginx 配置语法错误: {error_msg}")
|
||
return {'code': -1, 'message': f'Nginx 配置语法检查失败: {error_msg}'}, 400
|
||
|
||
return {
|
||
'code': 0,
|
||
'message': '规则修改成功',
|
||
'data': {
|
||
'domain': new_domain,
|
||
'protocol': new_protocol,
|
||
'target': new_target,
|
||
'config_file': config_file,
|
||
'ssl_cert': ssl_cert,
|
||
'ssl_key': ssl_key
|
||
}
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"修改规则失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/rules/<domain>', methods=['DELETE'])
|
||
def delete_rule(domain):
|
||
"""删除转发规则"""
|
||
try:
|
||
logger.info(f"删除规则: domain={domain}")
|
||
|
||
rule = nginx_service.find_rule(domain)
|
||
|
||
# 删除配置文件（如果存在）
|
||
deleted = nginx_service.delete_config(domain)
|
||
|
||
# 文件不存在但数据库有记录 → 允许继续（清理脏数据）
|
||
if not rule and not deleted:
|
||
logger.info(f"域名 {domain} 的规则文件不存在，跳过删除")
|
||
|
||
# 检查语法（删除后）
|
||
valid, error_msg = nginx_service.check_config()
|
||
if not valid:
|
||
logger.warning(f"删除后 Nginx 配置语法错误: {error_msg}")
|
||
|
||
return {'code': 0, 'message': '规则已删除', 'data': {'domain': domain}}
|
||
except Exception as e:
|
||
logger.error(f"删除规则失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 配置文件查看/编辑接口
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
@app.route('/api/nginx/rules/<domain>/config', methods=['GET'])
|
||
def get_config(domain):
|
||
"""获取指定域名的 Nginx 配置文件原始内容"""
|
||
try:
|
||
import os as _os
|
||
config_path = _os.path.join(nginx_conf_dir, f'{domain}.conf')
|
||
if not _os.path.exists(config_path):
|
||
return {'code': -1, 'message': f'{domain} 的配置文件不存在', 'data': None}, 404
|
||
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
return {
|
||
'code': 0,
|
||
'message': 'success',
|
||
'data': {
|
||
'domain': domain,
|
||
'config_file': config_path,
|
||
'content': content
|
||
}
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取配置文件失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/rules/<domain>/config', methods=['PUT'])
|
||
def update_config(domain):
|
||
"""直接写入 Nginx 配置文件（会进行语法检查，失败自动回滚）"""
|
||
try:
|
||
body = request.get_json(force=True)
|
||
new_content = body.get('content', '')
|
||
|
||
if not new_content.strip():
|
||
return {'code': -1, 'message': '配置内容不能为空'}, 400
|
||
|
||
import os as _os
|
||
config_path = _os.path.join(nginx_conf_dir, f'{domain}.conf')
|
||
old_content = None
|
||
old_exists = _os.path.exists(config_path)
|
||
|
||
# 备份旧内容（如果存在）
|
||
if old_exists:
|
||
with open(config_path, 'r', encoding='utf-8') as f:
|
||
old_content = f.read()
|
||
|
||
# 写入新内容
|
||
with open(config_path, 'w', encoding='utf-8') as f:
|
||
f.write(new_content)
|
||
logger.info(f"已写入配置文件: {config_path}")
|
||
|
||
# 语法检查
|
||
valid, error_msg = nginx_service.check_config()
|
||
if not valid:
|
||
# 回滚
|
||
if old_exists:
|
||
with open(config_path, 'w', encoding='utf-8') as f:
|
||
f.write(old_content)
|
||
else:
|
||
_os.remove(config_path)
|
||
logger.error(f"Nginx 配置语法错误，已回滚: {error_msg}")
|
||
return {'code': -1, 'message': f'配置语法检查失败，已自动回滚: {error_msg}'}, 400
|
||
|
||
# 重载 Nginx
|
||
nginx_service.reload()
|
||
logger.info(f"Nginx 配置已重载: {domain}")
|
||
|
||
return {
|
||
'code': 0,
|
||
'message': '配置已保存并重载生效',
|
||
'data': {'domain': domain, 'config_file': config_path}
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"更新配置文件失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/rules/<domain>/ssl', methods=['GET'])
|
||
def get_ssl_files(domain):
|
||
"""获取指定域名的 SSL 证书和密钥文件内容（仅查看，不可编辑）"""
|
||
try:
|
||
import os as _os
|
||
cert_path = _os.path.join(nginx_ssl_dir, f'{domain}.pem')
|
||
key_path = _os.path.join(nginx_ssl_dir, f'{domain}.key')
|
||
|
||
cert_content = None
|
||
key_content = None
|
||
|
||
if _os.path.exists(cert_path):
|
||
with open(cert_path, 'r', encoding='utf-8') as f:
|
||
cert_content = f.read()
|
||
if _os.path.exists(key_path):
|
||
with open(key_path, 'r', encoding='utf-8') as f:
|
||
key_content = f.read()
|
||
|
||
if not cert_content and not key_content:
|
||
return {'code': -1, 'message': f'{domain} 没有 SSL 证书文件', 'data': None}, 404
|
||
|
||
return {
|
||
'code': 0,
|
||
'message': 'success',
|
||
'data': {
|
||
'domain': domain,
|
||
'cert_path': cert_path,
|
||
'cert_content': cert_content,
|
||
'key_path': key_path,
|
||
'key_content': key_content,
|
||
'has_cert': bool(cert_content),
|
||
'has_key': bool(key_content)
|
||
}
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取SSL证书文件失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 内容压缩（gzip）接口
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
@app.route('/api/nginx/rules/<domain>/gzip', methods=['GET'])
|
||
def get_gzip_config(domain):
|
||
"""获取指定域名的 gzip 压缩配置"""
|
||
try:
|
||
gzip_config = nginx_service.parse_gzip_config(domain)
|
||
return {'code': 0, 'message': 'success', 'data': gzip_config}
|
||
except Exception as e:
|
||
logger.error(f"获取gzip配置失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/rules/<domain>/gzip', methods=['PUT'])
|
||
def update_gzip_config(domain):
|
||
"""更新指定域名的 gzip 压缩配置"""
|
||
try:
|
||
body = request.get_json(force=True)
|
||
enabled = body.get('enabled', False)
|
||
types = body.get('types', [])
|
||
comp_level = body.get('comp_level', 6)
|
||
min_length = body.get('min_length', '1000')
|
||
|
||
# 参数校验
|
||
if enabled and not isinstance(types, list):
|
||
return {'code': -1, 'message': 'types 必须是数组'}, 400
|
||
if enabled and comp_level not in range(1, 10):
|
||
return {'code': -1, 'message': '压缩级别必须在 1-9 之间'}, 400
|
||
if enabled and not min_length:
|
||
return {'code': -1, 'message': '最小压缩长度不能为空'}, 400
|
||
|
||
gzip_config = {
|
||
'enabled': enabled,
|
||
'types': types,
|
||
'comp_level': comp_level,
|
||
'min_length': min_length
|
||
}
|
||
|
||
success, message = nginx_service.set_gzip_config(domain, gzip_config)
|
||
if success:
|
||
return {'code': 0, 'message': message, 'data': gzip_config}
|
||
return {'code': -1, 'message': message}, 400
|
||
except Exception as e:
|
||
logger.error(f"更新gzip配置失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 域名日志接口
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
@app.route('/api/nginx/rules/<domain>/logs', methods=['GET'])
|
||
def get_domain_logs(domain):
|
||
"""获取指定域名的 Nginx 访问日志和错误日志"""
|
||
try:
|
||
lines = request.args.get('lines', 200, type=int)
|
||
# 限制最大行数
|
||
lines = min(lines, 1000)
|
||
logs = nginx_service.get_domain_logs(domain, lines=lines)
|
||
return {'code': 0, 'message': 'success', 'data': logs}
|
||
except Exception as e:
|
||
logger.error(f"获取域名日志失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/reload', methods=['POST'])
|
||
def reload_nginx():
|
||
"""重载 Nginx 配置"""
|
||
try:
|
||
logger.info("重载 Nginx 配置")
|
||
success, message = nginx_service.reload()
|
||
if success:
|
||
return {'code': 0, 'message': message}
|
||
return {'code': -1, 'message': message}, 500
|
||
except Exception as e:
|
||
logger.error(f"重载 Nginx 失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 证书管理接口（Let's Encrypt）
|
||
# ──────────────────────────────────────────────
|
||
|
||
@app.route('/api/nginx/certs', methods=['GET'])
|
||
def list_certs():
|
||
"""获取所有 SSL 证书信息列表（Let's Encrypt + 内部 CA）"""
|
||
try:
|
||
certs = []
|
||
# 扫描 SSL 目录中所有的 .pem 证书文件
|
||
import glob as _glob
|
||
# 要排除的文件（CA 根证书等）
|
||
exclude_files = {'ca.pem', 'ca.crt'}
|
||
for cert_file in _glob.glob(f'{nginx_ssl_dir}/*.pem'):
|
||
filename = os.path.basename(cert_file)
|
||
if filename in exclude_files:
|
||
continue
|
||
domain = filename.replace('.pem', '')
|
||
info = ssl_service.get_le_cert_info(domain)
|
||
certs.append(info)
|
||
return {'code': 0, 'message': 'success', 'data': certs}
|
||
except Exception as e:
|
||
logger.error(f"获取证书列表失败: {e}")
|
||
return {'code': -1, 'message': str(e), 'data': []}
|
||
|
||
|
||
@app.route('/api/nginx/certs/<domain>', methods=['GET'])
|
||
def get_cert_info(domain):
|
||
"""获取单个 SSL 证书详情（Let's Encrypt + 内部 CA）"""
|
||
try:
|
||
info = ssl_service.get_le_cert_info(domain)
|
||
return {'code': 0, 'message': 'success', 'data': info}
|
||
except Exception as e:
|
||
logger.error(f"获取证书信息失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/certs/<domain>/issue', methods=['POST'])
|
||
def issue_cert(domain):
|
||
"""首次签发或重新签发 Let's Encrypt 证书"""
|
||
try:
|
||
if not ssl_service.is_zgitm_domain(domain):
|
||
return {'code': -1, 'message': f'{domain} 不是 zgitm.com 域名'}, 400
|
||
|
||
logger.info(f"签发 LE 证书: {domain}")
|
||
cert_path, key_path = ssl_service.ensure_le_certificate(domain)
|
||
|
||
if cert_path and key_path:
|
||
# 获取最新证书信息
|
||
info = ssl_service.get_le_cert_info(domain)
|
||
return {'code': 0, 'message': f'{domain} 证书签发成功', 'data': info}
|
||
else:
|
||
return {'code': -1, 'message': f'{domain} 证书签发失败，请检查阿里云 AccessKey 和网络'}, 500
|
||
except Exception as e:
|
||
logger.error(f"签发证书失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/certs/<domain>/renew', methods=['POST'])
|
||
def renew_cert(domain):
|
||
"""手动续期 Let's Encrypt 证书"""
|
||
try:
|
||
if not ssl_service.is_zgitm_domain(domain):
|
||
return {'code': -1, 'message': f'{domain} 不是 zgitm.com 域名'}, 400
|
||
|
||
logger.info(f"手动续期 LE 证书: {domain}")
|
||
result = ssl_service.renew_le_certificate(domain)
|
||
|
||
if result['success']:
|
||
return {'code': 0, 'message': result['message'], 'data': {'log': result['log']}}
|
||
else:
|
||
return {'code': -1, 'message': result['message'], 'data': {'log': result['log']}}, 500
|
||
except Exception as e:
|
||
logger.error(f"续期证书失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# DNS 解析管理接口
|
||
# ──────────────────────────────────────────────
|
||
|
||
@app.route('/api/nginx/dns/records', methods=['GET'])
|
||
def list_dns_records():
|
||
"""获取 DNS 解析记录列表"""
|
||
try:
|
||
zone = request.args.get('zone', ssl_service.get_le_root_domain())
|
||
rr = request.args.get('rr', None)
|
||
rtype = request.args.get('type', None)
|
||
records = dns_service.list_records(zone, rr_keyword=rr, record_type=rtype)
|
||
return {'code': 0, 'message': 'success', 'data': records}
|
||
except Exception as e:
|
||
logger.error(f"获取DNS记录失败: {e}")
|
||
return {'code': -1, 'message': str(e), 'data': []}
|
||
|
||
|
||
@app.route('/api/nginx/dns/records', methods=['POST'])
|
||
def add_dns_record():
|
||
"""添加 DNS 解析记录"""
|
||
try:
|
||
body = request.get_json(force=True)
|
||
zone = body.get('zone', ssl_service.get_le_root_domain())
|
||
rr = body.get('rr', '').strip()
|
||
record_type = body.get('type', 'A').upper()
|
||
value = body.get('value', '').strip()
|
||
ttl = body.get('ttl', 600)
|
||
|
||
if not rr:
|
||
return {'code': -1, 'message': '主机记录(rr)不能为空'}, 400
|
||
if not value:
|
||
return {'code': -1, 'message': '记录值(value)不能为空'}, 400
|
||
|
||
result = dns_service.add_record(zone, rr, record_type, value, ttl)
|
||
return {'code': 0, 'message': f'{result["full_domain"]} 记录已添加', 'data': result}
|
||
except Exception as e:
|
||
logger.error(f"添加DNS记录失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/dns/records/<record_id>', methods=['DELETE'])
|
||
def delete_dns_record(record_id):
|
||
"""删除 DNS 解析记录"""
|
||
try:
|
||
dns_service.delete_record(record_id)
|
||
return {'code': 0, 'message': 'DNS记录已删除'}
|
||
except Exception as e:
|
||
logger.error(f"删除DNS记录失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
@app.route('/api/nginx/dns/records/<record_id>', methods=['PUT'])
|
||
def update_dns_record(record_id):
|
||
"""修改 DNS 解析记录"""
|
||
try:
|
||
body = request.get_json(force=True)
|
||
rr = body.get('rr', '@')
|
||
record_type = body.get('type', 'A').upper()
|
||
value = body.get('value', '').strip()
|
||
ttl = body.get('ttl', 600)
|
||
|
||
if not value:
|
||
return {'code': -1, 'message': '记录值(value)不能为空'}, 400
|
||
|
||
dns_service.update_record(record_id, rr, record_type, value, ttl)
|
||
return {'code': 0, 'message': 'DNS记录已更新'}
|
||
except Exception as e:
|
||
logger.error(f"修改DNS记录失败: {e}")
|
||
return {'code': -1, 'message': str(e)}, 500
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 启动入口
|
||
# ──────────────────────────────────────────────
|
||
|
||
if __name__ == '__main__':
|
||
logger.info("Nginx Manager API 启动在端口 5000")
|
||
app.run(host='0.0.0.0', port=5000, debug=False)
|