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

'''Direct, deduplicated writes to the single customer summary sheet.'''
from __future__ import annotations

import re
from copy import copy
from pathlib import Path
from typing import Any, Dict, Iterable, List, Tuple
from urllib.parse import urlparse

from openpyxl import load_workbook

SUMMARY_SHEET = '\u5ba2\u6237\u4fe1\u606f\u6c47\u603b\u8868'
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',
]

NO = '\u5426'
UNCONTACTED = '\u672a\u8054\u7cfb'
SOURCE_PREFIX = '\u5ba2\u6237\u6765\u6e90\uff1a'
MANUAL_FLAGS = [
    '\u4e3b\u4f53\u5f52\u5c5e\u5f85\u786e\u8ba4',
    '\u65b0\u8f66\u4e1a\u52a1\u5f85\u786e\u8ba4',
    '\u5e73\u53f0\u4e0e\u884c\u4e1a\u6e20\u9053\u4e3b\u4f53\u5f85\u786e\u8ba4',
    '\u5e73\u53f0\u4e0e\u884c\u4e1a\u6e20\u9053\u4f5c\u7528\u5f85\u786e\u8ba4',
    '\u4ec5\u7535\u8bdd/WhatsApp\u5f85\u4eba\u5de5\u786e\u8ba4',
    '\u8be6\u7ec6\u4fe1\u606f\u5f85\u786e\u8ba4',
    '\u6392\u4ed6\u534f\u8bae\u53ca\u65b0\u589e\u54c1\u724c\u6743\u9650\u5f85\u786e\u8ba4',
]

MULTI_FIELDS = {
    '\u5b98\u7f51\u94fe\u63a5', '\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', '\u8054\u7cfb\u4eba', '\u804c\u4f4d', '\u5907\u6ce8',
}
GRADE_ORDER = {'A': 3, 'B': 2, 'C': 1}
PLATFORM_HOSTS = ('facebook.com', 'linkedin.com', 'google.com', 'maps.app.goo.gl', 'instagram.com', 'wa.me', 'youtube.com', 'tiktok.com')


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


def first(record: Dict[str, Any], *keys: str) -> str:
    for key in keys:
        value = clean(record.get(key))
        if value:
            return value
    return ''


def split_values(value: Any) -> List[str]:
    if isinstance(value, list):
        raw = [clean(item) for item in value]
    else:
        raw = [clean(item) for item in re.split(r'[\uff1b;|\n]+', clean(value))]
    return list(dict.fromkeys(item for item in raw if item))


def merge_values(left: Any, right: Any) -> str:
    return '\uff1b'.join(dict.fromkeys([*split_values(left), *split_values(right)]))


def normalized_name(value: str) -> str:
    value = clean(value).casefold()
    return re.sub(r'[^0-9a-z\u00c0-\u024f\u0600-\u06ff\u4e00-\u9fff]+', '', value)


def normalized_url(value: str) -> str:
    value = clean(value).casefold().rstrip('/')
    parsed = urlparse(value if '://' in value else 'https://' + value)
    host = parsed.netloc.removeprefix('www.')
    path = parsed.path.rstrip('/')
    if 'google.' in host or 'maps.app.goo.gl' in host:
        return (host + path + ('?' + parsed.query if parsed.query else '')).rstrip('/')
    return (host + path).rstrip('/')


def is_company_website(value: str) -> bool:
    parsed = urlparse(clean(value) if '://' in clean(value) else 'https://' + clean(value))
    host = parsed.netloc.casefold().removeprefix('www.')
    return bool(host and not any(item in host for item in PLATFORM_HOSTS))


def host_contains(value: str, token: str) -> bool:
    parsed = urlparse(clean(value) if '://' in clean(value) else 'https://' + clean(value))
    return token in parsed.netloc.casefold()


def row_dict(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 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 ensure_summary_sheet(wb):
    if SUMMARY_SHEET in wb.sheetnames:
        ws = wb[SUMMARY_SHEET]
    else:
        ws = wb.create_sheet(SUMMARY_SHEET, 0)
    headers = {clean(cell.value): idx for idx, cell in enumerate(ws[1], start=1) if clean(cell.value)}
    if not headers:
        for idx, header in enumerate(SUMMARY_COLUMNS, start=1):
            ws.cell(1, idx, header)
        headers = {header: idx for idx, header in enumerate(SUMMARY_COLUMNS, start=1)}
    for header in SUMMARY_COLUMNS:
        if header not in headers:
            col_idx = ws.max_column + 1
            ws.cell(1, col_idx, header)
            headers[header] = col_idx
    return ws, headers


def note_text(record: Dict[str, Any], source_sheet: str) -> str:
    note = first(record, '\u5907\u6ce8', 'note', 'notes', 'remarks')
    source = first(record, '\u6765\u6e90\u7f51\u7ad9', 'source', 'source_site') or source_sheet
    if source and SOURCE_PREFIX not in note:
        note = merge_values(note, SOURCE_PREFIX + source)
    return note


def infer_lead_grade(record: Dict[str, Any]) -> str:
    explicit = first(record, '\u7ebf\u7d22\u7b49\u7ea7', '\u7b49\u7ea7', 'lead_grade').upper()
    if explicit in GRADE_ORDER:
        return explicit
    text = ' '.join(clean(v) for v in record.values()).casefold()
    if re.search(r'import|distribut|dealer network|reseau|r\u00e9seau|national|multibrand|multi-brand|multimarque|association|chamber|platform|marketplace|fleet|group', text):
        return 'A'
    if re.search(r'showroom|dealer|concession|auto|motors|vehicule|v\u00e9hicule|car sales|new car|used car|occasion', text):
        return 'B'
    return 'C'


def manual_review_value(record: Dict[str, Any], grade: str) -> str:
    explicit = first(record, '\u9700\u4eba\u5de5\u786e\u8ba4', 'manual_review')
    if explicit and explicit.casefold() not in {'no', 'false', '0', NO}:
        return explicit
    text = ' '.join(clean(v) for v in record.values())
    flags = [flag for flag in MANUAL_FLAGS if flag in text]
    risk_flags = record.get('risk_flags') or record.get('risk_flag') or record.get('\u98ce\u9669\u6807\u8bb0')
    if isinstance(risk_flags, list):
        flags.extend(clean(item) for item in risk_flags if clean(item))
    elif clean(risk_flags):
        flags.extend(split_values(risk_flags))
    if flags:
        return '\uff1b'.join(dict.fromkeys(flags))
    if grade == 'C':
        return '\u8be6\u7ec6\u4fe1\u606f\u5f85\u786e\u8ba4'
    return NO


def normalize_summary_record(record: Dict[str, Any], source_sheet: str) -> Dict[str, Any]:
    link = first(record, '\u4e3b\u9875/\u94fe\u63a5', 'page_url', 'profile_url', 'link', 'url', 'Link', 'URL')
    website = first(record, '\u5b98\u7f51\u94fe\u63a5', '\u516c\u53f8\u5b98\u7f51', 'website', 'company_website')
    if not website and link and is_company_website(link):
        website = link
    facebook = first(record, 'Facebook\u4e3b\u9875\u94fe\u63a5', 'Facebook\u94fe\u63a5', 'facebook_link', 'facebook_url')
    linkedin = first(record, 'linkined\u4e3b\u9875\u94fe\u63a5', 'LinkedIn\u4e3b\u9875\u94fe\u63a5', 'linkin\u94fe\u63a5', 'linkedin_link', 'linkedin_url')
    maps = first(record, 'google map\u94fe\u63a5', 'Google Maps\u94fe\u63a5', 'maps_link', 'maps_url', 'google_maps_url')
    if link and not is_company_website(link):
        if host_contains(link, 'facebook.com') and not facebook:
            facebook = link
        elif host_contains(link, 'linkedin.com') and not linkedin:
            linkedin = link
        elif host_contains(link, 'google.') or host_contains(link, 'maps.app.goo.gl'):
            maps = link

    public_email = first(record, '\u516c\u5171\u90ae\u7bb1', '\u90ae\u7bb1', '\u516c\u53f8\u516c\u5171\u90ae\u7bb1\uff08\u4efb\u4e00\u6709\u6548\u5373\u53ef\uff09', 'email')
    personal_email = first(record, '\u4e2a\u4eba\u90ae\u7bb1', '\u4e2a\u4eba\u90ae\u7bb1\uff08\u4e0d\u4e00\u5b9a\u6709\u6548\uff09', 'personal_email')
    public_phone = first(record, '\u516c\u5171\u7535\u8bdd/WhatsApp', '\u7535\u8bdd/WhatsApp', '\u516c\u53f8\u516c\u5171\u7535\u8bdd', 'phone', 'whatsapp')
    contact_phone = first(record, '\u8054\u7cfb\u4eba\u7535\u8bdd', 'contact_phone')
    grade = infer_lead_grade(record)
    normalized = {
        '\u516c\u53f8\u59d3\u540d': first(record, '\u516c\u53f8\u59d3\u540d', '\u5ba2\u6237\u59d3\u540d/\u516c\u53f8', '\u516c\u53f8\u540d\u79f0', 'name', 'dealer_name'),
        '\u56fd\u5bb6': first(record, '\u56fd\u5bb6', 'country'),
        '\u57ce\u5e02': first(record, '\u57ce\u5e02', 'city'),
        '\u5ba2\u6237\u7c7b\u578b': first(record, '\u5ba2\u6237\u7c7b\u578b', 'customer_type'),
        '\u5b98\u7f51\u94fe\u63a5': website,
        '\u8054\u7cfb\u4eba': first(record, '\u8054\u7cfb\u4eba', 'contact'),
        '\u804c\u4f4d': first(record, '\u804c\u4f4d', 'title', 'position'),
        '\u4e2a\u4eba\u90ae\u7bb1': personal_email,
        '\u8054\u7cfb\u4eba\u7535\u8bdd': contact_phone,
        'Facebook\u4e3b\u9875\u94fe\u63a5': facebook,
        'linkined\u4e3b\u9875\u94fe\u63a5': linkedin,
        'google map\u94fe\u63a5': maps,
        '\u516c\u5171\u7535\u8bdd/WhatsApp': public_phone,
        '\u516c\u5171\u90ae\u7bb1': public_email,
        '\u5ba2\u6237\u5c5e\u6027': first(record, '\u5ba2\u6237\u5c5e\u6027', 'customer_attribute'),
        '\u7ebf\u7d22\u7b49\u7ea7': grade,
        '\u9700\u4eba\u5de5\u786e\u8ba4': manual_review_value(record, grade),
        '\u5efa\u8054\u72b6\u6001': first(record, '\u5efa\u8054\u72b6\u6001', '\u5efa\u8054\u60c5\u51b5', 'status') or UNCONTACTED,
        '\u4e0b\u6b21\u8ddf\u8fdb': first(record, '\u4e0b\u6b21\u8ddf\u8fdb', 'next_follow_up'),
        '\u5907\u6ce8': note_text(record, source_sheet),
    }
    return normalized


def identity_keys(record: Dict[str, Any]) -> List[Tuple[str, str]]:
    keys: List[Tuple[str, str]] = []
    for field in ('\u516c\u5171\u90ae\u7bb1', '\u4e2a\u4eba\u90ae\u7bb1'):
        for value in split_values(record.get(field)):
            keys.append(('email', value.casefold()))
    for field in ('\u516c\u5171\u7535\u8bdd/WhatsApp', '\u8054\u7cfb\u4eba\u7535\u8bdd'):
        for value in split_values(record.get(field)):
            digits = re.sub(r'\D+', '', value)
            if len(digits) >= 7:
                keys.append(('phone', digits))
    for field in ('\u5b98\u7f51\u94fe\u63a5', 'Facebook\u4e3b\u9875\u94fe\u63a5', 'linkined\u4e3b\u9875\u94fe\u63a5', 'google map\u94fe\u63a5'):
        for value in split_values(record.get(field)):
            url_key = normalized_url(value)
            if url_key:
                keys.append(('url', url_key))
    name_key = normalized_name(clean(record.get('\u516c\u53f8\u59d3\u540d')))
    if name_key:
        keys.append(('name', name_key))
    return list(dict.fromkeys(keys))


def merge_record(existing: Dict[str, Any], incoming: Dict[str, Any]) -> Dict[str, Any]:
    merged = dict(existing)
    for field in SUMMARY_COLUMNS:
        left = clean(merged.get(field))
        right = clean(incoming.get(field))
        if not left and right:
            merged[field] = right
        elif field in MULTI_FIELDS and right:
            merged[field] = merge_values(left, right)
    if GRADE_ORDER.get(clean(incoming.get('\u7ebf\u7d22\u7b49\u7ea7')), 0) > GRADE_ORDER.get(clean(merged.get('\u7ebf\u7d22\u7b49\u7ea7')), 0):
        merged['\u7ebf\u7d22\u7b49\u7ea7'] = incoming.get('\u7ebf\u7d22\u7b49\u7ea7')
    left_manual = clean(merged.get('\u9700\u4eba\u5de5\u786e\u8ba4'))
    right_manual = clean(incoming.get('\u9700\u4eba\u5de5\u786e\u8ba4'))
    if right_manual and right_manual != NO:
        merged['\u9700\u4eba\u5de5\u786e\u8ba4'] = merge_values('' if left_manual == NO else left_manual, right_manual)
    elif not left_manual:
        merged['\u9700\u4eba\u5de5\u786e\u8ba4'] = NO
    return merged


def index_existing(ws, headers: Dict[str, int]) -> Dict[Tuple[str, str], int]:
    index: Dict[Tuple[str, str], int] = {}
    for row_idx in range(2, ws.max_row + 1):
        record = row_dict(ws, row_idx, headers)
        if not any(clean(record.get(col)) for col in SUMMARY_COLUMNS):
            continue
        for key in identity_keys(record):
            index.setdefault(key, row_idx)
    return index


def write_row(ws, headers: Dict[str, int], row_idx: int, record: Dict[str, Any]) -> None:
    for header in SUMMARY_COLUMNS:
        ws.cell(row_idx, headers[header], record.get(header, ''))


def append_summary_records(excel_path: str, records: Iterable[Dict[str, Any]], source_sheet: str = '') -> Dict[str, Any]:
    records = list(records)
    if not records:
        return {'appended': 0, 'merged': 0, 'skipped': 0, 'total': 0, 'sheet': SUMMARY_SHEET}
    path = Path(excel_path)
    if not path.exists():
        raise FileNotFoundError(f'Excel file not found: {excel_path}')
    wb = load_workbook(path)
    ws, headers = ensure_summary_sheet(wb)
    existing_index = index_existing(ws, headers)
    appended = 0
    merged = 0
    skipped = 0
    last_style_row = ws.max_row if ws.max_row > 1 else 1
    for raw in records:
        normalized = normalize_summary_record(raw, source_sheet)
        if not any(clean(normalized.get(field)) for field in ('\u516c\u53f8\u59d3\u540d', '\u5b98\u7f51\u94fe\u63a5', 'Facebook\u4e3b\u9875\u94fe\u63a5', 'linkined\u4e3b\u9875\u94fe\u63a5', 'google map\u94fe\u63a5', '\u516c\u5171\u90ae\u7bb1', '\u516c\u5171\u7535\u8bdd/WhatsApp')):
            skipped += 1
            continue
        match_row = None
        for key in identity_keys(normalized):
            if key in existing_index:
                match_row = existing_index[key]
                break
        if match_row:
            existing = row_dict(ws, match_row, headers)
            write_row(ws, headers, match_row, merge_record(existing, normalized))
            for key in identity_keys(row_dict(ws, match_row, headers)):
                existing_index.setdefault(key, match_row)
            merged += 1
            continue
        row_idx = ws.max_row + 1
        copy_row_style(ws, last_style_row, row_idx)
        write_row(ws, headers, row_idx, normalized)
        for key in identity_keys(normalized):
            existing_index.setdefault(key, row_idx)
        appended += 1
        last_style_row = row_idx
    wb.save(path)
    return {'appended': appended, 'merged': merged, 'skipped': skipped, 'total': max(ws.max_row - 1, 0), 'sheet': SUMMARY_SHEET, 'source': source_sheet}