一、环境准备与工具安装
在进行档案数据迁移之前，必须搭建一个稳定且隔离的运行环境。本指南采用Docker部署PostgreSQL作为目标数据库，使用Python作为ETL脚本语言。这种方式能确保环境一致性，避免因操作系统差异导致的依赖冲突。
1. 安装Docker与启动PostgreSQL
如果尚未安装Docker，请访问 https://www.docker.com/products/docker-desktop 下载对应操作系统的安装包并完成安装。安装完成后，打开终端或PowerShell，执行以下命令启动一个PostgreSQL 14版本的容器实例。
执行命令：
```bash
docker run --name archive-migration-db \
-e POSTGRES_PASSWORD=secure_archive_pass \
-e POSTGRES_DB=archive_db \
-p 5432:5432 \
-d postgres:14
```
参数解析：
- --name archive-migration-db：将容器命名为archive-migration-db，便于管理。
- -e POSTGRES_PASSWORD=...：设置数据库超级用户密码，生产环境请务必更换强密码。
- -e POSTGRES_DB=archive_db：初始化时直接创建名为archive_db的数据库。
- -p 5432:5432：将容器内的5432端口映射到宿主机，允许本地连接。
2. 配置Python运行环境
确保系统已安装Python 3.8或更高版本。推荐使用虚拟环境来管理项目依赖。在项目目录下执行以下命令：
```bash
创建虚拟环境
python -m venv venv
激活虚拟环境 (Windows)
venv\Scripts\activate
激活虚拟环境 (Mac/Linux)
source venv/bin/activate
```
激活环境后，安装必要的Python库。我们需要pandas进行数据处理，sqlalchemy和psycopg2用于数据库连接。
```bash
pip install pandas sqlalchemy psycopg2-binary
```
二、模拟源数据构建
实际迁移中，源数据通常是Excel、CSV或老旧的SQL Server导出文件。为了演示全流程，我们首先生成一份包含典型“脏数据”的CSV文件。这份数据模拟了旧档案系统中常见的日期格式不统一、字段缺失和编码混乱问题。
在项目根目录下创建名为generate_source_data.py的文件，并写入以下代码：
```python
import pandas as pd
import csv
模拟的脏数据
data = [
{"档案ID": "A001", "档案标题": "2022年度财务审计报告", "归档日期": "2022-12-31", "保管期限": "永久", "页数": 120},
{"档案ID": "A002", "档案标题": "项目验收文档-缺失页", "归档日期": "20230115", "保管期限": "长期", "页数": None}, 页数缺失
{"档案ID": "A003", "档案标题": "员工入职登记表_张三", "归档日期": "2023/05/20", "保管期限": "10年", "页数": 5},
{"档案ID": "A004", "档案标题": None, "归档日期": "2023-06-01", "保管期限": "30年", "页数": 12}, 标题缺失
{"档案ID": "A005", "档案标题": "合同扫描件", "归档日期": "错误日期格式", "保管期限": "长期", "页数": 45}, 日期错误
]
df = pd.DataFrame(data)
保存为CSV，使用utf-8-sig编码以兼容Excel打开
df.to_csv("legacy_archives.csv", index=False, encoding='utf-8-sig')
print("模拟数据已生成：legacy_archives.csv")
```
运行该脚本生成源文件：
```bash
python generate_source_data.py
```
三、目标数据库表结构设计
在将数据迁入新系统前，需要在PostgreSQL中建立规范的表结构。我们需要定义字段类型、约束条件以及主键。
连接到刚才启动的Docker容器中的数据库。可以直接使用docker exec执行SQL命令，或者使用DBeaver等工具连接。这里使用命令行直接执行：
```bash
docker exec -it archive-migration-db psql -U postgres -d archive_db -c "
CREATE TABLE IF NOT EXISTS archives (
id SERIAL PRIMARY KEY,
archive_code VARCHAR(50) UNIQUE NOT NULL,
title VARCHAR(255) NOT NULL,
archive_date DATE NOT NULL,
retention_period VARCHAR(20) NOT NULL,
page_count INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"
```
表结构说明：
- id：自增主键，新系统内部使用。
- archive_code：对应源数据的“档案ID”，设为唯一键防止重复导入。
- title：档案标题，设为非空约束。
- archive_date：标准化的DATE类型。
- page_count：页数，允许为空，但在应用层默认为0。
四、编写数据迁移核心脚本
这是迁移工作的核心。我们需要编写Python脚本来读取CSV，执行数据清洗（ETL中的Transform），最后写入PostgreSQL。清洗逻辑包括：统一日期格式、处理空值、标准化“保管期限”字段。
创建文件migrate.py：
```python
import pandas as pd
from sqlalchemy import create_engine, text
from datetime import datetime
数据库连接配置
用户名:postgres, 密码:secure_archive_pass, 主机:localhost, 端口:5432, 数据库:archive_db
DB_CONNECTION_STRING = "postgresql://postgres:secure_archive_pass@localhost:5432/archive_db"
def clean_date(date_str):
"""
清洗日期字段：处理 '2023-01-01', '20230101', '2023/01/01' 等格式
"""
if pd.isna(date_str):
return None
date_str = str(date_str).strip()
尝试常见格式
formats = ["%Y-%m-%d", "%Y%m%d", "%Y/%m/%d"]
for fmt in formats:
try:
dt = datetime.strptime(date_str, fmt)
return dt.date()
except ValueError:
continue
如果都解析失败，记录错误并返回None
print(f"警告：无法解析日期 '{date_str}'，将跳过该记录")
return None
def clean_retention(period):
"""
标准化保管期限：将 '永久', '长期' 映射为标准代码，'10年'等保持原样
"""
if pd.isna(period):
return "未知"
mapping = {
"永久": "Y",
"长期": "L"
}
return mapping.get(period.strip(), period.strip())
def main():
1. 读取源数据
print("正在读取源数据...")
try:
df = pd.read_csv("legacy_archives.csv", encoding='utf-8-sig')
except FileNotFoundError:
print("错误：未找到 legacy_archives.csv 文件")
return
2. 数据清洗与转换
print("正在执行数据清洗...")
clean_records = []
skipped_count = 0
for index, row in df.iterrows():
检查必填字段：标题
if pd.isna(row['档案标题']) or row['档案标题'].strip() == "":
print(f"跳过记录 {row['档案ID']}：标题为空")
skipped_count += 1
continue
清洗日期
std_date = clean_date(row['归档日期'])
if std_date is None:
print(f"跳过记录 {row['档案ID']}：日期格式无效")
skipped_count += 1
continue
构建符合目标表结构的字典
record = {
"archive_code": row['档案ID'],
"title": row['档案标题'].strip(),
"archive_date": std_date,
"retention_period": clean_retention(row['保管期限']),
"page_count": int(row['页数']) if pd.notna(row['页数']) else 0
}
clean_records.append(record)
3. 写入数据库
if not clean_records:
print("没有有效数据可导入。")
return
print(f"清洗完成，准备导入 {len(clean_records)} 条数据...")
engine = create_engine(DB_CONNECTION_STRING)
try:
使用 to_sql 批量插入，如果主键冲突则更新（upsert逻辑较复杂，这里演示简单的追加）
为避免重复导入，建议先清空测试表或使用 'append' 模式并忽略错误
这里采用逐条插入以演示错误处理和忽略重复
with engine.begin() as conn:
inserted_count = 0
for rec in clean_records:
try:
conn.execute(text("""
INSERT INTO archives (archive_code, title, archive_date, retention_period, page_count)
VALUES (:code, :title, :date, :period, :pages)
"""), {
"code": rec['archive_code'],
"title": rec['title'],
"date": rec['archive_date'],
"period": rec['retention_period'],
"pages": rec['page_count']
})
inserted_count += 1
except Exception as e:
捕获唯一键冲突等错误
if "duplicate key" in str(e):
print(f"记录 {rec['archive_code']} 已存在，跳过")
else:
print(f"插入记录 {rec['archive_code']} 失败: {e}")
print(f"成功！共导入 {inserted_count} 条新数据，跳过 {skipped_count} 条脏数据。")
except Exception as e:
print(f"数据库连接或操作失败: {e}")
if __name__ == "__main__":
main()
```
五、执行迁移与验证
代码编写完成后，直接运行迁移脚本。
1. 运行迁移脚本
在终端中执行：
```bash
python migrate.py
```
预期输出结果：
终端应当显示清洗过程中的警告信息（例如无法解析的日期），以及最终的导入统计。根据我们构建的模拟数据，ID为A005的记录会因为日期格式错误被跳过，A004因为标题为空被跳过。
2. 数据校验
迁移完成后，必须登录数据库核对数据的一致性。我们对比CSV中的有效记录数与数据库中的记录数，并抽查关键字段。
执行查询命令：
```bash
docker exec -it archive-migration-db psql -U postgres -d archive_db -c "
SELECT
archive_code,
title,
archive_date,
retention_period,
page_count
FROM archives
ORDER BY id;
"
```
校验要点：
- 日期格式：确认
archive_date列均为标准的YYYY-MM-DD格式。
- 空值处理：确认原CSV中页数为空的记录（A002），在数据库中显示为
0。
- 清洗结果：确认A005（日期错误）和A004（标题为空）未被导入。
- 编码正确：确认中文标题显示正常，无乱码。
3. 性能优化建议（针对大批量数据）
本指南使用了逐条插入以方便演示错误处理。在生产环境中迁移百万级档案数据时，应使用psycopg2.extras.execute_batch或Pandas的to_sql方法（设置chunksize）进行批量提交，这将显著提升迁移速度。
