118 lines
4.2 KiB
Python
118 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({'record': record, 'error': f'第{idx+2}行: 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({'record': record, 'error': f'第{idx+2}行: 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:
|
|
"""批量导入设备，存在则更新，不存在则新增"""
|
|
success_count = 0
|
|
skip_count = 0
|
|
invalid_count = 0
|
|
|
|
for record in records:
|
|
mac = record.get('mac_address', '')
|
|
if not mac:
|
|
skip_count += 1
|
|
continue
|
|
|
|
# 查询是否已存在该 MAC 地址
|
|
existing = self.db.query(ONUDevice).filter(ONUDevice.mac_address == mac).first()
|
|
|
|
if existing:
|
|
# 更新现有记录
|
|
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
|
|
success_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)
|
|
success_count += 1
|
|
|
|
self.db.commit()
|
|
return {'success': success_count}
|
|
|