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

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

Writes use openpyxl so the workbook structure, styles, data validations,
images, and non-target sheets are preserved.
"""
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

STANDARD_COLUMNS = [
    "序号", "客户姓名/公司", "国家", "城市", "客户属性", "客户类型",
    "主页/链接", "联系人", "职位", "电话/WhatsApp", "邮箱",
    "主营业务", "建联状态", "下次跟进", "备注",
]

FACEBOOK_COLUMNS = [*STANDARD_COLUMNS[:6], "公司官网", *STANDARD_COLUMNS[6:]]

LINKEDIN_COLUMNS = [
    "公司名称", "国家", "城市", "客户属性", "客户类型", "linkin链接",
    "联系人", "职位", "公司公共电话", "公司公共邮箱（任一有效即可）",
    "个人邮箱（不一定有效）", "公司主营业务", "建联状态", "备注",
]

GOOGLE_MAPS_COLUMNS = [
    "客户姓名/公司", "国家", "城市", "客户属性", "客户类型", "主页/链接",
    "联系人", "职位", "电话/WhatsApp", "邮箱", "主营业务",
    "建联状态", "下次跟进", "备注",
]

AUTO_WEBSITE_COLUMNS = [
    "序号", "客户姓名/公司", "国家", "城市", "客户属性", "客户类型", "主页/链接",
    "来源网站", "联系人", "职位", "电话/WhatsApp", "邮箱", "主营业务",
    "建联状态", "下次跟进", "备注",
]

COLUMN_ALIASES = {
    "客户姓名/公司": ["客户姓名/公司", "公司名称", "名称"],
    "公司名称": ["公司名称", "客户姓名/公司", "名称"],
    "主页/链接": ["主页/链接", "linkin链接", "LinkedIn链接", "链接", "网址"],
    "公司官网": ["公司官网", "官网", "官方网站", "Website", "Company Website"],
    "linkin链接": ["linkin链接", "LinkedIn链接", "主页/链接"],
    "电话/WhatsApp": ["电话/WhatsApp", "公司公共电话", "电话", "WhatsApp"],
    "公司公共电话": ["公司公共电话", "电话/WhatsApp", "电话"],
    "邮箱": ["邮箱", "公司公共邮箱（任一有效即可）", "公司公共邮箱", "公共邮箱"],
    "公司公共邮箱（任一有效即可）": ["公司公共邮箱（任一有效即可）", "公司公共邮箱", "公共邮箱", "邮箱"],
    "个人邮箱（不一定有效）": ["个人邮箱（不一定有效）", "个人邮箱"],
    "个人邮箱": ["个人邮箱", "个人邮箱（不一定有效）"],
    "主营业务": ["主营业务", "公司主营业务"],
    "公司主营业务": ["公司主营业务", "主营业务"],
    "建联状态": ["建联状态", "建联情况"],
    "建联情况": ["建联情况", "建联状态"],
}


def get_sheet_columns(sheet_name: str) -> List[str]:
    name = sheet_name.strip()
    if name == "Facebook":
        return FACEBOOK_COLUMNS
    if name == "LinkedIn":
        return LINKEDIN_COLUMNS
    if name == "Google Maps":
        return GOOGLE_MAPS_COLUMNS
    if name == "汽车网站精选线索":
        return AUTO_WEBSITE_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 文件不存在: {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 not records:
        return {"appended": 0, "skipped": 0, "total": 0}

    path = Path(excel_path)
    if not path.exists():
        raise FileNotFoundError(f"Excel 文件不存在: {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 == "序号" 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 文件不存在: {excel_path}")

    wb = load_workbook(path)
    if sheet_name not in wb.sheetnames:
        raise ValueError(f"Sheet 不存在: {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