import json
import os
import re
from collections import defaultdict
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from decimal import Decimal
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import psycopg2
import requests
from openpyxl import load_workbook


BASE_DIR = Path(__file__).resolve().parent
ENV_FILE = BASE_DIR / ".env"
DEFAULT_MAPPING_FILE = BASE_DIR / "店铺映射表.xlsx"
DEFAULT_START_DATE = "2026-03-01"
DEFAULT_END_DATE = "2026-03-01"
DEFAULT_STORE_FIELD_NAME = "业务店铺名称"
STORE_FIELD_CANDIDATES = ["业务店铺名称", "店铺名称", "门店名称", "店铺", "名称"]
SUPPORTED_PROJECTS = ["大眼睛", "羽心堂", "MMMJ", "EM", "春回盏", "BROKEN ZOO"]


@dataclass(frozen=True)
class MetricConfig:
    name: str
    description: str
    sql_template: str
    field_suffix: str


METRIC_CONFIGS = {
    "sales": MetricConfig(
        name="sales",
        description="销售额",
        field_suffix="总销售额",
        sql_template="""
SET TIMEZONE = 'Asia/Shanghai';

WITH base_data AS (
    SELECT
        f.支付时间::DATE AS 统计日期,
        f.店铺主键,
        s.店铺名称 AS 原始店铺名称,
        p.品牌名称,
        f.业务模式,
        f.实付分摊金额 AS 指标金额
    FROM analytics_layer.fact_sales f
    LEFT JOIN analytics_layer.dim_shop s
        ON f.店铺主键 = s.店铺主键
    LEFT JOIN analytics_layer.dim_product p
        ON f.商品主键 = p.商品主键
       AND f.统一规格编码 = p.统一规格编码
    WHERE f.支付时间 >= %(start_ts)s
      AND f.支付时间 < %(end_ts_exclusive)s
),
brand_split AS (
    SELECT
        统计日期,
        店铺主键,
        原始店铺名称,
        业务模式,
        CASE
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '羽心堂'
                THEN 原始店铺名称 || '_羽心堂'
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '春回盏'
                THEN 原始店铺名称 || '_春回盏'
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'Mama Mujer'
                THEN 原始店铺名称 || '_Mama Mujer'
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'ERBA MIGINA'
                THEN 原始店铺名称 || '_ERBA MIGINA'
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'BROKEN ZOO'
                THEN 原始店铺名称 || '_BROKEN ZOO'
            WHEN 店铺主键 IN (1,28,29)
                THEN 原始店铺名称 || '_其他'
            WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE '羽心堂'
                THEN 原始店铺名称 || '_羽心堂'
            WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE 'Mama Mujer'
                THEN 原始店铺名称 || '_Mama Mujer'
            WHEN 店铺主键 IN (3,9)
                THEN 原始店铺名称 || '_其他'
            WHEN 店铺主键 IS NULL OR 原始店铺名称 IS NULL
                THEN '分销/SEEDING'
            ELSE 原始店铺名称
        END AS 店铺名称,
        指标金额
    FROM base_data
),
shop_brand_detail AS (
    SELECT
        统计日期,
        店铺主键,
        店铺名称,
        SUM(指标金额) AS 指标值
    FROM brand_split
    GROUP BY 1,2,3
)
SELECT
    统计日期,
    店铺名称,
    指标值
FROM shop_brand_detail
WHERE NOT (
        店铺主键 IN (1,9,28,29)
    AND 店铺名称 LIKE '%%_其他'
)

UNION ALL

SELECT
    统计日期,
    原始店铺名称 AS 店铺名称,
    SUM(指标金额)
FROM base_data
WHERE 店铺主键 IN (1,28,29,3)
GROUP BY 1,2

UNION ALL

SELECT
    统计日期,
    CASE
        WHEN 业务模式 IN ('三方计算','三方分销')
            THEN 原始店铺名称 || '_三方'
        ELSE 原始店铺名称 || '_自营'
    END AS 店铺名称,
    SUM(指标金额)
FROM base_data
WHERE 店铺主键 = 1
GROUP BY 1,2

ORDER BY 统计日期 DESC, 指标值 DESC;
""",
    ),
    "refund": MetricConfig(
        name="refund",
        description="退款额",
        field_suffix="退款额",
        sql_template="""
SET TIMEZONE = 'Asia/Shanghai';

WITH base_data AS (
    SELECT
        f.完结时间::DATE AS 统计日期,
        f.店铺主键,
        s.店铺名称 AS 原始店铺名称,
        p.品牌名称,
        f.退款金额 AS 指标金额
    FROM analytics_layer.fact_returns f
    LEFT JOIN analytics_layer.dim_shop s
        ON f.店铺主键 = s.店铺主键
    LEFT JOIN analytics_layer.dim_product p
        ON f.商品主键 = p.商品主键
       AND f.统一规格编码 = p.统一规格编码
    WHERE f.完结时间 >= %(start_ts)s
      AND f.完结时间 < %(end_ts_exclusive)s
),
brand_split AS (
    SELECT
        统计日期,
        店铺主键,
        原始店铺名称,
        CASE
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '羽心堂'
                THEN 原始店铺名称 || '_羽心堂'
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '春回盏'
                THEN 原始店铺名称 || '_春回盏'
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'Mama Mujer'
                THEN 原始店铺名称 || '_Mama Mujer'
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'ERBA MIGINA'
                THEN 原始店铺名称 || '_ERBA MIGINA'
            WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'BROKEN ZOO'
                THEN 原始店铺名称 || '_BROKEN ZOO'
            WHEN 店铺主键 IN (1,28,29)
                THEN 原始店铺名称 || '_其他'
            WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE '羽心堂'
                THEN 原始店铺名称 || '_羽心堂'
            WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE 'Mama Mujer'
                THEN 原始店铺名称 || '_Mama Mujer'
            WHEN 店铺主键 IN (3,9)
                THEN 原始店铺名称 || '_其他'
            WHEN 店铺主键 IS NULL OR 原始店铺名称 IS NULL
                THEN '分销/SEEDING'
            ELSE 原始店铺名称
        END AS 店铺名称,
        指标金额
    FROM base_data
),
shop_brand_detail AS (
    SELECT
        统计日期,
        店铺主键,
        店铺名称,
        SUM(指标金额) AS 指标值
    FROM brand_split
    GROUP BY 1,2,3
)
SELECT
    统计日期,
    店铺名称,
    指标值
FROM shop_brand_detail
WHERE NOT (
        店铺主键 IN (1,9,28,29)
    AND 店铺名称 LIKE '%%_其他'
)

UNION ALL

SELECT
    统计日期,
    原始店铺名称 AS 店铺名称,
    SUM(指标金额)
FROM base_data
WHERE 店铺主键 IN (1,28,29,3)
GROUP BY 1,2

ORDER BY 统计日期 DESC, 指标值 DESC;
""",
    ),
}


@dataclass(frozen=True)
class MetricRow:
    stat_date: date
    bi_store_name: str
    amount: Decimal


@dataclass(frozen=True)
class TableInfo:
    table_id: str
    name: str


@dataclass(frozen=True)
class SyncResult:
    sql_only_stores: list[str]
    feishu_only_stores: list[str]


def log_step(message: str) -> None:
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{now}] {message}", flush=True)


def preview_list(values: list[Any], limit: int = 20) -> str:
    if len(values) <= limit:
        return str(values)
    return f"{values[:limit]} ... 共{len(values)}项"


def load_dotenv(env_file: Path = ENV_FILE) -> None:
    if not env_file.exists():
        log_step(f"未找到 .env 文件，跳过加载: {env_file}")
        return
    log_step(f"加载环境变量: {env_file}")
    for raw_line in env_file.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        os.environ.setdefault(key.strip(), value.strip())


def get_base_url() -> str:
    base_url = os.getenv("FEISHU_BASE_URL")
    if not base_url:
        raise ValueError("缺少环境变量 FEISHU_BASE_URL")
    return base_url


def get_mapping_file() -> Path:
    return Path(os.getenv("PROJECT_MAPPING_FILE", str(DEFAULT_MAPPING_FILE)))


def resolve_store_field_name(fields: list[dict[str, Any]], preferred_field_name: str) -> str:
    field_names = [str(item.get("field_name") or "") for item in fields]
    if preferred_field_name in field_names:
        return preferred_field_name
    for candidate in STORE_FIELD_CANDIDATES:
        if candidate in field_names:
            return candidate
    raise ValueError(f"未找到店铺名称字段: preferred={preferred_field_name}, 可用字段={field_names}")


def validate_date(value: str) -> date:
    return datetime.strptime(value, "%Y-%m-%d").date()


def normalize_text(value: str) -> str:
    return re.sub(r"[\s_\-（）()]+", "", value or "").lower()


def classify_project(store_name: str) -> str:
    name = store_name or ""
    prioritized_rules = [
        ("MMMJ", ["Mama", "mmmj", "MMMJ"]),
        ("羽心堂", ["羽心堂"]),
        ("EM", ["EM", "ERBA MIGINA"]),
        ("春回盏", ["春回盏"]),
        ("BROKEN ZOO", ["BROKEN ZOO"]),
    ]
    for project, keywords in prioritized_rules:
        if any(keyword in name for keyword in keywords):
            return project
    if "大眼睛" in name or "大西瓜籽妈妈" in name:
        return "大眼睛"
    return ""


def get_project_table_aliases(project: str) -> list[str]:
    aliases = {
        "大眼睛": ["大眼睛", "大眼睛买买买"],
        "羽心堂": ["羽心堂"],
        "MMMJ": ["MMMJ", "Mama Mujer", "MAMA MUJER"],
        "EM": ["EM", "ERBA MIGINA"],
        "春回盏": ["春回盏"],
        "BROKEN ZOO": ["BROKEN ZOO", "BROKENZOO", "葆福颂"],
    }
    return aliases.get(project, [project])


def load_store_mapping(path: Path) -> dict[str, str]:
    log_step(f"加载店铺映射表: {path}")
    wb = load_workbook(path, data_only=True)
    ws = wb.active
    mapping: dict[str, str] = {}
    for business_name, bi_name in ws.iter_rows(min_row=2, values_only=True):
        if not business_name or not bi_name:
            continue
        mapping[str(bi_name).strip()] = str(business_name).strip()
    log_step(f"店铺映射加载完成: {len(mapping)} 条")
    return mapping


def resolve_business_store_name(bi_store_name: str, store_mapping: dict[str, str]) -> str:
    return store_mapping.get(bi_store_name, "")


def get_db_connection() -> psycopg2.extensions.connection:
    required = [
        "ANALYTICS_DB_HOST",
        "ANALYTICS_DB_PORT",
        "ANALYTICS_DB_USER",
        "ANALYTICS_DB_PASSWORD",
        "ANALYTICS_DB_NAME",
    ]
    missing = [key for key in required if not os.getenv(key)]
    if missing:
        raise ValueError(f"缺少数据库环境变量: {', '.join(missing)}")
    log_step(
        f"连接 PostgreSQL: host={os.environ['ANALYTICS_DB_HOST']} port={os.environ['ANALYTICS_DB_PORT']} db={os.environ['ANALYTICS_DB_NAME']}"
    )
    return psycopg2.connect(
        host=os.environ["ANALYTICS_DB_HOST"],
        port=os.environ["ANALYTICS_DB_PORT"],
        user=os.environ["ANALYTICS_DB_USER"],
        password=os.environ["ANALYTICS_DB_PASSWORD"],
        dbname=os.environ["ANALYTICS_DB_NAME"],
    )


def fetch_metric_rows(start_date: date, end_date: date, metric_config: MetricConfig) -> list[MetricRow]:
    start_ts = f"{start_date.isoformat()} 00:00:00"
    end_ts_exclusive = f"{(end_date + timedelta(days=1)).isoformat()} 00:00:00"
    log_step(f"开始查询{metric_config.description}: {start_ts} -> {end_ts_exclusive} (右开区间)")
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(metric_config.sql_template, {"start_ts": start_ts, "end_ts_exclusive": end_ts_exclusive})
            rows = cur.fetchall()
    log_step(f"{metric_config.description}查询完成: {len(rows)} 行")
    return [
        MetricRow(stat_date=row[0], bi_store_name=(row[1] or "").strip(), amount=row[2] or Decimal("0"))
        for row in rows
    ]


def build_project_metrics(
    rows: list[MetricRow],
    store_mapping: dict[str, str],
    metric_config: MetricConfig,
) -> dict[str, dict[tuple[int, int], dict[date, dict[str, Decimal]]]]:
    log_step(f"开始按项目/月/日期聚合{metric_config.description}")
    grouped: dict[str, dict[tuple[int, int], dict[date, dict[str, Decimal]]]] = defaultdict(
        lambda: defaultdict(lambda: defaultdict(dict))
    )
    for row in rows:
        business_store_name = resolve_business_store_name(row.bi_store_name, store_mapping)
        if not business_store_name:
            continue
        project = classify_project(business_store_name)
        if not project:
            continue
        month_key = (row.stat_date.year, row.stat_date.month)
        current_amount = grouped[project][month_key][row.stat_date].get(business_store_name, Decimal("0"))
        grouped[project][month_key][row.stat_date][business_store_name] = current_amount + row.amount
    project_count = len(grouped)
    month_count = sum(len(months) for months in grouped.values())
    log_step(f"{metric_config.description}聚合完成: 项目={project_count}，项目月份组合={month_count}")
    return grouped


def build_target_project_months(
    start_date: date,
    end_date: date,
    grouped_metrics: dict[str, dict[tuple[int, int], dict[date, dict[str, Decimal]]]],
) -> dict[str, dict[tuple[int, int], dict[date, dict[str, Decimal]]]]:
    result = {
        project: {month_key: dict(date_map) for month_key, date_map in month_dict.items()}
        for project, month_dict in grouped_metrics.items()
    }
    current = start_date
    while current <= end_date:
        month_key = (current.year, current.month)
        for project in SUPPORTED_PROJECTS:
            project_months = result.setdefault(project, {})
            metrics_by_date = project_months.setdefault(month_key, {})
            metrics_by_date.setdefault(current, {})
        current += timedelta(days=1)
    return result


def get_feishu_token() -> str:
    app_id = os.getenv("FEISHU_APP_ID")
    app_secret = os.getenv("FEISHU_APP_SECRET")
    if not app_id or not app_secret:
        raise ValueError("缺少飞书环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET")
    log_step("开始获取飞书 tenant_access_token")
    resp = requests.post(
        "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
        json={"app_id": app_id, "app_secret": app_secret},
        timeout=30,
    )
    resp.raise_for_status()
    payload = resp.json()
    if payload.get("code") != 0:
        raise RuntimeError(f"获取飞书 token 失败: {payload}")
    log_step("飞书 tenant_access_token 获取成功")
    return payload["tenant_access_token"]


class FeishuBitableClient:
    def __init__(self, base_url: str) -> None:
        self.base_url = base_url
        self.app_token = self.extract_app_token(base_url)
        log_step(f"解析飞书 Base URL 成功: app_token={self.app_token}")
        self.token = get_feishu_token()
        self._fields_cache: dict[str, list[dict[str, Any]]] = {}
        self.session = requests.Session()
        self.session.headers.update(
            {
                "Authorization": f"Bearer {self.token}",
                "Content-Type": "application/json; charset=utf-8",
            }
        )

    @staticmethod
    def extract_app_token(base_url: str) -> str:
        path = urlparse(base_url).path.strip("/")
        match = re.search(r"base/([^/]+)", path)
        if not match:
            raise ValueError(f"无法从链接中解析 app_token: {base_url}")
        return match.group(1)

    def request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
        url = f"https://open.feishu.cn{path}"
        params = kwargs.get("params")
        if params:
            log_step(f"请求飞书接口: {method} {path} params={params}")
        else:
            log_step(f"请求飞书接口: {method} {path}")
        resp = self.session.request(method, url, timeout=60, **kwargs)
        try:
            payload = resp.json()
        except ValueError:
            resp.raise_for_status()
            raise RuntimeError(f"飞书接口返回了非 JSON 响应: status={resp.status_code}, body={resp.text[:500]}")
        if resp.status_code >= 400:
            error = payload.get("error") or {}
            raise RuntimeError(
                f"飞书接口 HTTP {resp.status_code} {path}: code={payload.get('code')} msg={payload.get('msg')} "
                f"log_id={error.get('log_id', '')} payload={payload}"
            )
        if payload.get("code") != 0:
            error = payload.get("error") or {}
            raise RuntimeError(
                f"飞书接口失败 {path}: code={payload.get('code')} msg={payload.get('msg')} "
                f"log_id={error.get('log_id', '')} payload={payload}"
            )
        return payload.get("data") or {}

    def list_tables(self) -> list[TableInfo]:
        log_step("开始获取多维表格数据表列表")
        items: list[TableInfo] = []
        page_token = ""
        while True:
            params: dict[str, Any] = {}
            if page_token:
                params["page_token"] = page_token
            data = self.request("GET", f"/open-apis/bitable/v1/apps/{self.app_token}/tables", params=params)
            for item in data.get("items") or []:
                items.append(TableInfo(table_id=item["table_id"], name=item["name"]))
            if not data.get("has_more", False):
                break
            page_token = data.get("page_token") or ""
            if not page_token:
                break
        log_step(f"飞书返回的数据表名称: {preview_list([item.name for item in items])}")
        log_step(f"数据表列表获取完成: {len(items)} 个")
        return items

    def list_fields(self, table_id: str) -> list[dict[str, Any]]:
        if table_id in self._fields_cache:
            log_step(f"命中字段缓存: table_id={table_id}, 字段数={len(self._fields_cache[table_id])}")
            return self._fields_cache[table_id]
        log_step(f"开始获取字段列表: table_id={table_id}")
        items: list[dict[str, Any]] = []
        page_token = ""
        seen_tokens: set[str] = set()
        page_index = 0
        while True:
            page_index += 1
            params = {"page_size": 100}
            if page_token:
                params["page_token"] = page_token
                if page_token in seen_tokens:
                    raise RuntimeError(f"字段列表分页出现重复 page_token，已中断: table_id={table_id}, page_token={page_token}")
                seen_tokens.add(page_token)
            log_step(f"获取字段列表分页: table_id={table_id}, page={page_index}, page_token={page_token or '<first>'}")
            data = self.request(
                "GET",
                f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/fields",
                params=params,
            )
            items.extend(data.get("items") or [])
            has_more = data.get("has_more", False)
            next_page_token = data.get("page_token") or ""
            log_step(
                f"字段列表分页完成: table_id={table_id}, page={page_index}, 累计字段数={len(items)}, "
                f"has_more={has_more}, next_page_token={next_page_token or '<empty>'}"
            )
            if not has_more:
                break
            if not next_page_token:
                raise RuntimeError(f"字段列表分页返回 has_more=True 但 page_token 为空: table_id={table_id}, 已获取字段数={len(items)}")
            page_token = next_page_token
            if page_index >= 100:
                raise RuntimeError(f"字段列表分页超过 100 页，已中断: table_id={table_id}")
        self._fields_cache[table_id] = items
        log_step(f"飞书返回的字段名称: table_id={table_id}, fields={preview_list([item.get('field_name', '') for item in items])}")
        log_step(f"字段列表获取完成: table_id={table_id}, 字段数={len(items)}")
        return items

    def ensure_number_field(self, table_id: str, field_name: str) -> None:
        log_step(f"检查字段是否存在: table_id={table_id}, field={field_name}")
        fields = self.list_fields(table_id)
        if any(item.get("field_name") == field_name for item in fields):
            log_step(f"字段已存在，跳过创建: table_id={table_id}, field={field_name}")
            return
        log_step(f"开始创建数字字段: table_id={table_id}, field={field_name}")
        self.request(
            "POST",
            f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/fields",
            json={"field_name": field_name, "type": 2},
        )
        self._fields_cache.pop(table_id, None)
        log_step(f"字段创建完成: table_id={table_id}, field={field_name}")

    def list_records(self, table_id: str) -> list[dict[str, Any]]:
        log_step(f"开始获取记录列表: table_id={table_id}")
        items: list[dict[str, Any]] = []
        page_token = ""
        while True:
            params = {"page_size": 500}
            if page_token:
                params["page_token"] = page_token
            data = self.request(
                "GET",
                f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/records",
                params=params,
            )
            items.extend(data.get("items") or [])
            page_token = data.get("page_token") or ""
            if not page_token:
                break
        log_step(f"记录列表获取完成: table_id={table_id}, 记录数={len(items)}")
        return items

    def update_record(self, table_id: str, record_id: str, fields: dict[str, Any]) -> None:
        log_step(f"更新记录: table_id={table_id}, record_id={record_id}, fields={list(fields.keys())}")
        self.request(
            "PUT",
            f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/records/{record_id}",
            json={"fields": fields},
        )


def load_table_name_overrides(path_text: str) -> dict[str, str]:
    if not path_text:
        return {}
    path = Path(path_text)
    if not path.exists():
        raise FileNotFoundError(f"未找到表名映射文件: {path}")
    payload = json.loads(path.read_text(encoding="utf-8"))
    return {str(key): str(value) for key, value in payload.items()}


def find_target_table(
    tables: list[TableInfo],
    project: str,
    year: int,
    month: int,
    overrides: dict[str, str] | None = None,
) -> TableInfo:
    log_step(f"开始匹配目标表: 项目={project}, 月份={year:04d}-{month:02d}")
    overrides = overrides or {}
    override_key = f"{project}:{year:04d}-{month:02d}"
    if override_key in overrides:
        target_name = overrides[override_key]
        for table in tables:
            if table.name == target_name:
                log_step(f"命中表名映射: {override_key} -> {table.name}")
                return table
        raise ValueError(f"表名映射命中了 {target_name}，但飞书中不存在该表")
    normalized_projects = [normalize_text(alias) for alias in get_project_table_aliases(project)]
    month_tokens = {
        str(month),
        f"{month}月",
        f"{month:02d}月",
        f"{year:04d}{month:02d}",
        f"{year:04d}年{month:02d}月",
        f"{year:04d}-{month:02d}",
        f"{year:04d}_{month:02d}",
        f"{year:04d}.{month:02d}",
    }
    log_step(f"项目表名别名: 项目={project}, aliases={get_project_table_aliases(project)}")
    candidates = []
    project_related_tables = []
    for table in tables:
        normalized_name = normalize_text(table.name)
        matched_project = any(alias in normalized_name for alias in normalized_projects)
        if matched_project:
            project_related_tables.append(table.name)
        if not matched_project:
            continue
        if any(normalize_text(token) in normalized_name for token in month_tokens):
            candidates.append(table)
    if len(candidates) == 1:
        log_step(f"匹配到目标表: 项目={project}, 月份={year:04d}-{month:02d}, 表={candidates[0].name}")
        return candidates[0]
    if not candidates:
        all_table_names = [table.name for table in tables]
        raise ValueError(
            f"未找到匹配表: 项目={project}, 月份={year:04d}-{month:02d}, 项目相关表={project_related_tables or '[]'}, 全部表={all_table_names}"
        )
    raise ValueError(
        f"匹配到多个表，请提供 table_map_file 指定: 项目={project}, 月份={year:04d}-{month:02d}, 表={', '.join(t.name for t in candidates)}"
    )


def decimal_to_number(value: Decimal) -> float:
    return float(value.quantize(Decimal("0.01")))


def get_metric_field_name(stat_date: date, metric_config: MetricConfig) -> str:
    return stat_date.strftime(f"%m-%d{metric_config.field_suffix}")


def parse_number(value: Any) -> float | None:
    if value in (None, ""):
        return None
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return round(float(value), 2)
    if isinstance(value, str):
        text = value.strip().replace(",", "")
        if not text:
            return None
        try:
            return round(float(text), 2)
        except ValueError:
            return None
    return None


def iter_dates(start_date: date, end_date: date) -> list[date]:
    dates = []
    current = start_date
    while current <= end_date:
        dates.append(current)
        current = date.fromordinal(current.toordinal() + 1)
    return dates


def sync_metric_table_for_month(
    client: FeishuBitableClient,
    table: TableInfo,
    store_field_name: str,
    metrics_by_date: dict[date, dict[str, Decimal]],
    dry_run: bool,
    metric_config: MetricConfig,
) -> SyncResult:
    log_step(f"开始同步数据表: {table.name}")
    table_fields = client.list_fields(table.table_id)
    store_field_name = resolve_store_field_name(table_fields, store_field_name)
    log_step(f"识别到店铺字段: 表={table.name}, field={store_field_name}")
    records = client.list_records(table.table_id)
    record_by_store: dict[str, dict[str, Any]] = {}
    preview_store_fields: list[str] = []
    for item in records:
        fields = item.get("fields", {})
        store_name = fields.get(store_field_name)
        if isinstance(store_name, list):
            store_name = store_name[0] if store_name else ""
        if not store_name:
            continue
        normalized_store_name = str(store_name).strip()
        normalized_store_key = normalize_text(normalized_store_name)
        preview_store_fields.append(normalized_store_name)
        record_by_store[normalized_store_key] = item
    log_step(f"飞书返回的记录店铺名称预览: table_id={table.table_id}, stores={preview_list(preview_store_fields)}")

    feishu_store_names = set(record_by_store)
    sql_store_names: set[str] = set()
    sql_store_names_raw: set[str] = set()
    for store_metrics in metrics_by_date.values():
        sql_store_names_raw.update(store_metrics.keys())
        sql_store_names.update(normalize_text(name) for name in store_metrics.keys())
    all_store_names = feishu_store_names | sql_store_names

    sql_name_map = {normalize_text(name): name for name in sql_store_names_raw}
    feishu_name_map = {normalize_text(name): name for name in preview_store_fields}
    sql_only_stores = sorted(sql_name_map[name] for name in (sql_store_names - feishu_store_names))
    feishu_only_stores = sorted(feishu_name_map[name] for name in (feishu_store_names - sql_store_names))

    log_step(f"SQL 聚合出的店铺名称: 表={table.name}, stores={preview_list(sorted(sql_store_names_raw))}")
    log_step(f"待处理店铺数: 表={table.name}, 店铺数={len(all_store_names)}, 日期数={len(metrics_by_date)}")
    log_step(
        f"店铺匹配情况: 表={table.name}, SQL有但飞书无={preview_list(sql_only_stores)}, "
        f"飞书有但SQL无={preview_list(feishu_only_stores)}"
    )

    for stat_date in sorted(metrics_by_date):
        field_name = get_metric_field_name(stat_date, metric_config)
        if dry_run:
            print(f"[DRY-RUN] 表 {table.name} 将新增字段: {field_name}")
        else:
            client.ensure_number_field(table.table_id, field_name)
        for store_key in sorted(all_store_names):
            store_name = sql_name_map.get(store_key) or feishu_name_map.get(store_key) or store_key
            amount = None
            for raw_name, raw_amount in metrics_by_date[stat_date].items():
                if normalize_text(raw_name) == store_key:
                    amount = raw_amount
                    break
            field_value = decimal_to_number(amount) if amount is not None else None
            existing = record_by_store.get(store_key)
            if dry_run:
                print(
                    f"[DRY-RUN] 表={table.name} 店铺={store_name} 字段={field_name} 值={field_value if field_value is not None else ''}"
                )
                continue
            if existing:
                client.update_record(table.table_id, existing["record_id"], {field_name: field_value})
                continue
            log_step(f"跳过新增记录: 表={table.name}, 店铺={store_name}, 字段={field_name}, 原因=飞书中不存在该店铺行")
    log_step(f"数据表同步完成: {table.name}")
    return SyncResult(sql_only_stores=sql_only_stores, feishu_only_stores=feishu_only_stores)


def run_metric_sync(
    metric_name: str,
    start_date: date,
    end_date: date,
    base_url: str,
    mapping_file: Path,
    store_field_name: str,
    table_map_file: str,
    dry_run: bool,
    skip_missing_tables: bool,
) -> None:
    metric_config = METRIC_CONFIGS[metric_name]
    mapping = load_store_mapping(mapping_file)
    metric_rows = fetch_metric_rows(start_date, end_date, metric_config)
    grouped_metrics = build_project_metrics(metric_rows, mapping, metric_config)
    grouped_metrics = build_target_project_months(start_date, end_date, grouped_metrics)
    table_overrides = load_table_name_overrides(table_map_file)
    if table_overrides:
        log_step(f"加载表名映射完成: {len(table_overrides)} 条")
    client = FeishuBitableClient(base_url)
    tables = client.list_tables()
    unmatched_summary: list[tuple[str, str, str, SyncResult]] = []
    for project in sorted(grouped_metrics):
        for (year, month), metrics_by_date in sorted(grouped_metrics[project].items()):
            try:
                table = find_target_table(tables, project, year, month, table_overrides)
            except ValueError as exc:
                if not skip_missing_tables:
                    raise
                log_step(f"跳过: {exc}")
                continue
            log_step(f"同步项目={project} 月份={year:04d}-{month:02d} -> 表={table.name}")
            sync_result = sync_metric_table_for_month(
                client=client,
                table=table,
                store_field_name=store_field_name,
                metrics_by_date=metrics_by_date,
                dry_run=dry_run,
                metric_config=metric_config,
            )
            unmatched_summary.append((project, f"{year:04d}-{month:02d}", table.name, sync_result))
    log_step("未匹配店铺汇总开始")
    for project, month, table_name, result in unmatched_summary:
        log_step(
            f"未匹配店铺汇总: 项目={project}, 月份={month}, 表={table_name}, "
            f"SQL有但飞书无={result.sql_only_stores}, 飞书有但SQL无={result.feishu_only_stores}"
        )


def get_sales_field_name(stat_date: date) -> str:
    return get_metric_field_name(stat_date, METRIC_CONFIGS["sales"])


def get_refund_field_name(stat_date: date) -> str:
    return get_metric_field_name(stat_date, METRIC_CONFIGS["refund"])


def get_gmv_field_name(stat_date: date) -> str:
    return stat_date.strftime("%m-%d GMV")


def sync_gmv_table_for_day(
    client: FeishuBitableClient,
    table: TableInfo,
    stat_date: date,
    preferred_store_field_name: str,
    dry_run: bool,
) -> None:
    log_step(f"开始计算 GMV: 表={table.name}, 日期={stat_date.isoformat()}")
    fields = client.list_fields(table.table_id)
    store_field_name = resolve_store_field_name(fields, preferred_store_field_name)
    sales_field_name = get_sales_field_name(stat_date)
    refund_field_name = get_refund_field_name(stat_date)
    gmv_field_name = get_gmv_field_name(stat_date)
    field_names = [str(item.get("field_name") or "") for item in fields]
    log_step(
        f"GMV 字段检查: 表={table.name}, 店铺字段={store_field_name}, 销售字段={sales_field_name in field_names}, "
        f"退款字段={refund_field_name in field_names}, GMV字段={gmv_field_name in field_names}"
    )
    if sales_field_name not in field_names or refund_field_name not in field_names:
        log_step(f"跳过 GMV 计算: 表={table.name}, 原因=缺少销售额或退款额字段")
        return
    if not dry_run:
        client.ensure_number_field(table.table_id, gmv_field_name)
    records = client.list_records(table.table_id)
    preview_stores = []
    updated_count = 0
    skipped_empty_count = 0
    for item in records:
        record_id = item["record_id"]
        record_fields = item.get("fields", {})
        store_name = record_fields.get(store_field_name)
        if isinstance(store_name, list):
            store_name = store_name[0] if store_name else ""
        store_name = str(store_name or "").strip()
        if store_name:
            preview_stores.append(store_name)
        sales_value = parse_number(record_fields.get(sales_field_name))
        refund_value = parse_number(record_fields.get(refund_field_name))
        if sales_value is None and refund_value is None:
            skipped_empty_count += 1
            continue
        gmv_value = round((sales_value or 0.0) - (refund_value or 0.0), 2)
        if dry_run:
            print(
                f"[DRY-RUN] 表={table.name} 店铺={store_name} 字段={gmv_field_name} "
                f"销售额={sales_value if sales_value is not None else ''} "
                f"退款额={refund_value if refund_value is not None else ''} GMV={gmv_value}"
            )
        else:
            client.update_record(table.table_id, record_id, {gmv_field_name: gmv_value})
        updated_count += 1
    log_step(f"飞书记录店铺预览: 表={table.name}, stores={preview_list(preview_stores)}")
    log_step(f"GMV 计算完成: 表={table.name}, 更新记录数={updated_count}, 双空跳过={skipped_empty_count}")


def run_gmv_sync(
    start_date: date,
    end_date: date,
    base_url: str,
    store_field_name: str,
    dry_run: bool,
    skip_missing_tables: bool,
) -> None:
    client = FeishuBitableClient(base_url)
    tables = client.list_tables()
    for stat_date in iter_dates(start_date, end_date):
        for project in SUPPORTED_PROJECTS:
            try:
                table = find_target_table(tables, project, stat_date.year, stat_date.month)
            except ValueError as exc:
                if not skip_missing_tables:
                    raise
                log_step(f"跳过: {exc}")
                continue
            sync_gmv_table_for_day(
                client=client,
                table=table,
                stat_date=stat_date,
                preferred_store_field_name=store_field_name,
                dry_run=dry_run,
            )