Spaces:
Runtime error
Runtime error
File size: 4,680 Bytes
0e8f624 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | #!/usr/bin/env python3
"""
初始化SQLite数据库并导入CSV数据
"""
import sqlite3
import csv
from pathlib import Path
from datetime import datetime
DB_PATH = "data/customs_data.db"
CSV_PATH = "data/standard_trade_records_sample.csv"
def create_database():
"""创建SQLite数据库和表结构"""
print(f"创建数据库: {DB_PATH}")
# 确保data目录存在
Path("data").mkdir(exist_ok=True)
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# 创建标准贸易记录表
cursor.execute("""
CREATE TABLE IF NOT EXISTS standard_trade_records (
record_id TEXT PRIMARY KEY,
source_record_id TEXT NOT NULL,
batch_no TEXT NOT NULL,
source_country TEXT NOT NULL,
trade_direction TEXT NOT NULL,
trade_date TIMESTAMP NOT NULL,
importer_name TEXT,
exporter_name TEXT,
hs_code TEXT,
product_name TEXT,
amount REAL,
currency TEXT,
weight REAL,
weight_unit TEXT,
origin_country TEXT,
destination_country TEXT,
departure_port TEXT,
arrival_port TEXT,
transport_mode TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP
)
""")
# 创建索引
cursor.execute("CREATE INDEX IF NOT EXISTS idx_source_country ON standard_trade_records(source_country)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_trade_date ON standard_trade_records(trade_date)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_hs_code ON standard_trade_records(hs_code)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_trade_direction ON standard_trade_records(trade_direction)")
conn.commit()
return conn
def import_csv_data(conn):
"""从CSV导入数据"""
print(f"导入CSV数据: {CSV_PATH}")
if not Path(CSV_PATH).exists():
print(f"警告: CSV文件不存在 {CSV_PATH}")
return
cursor = conn.cursor()
with open(CSV_PATH, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
count = 0
batch = []
for row in reader:
batch.append((
row['record_id'],
row['source_record_id'],
row['batch_no'],
row['source_country'],
row['trade_direction'],
row['trade_date'],
row['importer_name'] or None,
row['exporter_name'] or None,
row['hs_code'] or None,
row['product_name'] or None,
float(row['amount']) if row['amount'] else None,
row['currency'] or None,
float(row['weight']) if row['weight'] else None,
row['weight_unit'] or None,
row['origin_country'] or None,
row['destination_country'] or None,
row['departure_port'] or None,
row['arrival_port'] or None,
row['transport_mode'] or None,
row['created_at'],
row.get('updated_at')
))
count += 1
# 每1000条批量插入
if len(batch) >= 1000:
cursor.executemany("""
INSERT INTO standard_trade_records VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
)
""", batch)
conn.commit()
print(f"已导入 {count} 条记录...")
batch = []
# 插入剩余数据
if batch:
cursor.executemany("""
INSERT INTO standard_trade_records VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
)
""", batch)
conn.commit()
print(f"总共导入 {count} 条记录")
def main():
"""主函数"""
print("=== 初始化SQLite数据库 ===\n")
conn = create_database()
import_csv_data(conn)
# 验证数据
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM standard_trade_records")
total = cursor.fetchone()[0]
print(f"\n数据库中共有 {total} 条记录")
cursor.execute("SELECT COUNT(*) FROM standard_trade_records WHERE source_country='BR'")
br_count = cursor.fetchone()[0]
print(f"巴西数据: {br_count} 条")
conn.close()
print("\n✅ 数据库初始化完成！")
if __name__ == "__main__":
main()
|