break
370fd3e67f
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 10m49s
70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
||
"""生成奇成代理商店铺批量导入 SQL，不连接或写入目标 PostgreSQL。"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
||
from lib.shop_import import load_config, load_plan, write_errors, write_results, write_summary # noqa: E402
|
||
from lib.shop_sql import build_shop_sql # noqa: E402
|
||
|
||
|
||
def main() -> int:
|
||
base = Path(__file__).resolve().parent
|
||
parser = argparse.ArgumentParser(description="生成店铺批量导入 SQL 和审核产物")
|
||
parser.add_argument("--csv", default=str(base / "副本代理商信息汇总表(3).csv"), help="代理商 CSV 路径；相对路径按当前目录解析")
|
||
parser.add_argument("--config", default=str(base / "config/shop_bulk_import.yaml"), help="本地导入配置路径；相对路径按当前目录解析")
|
||
parser.add_argument("--output-dir", default=str(base / "output"), help="SQL 和审核产物目录；相对路径按当前目录解析")
|
||
args = parser.parse_args()
|
||
|
||
csv_path = _resolve(args.csv)
|
||
config_path = _resolve(args.config)
|
||
output_dir = _resolve(args.output_dir)
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
config, config_errors = load_config(config_path)
|
||
if config_errors:
|
||
error_path = output_dir / "shop_bulk_import_errors.csv"
|
||
write_errors(error_path, config_errors)
|
||
print(f"配置预检失败: {len(config_errors)} 项，错误清单: {error_path}")
|
||
return 1
|
||
|
||
plan = load_plan(csv_path, config)
|
||
prefix = f"shop_bulk_import_{config.batch_id}"
|
||
error_path = output_dir / f"{prefix}_errors.csv"
|
||
result_path = output_dir / f"{prefix}_results.csv"
|
||
summary_path = output_dir / f"{prefix}_summary.txt"
|
||
sql_path = output_dir / f"{prefix}.sql"
|
||
write_errors(error_path, plan.errors)
|
||
if plan.errors:
|
||
write_summary(summary_path, plan, sql_path)
|
||
print(f"CSV 预检失败: {len(plan.errors)} 项，错误清单: {error_path}")
|
||
return 1
|
||
|
||
try:
|
||
sql = build_shop_sql(plan)
|
||
except ModuleNotFoundError as exc:
|
||
if exc.name == "bcrypt":
|
||
print("缺少 bcrypt 依赖，请先执行: pip install -r requirements.txt")
|
||
return 1
|
||
raise
|
||
sql_path.write_text(sql, encoding="utf-8")
|
||
write_results(result_path, plan.results)
|
||
write_summary(summary_path, plan, sql_path)
|
||
print(f"SQL 已生成: {sql_path}")
|
||
print(f"审核结果: {result_path}")
|
||
print(f"摘要: {summary_path}")
|
||
print(f"计划导入: {len(plan.rows)} 条，父子关系: {sum(row.parent_line_no > 0 for row in plan.rows)} 条")
|
||
return 0
|
||
|
||
|
||
def _resolve(value: str) -> Path:
|
||
return Path(value).resolve()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|