wuling-overseas-dealer-expansion-skill/scripts/common/build_customer_summary.py

'''Build or repair the single customer summary sheet.'''
from __future__ import annotations

import argparse
import json
import re
from collections import Counter
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple

from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill
from openpyxl.utils import get_column_letter

try:
    from .artifact_manager import create_backup_once, resolve_artifact_path
    from .workbook_resolver import resolve_workbook_path
    from .direct_summary import SUMMARY_COLUMNS, SUMMARY_SHEET, identity_keys, merge_record, normalize_summary_record
except ImportError:  # pragma: no cover
    from artifact_manager import create_backup_once, resolve_artifact_path
    from workbook_resolver import resolve_workbook_path
    from direct_summary import SUMMARY_COLUMNS, SUMMARY_SHEET, identity_keys, merge_record, normalize_summary_record

CONVERSATION_SHEET = 'Facebook\u5bf9\u8bdd\u8bb0\u5f55'
SKIP_SHEETS = {'\u586b\u5199\u8bf4\u660e', '\u9644\u4ef6', CONVERSATION_SHEET}
LEGACY_SOURCE_SHEETS = {
    'Facebook', 'LinkedIn', 'Google Maps', 'TikTok', '\u534f\u4f1a\u5546\u4f1a',
    '\u672c\u5730\u6c7d\u8f66\u7f51\u7ad9', '\u6c7d\u8f66\u7f51\u7ad9\u7cbe\u9009\u7ebf\u7d22',
    'Sheet11', '\u653f\u5e9c\u91c7\u8d2d\u6295\u6807', '\u6d4b\u8bc4\u535a\u4e3b',
}


def clean(value: Any) -> str:
    if value is None:
        return ''
    return re.sub(r'\s+', ' ', str(value).strip())


def headers_for(ws) -> Dict[str, int]:
    return {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1) if clean(cell.value)}


def row_to_record(ws, row_idx: int, headers: Dict[str, int]) -> Dict[str, Any]:
    return {header: ws.cell(row_idx, col_idx).value for header, col_idx in headers.items()}


def has_customer_evidence(record: Dict[str, Any]) -> bool:
    evidence = [
        '\u516c\u53f8\u59d3\u540d', '\u5ba2\u6237\u59d3\u540d/\u516c\u53f8', '\u516c\u53f8\u540d\u79f0',
        '\u4e3b\u9875/\u94fe\u63a5', '\u5b98\u7f51\u94fe\u63a5', '\u516c\u53f8\u5b98\u7f51',
        'Facebook\u4e3b\u9875\u94fe\u63a5', 'linkined\u4e3b\u9875\u94fe\u63a5', 'google map\u94fe\u63a5',
        '\u516c\u5171\u90ae\u7bb1', '\u4e2a\u4eba\u90ae\u7bb1', '\u90ae\u7bb1',
        '\u516c\u5171\u7535\u8bdd/WhatsApp', '\u7535\u8bdd/WhatsApp', '\u8054\u7cfb\u4eba\u7535\u8bdd',
    ]
    return any(clean(record.get(key)) for key in evidence)


def source_sheets(wb) -> List[str]:
    ordered: List[str] = []
    if SUMMARY_SHEET in wb.sheetnames:
        ordered.append(SUMMARY_SHEET)
    for sheet in wb.sheetnames:
        if sheet == SUMMARY_SHEET or sheet in SKIP_SHEETS:
            continue
        if sheet in LEGACY_SOURCE_SHEETS:
            ordered.append(sheet)
    return ordered


def read_all_customer_records(wb) -> Tuple[List[Dict[str, Any]], Counter, int]:
    records: List[Dict[str, Any]] = []
    source_counts: Counter = Counter()
    skipped_blank = 0
    for sheet in source_sheets(wb):
        ws = wb[sheet]
        headers = headers_for(ws)
        if not headers:
            continue
        for row_idx in range(2, ws.max_row + 1):
            raw = row_to_record(ws, row_idx, headers)
            if not has_customer_evidence(raw):
                skipped_blank += 1
                continue
            normalized = normalize_summary_record(raw, sheet)
            if not has_customer_evidence(normalized):
                skipped_blank += 1
                continue
            records.append(normalized)
            source_counts[sheet] += 1
    return records, source_counts, skipped_blank


def merge_records(records: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
    output: List[Dict[str, Any]] = []
    key_to_index: Dict[Tuple[str, str], int] = {}
    for record in records:
        match_idx: Optional[int] = None
        for key in identity_keys(record):
            if key in key_to_index:
                match_idx = key_to_index[key]
                break
        if match_idx is None:
            output.append(record)
            match_idx = len(output) - 1
        else:
            output[match_idx] = merge_record(output[match_idx], record)
        for key in identity_keys(output[match_idx]):
            key_to_index.setdefault(key, match_idx)
    return output


def write_summary_sheet(wb, rows: Sequence[Dict[str, Any]], sheet_name: str) -> None:
    if sheet_name in wb.sheetnames:
        del wb[sheet_name]
    ws = wb.create_sheet(sheet_name, 0)
    ws.append(SUMMARY_COLUMNS)
    for row in rows:
        ws.append([row.get(header, '') for header in SUMMARY_COLUMNS])
    ws.freeze_panes = 'A2'
    ws.auto_filter.ref = ws.dimensions
    header_fill = PatternFill(fill_type='solid', fgColor='D9EAF7')
    for cell in ws[1]:
        cell.font = Font(bold=True)
        cell.fill = header_fill
    widths = [28, 12, 16, 22, 34, 18, 18, 28, 24, 34, 34, 34, 24, 28, 20, 12, 20, 18, 18, 80]
    for idx, width in enumerate(widths, start=1):
        ws.column_dimensions[get_column_letter(idx)].width = width


def lock_file_for(path: Path) -> Path:
    return path.with_name('~$' + path.name)


def build_report(workbook_path: Path, backup_path: Optional[Path], rows: Sequence[Dict[str, Any]], source_counts: Counter, skipped_blank: int, dry_run: bool, sheet_name: str) -> Dict[str, Any]:
    grade_col = '\u7ebf\u7d22\u7b49\u7ea7'
    manual_col = '\u9700\u4eba\u5de5\u786e\u8ba4'
    attr_col = '\u5ba2\u6237\u5c5e\u6027'
    type_col = '\u5ba2\u6237\u7c7b\u578b'
    return {
        'workbook': str(workbook_path),
        'summary_sheet': sheet_name,
        'summary_headers': SUMMARY_COLUMNS,
        'dry_run': dry_run,
        'backup': str(backup_path) if backup_path else '',
        'summary_rows': len(rows),
        'source_rows': dict(source_counts),
        'skipped_blank_or_reserved_rows': skipped_blank,
        'attribute_counts': dict(Counter(clean(row.get(attr_col)) for row in rows if clean(row.get(attr_col)))),
        'type_counts': dict(Counter(clean(row.get(type_col)) for row in rows if clean(row.get(type_col)))),
        'lead_grade_counts': dict(Counter(clean(row.get(grade_col)) for row in rows if clean(row.get(grade_col)))),
        'manual_review_counts': dict(Counter(clean(row.get(manual_col)) for row in rows if clean(row.get(manual_col)))),
        'missing_counts': {
            '\u5ba2\u6237\u5c5e\u6027': sum(1 for row in rows if not clean(row.get(attr_col))),
            '\u5ba2\u6237\u7c7b\u578b': sum(1 for row in rows if not clean(row.get(type_col))),
            '\u7ebf\u7d22\u7b49\u7ea7': sum(1 for row in rows if not clean(row.get(grade_col))),
            '\u9700\u4eba\u5de5\u786e\u8ba4': sum(1 for row in rows if not clean(row.get(manual_col))),
            '\u5907\u6ce8': sum(1 for row in rows if not clean(row.get('\u5907\u6ce8'))),
        },
    }


def save_json_report(report: Dict[str, Any], output: str) -> None:
    if not output:
        return
    path = resolve_artifact_path(output, kind='summary_report', default_name='report.json')
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8')


def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description='Build or preview the single customer summary sheet.')
    parser.add_argument('--excel', default='', help='Workbook path. If omitted, resolve the project workbook by skill rules.')
    parser.add_argument('--summary-sheet', default=SUMMARY_SHEET, help='Summary sheet name.')
    parser.add_argument('--write-summary', action='store_true', help='Write or overwrite the summary sheet. Omit for preview only.')
    parser.add_argument('--dry-run', action='store_true', help='Preview only; never saves the workbook.')
    parser.add_argument('--no-backup', action='store_true', help='Skip backup when writing.')
    parser.add_argument('--output', default='', help='Optional JSON report path.')
    parser.add_argument('--run-id', default='', help='Run ID used for artifact and backup paths.')
    return parser.parse_args(argv)


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = parse_args(argv)
    resolved = resolve_workbook_path(args.excel, create_from_template=False)
    workbook_path = resolved.get('path')
    if not workbook_path:
        raise FileNotFoundError('No outreach workbook found. Pass --excel or create one from the skill blank template in write-enabled workflows.')
    workbook_path = Path(workbook_path)
    sheet_name = clean(args.summary_sheet) or SUMMARY_SHEET
    should_write = bool(args.write_summary and not args.dry_run)
    if should_write and lock_file_for(workbook_path).exists():
        raise PermissionError(f'Workbook appears to be open in Excel: {lock_file_for(workbook_path)}')
    wb = load_workbook(workbook_path)
    records, source_counts, skipped_blank = read_all_customer_records(wb)
    rows = merge_records(records)
    backup_path: Optional[Path] = None
    if should_write:
        if not args.no_backup:
            backup_path = create_backup_once(workbook_path, purpose='summary', run_id=args.run_id or None)
        write_summary_sheet(wb, rows, sheet_name)
        wb.save(workbook_path)
    report = build_report(workbook_path, backup_path, rows, source_counts, skipped_blank, dry_run=not should_write, sheet_name=sheet_name)
    save_json_report(report, args.output)
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0


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