Your Name
278 lines
9.9 KiB
Python
278 lines
9.9 KiB
Python
from sqlalchemy.orm import Session
|
|
from typing import Optional, List, Dict, Any
|
|
from io import BytesIO, StringIO
|
|
import csv
|
|
from datetime import datetime
|
|
|
|
from app.models.network import IPAddress, Network, IPStatus
|
|
from app.models.snmp import NetworkDevice, ARPEntry
|
|
from app.models.alert import Alert
|
|
|
|
|
|
class ReportExportService:
|
|
"""报表导出服务"""
|
|
|
|
# ========== CSV 导出 ==========
|
|
|
|
@staticmethod
|
|
def export_ip_addresses_csv(db: Session,
|
|
network_id: Optional[int] = None,
|
|
status: Optional[IPStatus] = None,
|
|
include_online_only: bool = False) -> tuple[str, str]:
|
|
"""
|
|
导出IP地址表为CSV
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
network_id: 网段ID（可选，不指定则导出所有）
|
|
status: IP状态过滤
|
|
include_online_only: 仅导出在线IP
|
|
|
|
Returns:
|
|
(CSV内容字符串, 文件名)
|
|
"""
|
|
query = db.query(IPAddress)
|
|
|
|
if network_id:
|
|
query = query.filter(IPAddress.network_id == network_id)
|
|
if status:
|
|
query = query.filter(IPAddress.status == status)
|
|
if include_online_only:
|
|
query = query.filter(IPAddress.status == IPStatus.ONLINE)
|
|
|
|
ips = query.order_by(IPAddress.network_id, IPAddress.id).all()
|
|
|
|
output = StringIO()
|
|
writer = csv.writer(output)
|
|
|
|
# 写入表头
|
|
writer.writerow([
|
|
'ID', '网段ID', 'IP地址', 'MAC地址', '主机名', '状态',
|
|
'使用者', '业务类型', '备注', '厂商',
|
|
'最后发现时间', '创建时间'
|
|
])
|
|
|
|
# 写入数据
|
|
for ip in ips:
|
|
writer.writerow([
|
|
ip.id,
|
|
ip.network_id,
|
|
ip.ip_address,
|
|
ip.mac_address or '',
|
|
ip.hostname or '',
|
|
ip.status.value,
|
|
ip.owner or '',
|
|
ip.business_type or '',
|
|
ip.notes or '',
|
|
ip.vendor or '',
|
|
ip.last_seen.strftime('%Y-%m-%d %H:%M:%S') if ip.last_seen else '',
|
|
ip.created_at.strftime('%Y-%m-%d %H:%M:%S') if ip.created_at else ''
|
|
])
|
|
|
|
csv_content = output.getvalue()
|
|
filename = f"ip_addresses_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
|
|
|
|
return csv_content, filename
|
|
|
|
@staticmethod
|
|
def export_networks_csv(db: Session) -> tuple[str, str]:
|
|
"""导出网段汇总CSV"""
|
|
networks = db.query(Network).order_by(Network.id).all()
|
|
|
|
output = StringIO()
|
|
writer = csv.writer(output)
|
|
|
|
writer.writerow([
|
|
'ID', '网段CIDR', '名称', '描述', '分组',
|
|
'网关', 'VLAN', '总IP数', '已用IP数', '保留IP数',
|
|
'创建时间'
|
|
])
|
|
|
|
for net in networks:
|
|
writer.writerow([
|
|
net.id,
|
|
net.cidr,
|
|
net.name or '',
|
|
net.description or '',
|
|
net.group_name or '',
|
|
net.gateway or '',
|
|
net.vlan_id or '',
|
|
net.total_ips,
|
|
net.used_ips,
|
|
net.reserved_ips,
|
|
net.created_at.strftime('%Y-%m-%d %H:%M:%S') if net.created_at else ''
|
|
])
|
|
|
|
csv_content = output.getvalue()
|
|
filename = f"networks_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
|
|
|
|
return csv_content, filename
|
|
|
|
@staticmethod
|
|
def export_alerts_csv(db: Session, status: Optional[str] = None) -> tuple[str, str]:
|
|
"""导出告警CSV"""
|
|
query = db.query(Alert)
|
|
if status:
|
|
query = query.filter(Alert.status == status)
|
|
|
|
alerts = query.order_by(Alert.created_at.desc()).all()
|
|
|
|
output = StringIO()
|
|
writer = csv.writer(output)
|
|
|
|
writer.writerow([
|
|
'ID', '告警类型', '严重程度', '状态', '标题',
|
|
'详细信息', '相关IP', '相关MAC', '创建时间'
|
|
])
|
|
|
|
for alert in alerts:
|
|
writer.writerow([
|
|
alert.id,
|
|
alert.alert_type.value,
|
|
alert.severity.value,
|
|
alert.status.value,
|
|
alert.title,
|
|
alert.message or '',
|
|
alert.ip_address_str or '',
|
|
alert.mac_address or '',
|
|
alert.created_at.strftime('%Y-%m-%d %H:%M:%S') if alert.created_at else ''
|
|
])
|
|
|
|
csv_content = output.getvalue()
|
|
filename = f"alerts_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
|
|
|
|
return csv_content, filename
|
|
|
|
@staticmethod
|
|
def export_snmp_devices_csv(db: Session) -> tuple[str, str]:
|
|
"""导出SNMP设备CSV"""
|
|
devices = db.query(NetworkDevice).order_by(NetworkDevice.id).all()
|
|
|
|
output = StringIO()
|
|
writer = csv.writer(output)
|
|
|
|
writer.writerow([
|
|
'ID', '设备名称', 'IP地址', '端口', '设备类型',
|
|
'厂商', '型号', '位置', '最后轮询时间', '创建时间'
|
|
])
|
|
|
|
for dev in devices:
|
|
writer.writerow([
|
|
dev.id,
|
|
dev.name,
|
|
dev.ip_address,
|
|
dev.port,
|
|
dev.device_type.value,
|
|
dev.vendor or '',
|
|
dev.model or '',
|
|
dev.location or '',
|
|
dev.last_polled_at.strftime('%Y-%m-%d %H:%M:%S') if dev.last_polled_at else '',
|
|
dev.created_at.strftime('%Y-%m-%d %H:%M:%S') if dev.created_at else ''
|
|
])
|
|
|
|
csv_content = output.getvalue()
|
|
filename = f"snmp_devices_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
|
|
|
|
return csv_content, filename
|
|
|
|
@staticmethod
|
|
def export_arp_table_csv(db: Session, device_id: Optional[int] = None) -> tuple[str, str]:
|
|
"""导出ARP表CSV"""
|
|
query = db.query(ARPEntry)
|
|
if device_id:
|
|
query = query.filter(ARPEntry.device_id == device_id)
|
|
|
|
entries = query.order_by(ARPEntry.device_id, ARPEntry.id).all()
|
|
|
|
output = StringIO()
|
|
writer = csv.writer(output)
|
|
|
|
writer.writerow([
|
|
'ID', '设备ID', 'IP地址', 'MAC地址', '接口', '最后发现时间'
|
|
])
|
|
|
|
for entry in entries:
|
|
writer.writerow([
|
|
entry.id,
|
|
entry.device_id,
|
|
entry.ip_address,
|
|
entry.mac_address,
|
|
entry.interface or '',
|
|
entry.last_seen.strftime('%Y-%m-%d %H:%M:%S') if entry.last_seen else ''
|
|
])
|
|
|
|
csv_content = output.getvalue()
|
|
filename = f"arp_table_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
|
|
|
|
return csv_content, filename
|
|
|
|
# ========== 汇总统计报表 ==========
|
|
|
|
@staticmethod
|
|
def get_ipam_summary(db: Session) -> Dict[str, Any]:
|
|
"""获取IPAM系统汇总报表"""
|
|
# 网段统计
|
|
total_networks = db.query(Network).count()
|
|
total_ips = db.query(IPAddress).count()
|
|
online_ips = db.query(IPAddress).filter(IPAddress.status == IPStatus.ONLINE).count()
|
|
reserved_ips = db.query(IPAddress).filter(IPAddress.status == IPStatus.RESERVED).count()
|
|
|
|
# 带MAC/主机名的IP统计
|
|
with_mac = db.query(IPAddress).filter(IPAddress.mac_address.isnot(None)).count()
|
|
with_hostname = db.query(IPAddress).filter(IPAddress.hostname.isnot(None)).count()
|
|
|
|
# 网段使用率分布
|
|
networks = db.query(Network).all()
|
|
usage_distribution = {
|
|
'under_50': 0, # 50%以下
|
|
'50_70': 0, # 50%-70%
|
|
'70_90': 0, # 70%-90%
|
|
'over_90': 0 # 90%以上
|
|
}
|
|
|
|
for net in networks:
|
|
usage_rate = (net.used_ips / net.total_ips * 100) if net.total_ips > 0 else 0
|
|
if usage_rate < 50:
|
|
usage_distribution['under_50'] += 1
|
|
elif usage_rate < 70:
|
|
usage_distribution['50_70'] += 1
|
|
elif usage_rate < 90:
|
|
usage_distribution['70_90'] += 1
|
|
else:
|
|
usage_distribution['over_90'] += 1
|
|
|
|
# 告警统计
|
|
from app.models.alert import Alert, AlertStatus
|
|
active_alerts = db.query(Alert).filter(Alert.status == AlertStatus.ACTIVE).count()
|
|
resolved_alerts = db.query(Alert).filter(Alert.status == AlertStatus.RESOLVED).count()
|
|
|
|
# SNMP设备统计
|
|
snmp_devices = db.query(NetworkDevice).count()
|
|
active_devices = db.query(NetworkDevice).filter(NetworkDevice.last_polled_at.isnot(None)).count()
|
|
|
|
return {
|
|
"summary": {
|
|
"total_networks": total_networks,
|
|
"total_ip_addresses": total_ips,
|
|
"online_ips": online_ips,
|
|
"reserved_ips": reserved_ips,
|
|
"available_ips": total_ips - online_ips - reserved_ips,
|
|
"online_rate": round(online_ips / total_ips * 100, 2) if total_ips > 0 else 0
|
|
},
|
|
"discovery": {
|
|
"with_mac_address": with_mac,
|
|
"with_hostname": with_hostname,
|
|
"mac_discovery_rate": round(with_mac / total_ips * 100, 2) if total_ips > 0 else 0
|
|
},
|
|
"network_usage_distribution": usage_distribution,
|
|
"alerts": {
|
|
"active": active_alerts,
|
|
"resolved": resolved_alerts
|
|
},
|
|
"snmp": {
|
|
"total_devices": snmp_devices,
|
|
"active_devices": active_devices
|
|
},
|
|
"generated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
}
|