RailWise KB已发布
数据导出与格式转换模板
RailWise监测数据导出与格式转换标准化代码模板，支持CSV、Excel、JSON、GeoJSON、Parquet等多种格式的数据导出，实现监测数据的标准化交换与归档。
template-doc
数据导出与格式转换模板
Section titled “数据导出与格式转换模板”1. 模板概述
Section titled “1. 模板概述”1.1 用途
Section titled “1.1 用途”本模板提供监测数据从内部标准化格式到多种外部格式的完整导出与转换流程，支持：
- 结构化格式：CSV、Excel（.xlsx）、Parquet（列式存储）
- 数据交换格式：JSON、XML、YAML
- 地理信息格式：GeoJSON（GIS集成）
- 专业格式：监测日报格式、CASS格式、南方CASS格式
- 压缩归档：ZIP压缩、加密导出
1.2 适用场景
Section titled “1.2 适用场景”| 场景 | 目标格式 | 用途 | 数据量 |
|---|---|---|---|
| 日常数据导出 | CSV/Excel | 人工分析、报表制作 | 中小 |
| 系统间数据交换 | JSON/XML | API对接、第三方系统 | 中小 |
| GIS可视化 | GeoJSON | 地图展示、空间分析 | 中 |
| 大数据归档 | Parquet | 历史数据存储、分析 | 大 |
| 项目交付 | ZIP | 完整数据包交付 | 大 |
1.3 输入输出
Section titled “1.3 输入输出”- 输入：监测数据库记录、项目配置、导出参数
- 输出：各种格式的数据文件、导出日志、校验报告
- 中间产物：数据筛选结果、格式转换中间文件
2. 代码示例
Section titled “2. 代码示例”2.1 核心导出引擎
Section titled “2.1 核心导出引擎”"""RailWise 数据导出与格式转换引擎支持多种格式的监测数据导出与转换
作者: RailWise技术部版本: 1.0.0"""
import csvimport jsonimport loggingimport zipfilefrom abc import ABC, abstractmethodfrom dataclasses import dataclass, fieldfrom datetime import datetime, timezonefrom enum import Enumfrom pathlib import Pathfrom typing import Any, Dict, List, Optional, Union, Iterator
import pandas as pdimport pyarrow as paimport pyarrow.parquet as pqfrom openpyxl import Workbookfrom openpyxl.styles import Font, Alignment, PatternFill, Border, Sidefrom openpyxl.utils.dataframe import dataframe_to_rows
# 配置日志logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")logger = logging.getLogger("RailWise.DataExport")
class ExportFormat(Enum): """导出格式枚举""" CSV = "csv" EXCEL = "excel" JSON = "json" GEOJSON = "geojson" PARQUET = "parquet" XML = "xml" YAML = "yaml" CASS = "cass" ZIP = "zip"
class ExportMode(Enum): """导出模式枚举""" FULL = "full" # 全量导出 INCREMENTAL = "incremental" # 增量导出 FILTERED = "filtered" # 条件筛选导出 AGGREGATED = "aggregated" # 聚合导出
@dataclassclass ExportConfig: """导出配置""" project_id: str export_format: ExportFormat export_mode: ExportMode = ExportMode.FULL output_path: Optional[Path] = None start_time: Optional[datetime] = None end_time: Optional[datetime] = None point_ids: Optional[List[str]] = None point_types: Optional[List[str]] = None include_raw: bool = False include_metadata: bool = True compress: bool = False encrypt: bool = False password: Optional[str] = None encoding: str = "utf-8" delimiter: str = "," sheet_name: str = "监测数据"
@dataclassclass ExportResult: """导出结果""" success: bool file_path: Optional[Path] = None records_count: int = 0 file_size_bytes: int = 0 processing_time_ms: float = 0.0 checksum: Optional[str] = None errors: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list)
class DataExporter(ABC): """数据导出器抽象基类"""
def __init__(self, config: ExportConfig): self.config = config self.logger = logging.getLogger(self.__class__.__name__)
@abstractmethod def export(self, data: List[Dict[str, Any]]) -> ExportResult: """执行导出""" pass
@abstractmethod def validate_data(self, data: List[Dict[str, Any]]) -> bool: """验证数据格式""" pass
def preprocess_data(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """数据预处理""" # 过滤时间范围 if self.config.start_time or self.config.end_time: data = self._filter_by_time(data)
# 过滤测点 if self.config.point_ids: data = [r for r in data if r.get('point_id') in self.config.point_ids]
# 过滤类型 if self.config.point_types: data = [r for r in data if r.get('point_type') in self.config.point_types]
return data
def _filter_by_time(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """按时间范围过滤""" filtered = [] for record in data: obs_time = record.get('observation_time') if isinstance(obs_time, str): obs_time = datetime.fromisoformat(obs_time.replace('Z', '+00:00'))
if self.config.start_time and obs_time < self.config.start_time: continue if self.config.end_time and obs_time > self.config.end_time: continue
filtered.append(record)
return filtered
def _get_output_path(self, extension: str) -> Path: """获取输出文件路径""" if self.config.output_path: return self.config.output_path
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{self.config.project_id}_{self.config.export_mode.value}_{timestamp}.{extension}" return Path("./exports") / filename
class CSVExporter(DataExporter): """CSV格式导出器"""
def validate_data(self, data: List[Dict[str, Any]]) -> bool: """验证数据是否可以导出为CSV""" if not data: return False
# 检查所有记录是否有相同的键 keys = set(data[0].keys()) for record in data[1:10]: # 抽样检查 if set(record.keys()) != keys: self.logger.warning("数据字段不一致，CSV导出可能有问题") return True # 仍然允许导出
return True
def export(self, data: List[Dict[str, Any]]) -> ExportResult: """导出为CSV格式""" import time start_time = time.time()
result = ExportResult(success=False)
try: # 预处理数据 data = self.preprocess_data(data)
if not data: result.errors.append("无数据可导出") return result
# 确定输出路径 output_path = self._get_output_path("csv") output_path.parent.mkdir(parents=True, exist_ok=True)
# 处理数据（扁平化嵌套结构） flat_data = self._flatten_data(data)
# 写入CSV with open(output_path, 'w', newline='', encoding=self.config.encoding) as f: if flat_data: writer = csv.DictWriter( f, fieldnames=flat_data[0].keys(), delimiter=self.config.delimiter ) writer.writeheader() writer.writerows(flat_data)
# 计算结果 result.success = True result.file_path = output_path result.records_count = len(flat_data) result.file_size_bytes = output_path.stat().st_size result.processing_time_ms = (time.time() - start_time) * 1000
# 计算校验和 import hashlib with open(output_path, 'rb') as f: result.checksum = hashlib.md5(f.read()).hexdigest()
self.logger.info(f"CSV导出完成: {output_path}, {result.records_count}条记录")
except Exception as e: result.errors.append(str(e)) self.logger.error(f"CSV导出失败: {e}")
return result
def _flatten_data(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """扁平化嵌套数据结构""" flat_records = []
for record in data: flat_record = {}
for key, value in record.items(): if isinstance(value, dict): # 嵌套字典，展平为 key.subkey for sub_key, sub_value in value.items(): flat_record[f"{key}.{sub_key}"] = sub_value elif isinstance(value, (datetime,)): flat_record[key] = value.isoformat() elif isinstance(value, (list, tuple)): flat_record[key] = json.dumps(value, ensure_ascii=False) else: flat_record[key] = value
flat_records.append(flat_record)
return flat_records
class ExcelExporter(DataExporter): """Excel格式导出器"""
def validate_data(self, data: List[Dict[str, Any]]) -> bool: """验证数据是否可以导出为Excel""" return len(data) <= 1048576 # Excel行数限制
def export(self, data: List[Dict[str, Any]]) -> ExportResult: """导出为Excel格式""" import time start_time = time.time()
result = ExportResult(success=False)
try: data = self.preprocess_data(data)
if not data: result.errors.append("无数据可导出") return result
output_path = self._get_output_path("xlsx") output_path.parent.mkdir(parents=True, exist_ok=True)
# 转换为DataFrame df = pd.DataFrame(data)
# 创建Excel工作簿 wb = Workbook() ws = wb.active ws.title = self.config.sheet_name
# 写入数据 for r_idx, row in enumerate(dataframe_to_rows(df, index=False, header=True), 1): for c_idx, value in enumerate(row, 1): cell = ws.cell(row=r_idx, column=c_idx, value=value)
# 设置表头样式 if r_idx == 1: cell.font = Font(bold=True, color="FFFFFF") cell.fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid") cell.alignment = Alignment(horizontal="center", vertical="center")
# 设置数据行样式 else: cell.alignment = Alignment(horizontal="left", vertical="center")
# 交替行颜色 if r_idx % 2 == 0: cell.fill = PatternFill(start_color="F2F2F2", end_color="F2F2F2", fill_type="solid")
# 设置列宽 for column in ws.columns: max_length = 0 column_letter = column[0].column_letter for cell in column: try: if cell.value: max_length = max(max_length, len(str(cell.value))) except: pass adjusted_width = min(max_length + 2, 50) ws.column_dimensions[column_letter].width = adjusted_width
# 添加边框 thin_border = Border( left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'), bottom=Side(style='thin') )
for row in ws.iter_rows(): for cell in row: cell.border = thin_border
# 冻结首行 ws.freeze_panes = 'A2'
# 添加元数据工作表 if self.config.include_metadata: meta_ws = wb.create_sheet("元数据") metadata = [ ["项目编号", self.config.project_id], ["导出时间", datetime.now().isoformat()], ["导出格式", "Excel"], ["记录数", str(len(data))], ["导出模式", self.config.export_mode.value], ]
if self.config.start_time: metadata.append(["开始时间", self.config.start_time.isoformat()]) if self.config.end_time: metadata.append(["结束时间", self.config.end_time.isoformat()])
for row_idx, (key, value) in enumerate(metadata, 1): meta_ws.cell(row=row_idx, column=1, value=key).font = Font(bold=True) meta_ws.cell(row=row_idx, column=2, value=value)
meta_ws.column_dimensions['A'].width = 20 meta_ws.column_dimensions['B'].width = 40
# 保存 wb.save(str(output_path))
result.success = True result.file_path = output_path result.records_count = len(data) result.file_size_bytes = output_path.stat().st_size result.processing_time_ms = (time.time() - start_time) * 1000
self.logger.info(f"Excel导出完成: {output_path}")
except Exception as e: result.errors.append(str(e)) self.logger.error(f"Excel导出失败: {e}")
return result
class JSONExporter(DataExporter): """JSON格式导出器"""
def validate_data(self, data: List[Dict[str, Any]]) -> bool: """验证数据是否可以导出为JSON""" try: json.dumps(data, ensure_ascii=False, default=str) return True except (TypeError, ValueError): return False
def export(self, data: List[Dict[str, Any]]) -> ExportResult: """导出为JSON格式""" import time start_time = time.time()
result = ExportResult(success=False)
try: data = self.preprocess_data(data)
if not data: result.errors.append("无数据可导出") return result
output_path = self._get_output_path("json") output_path.parent.mkdir(parents=True, exist_ok=True)
# 构建JSON结构 export_data = { "metadata": { "project_id": self.config.project_id, "export_time": datetime.now().isoformat(), "export_format": "JSON", "records_count": len(data), "export_mode": self.config.export_mode.value }, "data": data }
# 写入JSON with open(output_path, 'w', encoding=self.config.encoding) as f: json.dump(export_data, f, ensure_ascii=False, indent=2, default=str)
result.success = True result.file_path = output_path result.records_count = len(data) result.file_size_bytes = output_path.stat().st_size result.processing_time_ms = (time.time() - start_time) * 1000
self.logger.info(f"JSON导出完成: {output_path}")
except Exception as e: result.errors.append(str(e)) self.logger.error(f"JSON导出失败: {e}")
return result
class GeoJSONExporter(DataExporter): """GeoJSON格式导出器（GIS集成）"""
def validate_data(self, data: List[Dict[str, Any]]) -> bool: """验证数据是否包含坐标信息""" if not data: return False
# 检查是否包含坐标字段 required_coords = ['longitude', 'latitude'] sample = data[0]
has_coords = all(coord in sample for coord in required_coords) if not has_coords: self.logger.warning("数据缺少坐标字段，GeoJSON导出将不包含几何信息")
return True
def export(self, data: List[Dict[str, Any]]) -> ExportResult: """导出为GeoJSON格式""" import time start_time = time.time()
result = ExportResult(success=False)
try: data = self.preprocess_data(data)
if not data: result.errors.append("无数据可导出") return result
output_path = self._get_output_path("geojson") output_path.parent.mkdir(parents=True, exist_ok=True)
# 构建GeoJSON FeatureCollection features = []
for record in data: # 提取坐标 longitude = record.get('longitude') or record.get('x') latitude = record.get('latitude') or record.get('y') elevation = record.get('elevation') or record.get('z')
# 构建几何对象 if longitude and latitude: geometry = { "type": "Point", "coordinates": [ float(longitude), float(latitude), float(elevation) if elevation else 0 ] } else: geometry = None
# 构建属性 properties = {k: v for k, v in record.items() if k not in ['longitude', 'latitude', 'elevation', 'x', 'y', 'z']}
feature = { "type": "Feature", "geometry": geometry, "properties": properties }
features.append(feature)
geojson = { "type": "FeatureCollection", "metadata": { "project_id": self.config.project_id, "export_time": datetime.now().isoformat(), "records_count": len(data) }, "features": features }
# 写入GeoJSON with open(output_path, 'w', encoding=self.config.encoding) as f: json.dump(geojson, f, ensure_ascii=False, indent=2, default=str)
result.success = True result.file_path = output_path result.records_count = len(data) result.file_size_bytes = output_path.stat().st_size result.processing_time_ms = (time.time() - start_time) * 1000
self.logger.info(f"GeoJSON导出完成: {output_path}")
except Exception as e: result.errors.append(str(e)) self.logger.error(f"GeoJSON导出失败: {e}")
return result
class ParquetExporter(DataExporter): """Parquet格式导出器（大数据归档）"""
def validate_data(self, data: List[Dict[str, Any]]) -> bool: """验证数据是否可以导出为Parquet""" try: df = pd.DataFrame(data) # 尝试转换为Arrow表 pa.Table.from_pandas(df) return True except Exception: return False
def export(self, data: List[Dict[str, Any]]) -> ExportResult: """导出为Parquet格式""" import time start_time = time.time()
result = ExportResult(success=False)
try: data = self.preprocess_data(data)
if not data: result.errors.append("无数据可导出") return result
output_path = self._get_output_path("parquet") output_path.parent.mkdir(parents=True, exist_ok=True)
# 转换为DataFrame df = pd.DataFrame(data)
# 转换为Arrow表并写入Parquet table = pa.Table.from_pandas(df)
pq.write_table( table, str(output_path), compression='zstd', # 使用ZSTD压缩 row_group_size=10000 )
result.success = True result.file_path = output_path result.records_count = len(data) result.file_size_bytes = output_path.stat().st_size result.processing_time_ms = (time.time() - start_time) * 1000
self.logger.info(f"Parquet导出完成: {output_path}")
except Exception as e: result.errors.append(str(e)) self.logger.error(f"Parquet导出失败: {e}")
return result
class ZipExporter(DataExporter): """ZIP压缩导出器"""
def __init__(self, config: ExportConfig, files_to_archive: List[Path]): super().__init__(config) self.files_to_archive = files_to_archive
def validate_data(self, data: List[Dict[str, Any]]) -> bool: """验证要归档的文件是否存在""" return all(f.exists() for f in self.files_to_archive)
def export(self, data: List[Dict[str, Any]]) -> ExportResult: """导出为ZIP压缩包""" import time start_time = time.time()
result = ExportResult(success=False)
try: output_path = self._get_output_path("zip") output_path.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zf: for file_path in self.files_to_archive: arcname = file_path.name zf.write(file_path, arcname)
# 添加密码保护（如果需要） if self.config.encrypt and self.config.password: # 注意：标准zipfile不支持AES加密，需使用pyzipper等库 pass
result.success = True result.file_path = output_path result.records_count = len(self.files_to_archive) result.file_size_bytes = output_path.stat().st_size result.processing_time_ms = (time.time() - start_time) * 1000
self.logger.info(f"ZIP导出完成: {output_path}")
except Exception as e: result.errors.append(str(e)) self.logger.error(f"ZIP导出失败: {e}")
return result
# 导出器工厂def create_exporter(config: ExportConfig) -> DataExporter: """创建导出器实例""" exporters = { ExportFormat.CSV: CSVExporter, ExportFormat.EXCEL: ExcelExporter, ExportFormat.JSON: JSONExporter, ExportFormat.GEOJSON: GeoJSONExporter, ExportFormat.PARQUET: ParquetExporter, }
exporter_class = exporters.get(config.export_format) if not exporter_class: raise ValueError(f"不支持的导出格式: {config.export_format}")
return exporter_class(config)2.2 批量导出与转换模块
Section titled “2.2 批量导出与转换模块”"""批量导出与格式转换模块"""
from concurrent.futures import ThreadPoolExecutorfrom typing import List, Dict
class BatchExporter: """批量导出器"""
def __init__(self, max_workers: int = 4): self.max_workers = max_workers self.logger = logging.getLogger("BatchExporter")
def export_multiple_formats( self, data: List[Dict[str, Any]], project_id: str, formats: List[ExportFormat], base_path: Path = Path("./exports") ) -> Dict[ExportFormat, ExportResult]: """ 同时导出多种格式
Args: data: 要导出的数据 project_id: 项目编号 formats: 要导出的格式列表 base_path: 基础输出路径
Returns: 各格式的导出结果 """ results = {}
with ThreadPoolExecutor(max_workers=self.max_workers) as executor: futures = {}
for fmt in formats: config = ExportConfig( project_id=project_id, export_format=fmt, output_path=base_path / f"{project_id}.{fmt.value}" )
exporter = create_exporter(config) future = executor.submit(exporter.export, data) futures[future] = fmt
for future in futures: fmt = futures[future] try: result = future.result() results[fmt] = result except Exception as e: self.logger.error(f"{fmt.value}导出失败: {e}") results[fmt] = ExportResult(success=False, errors=[str(e)])
return results
def create_export_package( self, data: List[Dict[str, Any]], project_id: str, formats: List[ExportFormat] = None, include_raw: bool = False ) -> Path: """ 创建完整导出包（包含多种格式）
Args: data: 要导出的数据 project_id: 项目编号 formats: 要包含的格式（默认全部） include_raw: 是否包含原始数据
Returns: ZIP文件路径 """ if formats is None: formats = [ExportFormat.CSV, ExportFormat.EXCEL, ExportFormat.JSON]
base_path = Path("./exports") / project_id base_path.mkdir(parents=True, exist_ok=True)
# 导出各格式 results = self.export_multiple_formats(data, project_id, formats, base_path)
# 收集成功导出的文件 files_to_archive = [] for fmt, result in results.items(): if result.success and result.file_path: files_to_archive.append(result.file_path)
# 创建ZIP包 if files_to_archive: zip_config = ExportConfig( project_id=project_id, export_format=ExportFormat.ZIP, output_path=base_path.parent / f"{project_id}_complete.zip" )
zip_exporter = ZipExporter(zip_config, files_to_archive) zip_result = zip_exporter.export(data)
if zip_result.success: return zip_result.file_path
return None
def convert_format( input_path: Path, input_format: ExportFormat, output_format: ExportFormat, output_path: Optional[Path] = None) -> ExportResult: """ 格式转换函数
Args: input_path: 输入文件路径 input_format: 输入格式 output_format: 输出格式 output_path: 输出路径（可选）
Returns: 导出结果 """ # 读取数据 if input_format == ExportFormat.CSV: data = pd.read_csv(input_path).to_dict('records') elif input_format == ExportFormat.JSON: with open(input_path, 'r', encoding='utf-8') as f: json_data = json.load(f) data = json_data.get('data', json_data) elif input_format == ExportFormat.EXCEL: data = pd.read_excel(input_path).to_dict('records') else: raise ValueError(f"不支持的输入格式: {input_format}")
# 导出为指定格式 project_id = "converted" config = ExportConfig( project_id=project_id, export_format=output_format, output_path=output_path )
exporter = create_exporter(config) return exporter.export(data)3. 参数说明
Section titled “3. 参数说明”3.1 ExportConfig 导出配置参数
Section titled “3.1 ExportConfig 导出配置参数”| 参数名 | 类型 | 必填 | 默认值 | 说明 |
|---|---|---|---|---|
project_id |
str | 是 | - | 项目编号 |
export_format |
ExportFormat | 是 | - | 导出格式 |
export_mode |
ExportMode | 否 | FULL | 导出模式（全量/增量/筛选/聚合） |
output_path |
Path | 否 | None | 输出文件路径（自动命名） |
start_time |
datetime | 否 | None | 开始时间筛选 |
end_time |
datetime | 否 | None | 结束时间筛选 |
point_ids |
List[str] | 否 | None | 测点编号筛选 |
point_types |
List[str] | 否 | None | 测点类型筛选 |
include_raw |
bool | 否 | False | 是否包含原始数据 |
include_metadata |
bool | 否 | True | 是否包含元数据 |
compress |
bool | 否 | False | 是否压缩 |
encrypt |
bool | 否 | False | 是否加密 |
password |
str | 否 | None | 加密密码 |
encoding |
str | 否 | utf-8 | 文件编码 |
delimiter |
str | 否 | , | CSV分隔符 |
sheet_name |
str | 否 | 监测数据 | Excel工作表名称 |
3.2 ExportResult 导出结果参数
Section titled “3.2 ExportResult 导出结果参数”| 参数名 | 类型 | 说明 |
|---|---|---|
success |
bool | 导出是否成功 |
file_path |
Path | 导出文件路径 |
records_count |
int | 导出记录数 |
file_size_bytes |
int | 文件大小（字节） |
processing_time_ms |
float | 处理耗时（毫秒） |
checksum |
str | 文件MD5校验和 |
errors |
List[str] | 错误信息列表 |
warnings |
List[str] | 警告信息列表 |
4. 使用示例
Section titled “4. 使用示例”4.1 基本导出流程
Section titled “4.1 基本导出流程”from pathlib import Pathfrom template_data_export import ( ExportConfig, ExportFormat, ExportMode, create_exporter)
# 1. 准备数据（模拟从数据库查询）sample_data = [ { "record_id": "RW001_001_20240708080000", "project_id": "RW-2024-001", "point_id": "JC-01", "point_name": "3号线K12+350左线", "observation_time": "2024-07-08T08:00:00+00:00", "value_x": 0.15, "value_y": -0.05, "value_z": 0.01, "longitude": 121.55, "latitude": 29.87, "alert_level": "正常" }, { "record_id": "RW001_002_20240708080000", "project_id": "RW-2024-001", "point_id": "JC-02", "point_name": "3号线K12+350右线", "observation_time": "2024-07-08T08:00:00+00:00", "value_x": 0.22, "value_y": -0.03, "value_z": 0.01, "longitude": 121.55, "latitude": 29.87, "alert_level": "正常" }]
# 2. 配置导出参数config = ExportConfig( project_id="RW-2024-001", export_format=ExportFormat.CSV, output_path=Path("./exports/monitoring_data.csv"))
# 3. 创建导出器并执行导出exporter = create_exporter(config)result = exporter.export(sample_data)
if result.success: print(f"导出成功: {result.file_path}") print(f"记录数: {result.records_count}") print(f"文件大小: {result.file_size_bytes} bytes") print(f"校验和: {result.checksum}")else: print(f"导出失败: {result.errors}")4.2 导出为Excel（带样式）
Section titled “4.2 导出为Excel（带样式）”from template_data_export import ExportConfig, ExportFormat
config = ExportConfig( project_id="RW-2024-001", export_format=ExportFormat.EXCEL, output_path=Path("./exports/monitoring_report.xlsx"), sheet_name="监测日报", include_metadata=True)
exporter = create_exporter(config)result = exporter.export(sample_data)
if result.success: print(f"Excel导出成功: {result.file_path}")4.3 导出为GeoJSON（GIS集成）
Section titled “4.3 导出为GeoJSON（GIS集成）”from template_data_export import ExportConfig, ExportFormat
# 确保数据包含经纬度坐标geo_data = [ { "point_id": "JC-01", "point_name": "3号线K12+350左线", "longitude": 121.550123, "latitude": 29.870456, "elevation": 5.23, "value_x": 0.15, "value_y": -0.05, "value_z": 0.01, "observation_time": "2024-07-08T08:00:00+00:00" }]
config = ExportConfig( project_id="RW-2024-001", export_format=ExportFormat.GEOJSON, output_path=Path("./exports/monitoring_geo.json"))
exporter = create_exporter(config)result = exporter.export(geo_data)
if result.success: print(f"GeoJSON导出成功，可在QGIS/ArcGIS中打开")4.4 批量导出多种格式
Section titled “4.4 批量导出多种格式”from template_data_export import BatchExporter, ExportFormat
# 创建批量导出器batch_exporter = BatchExporter(max_workers=4)
# 同时导出CSV、Excel、JSONformats = [ExportFormat.CSV, ExportFormat.EXCEL, ExportFormat.JSON]results = batch_exporter.export_multiple_formats( data=sample_data, project_id="RW-2024-001", formats=formats)
# 查看各格式导出结果for fmt, result in results.items(): status = "成功" if result.success else "失败" print(f"{fmt.value}: {status} - {result.file_path}")4.5 创建完整导出包
Section titled “4.5 创建完整导出包”from template_data_export import BatchExporter, ExportFormat
# 创建包含多种格式的ZIP包batch_exporter = BatchExporter()
zip_path = batch_exporter.create_export_package( data=sample_data, project_id="RW-2024-001", formats=[ExportFormat.CSV, ExportFormat.EXCEL, ExportFormat.JSON, ExportFormat.GEOJSON], include_raw=False)
if zip_path: print(f"导出包已创建: {zip_path}") print(f"文件大小: {zip_path.stat().st_size / 1024 / 1024:.2f} MB")4.6 格式转换
Section titled “4.6 格式转换”from template_data_export import convert_format, ExportFormatfrom pathlib import Path
# CSV转Excelresult = convert_format( input_path=Path("./exports/monitoring_data.csv"), input_format=ExportFormat.CSV, output_format=ExportFormat.EXCEL, output_path=Path("./exports/converted.xlsx"))
if result.success: print(f"转换成功: {result.file_path}")5. 输出示例
Section titled “5. 输出示例”5.1 CSV输出示例
Section titled “5.1 CSV输出示例”record_id,project_id,point_id,point_name,observation_time,value_x,value_y,value_z,longitude,latitude,alert_levelRW001_001_20240708080000,RW-2024-001,JC-01,3号线K12+350左线,2024-07-08T08:00:00+00:00,0.15,-0.05,0.01,121.55,29.87,正常RW001_002_20240708080000,RW-2024-001,JC-02,3号线K12+350右线,2024-07-08T08:00:00+00:00,0.22,-0.03,0.01,121.55,29.87,正常5.2 JSON输出示例
Section titled “5.2 JSON输出示例”{ "metadata": { "project_id": "RW-2024-001", "export_time": "2024-07-08T12:00:00+00:00", "export_format": "JSON", "records_count": 2, "export_mode": "full" }, "data": [ { "record_id": "RW001_001_20240708080000", "project_id": "RW-2024-001", "point_id": "JC-01", "value_x": 0.15, "value_y": -0.05, "value_z": 0.01 } ]}5.3 GeoJSON输出示例
Section titled “5.3 GeoJSON输出示例”{ "type": "FeatureCollection", "metadata": { "project_id": "RW-2024-001", "export_time": "2024-07-08T12:00:00+00:00", "records_count": 2 }, "features": [ { "type": "Feature", "geometry": { "type": "Point", "coordinates": [121.550123, 29.870456, 5.23] }, "properties": { "point_id": "JC-01", "point_name": "3号线K12+350左线", "value_x": 0.15, "value_y": -0.05, "value_z": 0.01 } } ]}6. 扩展指南
Section titled “6. 扩展指南”6.1 添加新的导出格式
Section titled “6.1 添加新的导出格式”from template_data_export import DataExporter, ExportConfig, ExportResult
class KMLExporter(DataExporter): """KML格式导出器（Google Earth）"""
def validate_data(self, data): return all('longitude' in r and 'latitude' in r for r in data)
def export(self, data): # 实现KML导出逻辑 kml_content = '<?xml version="1.0" encoding="UTF-8"?>\n' kml_content += '<kml xmlns="http://www.opengis.net/kml/2.2">\n' kml_content += '<Document>\n'
for record in data: kml_content += f'<Placemark>\n' kml_content += f'<name>{record.get("point_id")}</name>\n' kml_content += f'<Point>\n' kml_content += f'<coordinates>{record["longitude"]},{record["latitude"]},0</coordinates>\n' kml_content += f'</Point>\n' kml_content += f'</Placemark>\n'
kml_content += '</Document>\n</kml>'
output_path = self._get_output_path("kml") with open(output_path, 'w', encoding='utf-8') as f: f.write(kml_content)
return ExportResult(success=True, file_path=output_path, records_count=len(data))
# 注册到工厂from template_data_export import create_exporter
# 扩展工厂函数def create_exporter_extended(config: ExportConfig): if config.export_format.value == "kml": return KMLExporter(config) return create_exporter(config)6.2 自定义数据筛选
Section titled “6.2 自定义数据筛选”class FilteredExporter(CSVExporter): """自定义筛选导出器"""
def preprocess_data(self, data): # 先执行标准预处理 data = super().preprocess_data(data)
# 添加自定义筛选：只导出预警测点 data = [r for r in data if r.get('alert_level') != '正常']
# 添加自定义排序：按变化量从大到小 data.sort(key=lambda r: abs(r.get('value_d', 0)), reverse=True)
return data6.3 集成到数据导入流程
Section titled “6.3 集成到数据导入流程”from template_data_import import MonitoringDatabasefrom template_data_export import ExportConfig, ExportFormat, create_exporter
def export_project_data(project_id: str, output_dir: Path): """导出项目全部数据""" # 从数据库查询 db = MonitoringDatabase() records = db.query_by_project(project_id) # 假设有此方法
# 转换为字典列表 data = [r.dict() for r in records]
# 导出为多种格式 for fmt in [ExportFormat.CSV, ExportFormat.EXCEL, ExportFormat.JSON]: config = ExportConfig( project_id=project_id, export_format=fmt, output_path=output_dir / f"{project_id}.{fmt.value}" )
exporter = create_exporter(config) result = exporter.export(data)
if result.success: print(f"{fmt.value}导出成功: {result.file_path}")6.4 定时自动归档
Section titled “6.4 定时自动归档”from apscheduler.schedulers.background import BackgroundSchedulerfrom datetime import datetime, timedelta
def daily_archive(project_id: str): """每日自动归档""" yesterday = datetime.now() - timedelta(days=1)
config = ExportConfig( project_id=project_id, export_format=ExportFormat.PARQUET, export_mode=ExportMode.INCREMENTAL, start_time=yesterday.replace(hour=0, minute=0, second=0), end_time=yesterday.replace(hour=23, minute=59, second=59), output_path=Path(f"./archive/{project_id}/{yesterday.strftime('%Y%m%d')}.parquet") )
# 从数据库获取数据并导出 # ...
# 每日凌晨2点执行归档scheduler = BackgroundScheduler()scheduler.add_job(daily_archive, 'cron', hour=2, minute=0, args=["RW-2024-001"])scheduler.start()7. 常见问题
Section titled “7. 常见问题”Q1: 导出大数据量时内存不足怎么办？
Section titled “Q1: 导出大数据量时内存不足怎么办？”A: 使用分块导出或流式处理：
# 分块导出CSVchunk_size = 10000chunk_idx = 0
for chunk in pd.read_csv(large_file, chunksize=chunk_size): output_path = Path(f"./exports/chunk_{chunk_idx}.csv") chunk.to_csv(output_path, index=False) chunk_idx += 1Q2: 如何导出带密码保护的Excel？
Section titled “Q2: 如何导出带密码保护的Excel？”A: 使用openpyxl的加密功能：
from openpyxl import Workbookfrom openpyxl.worksheet.protection import SheetProtection
wb = Workbook()ws = wb.active
# 设置工作表保护密码ws.protection = SheetProtection(password="your_password")
# 保存wb.save("protected.xlsx")Q3: 中文文件名乱码怎么办？
Section titled “Q3: 中文文件名乱码怎么办？”A: 确保使用UTF-8编码，并在Windows上处理编码问题：
import sysimport os
# Windows系统设置编码if sys.platform == 'win32': os.environ['PYTHONIOENCODING'] = 'utf-8'
# 导出时使用正确的编码with open(output_path, 'w', encoding='utf-8-sig') as f: # utf-8-sig添加BOM，Excel可识别 passQ4: 如何导出符合CASS格式的数据？
Section titled “Q4: 如何导出符合CASS格式的数据？”A: CASS格式有特定要求，需要自定义导出器：
class CASSExporter(DataExporter): """南方CASS格式导出器"""
def export(self, data): # CASS格式: 点号,编码,东坐标,北坐标,高程 lines = [] for record in data: line = f"{record['point_id']},{record.get('code', '')},{record['x']},{record['y']},{record['z']}" lines.append(line)
output_path = self._get_output_path("dat") with open(output_path, 'w', encoding='gbk') as f: # CASS通常使用GBK编码 f.write('\n'.join(lines))
return ExportResult(success=True, file_path=output_path, records_count=len(data))Q5: 如何验证导出数据的完整性？
Section titled “Q5: 如何验证导出数据的完整性？”A: 使用校验和验证：
import hashlib
def verify_export(file_path: Path, expected_checksum: str) -> bool: """验证导出文件完整性""" with open(file_path, 'rb') as f: actual_checksum = hashlib.md5(f.read()).hexdigest()
return actual_checksum == expected_checksum
# 使用result = exporter.export(data)if result.success and result.checksum: is_valid = verify_export(result.file_path, result.checksum) print(f"文件完整性验证: {'通过' if is_valid else '失败'}")相关文档链接
Section titled “相关文档链接”- 监测数据导入标准化模板 — 数据导入与清洗
- 监测报告自动生成模板 — 基于导出数据生成报告
- 批量数据处理模板 — 大规模数据批量处理
- API接口集成模板 — 通过API导出数据
ai_tags: domain: ["工程监测", "数据管理", "数据交换"] technology: ["Python", "Pandas", "OpenPyXL", "PyArrow", "GeoJSON"] standard: ["GB 50497", "CJJ/T 202", "JGJ 8"] product: ["RAILWISE-TSM", "RAILWISE-CLI", "WorkWise"] difficulty: "intermediate" role: ["监测工程师", "数据分析师", "GIS工程师"] task_type: ["数据导出", "格式转换", "数据归档"]