#!/usr/bin/env python3 """将平台账单 Excel 导入本地 SQLite，供后续按店铺和时间汇总。 示例（仓库根目录）： python3 scripts/import_bills_sqlite.py python3 scripts/import_bills_sqlite.py --input-dir /path/to/账单数据 \ --db runtime/bills.sqlite --reset 运行环境需要 openpyxl；项目桌面运行时可使用 Codex bundled Python。 """ from __future__ import annotations import argparse import json import re import sqlite3 import sys from datetime import date, datetime, time, timedelta from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any, Iterable PLATFORMS: tuple[tuple[str, str, tuple[str, ...]], ...] = ( ("dewu", "得物", ("得物",)), ("douyin", "抖音", ("头条放心购", "抖音")), ("taobao", "淘系", ("淘系",)), ("jd", "京东", ("京东商城", "京东")), ("pinduoduo", "拼多多", ("拼多多",)), ) TIME_HEADERS = { "dewu": "支付/退款时间", "douyin": "结算时间", "taobao": "入账时间", "jd": "费用结算时间", "pinduoduo": "发生时间", } def platform_for_file(path: Path) -> tuple[str, str]: for key, label, markers in PLATFORMS: if any(marker in path.name for marker in markers): return key, label raise ValueError(f"无法从文件名识别平台：{path.name}") def clean(value: Any) -> str | None: if value is None: return None text = str(value).strip() return text or None def money(value: Any) -> str: if value is None or value == "": return "0.00" text = str(value).replace(",", "").replace("￥", "").replace("¥", "").strip() try: return format(Decimal(text), ".2f") except InvalidOperation as exc: raise ValueError(f"金额无法解析：{value!r}") from exc def iso_time(value: Any) -> str | None: if value is None or value == "": 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)): # Excel 1900 date system; openpyxl normally converts typed date cells, # but this covers numeric exports as well. converted = datetime(1899, 12, 30) + timedelta(days=float(value)) return converted.strftime("%Y-%m-%d %H:%M:%S") text = str(value).strip().replace("/", "-") for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"): try: return datetime.strptime(text, fmt).strftime("%Y-%m-%d %H:%M:%S") except ValueError: continue # 得物偶尔出现 2026/8/13 14:11:19 这种非零填充形式。 match = re.fullmatch(r"(\d{4})-(\d{1,2})-(\d{1,2})(?:\s+|T)(\d{1,2}):(\d{2})(?::(\d{2}))?", text) if match: y, m, d, hh, mm, ss = (int(x or 0) for x in match.groups()) return f"{y:04d}-{m:02d}-{d:02d} {hh:02d}:{mm:02d}:{ss:02d}" raise ValueError(f"时间无法解析：{value!r}") def json_default(value: Any) -> str: if isinstance(value, (datetime, date)): return value.isoformat(sep=" ") return str(value) def init_db(connection: sqlite3.Connection) -> None: connection.executescript( """ PRAGMA journal_mode = WAL; CREATE TABLE IF NOT EXISTS bills ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform TEXT NOT NULL, type TEXT NOT NULL, store_name TEXT NOT NULL, occurred_at TEXT, income NUMERIC NOT NULL DEFAULT 0, expense NUMERIC NOT NULL DEFAULT 0, bill_item TEXT, record_code TEXT NOT NULL, source_file TEXT NOT NULL, raw_json TEXT NOT NULL, imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(platform, record_code) ); CREATE INDEX IF NOT EXISTS idx_bills_store_time ON bills(store_name, occurred_at); CREATE INDEX IF NOT EXISTS idx_bills_platform_store_time ON bills(platform, store_name, occurred_at); CREATE VIEW IF NOT EXISTS bill_daily_summary AS SELECT platform, type, store_name, substr(occurred_at, 1, 10) AS bill_date, ROUND(SUM(income), 2) AS income, ROUND(SUM(expense), 2) AS expense, COUNT(*) AS row_count FROM bills WHERE occurred_at IS NOT NULL AND occurred_at <> '' GROUP BY platform, type, store_name, substr(occurred_at, 1, 10); """ ) def rows_from_file(path: Path, platform: str, label: str) -> Iterable[tuple[Any, ...]]: try: import openpyxl except ImportError as exc: raise RuntimeError("缺少 openpyxl，请使用项目提供的 bundled Python 运行此脚本。") from exc workbook = openpyxl.load_workbook(path, read_only=True, data_only=True) try: for sheet in workbook.worksheets: iterator = sheet.iter_rows(values_only=True) header_row = next(iterator, None) if not header_row: continue headers = [clean(item) or "" for item in header_row] required = {"店铺", "记录编码", TIME_HEADERS[platform], "收入金额" if platform != "douyin" else "收入", "支出金额" if platform != "douyin" else "支出"} missing = required.difference(headers) if missing: raise ValueError(f"{path.name}/{sheet.title} 缺少列：{', '.join(sorted(missing))}") index = {name: position for position, name in enumerate(headers)} income_header = "收入" if platform == "douyin" else "收入金额" expense_header = "支出" if platform == "douyin" else "支出金额" for row in iterator: if not row or all(item in (None, "") for item in row): continue get = lambda name: row[index[name]] if index[name] < len(row) else None record_code = clean(get("记录编码")) store_name = clean(get("店铺")) if not record_code or not store_name: continue raw = {headers[i]: row[i] for i in range(min(len(headers), len(row))) if row[i] not in (None, "")} yield (platform, label, store_name, iso_time(get(TIME_HEADERS[platform])), money(get(income_header)), money(get(expense_header)), clean(get("账单项目")), record_code, path.name, json.dumps(raw, ensure_ascii=False, default=json_default)) finally: workbook.close() def main() -> int: parser = argparse.ArgumentParser(description="导入得物/抖音/淘系/京东/拼多多账单到 SQLite") parser.add_argument("--input-dir", type=Path, default=Path("/Users/dodo/AI系统/账单数据")) parser.add_argument("--db", type=Path, default=Path("runtime/bills.sqlite")) parser.add_argument("--reset", action="store_true", help="清空已有 bills 后重新导入") args = parser.parse_args() files = sorted(args.input_dir.glob("*.xlsx")) if not files: print(f"未找到 Excel 文件：{args.input_dir}", file=sys.stderr) return 2 args.db.parent.mkdir(parents=True, exist_ok=True) connection = sqlite3.connect(args.db) init_db(connection) if args.reset: connection.execute("DELETE FROM bills") counts: dict[str, int] = {} try: for path in files: platform, label = platform_for_file(path) count = 0 for row in rows_from_file(path, platform, label): connection.execute( """INSERT INTO bills(platform,type,store_name,occurred_at,income,expense,bill_item,record_code,source_file,raw_json) VALUES(?,?,?,?,?,?,?,?,?,?) ON CONFLICT(platform,record_code) DO UPDATE SET type=excluded.type, store_name=excluded.store_name, occurred_at=excluded.occurred_at, income=excluded.income, expense=excluded.expense, bill_item=excluded.bill_item, source_file=excluded.source_file, raw_json=excluded.raw_json, imported_at=CURRENT_TIMESTAMP""", row, ) count += 1 counts[platform] = counts.get(platform, 0) + count connection.commit() finally: connection.close() print(f"导入完成：{args.db}") for platform, count in counts.items(): print(f" {platform}: {count} 条") return 0 if __name__ == "__main__": raise SystemExit(main())