9cd3c0f231
- 任务详情上传支持"本批素材无条码"开关，跳过条码必填校验 - SKU卡片/主图网格从4列改5列，图片高度压缩，去掉冗余标签和提示 - 详情图/PSD上传区去掉大占位框，改成紧凑按钮+标签 - 补充历史导入相关脚本和工具文件 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
341 lines
12 KiB
Python
341 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Discover minimal package candidates inside needs_review SPU roots.
|
|
|
|
This is a read-only second pass. It is stricter than grouping by SPU root: a
|
|
candidate package is the smallest directory subtree that contains main, SKU and
|
|
detail assets. That handles cross-level asset folders while avoiding merging
|
|
multiple versions under the same SPU root.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
from collections import Counter, defaultdict
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from review_needs_review_spu_groups import choose_spu_root_path
|
|
from scan_spu_ledger import (
|
|
EXCLUDED_DIR_NAMES,
|
|
RELEVANT_EXTENSIONS,
|
|
build_sku_code,
|
|
classify_file,
|
|
extract_file_spu_candidates,
|
|
has_hard_skip,
|
|
is_compatible_spu_token,
|
|
normalize_spu,
|
|
safe_relative,
|
|
)
|
|
|
|
|
|
CAUTION_RE = re.compile(
|
|
r"(^|[\\/])3D($|[\\/])|3D建模|3D原图|3D渲染原图|建模|建模图|建模原图|渲染原图|原图建模|"
|
|
r"源文件|psd源文件|PSD源文件|PS源文件|"
|
|
r"大图|AI合成|AI原图|图片有误|错误勿用|勿用|废稿|作废|弃用|不要",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
HEADERS = [
|
|
"decision",
|
|
"spu",
|
|
"store",
|
|
"product_category",
|
|
"package_path",
|
|
"relative_package_path",
|
|
"main_count",
|
|
"sku_count",
|
|
"detail_count",
|
|
"source_count",
|
|
"needs_review_rows_under_root",
|
|
"caution_reason",
|
|
"related_workspace_samples",
|
|
"main_samples",
|
|
"sku_samples",
|
|
"detail_samples",
|
|
]
|
|
|
|
|
|
def json_dumps(value: Any) -> 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"])
|
|
|
|
|
|
class Node:
|
|
def __init__(self, path: str, root: str) -> None:
|
|
self.path = path
|
|
self.root = root
|
|
self.children: list[str] = []
|
|
self.files: list[dict[str, Any]] = []
|
|
self.direct_counts: Counter[str] = Counter()
|
|
self.subtree_counts: Counter[str] = Counter()
|
|
self.samples: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
self.has_caution = False
|
|
self.conflicts: list[dict[str, str]] = []
|
|
|
|
|
|
def scan_tree(root_path: str, spu: str) -> dict[str, Node]:
|
|
root = Path(root_path)
|
|
nodes: dict[str, Node] = {}
|
|
if not root.exists():
|
|
return nodes
|
|
|
|
for current, dir_names, file_names in os.walk(root):
|
|
current_path = Path(current)
|
|
current_str = str(current_path)
|
|
node = nodes.setdefault(current_str, Node(current_str, root_path))
|
|
node.has_caution = node.has_caution or bool(CAUTION_RE.search(current_str))
|
|
|
|
kept_dirs: list[str] = []
|
|
for name in dir_names:
|
|
child = current_path / name
|
|
if name in EXCLUDED_DIR_NAMES or has_hard_skip(str(child)):
|
|
continue
|
|
kept_dirs.append(name)
|
|
child_str = str(child)
|
|
node.children.append(child_str)
|
|
nodes.setdefault(child_str, Node(child_str, root_path))
|
|
dir_names[:] = kept_dirs
|
|
|
|
for file_name in file_names:
|
|
file_path = current_path / file_name
|
|
extension = file_path.suffix.lower()
|
|
if extension not in RELEVANT_EXTENSIONS or has_hard_skip(str(file_path)):
|
|
continue
|
|
try:
|
|
file_size = file_path.stat().st_size
|
|
except OSError:
|
|
continue
|
|
|
|
relative_path = safe_relative(file_path, root)
|
|
group = classify_file(relative_path, file_name, extension)
|
|
record = {
|
|
"filePath": str(file_path),
|
|
"relativePath": relative_path,
|
|
"fileName": file_name,
|
|
"fileSize": int(file_size),
|
|
"group": group,
|
|
}
|
|
if group == "sku":
|
|
record["skuCode"] = build_sku_code(spu, file_name, node.direct_counts[group])
|
|
record["barcodes"] = []
|
|
node.files.append(record)
|
|
node.direct_counts[group] += 1
|
|
if group in {"main", "sku", "detail", "source"} and len(node.samples[group]) < 8:
|
|
node.samples[group].append(record)
|
|
if bool(CAUTION_RE.search(str(file_path))):
|
|
node.has_caution = True
|
|
|
|
if group in {"main", "detail"}:
|
|
tokens = extract_file_spu_candidates(file_name)
|
|
if tokens and not any(is_compatible_spu_token(spu, token) for token in tokens):
|
|
node.conflicts.append(
|
|
{
|
|
"group": group,
|
|
"file": relative_path,
|
|
"tokens": ",".join(tokens),
|
|
}
|
|
)
|
|
|
|
for path in sorted(nodes, key=lambda item: len(Path(item).parts), reverse=True):
|
|
node = nodes[path]
|
|
node.subtree_counts.update(node.direct_counts)
|
|
for group, records in node.samples.items():
|
|
node.samples[group] = records[:8]
|
|
for child_path in node.children:
|
|
child = nodes[child_path]
|
|
node.subtree_counts.update(child.subtree_counts)
|
|
node.has_caution = node.has_caution or child.has_caution
|
|
if len(node.conflicts) < 12:
|
|
node.conflicts.extend(child.conflicts[: max(0, 12 - len(node.conflicts))])
|
|
for group, records in child.samples.items():
|
|
existing = node.samples[group]
|
|
for record in records:
|
|
if len(existing) >= 8:
|
|
break
|
|
existing.append(record)
|
|
return nodes
|
|
|
|
|
|
def has_all_three(node: Node) -> bool:
|
|
return (
|
|
node.subtree_counts["main"] > 0
|
|
and node.subtree_counts["sku"] > 0
|
|
and node.subtree_counts["detail"] > 0
|
|
)
|
|
|
|
|
|
def minimal_package_nodes(nodes: dict[str, Node]) -> list[Node]:
|
|
candidates = [node for node in nodes.values() if has_all_three(node)]
|
|
candidate_paths = {node.path for node in candidates}
|
|
minimal: list[Node] = []
|
|
for node in candidates:
|
|
descendant_has_all_three = False
|
|
stack = list(node.children)
|
|
while stack:
|
|
child_path = stack.pop()
|
|
if child_path in candidate_paths:
|
|
descendant_has_all_three = True
|
|
break
|
|
child = nodes.get(child_path)
|
|
if child:
|
|
stack.extend(child.children)
|
|
if not descendant_has_all_three:
|
|
minimal.append(node)
|
|
minimal.sort(key=lambda item: item.path.lower())
|
|
return minimal
|
|
|
|
|
|
def package_decision(node: Node) -> tuple[str, str]:
|
|
reasons: list[str] = []
|
|
if node.has_caution:
|
|
reasons.append("路径或文件名包含建模/源文件/大图/AI原图/错误等谨慎标识")
|
|
if node.conflicts:
|
|
tokens = sorted({conflict["tokens"] for conflict in node.conflicts})[:4]
|
|
reasons.append("主图/详情文件名出现不同SPU: " + ";".join(tokens))
|
|
if node.subtree_counts["main"] == 0 or node.subtree_counts["sku"] == 0 or node.subtree_counts["detail"] == 0:
|
|
reasons.append("缺少主图/SKU/详情之一")
|
|
if reasons:
|
|
return "package_caution_review", " | ".join(reasons)
|
|
return "package_ready_review", ""
|
|
|
|
|
|
def row_for(node: Node, item_group: list[sqlite3.Row], spu: str, store: str, category: str) -> dict[str, Any]:
|
|
package = Path(node.path)
|
|
root = Path(node.root)
|
|
decision, caution_reason = package_decision(node)
|
|
related = [
|
|
str(item["workspace_path"])
|
|
for item in item_group
|
|
if str(item["workspace_path"]).lower().startswith(node.path.lower())
|
|
][:8]
|
|
if not related:
|
|
related = [str(item["workspace_path"]) for item in item_group[:8]]
|
|
|
|
return {
|
|
"decision": decision,
|
|
"spu": spu,
|
|
"store": store,
|
|
"product_category": category,
|
|
"package_path": str(package),
|
|
"relative_package_path": safe_relative(package, root),
|
|
"main_count": node.subtree_counts["main"],
|
|
"sku_count": node.subtree_counts["sku"],
|
|
"detail_count": node.subtree_counts["detail"],
|
|
"source_count": node.subtree_counts["source"],
|
|
"needs_review_rows_under_root": len(item_group),
|
|
"caution_reason": caution_reason,
|
|
"related_workspace_samples": " || ".join(related),
|
|
"main_samples": " || ".join(record["relativePath"] for record in node.samples.get("main", [])),
|
|
"sku_samples": " || ".join(record["relativePath"] for record in node.samples.get("sku", [])),
|
|
"detail_samples": " || ".join(record["relativePath"] for record in node.samples.get("detail", [])),
|
|
}
|
|
|
|
|
|
def write_csv(path: Path, rows: list[dict[str, Any]]) -> 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()
|
|
writer.writerows(rows)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Discover package candidates for needs_review rows.")
|
|
parser.add_argument("--db-path", default="output/tianlan-spu-ledger.db")
|
|
parser.add_argument("--run-id", default="")
|
|
parser.add_argument("--output-dir", default="")
|
|
args = parser.parse_args()
|
|
|
|
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
output_dir = Path(args.output_dir or f"output/tianlan-spu-ledger/needs-review-package-candidates-{stamp}")
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
conn = sqlite3.connect(str(Path(args.db_path)))
|
|
conn.row_factory = sqlite3.Row
|
|
run_id = args.run_id or latest_run_id(conn)
|
|
items = conn.execute(
|
|
"""
|
|
SELECT *
|
|
FROM spu_ledger_items
|
|
WHERE run_id = ? AND status = 'needs_review'
|
|
ORDER BY source_root_name, relative_workspace_path, spu
|
|
""",
|
|
(run_id,),
|
|
).fetchall()
|
|
|
|
groups: dict[tuple[str, str, str, str], list[sqlite3.Row]] = defaultdict(list)
|
|
for item in items:
|
|
spu = normalize_spu(item["spu"])
|
|
root_path = choose_spu_root_path(item)
|
|
key = (str(item["store"] or ""), spu, str(item["product_category"] or ""), root_path)
|
|
groups[key].append(item)
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
detail_payload: list[dict[str, Any]] = []
|
|
roots_without_package = 0
|
|
for (store, spu, category, root_path), item_group in sorted(groups.items()):
|
|
nodes = scan_tree(root_path, spu)
|
|
packages = minimal_package_nodes(nodes)
|
|
if not packages:
|
|
roots_without_package += 1
|
|
continue
|
|
for node in packages:
|
|
row = row_for(node, item_group, spu, store, category)
|
|
rows.append(row)
|
|
detail_payload.append(
|
|
{
|
|
**row,
|
|
"samples": node.samples,
|
|
"conflicts": node.conflicts[:12],
|
|
"workspaces": [
|
|
{
|
|
"fingerprint": item["fingerprint"],
|
|
"workspacePath": item["workspace_path"],
|
|
"relativeWorkspacePath": item["relative_workspace_path"],
|
|
"versionLabel": item["version_label"],
|
|
"mainCount": item["main_count"],
|
|
"skuCount": item["sku_count"],
|
|
"detailCount": item["detail_count"],
|
|
"statusReason": item["status_reason"],
|
|
}
|
|
for item in item_group
|
|
],
|
|
}
|
|
)
|
|
|
|
write_csv(output_dir / "package-candidates.csv", rows)
|
|
(output_dir / "package-candidates.json").write_text(json_dumps(detail_payload), encoding="utf-8")
|
|
summary = {
|
|
"runId": run_id,
|
|
"needsReviewItems": len(items),
|
|
"spuRootGroups": len(groups),
|
|
"rootsWithoutPackage": roots_without_package,
|
|
"packageCandidates": len(rows),
|
|
"decisionCounts": dict(Counter(row["decision"] for row in rows)),
|
|
"outputCsv": str(output_dir / "package-candidates.csv"),
|
|
"outputJson": str(output_dir / "package-candidates.json"),
|
|
}
|
|
(output_dir / "summary.json").write_text(json_dumps(summary), encoding="utf-8")
|
|
print(json_dumps(summary))
|
|
conn.close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|