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') }