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

'''
Excel workbook helpers for the Wuling dealer outreach skill.

New customer records are written directly to the single summary sheet. Legacy
channel sheet names are accepted as source labels for compatibility.
'''
from __future__ import annotations

from copy import copy
from pathlib import Path
from typing import Any, Dict, List, Optional

import pandas as pd
from openpyxl import load_workbook

SUMMARY_SHEET = '\u5ba2\u6237\u4fe1\u606f\u6c47\u603b\u8868'
CONVERSATION_SHEET = 'Facebook\u5bf9\u8bdd\u8bb0\u5f55'
CHANNEL_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',
}

SUMMARY_COLUMNS = [
    '\u516c\u53f8\u59d3\u540d', '\u56fd\u5bb6', '\u57ce\u5e02', '\u5ba2\u6237\u7c7b\u578b', '\u5b98\u7f51\u94fe\u63a5',
    '\u8054\u7cfb\u4eba', '\u804c\u4f4d', '\u4e2a\u4eba\u90ae\u7bb1', '\u8054\u7cfb\u4eba\u7535\u8bdd',
    'Facebook\u4e3b\u9875\u94fe\u63a5', 'linkined\u4e3b\u9875\u94fe\u63a5', 'google map\u94fe\u63a5',
    '\u516c\u5171\u7535\u8bdd/WhatsApp', '\u516c\u5171\u90ae\u7bb1', '\u5ba2\u6237\u5c5e\u6027',
    '\u7ebf\u7d22\u7b49\u7ea7', '\u9700\u4eba\u5de5\u786e\u8ba4', '\u5efa\u8054\u72b6\u6001',
    '\u4e0b\u6b21\u8ddf\u8fdb', '\u5907\u6ce8',
]

STANDARD_COLUMNS = [
    '\u5e8f\u53f7', '\u5ba2\u6237\u59d3\u540d/\u516c\u53f8', '\u56fd\u5bb6', '\u57ce\u5e02', '\u5ba2\u6237\u5c5e\u6027',
    '\u5ba2\u6237\u7c7b\u578b', '\u4e3b\u9875/\u94fe\u63a5', '\u8054\u7cfb\u4eba', '\u804c\u4f4d',
    '\u7535\u8bdd/WhatsApp', '\u90ae\u7bb1', '\u4e3b\u8425\u4e1a\u52a1', '\u5efa\u8054\u72b6\u6001',
    '\u4e0b\u6b21\u8ddf\u8fdb', '\u5907\u6ce8',
]
FACEBOOK_COLUMNS = [*STANDARD_COLUMNS[:6], '\u516c\u53f8\u5b98\u7f51', *STANDARD_COLUMNS[6:]]
LINKEDIN_COLUMNS = [
    '\u516c\u53f8\u540d\u79f0', '\u56fd\u5bb6', '\u57ce\u5e02', '\u5ba2\u6237\u5c5e\u6027', '\u5ba2\u6237\u7c7b\u578b',
    'linkin\u94fe\u63a5', '\u8054\u7cfb\u4eba', '\u804c\u4f4d', '\u516c\u53f8\u516c\u5171\u7535\u8bdd',
    '\u516c\u53f8\u516c\u5171\u90ae\u7bb1\uff08\u4efb\u4e00\u6709\u6548\u5373\u53ef\uff09',
    '\u4e2a\u4eba\u90ae\u7bb1\uff08\u4e0d\u4e00\u5b9a\u6709\u6548\uff09', '\u516c\u53f8\u4e3b\u8425\u4e1a\u52a1',
    '\u5efa\u8054\u72b6\u6001', '\u5907\u6ce8',
]
GOOGLE_MAPS_COLUMNS = [
    '\u5ba2\u6237\u59d3\u540d/\u516c\u53f8', '\u56fd\u5bb6', '\u57ce\u5e02', '\u5ba2\u6237\u5c5e\u6027', '\u5ba2\u6237\u7c7b\u578b',
    '\u4e3b\u9875/\u94fe\u63a5', '\u8054\u7cfb\u4eba', '\u804c\u4f4d', '\u7535\u8bdd/WhatsApp', '\u90ae\u7bb1',
    '\u4e3b\u8425\u4e1a\u52a1', '\u5efa\u8054\u72b6\u6001', '\u4e0b\u6b21\u8ddf\u8fdb', '\u5907\u6ce8',
]
AUTO_WEBSITE_COLUMNS = [
    '\u5e8f\u53f7', '\u5ba2\u6237\u59d3\u540d/\u516c\u53f8', '\u56fd\u5bb6', '\u57ce\u5e02', '\u5ba2\u6237\u5c5e\u6027',
    '\u5ba2\u6237\u7c7b\u578b', '\u4e3b\u9875/\u94fe\u63a5', '\u6765\u6e90\u7f51\u7ad9', '\u8054\u7cfb\u4eba',
    '\u804c\u4f4d', '\u7535\u8bdd/WhatsApp', '\u90ae\u7bb1', '\u4e3b\u8425\u4e1a\u52a1', '\u5efa\u8054\u72b6\u6001',
    '\u4e0b\u6b21\u8ddf\u8fdb', '\u5907\u6ce8',
]
FACEBOOK_CONVERSATION_COLUMNS = [
    '\u8bb0\u5f55ID', '\u5ba2\u6237\u5e8f\u53f7', '\u5ba2\u6237\u59d3\u540d/\u516c\u53f8', 'Facebook\u4e3b\u9875\u94fe\u63a5',
    'Messenger\u7ebf\u7a0bID', '\u6d88\u606f\u65f6\u95f4', '\u6d88\u606f\u65b9\u5411', '\u53d1\u4ef6\u4eba',
    '\u539f\u6587\u8bed\u8a00', '\u5bf9\u8bdd\u539f\u6587', '\u4e2d\u6587\u7ffb\u8bd1', '\u6d88\u606f\u7c7b\u578b',
    '\u662f\u5426\u6709\u6548\u5ba2\u6237\u56de\u590d', '\u5408\u4f5c\u610f\u5411', '\u610f\u5411\u5224\u65ad\u4f9d\u636e',
    '\u4e0b\u4e00\u6b65\u5efa\u8bae', '\u540c\u6b65\u65f6\u95f4', '\u6765\u6e90\u8d26\u53f7/Profile ID', '\u98ce\u9669\u6807\u8bb0',
]

COLUMN_ALIASES = {
    '\u5ba2\u6237\u59d3\u540d/\u516c\u53f8': ['\u5ba2\u6237\u59d3\u540d/\u516c\u53f8', '\u516c\u53f8\u540d\u79f0', '\u516c\u53f8\u59d3\u540d', '\u540d\u79f0'],
    '\u516c\u53f8\u540d\u79f0': ['\u516c\u53f8\u540d\u79f0', '\u5ba2\u6237\u59d3\u540d/\u516c\u53f8', '\u516c\u53f8\u59d3\u540d', '\u540d\u79f0'],
    '\u4e3b\u9875/\u94fe\u63a5': ['\u4e3b\u9875/\u94fe\u63a5', 'linkin\u94fe\u63a5', 'LinkedIn\u94fe\u63a5', '\u94fe\u63a5', '\u7f51\u5740'],
    '\u516c\u53f8\u5b98\u7f51': ['\u516c\u53f8\u5b98\u7f51', '\u5b98\u7f51\u94fe\u63a5', '\u5b98\u7f51', 'Website', 'Company Website'],
    'linkin\u94fe\u63a5': ['linkin\u94fe\u63a5', 'LinkedIn\u94fe\u63a5', 'linkined\u4e3b\u9875\u94fe\u63a5', '\u4e3b\u9875/\u94fe\u63a5'],
    '\u7535\u8bdd/WhatsApp': ['\u7535\u8bdd/WhatsApp', '\u516c\u53f8\u516c\u5171\u7535\u8bdd', '\u516c\u5171\u7535\u8bdd/WhatsApp', '\u7535\u8bdd', 'WhatsApp'],
    '\u90ae\u7bb1': ['\u90ae\u7bb1', '\u516c\u5171\u90ae\u7bb1', '\u516c\u53f8\u516c\u5171\u90ae\u7bb1\uff08\u4efb\u4e00\u6709\u6548\u5373\u53ef\uff09', '\u516c\u53f8\u516c\u5171\u90ae\u7bb1'],
    '\u4e3b\u8425\u4e1a\u52a1': ['\u4e3b\u8425\u4e1a\u52a1', '\u516c\u53f8\u4e3b\u8425\u4e1a\u52a1'],
    '\u5efa\u8054\u72b6\u6001': ['\u5efa\u8054\u72b6\u6001', '\u5efa\u8054\u60c5\u51b5'],
}


def get_sheet_columns(sheet_name: str) -> List[str]:
    name = sheet_name.strip()
    if name == SUMMARY_SHEET:
        return SUMMARY_COLUMNS
    if name == 'Facebook':
        return FACEBOOK_COLUMNS
    if name == 'LinkedIn':
        return LINKEDIN_COLUMNS
    if name == 'Google Maps':
        return GOOGLE_MAPS_COLUMNS
    if name == '\u6c7d\u8f66\u7f51\u7ad9\u7cbe\u9009\u7ebf\u7d22':
        return AUTO_WEBSITE_COLUMNS
    if name == CONVERSATION_SHEET:
        return FACEBOOK_CONVERSATION_COLUMNS
    return STANDARD_COLUMNS


def _aliases(column: str) -> List[str]:
    values = COLUMN_ALIASES.get(column, [column])
    return list(dict.fromkeys([column, *values]))


def _first_value(record: Dict[str, Any], column: str) -> Any:
    for key in _aliases(column):
        value = record.get(key, '')
        if value not in (None, ''):
            return value
    return ''


def _headers(ws) -> List[str]:
    values = []
    for cell in ws[1]:
        value = '' if cell.value is None else str(cell.value).strip()
        if value:
            values.append(value)
    return values


def _header_index(ws) -> Dict[str, int]:
    return {header: idx for idx, header in enumerate(_headers(ws), start=1)}


def _find_header(headers: Dict[str, int], column: str) -> Optional[str]:
    for alias in _aliases(column):
        if alias in headers:
            return alias
    return None


def _cell_value(row_values: Dict[str, Any], column: str) -> str:
    for alias in _aliases(column):
        value = row_values.get(alias, '')
        if value not in (None, ''):
            return str(value).strip()
    return ''


def _copy_row_style(ws, source_row: int, target_row: int) -> None:
    if source_row < 1 or source_row == target_row:
        return
    for col in range(1, ws.max_column + 1):
        source = ws.cell(source_row, col)
        target = ws.cell(target_row, col)
        if source.has_style:
            target._style = copy(source._style)
        if source.number_format:
            target.number_format = source.number_format
        if source.alignment:
            target.alignment = copy(source.alignment)
        if source.protection:
            target.protection = copy(source.protection)


def read_sheet(excel_path: str, sheet_name: str) -> pd.DataFrame:
    path = Path(excel_path)
    if not path.exists():
        raise FileNotFoundError(f'Excel file not found: {excel_path}')
    df = pd.read_excel(excel_path, sheet_name=sheet_name)
    expected_cols = get_sheet_columns(sheet_name)
    for col in expected_cols:
        if col in df.columns:
            continue
        for alias in _aliases(col):
            if alias in df.columns:
                df[col] = df[alias]
                break
        else:
            df[col] = ''
    return df


def normalize_record(record: Dict[str, Any], sheet_name: str) -> Dict[str, Any]:
    normalized: Dict[str, Any] = {}
    for col in get_sheet_columns(sheet_name):
        normalized[col] = _first_value(record, col)
    return normalized


def append_records(excel_path: str, sheet_name: str, records: List[Dict[str, Any]], dedup_keys: Optional[List[str]] = None) -> Dict[str, Any]:
    if sheet_name in CHANNEL_SOURCE_SHEETS or sheet_name == SUMMARY_SHEET:
        try:
            from .direct_summary import append_summary_records
        except ImportError:
            from direct_summary import append_summary_records
        return append_summary_records(excel_path, records, source_sheet=sheet_name)
    if not records:
        return {'appended': 0, 'skipped': 0, 'total': 0}
    path = Path(excel_path)
    if not path.exists():
        raise FileNotFoundError(f'Excel file not found: {excel_path}')
    wb = load_workbook(path)
    if sheet_name not in wb.sheetnames:
        ws = wb.create_sheet(sheet_name)
        for idx, header in enumerate(get_sheet_columns(sheet_name), start=1):
            ws.cell(1, idx, header)
    else:
        ws = wb[sheet_name]
    headers = _header_index(ws)
    if not headers:
        for idx, header in enumerate(get_sheet_columns(sheet_name), start=1):
            ws.cell(1, idx, header)
        headers = _header_index(ws)
    for expected_header in get_sheet_columns(sheet_name):
        if expected_header not in headers and not _find_header(headers, expected_header):
            next_col = ws.max_column + 1
            ws.cell(1, next_col, expected_header)
            headers[expected_header] = next_col
    keys = [key for key in (dedup_keys or []) if _find_header(headers, key)]
    existing_signatures = set()
    if keys:
        for row in ws.iter_rows(min_row=2, values_only=False):
            values = {header: row[col_idx - 1].value for header, col_idx in headers.items() if col_idx <= len(row)}
            sig = '|'.join(_cell_value(values, key).casefold() for key in keys)
            if sig.strip('|'):
                existing_signatures.add(sig)
    appended = 0
    skipped = 0
    last_style_row = ws.max_row if ws.max_row > 1 else 1
    for record in records:
        normalized = normalize_record(record, sheet_name)
        if keys:
            sig = '|'.join(str(_first_value(normalized, key)).strip().casefold() for key in keys)
            if sig.strip('|') and sig in existing_signatures:
                skipped += 1
                continue
            if sig.strip('|'):
                existing_signatures.add(sig)
        target_row = ws.max_row + 1
        _copy_row_style(ws, last_style_row, target_row)
        for header, col_idx in headers.items():
            if header.startswith('Unnamed'):
                continue
            value = _first_value(normalized, header)
            if header == '\u5e8f\u53f7' and value == '':
                value = target_row - 1
            ws.cell(target_row, col_idx, value)
        appended += 1
        last_style_row = target_row
    wb.save(path)
    return {'appended': appended, 'skipped': skipped, 'total': max(ws.max_row - 1, 0)}


def update_status(excel_path: str, sheet_name: str, filters: Dict[str, Any], updates: Dict[str, Any]) -> int:
    path = Path(excel_path)
    if not path.exists():
        raise FileNotFoundError(f'Excel file not found: {excel_path}')
    wb = load_workbook(path)
    if sheet_name not in wb.sheetnames:
        raise ValueError(f'Sheet not found: {sheet_name}')
    ws = wb[sheet_name]
    headers = _header_index(ws)
    updated = 0
    for row_idx in range(2, ws.max_row + 1):
        row_values = {header: ws.cell(row_idx, col_idx).value for header, col_idx in headers.items()}
        matched = True
        for key, expected in filters.items():
            actual = _cell_value(row_values, key)
            if actual != str(expected):
                matched = False
                break
        if not matched:
            continue
        for key, value in updates.items():
            header = _find_header(headers, key)
            if header:
                ws.cell(row_idx, headers[header], value)
        updated += 1
    wb.save(path)
    return updated