v6ole
5aadbc78c6
- 新增 RecordsLog.vue 操作记录日志页面 - 新增 ReplacementRecords.vue 设备更换记录页面 - 新增 imc_service.py IMC网管系统集成服务 - 新增 datetime.js 前端日期时间工具函数 - 新增 OLT时间同步.md 文档 - 扩展 devices.py API：设备更换记录、批量操作等 - 扩展 ssh_service.py：OLT时间同步功能 - 扩展 olt.py：新增时间同步相关接口 - 更新 DeviceList.vue：增强设备列表功能 - 更新路由和导航菜单 - 将 .claude/ 和 .mcp.json 加入 .gitignore Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
116 lines
4.2 KiB
Python
116 lines
4.2 KiB
Python
"""Excel 导入服务"""
|
|
import pandas as pd
|
|
from typing import List, Dict
|
|
from sqlalchemy.orm import Session
|
|
from app.models.device import ONUDevice
|
|
import math
|
|
|
|
# 定义字段映射：Excel列名 -> 数据库字段名
|
|
FIELD_MAPPING = {
|
|
'mac_address': 'mac_address',
|
|
'region': 'region',
|
|
'school_name': 'school_name',
|
|
'building': 'building',
|
|
'place_type': 'place_type',
|
|
'room_number': 'room_number',
|
|
'notes': 'notes',
|
|
}
|
|
|
|
|
|
def clean_value(value) -> str:
|
|
"""清理单元格值，处理NaN和None"""
|
|
if value is None:
|
|
return ''
|
|
if isinstance(value, float) and math.isnan(value):
|
|
return ''
|
|
return str(value).strip()
|
|
|
|
|
|
class ImportService:
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def parse_excel(self, file_path: str) -> List[Dict]:
|
|
"""解析 Excel 文件"""
|
|
df = pd.read_excel(file_path)
|
|
# 标准化列名（去除空格，转小写）
|
|
df.columns = [col.strip().lower() for col in df.columns]
|
|
# 转换为记录列表
|
|
records = []
|
|
for _, row in df.iterrows():
|
|
record = {}
|
|
for col_name, db_field in FIELD_MAPPING.items():
|
|
if col_name in row:
|
|
record[db_field] = clean_value(row[col_name])
|
|
records.append(record)
|
|
return records
|
|
|
|
def validate_data(self, records: List[Dict]) -> Dict:
|
|
"""验证数据"""
|
|
valid = []
|
|
invalid = []
|
|
|
|
for idx, record in enumerate(records):
|
|
mac = record.get('mac_address', '').strip()
|
|
if not mac:
|
|
invalid.append({'row': idx + 2, 'mac': mac, 'reason': 'MAC地址缺失'})
|
|
continue
|
|
|
|
# 标准化MAC地址：统一使用横杠分隔小写格式
|
|
# 支持格式：AA:BB:CC:DD:EE:FF, AA-BB-CC-DD-EE-FF, AABBCCDDEEFF, aa:bb:cc:dd:ee:ff
|
|
mac_clean = mac.upper().replace(':', '-')
|
|
|
|
# 验证基本格式：12个十六进制字符（可能有分隔符）
|
|
hex_chars = mac_clean.replace('-', '')
|
|
if len(hex_chars) != 12 or not all(c in '0123456789ABCDEF' for c in hex_chars):
|
|
invalid.append({'row': idx + 2, 'mac': mac, 'reason': f'MAC地址格式错误 "{mac}"'})
|
|
continue
|
|
|
|
# 转换为标准格式 34dc-99c8-56e0（小写4位分组）
|
|
mac_formatted = '-'.join([hex_chars[i:i+4].lower() for i in range(0, 12, 4)])
|
|
record['mac_address'] = mac_formatted
|
|
valid.append(record)
|
|
|
|
return {'valid': valid, 'invalid': invalid}
|
|
|
|
def import_devices(self, records: List[Dict], olt_id: int = None) -> Dict:
|
|
"""批量导入设备，存在则更新，不存在则新增"""
|
|
created_count = 0
|
|
updated_count = 0
|
|
|
|
for record in records:
|
|
mac = record.get('mac_address', '')
|
|
if not mac:
|
|
continue
|
|
|
|
# 查询是否已存在该 MAC 地址
|
|
existing = self.db.query(ONUDevice).filter(ONUDevice.mac_address == mac).first()
|
|
|
|
if existing:
|
|
# 更新现有记录（MAC 地址不变）
|
|
existing.region = record.get('region', '')
|
|
existing.school_name = record.get('school_name', '')
|
|
existing.building = record.get('building') or None
|
|
existing.place_type = record.get('place_type') or None
|
|
existing.room_number = record.get('room_number') or None
|
|
existing.notes = record.get('notes') or None
|
|
updated_count += 1
|
|
else:
|
|
# 新增记录
|
|
device = ONUDevice(
|
|
mac_address=mac,
|
|
olt_id=olt_id,
|
|
region=record.get('region', ''),
|
|
school_name=record.get('school_name', ''),
|
|
building=record.get('building') or None,
|
|
place_type=record.get('place_type') or None,
|
|
room_number=record.get('room_number') or None,
|
|
notes=record.get('notes') or None
|
|
)
|
|
self.db.add(device)
|
|
created_count += 1
|
|
|
|
self.db.commit()
|
|
return {'success': created_count + updated_count, 'created': created_count, 'updated': updated_count}
|
|
|