读取DM数据库
This commit is contained in:
4
.env.example
Normal file
4
.env.example
Normal file
@@ -0,0 +1,4 @@
|
||||
# 达梦数据库密码(仅 source=db 时需要)
|
||||
# 复制为 .env 后填入真实密码,或将下面一行加入 ~/.bashrc 等
|
||||
# export DM_PASSWORD=你的密码
|
||||
DM_PASSWORD=
|
||||
5
.gitignore
vendored
Normal file
5
.gitignore
vendored
Normal file
@@ -0,0 +1,5 @@
|
||||
.env
|
||||
*.pyc
|
||||
__pycache__/
|
||||
.venv/
|
||||
venv/
|
||||
18
.vscode/c_cpp_properties.json
vendored
Normal file
18
.vscode/c_cpp_properties.json
vendored
Normal file
@@ -0,0 +1,18 @@
|
||||
{
|
||||
"configurations": [
|
||||
{
|
||||
"name": "linux-gcc-x64",
|
||||
"includePath": [
|
||||
"${workspaceFolder}/**"
|
||||
],
|
||||
"compilerPath": "/usr/bin/gcc",
|
||||
"cStandard": "${default}",
|
||||
"cppStandard": "${default}",
|
||||
"intelliSenseMode": "linux-gcc-x64",
|
||||
"compilerArgs": [
|
||||
""
|
||||
]
|
||||
}
|
||||
],
|
||||
"version": 4
|
||||
}
|
||||
24
.vscode/launch.json
vendored
Normal file
24
.vscode/launch.json
vendored
Normal file
@@ -0,0 +1,24 @@
|
||||
{
|
||||
"version": "0.2.0",
|
||||
"configurations": [
|
||||
{
|
||||
"name": "C/C++ Runner: Debug Session",
|
||||
"type": "cppdbg",
|
||||
"request": "launch",
|
||||
"args": [],
|
||||
"stopAtEntry": false,
|
||||
"externalConsole": false,
|
||||
"cwd": "/home/huangfukk/module1_3/.vscode",
|
||||
"program": "/home/huangfukk/module1_3/.vscode/build/Debug/outDebug",
|
||||
"MIMode": "gdb",
|
||||
"miDebuggerPath": "gdb",
|
||||
"setupCommands": [
|
||||
{
|
||||
"description": "Enable pretty-printing for gdb",
|
||||
"text": "-enable-pretty-printing",
|
||||
"ignoreFailures": true
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
59
.vscode/settings.json
vendored
Normal file
59
.vscode/settings.json
vendored
Normal file
@@ -0,0 +1,59 @@
|
||||
{
|
||||
"C_Cpp_Runner.cCompilerPath": "gcc",
|
||||
"C_Cpp_Runner.cppCompilerPath": "g++",
|
||||
"C_Cpp_Runner.debuggerPath": "gdb",
|
||||
"C_Cpp_Runner.cStandard": "",
|
||||
"C_Cpp_Runner.cppStandard": "",
|
||||
"C_Cpp_Runner.msvcBatchPath": "",
|
||||
"C_Cpp_Runner.useMsvc": false,
|
||||
"C_Cpp_Runner.warnings": [
|
||||
"-Wall",
|
||||
"-Wextra",
|
||||
"-Wpedantic",
|
||||
"-Wshadow",
|
||||
"-Wformat=2",
|
||||
"-Wcast-align",
|
||||
"-Wconversion",
|
||||
"-Wsign-conversion",
|
||||
"-Wnull-dereference"
|
||||
],
|
||||
"C_Cpp_Runner.msvcWarnings": [
|
||||
"/W4",
|
||||
"/permissive-",
|
||||
"/w14242",
|
||||
"/w14287",
|
||||
"/w14296",
|
||||
"/w14311",
|
||||
"/w14826",
|
||||
"/w44062",
|
||||
"/w44242",
|
||||
"/w14905",
|
||||
"/w14906",
|
||||
"/w14263",
|
||||
"/w44265",
|
||||
"/w14928"
|
||||
],
|
||||
"C_Cpp_Runner.enableWarnings": true,
|
||||
"C_Cpp_Runner.warningsAsError": false,
|
||||
"C_Cpp_Runner.compilerArgs": [],
|
||||
"C_Cpp_Runner.linkerArgs": [],
|
||||
"C_Cpp_Runner.includePaths": [],
|
||||
"C_Cpp_Runner.includeSearch": [
|
||||
"*",
|
||||
"**/*"
|
||||
],
|
||||
"C_Cpp_Runner.excludeSearch": [
|
||||
"**/build",
|
||||
"**/build/**",
|
||||
"**/.*",
|
||||
"**/.*/**",
|
||||
"**/.vscode",
|
||||
"**/.vscode/**"
|
||||
],
|
||||
"C_Cpp_Runner.useAddressSanitizer": false,
|
||||
"C_Cpp_Runner.useUndefinedSanitizer": false,
|
||||
"C_Cpp_Runner.useLeakSanitizer": false,
|
||||
"C_Cpp_Runner.showCompilationTime": false,
|
||||
"C_Cpp_Runner.useLinkTimeOptimization": false,
|
||||
"C_Cpp_Runner.msvcSecureNoWarnings": false
|
||||
}
|
||||
58
README.md
58
README.md
@@ -6,13 +6,20 @@
|
||||
## 目录结构
|
||||
```
|
||||
project/
|
||||
├── config/
|
||||
│ ├── config.yaml # 配置文件(推荐,支持注释)
|
||||
│ └── config.json # 配置文件备选
|
||||
├── data/
|
||||
│ ├── raw_data_sample.json # 原始输入数据(包含噪声和错误)
|
||||
│ ├── raw_data_sample.json # 原始输入数据(JSON 模式)
|
||||
│ └── cleaned_data_final.json # 清洗后的输出数据
|
||||
├── src/
|
||||
│ └── cleaner.py # 清洗运行程序
|
||||
│ ├── cleaner.py # 清洗运行程序
|
||||
│ └── data_loader.py # 数据加载(JSON/达梦)
|
||||
├── report/
|
||||
│ └── detailed_cleaning_report.json # 详细清洗报告
|
||||
├── scripts/
|
||||
│ └── inspect_dm_tables.py # 达梦表结构查看(部署时用)
|
||||
├── .env.example # 环境变量示例(含 DM_PASSWORD)
|
||||
└── README.md # 项目说明文档
|
||||
```
|
||||
|
||||
@@ -50,15 +57,54 @@ project/
|
||||
- 时间戳统一为 ISO-8601 格式
|
||||
- 输出清洗后 JSON 及详细报告(含各步骤的统计与示例)
|
||||
|
||||
## 数据源配置
|
||||
|
||||
程序支持两种数据源,通过 `config/config.yaml` 或 `config/config.json` 中的 `source` 切换:
|
||||
|
||||
| source | 适用场景 | 说明 |
|
||||
|--------|--------------|------|
|
||||
| `json` | 开发、无数据库 | 从 `data/raw_data_sample.json` 读取 |
|
||||
| `db` | 生产、有达梦库 | 从达梦数据库指定表读取 |
|
||||
|
||||
### 开发阶段(当前设备无达梦)
|
||||
|
||||
保持 `source: json`,直接使用本地 JSON 文件即可。
|
||||
|
||||
### 生产阶段(部署到有达梦的设备)
|
||||
|
||||
`OBJECTIVE_INSTANCE` 表结构与 `raw_data_sample.json` 一致,配置中已设为 `db.table: OBJECTIVE_INSTANCE`,直接查询即可。
|
||||
|
||||
1. **修改配置**:`source: db`,端口 `5080`、表 `OBJECTIVE_INSTANCE` 已预设;
|
||||
2. **设置密码**:`export DM_PASSWORD=你的密码`;
|
||||
3. **运行清洗**:`python src/cleaner.py`。
|
||||
|
||||
若需自定义查询(如加 WHERE 条件),可清空 `table`,在 `db.query` 中写完整 SQL。
|
||||
|
||||
### 密码注入与安全说明
|
||||
|
||||
**⚠️ 安全原则:切勿将达梦密码写入配置文件或源代码。**
|
||||
|
||||
达梦模式通过环境变量 `DM_PASSWORD` 注入密码,支持以下方式:
|
||||
|
||||
| 方式 | 说明 |
|
||||
|------|------|
|
||||
| **终端** | `export DM_PASSWORD=你的密码`(Linux/Mac)或 `set DM_PASSWORD=你的密码`(Windows CMD) |
|
||||
| **PyCharm** | Run → Edit Configurations → Environment variables → 添加 `DM_PASSWORD=你的密码` |
|
||||
| **.env 文件** | 复制 `.env.example` 为 `.env`,填入 `DM_PASSWORD=你的密码`,并用 `python-dotenv` 加载(需自行集成) |
|
||||
|
||||
`inspect_dm_tables.py` 支持两种密码来源:优先环境变量 `DM_PASSWORD`,其次脚本内 `LOCAL_PASSWORD`。若使用 `LOCAL_PASSWORD`,务必执行 `git update-index --assume-unchanged scripts/inspect_dm_tables.py` 避免误提交。
|
||||
|
||||
配置文件(`config.yaml` / `config.json`)中仅配置 `host`、`port`、`user`、`schema`、`table` 等,**不包含 password**。本项目源代码中不包含任何硬编码密码。
|
||||
|
||||
## 快速开始
|
||||
|
||||
### 1. 环境依赖
|
||||
本项目仅依赖 Python 标准库及 NumPy/Pandas:
|
||||
|
||||
```bash
|
||||
pip install numpy pandas
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
核心依赖:`numpy`、`pandas`、`PyYAML`。达梦模式需额外安装 `dmPython`。
|
||||
|
||||
### 2. 运行清洗
|
||||
在项目根目录下执行:
|
||||
|
||||
@@ -66,7 +112,7 @@ pip install numpy pandas
|
||||
python src/cleaner.py
|
||||
```
|
||||
|
||||
程序会读取 `data/raw_data_sample.json`,处理后生成:
|
||||
默认从 JSON 读取,处理后生成:
|
||||
- `data/cleaned_data_final.json`(清洗后的数据)
|
||||
- `report/detailed_cleaning_report.json`(详细清洗报告)
|
||||
|
||||
|
||||
16
config/config.json
Normal file
16
config/config.json
Normal file
@@ -0,0 +1,16 @@
|
||||
{
|
||||
"source": "json",
|
||||
"json": {
|
||||
"input_file": "data/raw_data_sample.json",
|
||||
"output_file": "data/cleaned_data_final.json",
|
||||
"report_file": "report/detailed_cleaning_report.json"
|
||||
},
|
||||
"db": {
|
||||
"host": "127.0.0.1",
|
||||
"port": 5080,
|
||||
"user": "SYSDBA",
|
||||
"schema": "SYSDBA",
|
||||
"table": "OBJECTIVE_INSTANCE",
|
||||
"query": null
|
||||
}
|
||||
}
|
||||
21
config/config.yaml
Normal file
21
config/config.yaml
Normal file
@@ -0,0 +1,21 @@
|
||||
# 数据清洗系统配置文件
|
||||
# 通过 source 切换:json(开发) / db(生产-达梦)
|
||||
|
||||
source: json # 开发用 json;部署达梦时改为 db(或直接改 config.json)
|
||||
|
||||
# JSON 文件模式(source=json 时生效)
|
||||
json:
|
||||
input_file: data/raw_data_sample.json
|
||||
output_file: data/cleaned_data_final.json
|
||||
report_file: report/detailed_cleaning_report.json
|
||||
|
||||
# 达梦数据库模式(source=db 时生效)
|
||||
# 密码通过环境变量 DM_PASSWORD 传入,切勿写入配置文件
|
||||
db:
|
||||
host: 127.0.0.1
|
||||
port: 5080
|
||||
user: SYSDBA
|
||||
schema: SYSDBA
|
||||
# OBJECTIVE_INSTANCE 表结构与 raw_data_sample.json 一致,直接查询即可
|
||||
table: OBJECTIVE_INSTANCE
|
||||
query: null
|
||||
5
requirements.txt
Normal file
5
requirements.txt
Normal file
@@ -0,0 +1,5 @@
|
||||
numpy>=1.20.0
|
||||
pandas>=1.3.0
|
||||
PyYAML>=5.4
|
||||
# 达梦数据库(仅 source=db 时需要,开发阶段可不安)
|
||||
# pip install dmPython
|
||||
102
scripts/inspect_dm_tables.py
Normal file
102
scripts/inspect_dm_tables.py
Normal file
@@ -0,0 +1,102 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
在达梦数据库机器上运行,查看各表结构,用于编写 config 中的 db.query
|
||||
用法: python scripts/inspect_dm_tables.py
|
||||
|
||||
密码获取顺序:环境变量 DM_PASSWORD > 下方 LOCAL_PASSWORD
|
||||
若使用 LOCAL_PASSWORD,切勿提交到 git!可执行:git update-index --assume-unchanged scripts/inspect_dm_tables.py
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
# 可选:在此直接填写密码(仅限本机调试,切勿提交到版本库!)
|
||||
LOCAL_PASSWORD = None # 例如: "你的密码"
|
||||
|
||||
# 支持从项目根或 scripts 目录运行
|
||||
_script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
_project_root = os.path.dirname(_script_dir)
|
||||
if _project_root not in sys.path:
|
||||
sys.path.insert(0, _project_root)
|
||||
|
||||
try:
|
||||
import dmPython
|
||||
except ImportError:
|
||||
print("请先安装: pip install dmPython")
|
||||
sys.exit(1)
|
||||
|
||||
CONFIG = {
|
||||
"user": "SYSDBA",
|
||||
"password": os.environ.get("DM_PASSWORD") or LOCAL_PASSWORD,
|
||||
"server": "127.0.0.1",
|
||||
"port": 5080,
|
||||
"schema": "SYSDBA",
|
||||
}
|
||||
|
||||
TABLES = [
|
||||
"OBJECTIVE_INSTANCE",
|
||||
"METRIC_VALUE_INSTANCE",
|
||||
"METRIC_PROFILE",
|
||||
"ROLE_PROFILE",
|
||||
"TASK_PROFILE",
|
||||
"RELATION_INSTANCE",
|
||||
]
|
||||
|
||||
|
||||
def main():
|
||||
if not CONFIG["password"]:
|
||||
print("请设置环境变量 DM_PASSWORD 或在脚本中填写 LOCAL_PASSWORD")
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
conn = dmPython.connect(
|
||||
user=CONFIG["user"],
|
||||
password=CONFIG["password"],
|
||||
server=CONFIG["server"],
|
||||
port=CONFIG["port"],
|
||||
schema=CONFIG["schema"],
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"连接失败: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
cursor = conn.cursor()
|
||||
print("=" * 60)
|
||||
print("达梦表结构(用于编写 config/db.query)")
|
||||
print("=" * 60)
|
||||
|
||||
for table in TABLES:
|
||||
try:
|
||||
cursor.execute(f'SELECT * FROM "{CONFIG["schema"]}"."{table}" WHERE 1=0')
|
||||
cols = [d[0] for d in cursor.description]
|
||||
print(f"\n【{table}】 列: {cols}")
|
||||
except Exception as e:
|
||||
print(f"\n【{table}】 查询失败: {e}")
|
||||
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE
|
||||
FROM DBA_TAB_COLUMNS
|
||||
WHERE OWNER = ?
|
||||
ORDER BY TABLE_NAME, COLUMN_ID
|
||||
""",
|
||||
(CONFIG["schema"],),
|
||||
)
|
||||
rows = cursor.fetchall()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("所有列详情 (TABLE_NAME | COLUMN_NAME | DATA_TYPE)")
|
||||
print("=" * 60)
|
||||
cur_table = None
|
||||
for table, col, dtype in rows:
|
||||
if table != cur_table:
|
||||
cur_table = table
|
||||
print(f"\n--- {table} ---")
|
||||
print(f" {col}: {dtype}")
|
||||
|
||||
cursor.close()
|
||||
conn.close()
|
||||
print("\n完成。请根据上述结构在 config 中编写 db.query。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
102
src/cleaner.py
102
src/cleaner.py
@@ -3,13 +3,28 @@ import numpy as np
|
||||
import pandas as pd
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
def _json_serializer(obj):
|
||||
"""处理 Timestamp、datetime、numpy、NaN 等不可直接 JSON 序列化的类型"""
|
||||
if hasattr(obj, 'isoformat'):
|
||||
return obj.isoformat()
|
||||
if hasattr(obj, 'item'):
|
||||
return obj.item()
|
||||
if isinstance(obj, float) and (obj != obj or obj == float('inf') or obj == float('-inf')):
|
||||
return None # NaN, Inf -> null
|
||||
raise TypeError(f'Object of type {type(obj).__name__} is not JSON serializable')
|
||||
|
||||
class AdvancedDataCleaner:
|
||||
def __init__(self, input_file, output_file, report_file):
|
||||
self.input_file = input_file
|
||||
def __init__(self, output_file, report_file, data=None, input_file=None):
|
||||
"""
|
||||
支持两种初始化方式:
|
||||
1. data=... 直接传入数据列表(从配置/数据库加载时使用)
|
||||
2. input_file=... 传入 JSON 文件路径(兼容旧用法)
|
||||
"""
|
||||
self.output_file = output_file
|
||||
self.report_file = report_file
|
||||
self.data = []
|
||||
self.df = None
|
||||
self.data = data if data is not None else []
|
||||
self.input_file = input_file
|
||||
|
||||
# 报告结构
|
||||
self.report = {
|
||||
@@ -40,8 +55,9 @@ class AdvancedDataCleaner:
|
||||
]
|
||||
|
||||
def load_data(self):
|
||||
with open(self.input_file, 'r', encoding='utf-8') as f:
|
||||
self.data = json.load(f)
|
||||
if not self.data and self.input_file:
|
||||
with open(self.input_file, 'r', encoding='utf-8') as f:
|
||||
self.data = json.load(f)
|
||||
self.report["summary"]["total_records"] = len(self.data)
|
||||
self.df = pd.DataFrame(self.data)
|
||||
|
||||
@@ -138,34 +154,72 @@ class AdvancedDataCleaner:
|
||||
self.report["details"]["standardization"].append("Coordinates normalized to 2 decimal places")
|
||||
self.report["details"]["standardization"].append("Timestamps formatted to ISO-8601")
|
||||
|
||||
# 保存数据
|
||||
# 保存数据(确保输出目录存在)
|
||||
import os
|
||||
output_dir = os.path.dirname(self.output_file)
|
||||
if output_dir:
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
result_data = self.df.to_dict('records')
|
||||
with open(self.output_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(result_data, f, ensure_ascii=False, indent=2)
|
||||
json.dump(result_data, f, ensure_ascii=False, indent=2, default=_json_serializer)
|
||||
|
||||
# 保存详细报告
|
||||
# 保存详细报告(确保目录存在)
|
||||
report_dir = os.path.dirname(self.report_file)
|
||||
if report_dir:
|
||||
os.makedirs(report_dir, exist_ok=True)
|
||||
with open(self.report_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(self.report, f, ensure_ascii=False, indent=2)
|
||||
json.dump(self.report, f, ensure_ascii=False, indent=2, default=_json_serializer)
|
||||
|
||||
print(f"完成!报告已生成至 {self.report_file}")
|
||||
|
||||
def load_config(project_root: str) -> dict:
|
||||
"""加载 config/config.json 或 config.yaml(json 优先,便于部署时只改 json)"""
|
||||
import os
|
||||
config_dir = os.path.join(project_root, 'config')
|
||||
json_path = os.path.join(config_dir, 'config.json')
|
||||
yaml_path = os.path.join(config_dir, 'config.yaml')
|
||||
if os.path.exists(json_path):
|
||||
with open(json_path, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
if os.path.exists(yaml_path):
|
||||
try:
|
||||
import yaml
|
||||
with open(yaml_path, 'r', encoding='utf-8') as f:
|
||||
return yaml.safe_load(f)
|
||||
except ImportError:
|
||||
pass
|
||||
return {
|
||||
'source': 'json',
|
||||
'json': {
|
||||
'input_file': 'data/raw_data_sample.json',
|
||||
'output_file': 'data/cleaned_data_final.json',
|
||||
'report_file': 'report/detailed_cleaning_report.json',
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import os
|
||||
# 基于脚本位置计算项目根目录,保证无论从哪里运行都能正确找到文件
|
||||
from data_loader import load_data as load_data_source
|
||||
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.dirname(script_dir)
|
||||
input_path = os.path.join(project_root, 'data', 'raw_data_sample.json')
|
||||
output_path = os.path.join(project_root, 'data', 'cleaned_data_final.json')
|
||||
report_path = os.path.join(project_root, 'report', 'detailed_cleaning_report.json')
|
||||
|
||||
# 确保 report 目录存在
|
||||
os.makedirs(os.path.dirname(report_path), exist_ok=True)
|
||||
config = load_config(project_root)
|
||||
source = config.get('source', 'json')
|
||||
print(f"数据源模式: {source}")
|
||||
|
||||
# 增加一个检查,防止路径错误
|
||||
if not os.path.exists(input_path):
|
||||
print(f"错误:找不到文件 {input_path}")
|
||||
print(f"当前工作目录是:{os.getcwd()}")
|
||||
print("请检查文件路径或确保已运行数据生成脚本。")
|
||||
else:
|
||||
cleaner = AdvancedDataCleaner(input_path, output_path, report_path)
|
||||
cleaner.run()
|
||||
try:
|
||||
data, paths = load_data_source(config, project_root)
|
||||
except (FileNotFoundError, ValueError) as e:
|
||||
print(f"错误:{e}")
|
||||
exit(1)
|
||||
|
||||
os.makedirs(os.path.dirname(paths['report']), exist_ok=True)
|
||||
|
||||
cleaner = AdvancedDataCleaner(
|
||||
output_file=paths['output'],
|
||||
report_file=paths['report'],
|
||||
data=data,
|
||||
)
|
||||
cleaner.run()
|
||||
101
src/data_loader.py
Normal file
101
src/data_loader.py
Normal file
@@ -0,0 +1,101 @@
|
||||
"""
|
||||
数据加载模块:支持从 JSON 文件或达梦数据库读取数据
|
||||
开发阶段使用 JSON,生产环境切换到达梦时只需修改 config 中的 source
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
from typing import Tuple
|
||||
|
||||
|
||||
def load_from_json(file_path: str, project_root: str = None) -> list:
|
||||
"""从 JSON 文件加载数据"""
|
||||
path = file_path
|
||||
if project_root and not os.path.isabs(file_path):
|
||||
path = os.path.join(project_root, file_path)
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def load_from_dm(config: dict, project_root: str = None) -> list:
|
||||
"""
|
||||
从达梦数据库加载数据
|
||||
需要安装 dmPython 及达梦客户端库,仅在 source=db 时调用
|
||||
"""
|
||||
try:
|
||||
import dmPython
|
||||
except ImportError as e:
|
||||
raise RuntimeError(
|
||||
"达梦数据库模式需要安装 dmPython。\n"
|
||||
"请执行: pip install dmPython\n"
|
||||
"并在有达梦客户端的环境运行(需配置 DM_HOME/LD_LIBRARY_PATH)"
|
||||
) from e
|
||||
|
||||
db_cfg = config.get('db', {})
|
||||
password = os.environ.get('DM_PASSWORD')
|
||||
if not password:
|
||||
raise ValueError(
|
||||
"达梦模式需要设置环境变量 DM_PASSWORD。\n"
|
||||
"示例: export DM_PASSWORD=你的密码"
|
||||
)
|
||||
|
||||
conn = dmPython.connect(
|
||||
user=db_cfg.get('user', 'SYSDBA'),
|
||||
password=password,
|
||||
server=db_cfg.get('host', '127.0.0.1'),
|
||||
port=int(db_cfg.get('port', 5080)),
|
||||
schema=db_cfg.get('schema', 'SYSDBA'),
|
||||
)
|
||||
|
||||
try:
|
||||
cursor = conn.cursor()
|
||||
query = (db_cfg.get('query') or '').strip()
|
||||
table = db_cfg.get('table')
|
||||
if query:
|
||||
cursor.execute(query)
|
||||
elif table:
|
||||
schema = db_cfg.get('schema', 'SYSDBA')
|
||||
cursor.execute(f'SELECT * FROM "{schema}"."{table}"')
|
||||
else:
|
||||
raise ValueError(
|
||||
"达梦模式需配置 db.query(自定义 SQL)或 db.table。\n"
|
||||
"当前达梦为规范化表,需通过 query 编写 JOIN 得到扁平化数据。"
|
||||
)
|
||||
columns = [desc[0] for desc in cursor.description]
|
||||
rows = cursor.fetchall()
|
||||
cursor.close()
|
||||
return [dict(zip(columns, row)) for row in rows]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def load_data(config: dict, project_root: str) -> Tuple[list, dict]:
|
||||
"""
|
||||
根据配置加载数据,返回 (数据列表, 路径配置)
|
||||
"""
|
||||
source = config.get('source', 'json')
|
||||
project_root = project_root or os.getcwd()
|
||||
|
||||
if source == 'json':
|
||||
json_cfg = config.get('json', {})
|
||||
data = load_from_json(json_cfg.get('input_file', 'data/raw_data_sample.json'), project_root)
|
||||
paths = {
|
||||
'output': _resolve_path(json_cfg.get('output_file', 'data/cleaned_data_final.json'), project_root),
|
||||
'report': _resolve_path(json_cfg.get('report_file', 'report/detailed_cleaning_report.json'), project_root),
|
||||
}
|
||||
return data, paths
|
||||
elif source == 'db':
|
||||
data = load_from_dm(config, project_root)
|
||||
json_cfg = config.get('json', {})
|
||||
paths = {
|
||||
'output': _resolve_path(json_cfg.get('output_file', 'data/cleaned_data_final.json'), project_root),
|
||||
'report': _resolve_path(json_cfg.get('report_file', 'report/detailed_cleaning_report.json'), project_root),
|
||||
}
|
||||
return data, paths
|
||||
else:
|
||||
raise ValueError(f"不支持的 source 类型: {source},应为 json 或 db")
|
||||
|
||||
|
||||
def _resolve_path(path: str, project_root: str) -> str:
|
||||
if os.path.isabs(path):
|
||||
return path
|
||||
return os.path.join(project_root, path)
|
||||
Reference in New Issue
Block a user