gongxiangp2/scripts/review_needs_review_spu_groups.py

#!/usr/bin/env python3
"""
Build a second-pass review table for needs_review SPU ledger items.

The first SPU ledger intentionally keeps uncertain workspaces out of the import
path. This script does not change the ledger or production data; it groups those
uncertain rows by the real SPU root and rescans that root so cross-level
main/SKU/detail folders can be reviewed together.
"""

from __future__ import annotations

import argparse
import csv
import json
import os
import sqlite3
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any

from scan_spu_ledger import (
    EXCLUDED_DIR_NAMES,
    IMAGE_EXTENSIONS,
    OTHER_IMAGE_EXTENSIONS,
    PS_SOURCE_EXTENSIONS,
    RELEVANT_EXTENSIONS,
    SOURCE_ARCHIVE_EXTENSIONS,
    build_sku_code,
    classify_file,
    extract_spu_candidates,
    has_hard_skip,
    has_modeling_source_segment,
    is_compatible_spu_token,
    normalize_spu,
    safe_relative,
)


REVIEW_HEADERS = [
    "decision",
    "spu",
    "store",
    "product_category",
    "spu_root_path",
    "needs_review_rows",
    "direct_all_three_rows",
    "aggregate_main_count",
    "aggregate_sku_count",
    "aggregate_detail_count",
    "aggregate_source_count",
    "hard_skip_file_count",
    "modeling_file_count",
    "status_reasons",
    "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"])


def choose_spu_root_path(item: sqlite3.Row) -> str:
    source_root = Path(item["source_root"])
    workspace_path = Path(item["workspace_path"])
    relative = Path(item["relative_workspace_path"])
    spu = normalize_spu(item["spu"])

    parts = list(relative.parts)
    for index, part in enumerate(parts):
        tokens = extract_spu_candidates(part, allow_unbranded=True)
        if any(is_compatible_spu_token(spu, token) for token in tokens):
            return str(source_root.joinpath(*parts[: index + 1]))

    # File-inferred and non-standard folders can have no SPU-looking segment.
    # In that case use the nearest useful parent instead of the leaf itself so
    # sibling main/SKU/detail folders can still be discovered.
    if len(parts) >= 2:
        return str(source_root.joinpath(*parts[:-1]))
    return str(workspace_path)


def scan_spu_root(root_path: str, spu: str) -> dict[str, Any]:
    root = Path(root_path)
    counts: Counter[str] = Counter()
    samples: dict[str, list[dict[str, Any]]] = defaultdict(list)
    hard_skip_file_count = 0
    modeling_file_count = 0

    if not root.exists():
        return {
            "counts": counts,
            "samples": samples,
            "hard_skip_file_count": 0,
            "modeling_file_count": 0,
            "exists": False,
        }

    for current, dir_names, file_names in os.walk(root):
        dir_names[:] = [
            name
            for name in dir_names
            if name not in EXCLUDED_DIR_NAMES and not has_hard_skip(str(Path(current) / name))
        ]

        current_path = Path(current)
        current_is_modeling = has_modeling_source_segment(str(current_path))
        for file_name in file_names:
            file_path = current_path / file_name
            extension = file_path.suffix.lower()
            if extension not in RELEVANT_EXTENSIONS:
                continue

            relative_path = safe_relative(file_path, root)
            if has_hard_skip(str(file_path)):
                hard_skip_file_count += 1
                continue
            if current_is_modeling:
                modeling_file_count += 1

            group = classify_file(relative_path, file_name, extension)
            if group == "other_source_archive" and extension in SOURCE_ARCHIVE_EXTENSIONS:
                # Keep archives visible, but do not promote generic zip files to
                # source unless the first-pass classifier saw source evidence.
                pass

            counts[group] += 1
            if group in {"main", "sku", "detail", "source"} and len(samples[group]) < 8:
                record = {
                    "filePath": str(file_path),
                    "relativePath": relative_path,
                    "fileName": file_name,
                    "fileSize": file_path.stat().st_size,
                }
                if group == "sku":
                    record["skuCode"] = build_sku_code(spu, file_name, counts[group] - 1)
                    record["barcodes"] = []
                samples[group].append(record)

    return {
        "counts": counts,
        "samples": samples,
        "hard_skip_file_count": hard_skip_file_count,
        "modeling_file_count": modeling_file_count,
        "exists": True,
    }


def decide_group(items: list[sqlite3.Row], aggregate: dict[str, Any]) -> str:
    counts: Counter[str] = aggregate["counts"]
    has_all_three = counts["main"] > 0 and counts["sku"] > 0 and counts["detail"] > 0
    direct_all_three = [
        item
        for item in items
        if int(item["main_count"] or 0) > 0
        and int(item["sku_count"] or 0) > 0
        and int(item["detail_count"] or 0) > 0
    ]
    reasons = " | ".join(sorted({str(item["status_reason"] or "") for item in items}))
    has_modeling_reason = "建模" in reasons or "3D" in reasons or aggregate["modeling_file_count"] > 0

    if not aggregate["exists"]:
        return "manual_missing_root"
    if direct_all_three and not has_modeling_reason:
        return "direct_ready_review"
    if has_all_three and not has_modeling_reason:
        return "cross_level_ready_review"
    if has_modeling_reason and not has_all_three:
        return "skip_likely_modeling_or_partial"
    if counts["detail"] == 0:
        return "skip_likely_no_detail"
    if counts["main"] == 0 or counts["sku"] == 0:
        return "partial_needs_pairing"
    return "manual_review"


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=REVIEW_HEADERS)
        writer.writeheader()
        writer.writerows(rows)


def main() -> int:
    parser = argparse.ArgumentParser(description="Second-pass review for needs_review SPU groups.")
    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()

    db_path = Path(args.db_path)
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    output_dir = Path(args.output_dir or f"output/tianlan-spu-ledger/needs-review-cross-level-{stamp}")
    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 = 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_root_path = choose_spu_root_path(item)
        key = (
            str(item["store"] or ""),
            normalize_spu(item["spu"]),
            str(item["source_root_name"] or ""),
            spu_root_path,
        )
        groups[key].append(item)

    review_rows: list[dict[str, Any]] = []
    details: list[dict[str, Any]] = []
    for (store, spu, source_root_name, spu_root_path), grouped_items in sorted(groups.items()):
        aggregate = scan_spu_root(spu_root_path, spu)
        counts: Counter[str] = aggregate["counts"]
        decision = decide_group(grouped_items, aggregate)
        status_reasons = sorted({str(item["status_reason"] or "") for item in grouped_items})
        workspace_samples = [str(item["workspace_path"]) for item in grouped_items[:8]]
        direct_all_three_rows = sum(
            1
            for item in grouped_items
            if int(item["main_count"] or 0) > 0
            and int(item["sku_count"] or 0) > 0
            and int(item["detail_count"] or 0) > 0
        )

        row = {
            "decision": decision,
            "spu": spu,
            "store": store,
            "product_category": grouped_items[0]["product_category"],
            "spu_root_path": spu_root_path,
            "needs_review_rows": len(grouped_items),
            "direct_all_three_rows": direct_all_three_rows,
            "aggregate_main_count": counts["main"],
            "aggregate_sku_count": counts["sku"],
            "aggregate_detail_count": counts["detail"],
            "aggregate_source_count": counts["source"],
            "hard_skip_file_count": aggregate["hard_skip_file_count"],
            "modeling_file_count": aggregate["modeling_file_count"],
            "status_reasons": " | ".join(status_reasons),
            "workspace_samples": " || ".join(workspace_samples),
            "main_samples": " || ".join(item["relativePath"] for item in aggregate["samples"].get("main", [])),
            "sku_samples": " || ".join(item["relativePath"] for item in aggregate["samples"].get("sku", [])),
            "detail_samples": " || ".join(item["relativePath"] for item in aggregate["samples"].get("detail", [])),
        }
        review_rows.append(row)
        details.append(
            {
                **row,
                "sourceRootName": source_root_name,
                "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"],
                        "sourceCount": item["source_count"],
                        "statusReason": item["status_reason"],
                    }
                    for item in grouped_items
                ],
                "samples": aggregate["samples"],
            }
        )

    write_csv(output_dir / "needs-review-cross-level-groups.csv", review_rows)
    (output_dir / "needs-review-cross-level-groups.json").write_text(
        json_dumps(details), encoding="utf-8"
    )

    summary = {
        "runId": run_id,
        "dbPath": str(db_path),
        "needsReviewItems": len(items),
        "groups": len(review_rows),
        "decisionCounts": dict(Counter(row["decision"] for row in review_rows)),
        "outputCsv": str(output_dir / "needs-review-cross-level-groups.csv"),
        "outputJson": str(output_dir / "needs-review-cross-level-groups.json"),
    }
    (output_dir / "summary.json").write_text(json_dumps(summary), encoding="utf-8")

    conn.close()
    print(json_dumps(summary))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())