116 lines
4.9 KiB
Python
116 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
||
"""导入店铺存续时间表到 runtime/bills.sqlite。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sqlite3
|
||
from datetime import date, datetime, time, timedelta
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
def text(value: Any) -> str | None:
|
||
if value is None:
|
||
return None
|
||
value = str(value).strip()
|
||
return value or None
|
||
|
||
|
||
def sqlite_time(value: Any) -> str | None:
|
||
if value in (None, ""):
|
||
return None
|
||
if isinstance(value, datetime):
|
||
return value.strftime("%Y-%m-%d %H:%M:%S")
|
||
if isinstance(value, date):
|
||
return datetime.combine(value, time()).strftime("%Y-%m-%d %H:%M:%S")
|
||
if isinstance(value, (int, float)):
|
||
value = datetime(1899, 12, 30) + timedelta(days=float(value))
|
||
return value.strftime("%Y-%m-%d %H:%M:%S")
|
||
value = str(value).strip().replace("/", "-")
|
||
if value in {"存续中", "至今", "长期", "无"}:
|
||
return None
|
||
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
|
||
try:
|
||
return datetime.strptime(value, fmt).strftime("%Y-%m-%d %H:%M:%S")
|
||
except ValueError:
|
||
pass
|
||
raise ValueError(f"时间无法解析：{value!r}")
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="导入店铺存续时间 Excel 到 SQLite")
|
||
parser.add_argument("file", type=Path, default=Path("/Users/dodo/Documents/店铺存续时间.xlsx"), nargs="?")
|
||
parser.add_argument("--db", type=Path, default=Path("runtime/bills.sqlite"))
|
||
parser.add_argument("--reset", action="store_true", help="清空已有店铺存续时间后重新导入")
|
||
args = parser.parse_args()
|
||
try:
|
||
import openpyxl
|
||
except ImportError as exc:
|
||
raise SystemExit("缺少 openpyxl，请使用项目提供的 bundled Python 运行此脚本。") from exc
|
||
if not args.file.is_file():
|
||
raise SystemExit(f"文件不存在：{args.file}")
|
||
args.db.parent.mkdir(parents=True, exist_ok=True)
|
||
db = sqlite3.connect(args.db)
|
||
db.executescript("""
|
||
CREATE TABLE IF NOT EXISTS store_lifecycles (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
erp_store_name TEXT NOT NULL,
|
||
company_name TEXT,
|
||
effective_at TEXT,
|
||
expires_at TEXT,
|
||
source_file TEXT NOT NULL,
|
||
raw_json TEXT NOT NULL,
|
||
imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
UNIQUE(erp_store_name, effective_at, expires_at)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_store_lifecycles_lookup
|
||
ON store_lifecycles(erp_store_name, effective_at, expires_at);
|
||
""")
|
||
if args.reset:
|
||
db.execute("DELETE FROM store_lifecycles")
|
||
workbook = openpyxl.load_workbook(args.file, read_only=True, data_only=True)
|
||
inserted = 0
|
||
try:
|
||
for sheet in workbook.worksheets:
|
||
rows = sheet.iter_rows(values_only=True)
|
||
header = next(rows, None)
|
||
if not header or all(value in (None, "") for value in header):
|
||
continue
|
||
headers = [text(value) or "" for value in header]
|
||
required = ["ERP系统店铺名称", "公司名称", "生效时间", "失效时间"]
|
||
missing = [name for name in required if name not in headers]
|
||
if missing:
|
||
raise SystemExit(f"{args.file.name}/{sheet.title} 缺少列：{', '.join(missing)}")
|
||
index = {name: headers.index(name) for name in required}
|
||
for row in rows:
|
||
if not row or all(value in (None, "") for value in row):
|
||
continue
|
||
get = lambda name: row[index[name]] if index[name] < len(row) else None
|
||
store = text(get("ERP系统店铺名称"))
|
||
if not store:
|
||
continue
|
||
raw = {headers[i]: row[i] for i in range(min(len(headers), len(row))) if row[i] not in (None, "")}
|
||
db.execute("""INSERT INTO store_lifecycles
|
||
(erp_store_name,company_name,effective_at,expires_at,source_file,raw_json)
|
||
VALUES (?,?,?,?,?,?)
|
||
ON CONFLICT(erp_store_name,effective_at,expires_at) DO UPDATE SET
|
||
company_name=excluded.company_name, source_file=excluded.source_file,
|
||
raw_json=excluded.raw_json, imported_at=CURRENT_TIMESTAMP""",
|
||
(store, text(get("公司名称")), sqlite_time(get("生效时间")),
|
||
sqlite_time(get("失效时间")), args.file.name,
|
||
json.dumps(raw, ensure_ascii=False, default=str)))
|
||
inserted += 1
|
||
finally:
|
||
workbook.close()
|
||
db.commit()
|
||
total = db.execute("SELECT COUNT(*) FROM store_lifecycles").fetchone()[0]
|
||
db.close()
|
||
print(f"导入完成：{args.db}，本次处理 {inserted} 条，表内共 {total} 条")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|