9cd3c0f231
- 任务详情上传支持"本批素材无条码"开关，跳过条码必填校验 - SKU卡片/主图网格从4列改5列，图片高度压缩，去掉冗余标签和提示 - 详情图/PSD上传区去掉大占位框，改成紧凑按钮+标签 - 补充历史导入相关脚本和工具文件 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
260 lines
9.5 KiB
Python
260 lines
9.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Export the SPU ledger into review CSV files and ready import JSON.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import json
|
||
import sqlite3
|
||
from pathlib import Path
|
||
|
||
|
||
STATUS_EXPORTS = {
|
||
"ready": "ready",
|
||
"needs_review": "needs-review",
|
||
"skipped": "skipped",
|
||
"duplicate": "duplicate",
|
||
"delete_candidate": "delete-candidate",
|
||
}
|
||
|
||
|
||
def json_dumps(value) -> str:
|
||
return json.dumps(value, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def latest_run_id(conn: sqlite3.Connection) -> str:
|
||
row = conn.execute(
|
||
"SELECT run_id FROM spu_ledger_runs ORDER BY started_at DESC LIMIT 1"
|
||
).fetchone()
|
||
if not row:
|
||
raise SystemExit("台账数据库里没有扫描 run")
|
||
return str(row["run_id"])
|
||
|
||
|
||
def read_items(conn: sqlite3.Connection, run_id: str) -> list[sqlite3.Row]:
|
||
return conn.execute(
|
||
"""
|
||
SELECT *
|
||
FROM spu_ledger_items
|
||
WHERE run_id = ?
|
||
ORDER BY source_root_name, relative_workspace_path, spu
|
||
""",
|
||
(run_id,),
|
||
).fetchall()
|
||
|
||
|
||
def read_files(conn: sqlite3.Connection, item_id: int, asset_group: str) -> list[sqlite3.Row]:
|
||
return conn.execute(
|
||
"""
|
||
SELECT *
|
||
FROM spu_ledger_files
|
||
WHERE item_id = ? AND asset_group = ?
|
||
ORDER BY sort_order, relative_file_path
|
||
""",
|
||
(item_id, asset_group),
|
||
).fetchall()
|
||
|
||
|
||
def csv_headers() -> list[str]:
|
||
return [
|
||
"status（状态）",
|
||
"spu（SPU）",
|
||
"store（店铺）",
|
||
"product_category（产品品类）",
|
||
"author（作者）",
|
||
"task_detail_source_type（出图方式）",
|
||
"version_label（版本/工作区）",
|
||
"main_count（主图数量）",
|
||
"sku_count（SKU图数量）",
|
||
"detail_count（详情图数量）",
|
||
"source_count（PS源文件数量）",
|
||
"status_reason（判断原因）",
|
||
"spu_source（SPU来源）",
|
||
"store_evidence（店铺证据）",
|
||
"author_evidence（作者证据）",
|
||
"task_detail_source_type_evidence（出图方式证据）",
|
||
"source_batch（来源批次）",
|
||
"source_root（来源根目录）",
|
||
"workspace_path（工作区完整路径）",
|
||
"relative_workspace_path（工作区相对路径）",
|
||
"fingerprint（台账指纹）",
|
||
]
|
||
|
||
|
||
def item_to_csv_row(item: sqlite3.Row) -> dict[str, object]:
|
||
return {
|
||
"status（状态）": item["status"],
|
||
"spu（SPU）": item["spu"],
|
||
"store（店铺）": item["store"],
|
||
"product_category（产品品类）": item["product_category"],
|
||
"author（作者）": item["author"],
|
||
"task_detail_source_type（出图方式）": item["task_detail_source_type"],
|
||
"version_label（版本/工作区）": item["version_label"],
|
||
"main_count（主图数量）": item["main_count"],
|
||
"sku_count（SKU图数量）": item["sku_count"],
|
||
"detail_count（详情图数量）": item["detail_count"],
|
||
"source_count（PS源文件数量）": item["source_count"],
|
||
"status_reason（判断原因）": item["status_reason"],
|
||
"spu_source（SPU来源）": item["spu_source"],
|
||
"store_evidence（店铺证据）": item["store_evidence"],
|
||
"author_evidence（作者证据）": item["author_evidence"],
|
||
"task_detail_source_type_evidence（出图方式证据）": item["task_detail_source_type_evidence"],
|
||
"source_batch（来源批次）": item["source_batch"],
|
||
"source_root（来源根目录）": item["source_root"],
|
||
"workspace_path（工作区完整路径）": item["workspace_path"],
|
||
"relative_workspace_path（工作区相对路径）": item["relative_workspace_path"],
|
||
"fingerprint（台账指纹）": item["fingerprint"],
|
||
}
|
||
|
||
|
||
def write_csv(path: Path, rows: list[dict[str, object]], headers: list[str]) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with path.open("w", encoding="utf-8-sig", newline="") as handle:
|
||
writer = csv.DictWriter(handle, fieldnames=headers)
|
||
writer.writeheader()
|
||
for row in rows:
|
||
writer.writerow(row)
|
||
|
||
|
||
def to_image_input(file_row: sqlite3.Row) -> dict[str, object]:
|
||
payload = {
|
||
"localFilePath": file_row["file_path"],
|
||
"fileName": file_row["file_name"],
|
||
"fileSize": int(file_row["file_size"] or 0),
|
||
"sortOrder": int(file_row["sort_order"] or 0),
|
||
}
|
||
if file_row["asset_group"] == "sku":
|
||
payload["skuCode"] = file_row["sku_code"] or ""
|
||
payload["barcodes"] = []
|
||
return payload
|
||
|
||
|
||
def to_source_input(file_row: sqlite3.Row) -> dict[str, object]:
|
||
return {
|
||
"localFilePath": file_row["file_path"],
|
||
"fileName": file_row["file_name"],
|
||
"fileSize": int(file_row["file_size"] or 0),
|
||
}
|
||
|
||
|
||
def build_import_record(conn: sqlite3.Connection, item: sqlite3.Row) -> dict[str, object]:
|
||
main = [to_image_input(row) for row in read_files(conn, item["id"], "main")]
|
||
sku = [to_image_input(row) for row in read_files(conn, item["id"], "sku")]
|
||
detail = [to_image_input(row) for row in read_files(conn, item["id"], "detail")]
|
||
sources = [to_source_input(row) for row in read_files(conn, item["id"], "source")]
|
||
|
||
return {
|
||
"fingerprint": item["fingerprint"],
|
||
"spu": item["spu"],
|
||
"sourcePath": item["workspace_path"],
|
||
"relativeSourcePath": item["relative_workspace_path"],
|
||
"guessedStore": item["store"],
|
||
"productCategoryName": item["product_category"],
|
||
"authorName": item["author"] or "历史导入",
|
||
"taskDetailSourceType": item["task_detail_source_type"] or "渲染",
|
||
"storeConfidence": 1,
|
||
"storeCandidates": [item["store"]] if item["store"] else [],
|
||
"mainCount": item["main_count"],
|
||
"skuCount": item["sku_count"],
|
||
"detailCount": item["detail_count"],
|
||
"sourceCount": item["source_count"],
|
||
"status": item["status"],
|
||
"skipReason": None if item["status"] == "ready" else item["status_reason"],
|
||
"mainImages": main,
|
||
"skuImages": sku,
|
||
"detailImages": detail,
|
||
"psdSources": sources,
|
||
"seenInYya": False,
|
||
"ledger": {
|
||
"runId": item["run_id"],
|
||
"sourceBatch": item["source_batch"],
|
||
"spuSource": item["spu_source"],
|
||
"spuEvidence": json.loads(item["spu_evidence_json"] or "[]"),
|
||
"pathFlags": json.loads(item["path_flags_json"] or "{}"),
|
||
},
|
||
}
|
||
|
||
|
||
def write_batch_csvs(output_dir: Path, prefix: str, items: list[sqlite3.Row], batch_size: int) -> None:
|
||
if batch_size <= 0:
|
||
return
|
||
headers = csv_headers()
|
||
pending = [item for item in items if item["status"] in ("ready", "needs_review")]
|
||
for start in range(0, len(pending), batch_size):
|
||
batch_no = start // batch_size + 1
|
||
batch = pending[start : start + batch_size]
|
||
rows = [item_to_csv_row(item) for item in batch]
|
||
write_csv(output_dir / f"{prefix}-batch-{batch_no:03d}.csv", rows, headers)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="Export SPU ledger review files.")
|
||
parser.add_argument("--db-path", required=True)
|
||
parser.add_argument("--output-dir", default=str(Path.cwd() / "output"))
|
||
parser.add_argument("--prefix", default="spu-ledger")
|
||
parser.add_argument("--run-id", default="")
|
||
parser.add_argument("--batch-size", type=int, default=20)
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> int:
|
||
args = parse_args()
|
||
db_path = Path(args.db_path)
|
||
output_dir = Path(args.output_dir)
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
conn = sqlite3.connect(str(db_path))
|
||
conn.row_factory = sqlite3.Row
|
||
run_id = args.run_id or latest_run_id(conn)
|
||
items = read_items(conn, run_id)
|
||
|
||
headers = csv_headers()
|
||
status_counts: dict[str, int] = {}
|
||
store_counts: dict[str, int] = {}
|
||
category_counts: dict[str, int] = {}
|
||
|
||
for item in items:
|
||
status_counts[item["status"]] = status_counts.get(item["status"], 0) + 1
|
||
store_counts[item["store"]] = store_counts.get(item["store"], 0) + 1
|
||
category = item["product_category"] or "未识别"
|
||
category_counts[category] = category_counts.get(category, 0) + 1
|
||
|
||
for status, suffix in STATUS_EXPORTS.items():
|
||
rows = [item_to_csv_row(item) for item in items if item["status"] == status]
|
||
write_csv(output_dir / f"{args.prefix}-{suffix}.csv", rows, headers)
|
||
|
||
write_csv(output_dir / f"{args.prefix}-all.csv", [item_to_csv_row(item) for item in items], headers)
|
||
write_batch_csvs(output_dir, args.prefix, items, args.batch_size)
|
||
|
||
ready_records = [build_import_record(conn, item) for item in items if item["status"] == "ready"]
|
||
ready_json_path = output_dir / f"{args.prefix}-ready-import-records.json"
|
||
ready_json_path.write_text(json_dumps(ready_records), encoding="utf-8")
|
||
|
||
summary = {
|
||
"runId": run_id,
|
||
"dbPath": str(db_path),
|
||
"total": len(items),
|
||
"status": status_counts,
|
||
"store": store_counts,
|
||
"productCategory": category_counts,
|
||
"readyImportRecords": str(ready_json_path),
|
||
}
|
||
summary_path = output_dir / f"{args.prefix}-summary.json"
|
||
summary_path.write_text(json_dumps(summary), encoding="utf-8")
|
||
|
||
conn.close()
|
||
|
||
print("SPU 台账导出完成")
|
||
print(f"- run_id: {run_id}")
|
||
print(f"- total: {len(items)}")
|
||
print(f"- ready_import_records: {ready_json_path}")
|
||
print(f"- summary: {summary_path}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|