logistics-service/import_logistics_to_feishu_bitable.py

#!/usr/bin/env python3
import json
import os
import time
from datetime import datetime
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple
from urllib import error, parse, request

try:
    import psycopg  # type: ignore
except ImportError:
    psycopg = None

try:
    import psycopg2  # type: ignore
    from psycopg2.extras import RealDictCursor  # type: ignore
except ImportError:
    psycopg2 = None
    RealDictCursor = None


ENV_PATH = Path(".env")
DEFAULT_BATCH_SIZE = 500
DEFAULT_HTTP_RETRIES = 3
DEFAULT_HTTP_RETRY_DELAY = 2.0

DB_QUERY = """
SELECT
    fs.支付时间      AS 支付时间,
    fs.平台订单号    AS 订单号,
    fs.有赞分销单号  AS 分销单号,
    dw.仓库名称      AS "仓库名称/供应商",
    ds.店铺名称      AS 店铺名称,
    dp.商品名称      AS 商品名称,
    fs.销售数量      AS 商品数量,
    fs.客服备注      AS 客服备注,
    fs.买家留言      AS 买家留言,
    dc.手机号_脱敏   AS 客户手机号,
    CONCAT(fs.收货省份, fs.收货城市, fs.收货区域) AS 地址

FROM analytics_layer.fact_sales fs

-- 商品
LEFT JOIN analytics_layer.dim_product dp
    ON fs.商品主键 = dp.商品主键

-- 仓库
LEFT JOIN analytics_layer.dim_warehouse dw
    ON fs.仓库主键 = dw.仓库主键
	
-- 客户
LEFT JOIN analytics_layer.dim_customer dc
    ON fs.客户主键 = dc.客户主键

-- 店铺
LEFT JOIN analytics_layer.dim_shop ds
    ON fs.店铺主键 = ds.店铺主键

WHERE 
    -- 48小时未发货
    fs.发货时间 IS NULL
    AND fs.支付时间 <= CURRENT_TIMESTAMP - INTERVAL '48' HOUR

    -- 排除平台已发货
    AND COALESCE(fs.平台发货状态, 0) != 1

    -- 排除关闭类订单
    AND COALESCE(fs.订单状态, '') NOT IN ('已关闭', '交易关闭', '已取消')

    -- 排除已发货/完成
    AND COALESCE(fs.订单状态, '') NOT IN ('已发货', '已完成')

    -- ❗新增：排除存在退款记录的订单
    AND NOT EXISTS (
        SELECT 1
        FROM analytics_layer.fact_returns fr
        WHERE fr.平台订单号 = fs.平台订单号
          AND fr.子单号 = fs.子单号
    )

ORDER BY fs.支付时间 DESC;
"""

# Feishu Bitable field type IDs.
FIELD_TYPE_TEXT = 1
FIELD_TYPE_NUMBER = 2
FIELD_TYPE_SINGLE_SELECT = 3
FIELD_TYPE_MULTI_SELECT = 4
FIELD_TYPE_DATETIME = 5
FIELD_TYPE_CHECKBOX = 7
FIELD_TYPE_PHONE = 13
FIELD_TYPE_URL = 15

FIELD_TYPE_HINTS = {
    "支付时间": FIELD_TYPE_DATETIME,
    "销售数量": FIELD_TYPE_NUMBER,
    "商品数量": FIELD_TYPE_NUMBER,
}


def log(message: str, *, level: str = "INFO") -> None:
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{timestamp}] [{level}] {message}")


def load_env(path: Path) -> None:
    if not path.exists():
        return
    for raw_line in path.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_db_config() -> Dict[str, str]:
    mapping = {
        "host": "ANALYTICS_DB_HOST",
        "port": "ANALYTICS_DB_PORT",
        "user": "ANALYTICS_DB_USER",
        "password": "ANALYTICS_DB_PASSWORD",
        "dbname": "ANALYTICS_DB_NAME",
    }
    config: Dict[str, str] = {}
    missing: List[str] = []
    for key, env_name in mapping.items():
        value = os.getenv(env_name)
        if not value:
            missing.append(env_name)
        else:
            config[key] = value
    if missing:
        raise RuntimeError(f"缺少数据库配置: {', '.join(missing)}")
    return config


def parse_base_url(base_url: str) -> Tuple[str, str, Optional[str]]:
    parsed = parse.urlparse(base_url)
    path_parts = [part for part in parsed.path.split("/") if part]
    if len(path_parts) < 2 or path_parts[0] != "base":
        raise ValueError(f"无法从 FEISHU_BASE_URL 解析 app_token: {base_url}")

    query = parse.parse_qs(parsed.query)
    app_token = path_parts[1]
    table_id = query.get("table", [None])[0]
    view_id = query.get("view", [None])[0]
    if not table_id:
        raise ValueError(f"无法从 FEISHU_BASE_URL 解析 table_id: {base_url}")
    return app_token, table_id, view_id


def feishu_request(
    method: str,
    url: str,
    *,
    token: Optional[str] = None,
    payload: Optional[dict] = None,
    retries: int = DEFAULT_HTTP_RETRIES,
    retry_delay: float = DEFAULT_HTTP_RETRY_DELAY,
) -> dict:
    headers = {"Content-Type": "application/json; charset=utf-8"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    data = None
    if payload is not None:
        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")

    for attempt in range(1, retries + 1):
        req = request.Request(url=url, method=method, headers=headers, data=data)
        try:
            with request.urlopen(req, timeout=60) as resp:
                body = resp.read().decode("utf-8")
            break
        except error.HTTPError as exc:
            detail = exc.read().decode("utf-8", errors="replace")
            should_retry = exc.code in {429, 500, 502, 503, 504}
            log(
                f"Feishu API HTTP 异常: method={method} status={exc.code} "
                f"url={url} attempt={attempt}/{retries} detail={detail[:500]}",
                level="ERROR",
            )
            if should_retry and attempt < retries:
                sleep_seconds = retry_delay * attempt
                log(
                    f"Feishu API 将重试: method={method} status={exc.code} "
                    f"next_retry_in={sleep_seconds:.1f}s",
                    level="WARN",
                )
                time.sleep(sleep_seconds)
                continue
            raise RuntimeError(f"Feishu API 请求失败: {exc.code} {detail}") from exc
        except error.URLError as exc:
            log(
                f"Feishu API 网络异常: method={method} url={url} "
                f"attempt={attempt}/{retries} error={exc}",
                level="ERROR",
            )
            if attempt < retries:
                sleep_seconds = retry_delay * attempt
                log(
                    f"Feishu API 网络异常后重试: method={method} next_retry_in={sleep_seconds:.1f}s",
                    level="WARN",
                )
                time.sleep(sleep_seconds)
                continue
            raise RuntimeError(f"无法连接飞书 API: {exc}") from exc

    result = json.loads(body)
    if result.get("code") != 0:
        log(
            f"Feishu API 业务异常: method={method} url={url} "
            f"code={result.get('code')} msg={result.get('msg')}",
            level="ERROR",
        )
        raise RuntimeError(
            f"Feishu API 返回错误: code={result.get('code')} msg={result.get('msg')} body={body}"
        )
    return result


def get_response_data(result: dict) -> dict:
    data = result.get("data")
    return data if isinstance(data, dict) else {}


def get_tenant_access_token(app_id: str, app_secret: str) -> str:
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    result = feishu_request(
        "POST",
        url,
        payload={"app_id": app_id, "app_secret": app_secret},
    )
    return result["tenant_access_token"]


def send_feishu_group_message(text: str) -> None:
    app_id = os.getenv("FEISHU_APP_ID")
    app_secret = os.getenv("FEISHU_APP_SECRET")
    chat_id = os.getenv("FEISHU_CHAT_ID")
    if not app_id or not app_secret or not chat_id:
        raise RuntimeError("缺少 FEISHU_APP_ID / FEISHU_APP_SECRET / FEISHU_CHAT_ID")

    token = get_tenant_access_token(app_id, app_secret)
    url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id"
    payload = {
        "receive_id": chat_id,
        "msg_type": "text",
        "content": json.dumps({"text": text}, ensure_ascii=False),
    }
    feishu_request("POST", url, token=token, payload=payload)


def build_success_message(summary: Dict[str, int]) -> str:
    base_url = os.getenv("FEISHU_BASE_URL", "")
    started_at = summary.get("started_at", "")
    finished_at = summary.get("finished_at", "")
    message = (
        "截止到中午12点，超48小时未发货的订单数据同步完成\n"
        f"新增记录数: {summary['created']}\n"
    )
    if base_url:
        message += f"\n多维表链接: {base_url}"
    return message


def build_failure_message(error_text: str, finished_at: str) -> str:
    return (
        "物流多维表同步失败\n"
        f"执行时间: {finished_at}\n"
        f"错误信息: {error_text}"
    )


def paginate_fields(app_token: str, table_id: str, token: str) -> List[dict]:
    fields: List[dict] = []
    page_token: Optional[str] = None

    while True:
        query = {"page_size": 100}
        if page_token:
            query["page_token"] = page_token
        url = (
            f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}"
            f"/tables/{table_id}/fields?{parse.urlencode(query)}"
        )
        result = feishu_request("GET", url, token=token)
        data = get_response_data(result)
        items = data.get("items", [])
        fields.extend(items)
        if not data.get("has_more"):
            break
        page_token = data.get("page_token")

    log(f"字段读取完成: total={len(fields)}")
    return fields


def list_all_record_ids(app_token: str, table_id: str, token: str) -> List[str]:
    record_ids: List[str] = []
    page_token: Optional[str] = None

    while True:
        query = {"page_size": 100}
        if page_token:
            query["page_token"] = page_token
        url = (
            f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}"
            f"/tables/{table_id}/records?{parse.urlencode(query)}"
        )
        result = feishu_request("GET", url, token=token)
        data = get_response_data(result)
        items = data.get("items", [])
        for item in items:
            record_id = item.get("record_id") or item.get("id")
            if record_id:
                record_ids.append(record_id)
        if not data.get("has_more"):
            break
        page_token = data.get("page_token")
        if not page_token:
            break

    log(f"目标表现有记录统计: total={len(record_ids)}")
    return record_ids


def create_field(app_token: str, table_id: str, token: str, field_name: str, field_type: int) -> dict:
    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/fields"
    payload = {
        "field_name": field_name,
        "type": field_type,
    }
    result = feishu_request("POST", url, token=token, payload=payload)
    data = get_response_data(result)
    field = data.get("field")
    if not isinstance(field, dict):
        raise RuntimeError(f"创建字段返回异常，缺少 field: {json.dumps(result, ensure_ascii=False)}")
    return field


def convert_for_field(value: Optional[str], field_type: int):
    if value is None:
        return None

    if field_type in {FIELD_TYPE_TEXT, FIELD_TYPE_PHONE, FIELD_TYPE_URL}:
        return value

    if field_type == FIELD_TYPE_NUMBER:
        try:
            return float(value) if "." in value else int(value)
        except ValueError:
            return value

    if field_type == FIELD_TYPE_SINGLE_SELECT:
        return value

    if field_type == FIELD_TYPE_MULTI_SELECT:
        return [part.strip() for part in value.split(",") if part.strip()]

    if field_type == FIELD_TYPE_CHECKBOX:
        return value.lower() in {"1", "true", "yes", "y", "是"}

    if field_type == FIELD_TYPE_DATETIME:
        for pattern in ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S+08", "%Y-%m-%d %H:%M:%S"):
            try:
                dt = datetime.strptime(value, pattern)
                if pattern == "%Y-%m-%d":
                    dt = datetime(dt.year, dt.month, dt.day)
                return int(dt.timestamp() * 1000)
            except ValueError:
                continue
        return value

    return value


def normalize_db_value(value) -> Optional[str]:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.strftime("%Y-%m-%d")
    return str(value).strip() or None


def query_db_rows(sql: str) -> Tuple[List[str], List[Dict[str, Optional[str]]]]:
    config = get_db_config()
    if psycopg is not None:
        with psycopg.connect(**config) as conn:
            with conn.cursor() as cursor:
                cursor.execute("SET TIMEZONE = 'Asia/Shanghai'")
                cursor.execute(sql)
                headers = [desc.name for desc in cursor.description or []]
                result_rows = cursor.fetchall()
        rows = []
        for item in result_rows:
            rows.append({header: normalize_db_value(value) for header, value in zip(headers, item)})
        return headers, rows

    if psycopg2 is not None and RealDictCursor is not None:
        with psycopg2.connect(**config) as conn:
            with conn.cursor(cursor_factory=RealDictCursor) as cursor:
                cursor.execute("SET TIMEZONE = 'Asia/Shanghai'")
                cursor.execute(sql)
                raw_rows = cursor.fetchall()
                headers = [desc.name for desc in cursor.description or []]
        rows = []
        for item in raw_rows:
            rows.append({header: normalize_db_value(item.get(header)) for header in headers})
        return headers, rows

    raise RuntimeError("缺少 PostgreSQL 驱动，请安装 psycopg 或 psycopg2")


def ensure_fields(
    app_token: str,
    table_id: str,
    token: str,
    csv_headers: Iterable[str],
) -> Tuple[Dict[str, dict], List[str]]:
    fields = paginate_fields(app_token, table_id, token)
    field_map = {field["field_name"]: field for field in fields}
    missing_headers: List[str] = []

    for header in csv_headers:
        if header not in field_map:
            try:
                created = create_field(
                    app_token,
                    table_id,
                    token,
                    header,
                    FIELD_TYPE_HINTS.get(header, FIELD_TYPE_TEXT),
                )
                field_map[header] = created
                log(f"字段已创建: field_name={header}")
            except RuntimeError as exc:
                missing_headers.append(header)
                log(f"字段创建失败，已跳过: field_name={header} error={exc}", level="WARN")

    return field_map, missing_headers


def chunked(items: List[dict], size: int) -> Iterable[List[dict]]:
    for index in range(0, len(items), size):
        yield items[index : index + size]


def build_records(
    rows: List[Dict[str, Optional[str]]],
    field_map: Dict[str, dict],
    skipped_headers: Iterable[str],
) -> List[dict]:
    skipped = set(skipped_headers)
    records: List[dict] = []
    for row in rows:
        fields_payload = {}
        for key, raw_value in row.items():
            if key in skipped or key not in field_map:
                continue
            if raw_value is None:
                continue
            field_type = field_map[key]["type"]
            value = convert_for_field(raw_value, field_type)
            if value is not None:
                fields_payload[key] = value
        if fields_payload:
            records.append({"fields": fields_payload})
    return records


def batch_create_records(
    app_token: str,
    table_id: str,
    token: str,
    records: List[dict],
    batch_size: int,
) -> int:
    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_create"
    created = 0
    total_batches = (len(records) + batch_size - 1) // batch_size

    for batch_index, batch in enumerate(chunked(records, batch_size), start=1):
        log(f"开始写入批次: batch={batch_index}/{total_batches} size={len(batch)}")
        feishu_request("POST", url, token=token, payload={"records": batch})
        created += len(batch)
        log(f"写入批次完成: batch={batch_index}/{total_batches} created_total={created}")
        time.sleep(0.15)

    return created


def batch_delete_records(
    app_token: str,
    table_id: str,
    token: str,
    record_ids: List[str],
    batch_size: int,
) -> int:
    if not record_ids:
        log("目标表当前没有历史记录，无需删除")
        return 0

    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_delete"
    deleted = 0
    total_batches = (len(record_ids) + batch_size - 1) // batch_size

    for batch_index, batch in enumerate(chunked(record_ids, batch_size), start=1):
        log(f"开始删除批次: batch={batch_index}/{total_batches} size={len(batch)}")
        result = feishu_request("POST", url, token=token, payload={"records": batch})
        data = get_response_data(result)
        response_records = data.get("records", [])
        if response_records:
            failed = [item.get("record_id") for item in response_records if not item.get("deleted", False)]
            if failed:
                log(f"删除批次存在失败记录: failed_count={len(failed)}", level="ERROR")
                raise RuntimeError(f"删除记录失败，record_ids={failed}")
        deleted += len(batch)
        log(f"删除批次完成: batch={batch_index}/{total_batches} deleted_total={deleted}")
        time.sleep(0.15)

    return deleted


def print_permission_help(message: str) -> None:
    log("飞书返回权限不足，当前应用还不能向这个多维表新增字段或记录。", level="ERROR")
    log(f"原始错误: {message}", level="ERROR")
    log(
        "请在飞书开放平台或多维表权限设置中，确认这个应用已被添加到该 Base，且具备多维表读取/写入权限。",
        level="ERROR",
    )
    log(
        "如果你手动补齐缺失字段并授权写入，重新执行 `python3 import_logistics_to_feishu_bitable.py` 即可继续导入。",
        level="ERROR",
    )


def execute_sync() -> Dict[str, int]:
    load_env(ENV_PATH)

    app_id = os.getenv("FEISHU_APP_ID")
    app_secret = os.getenv("FEISHU_APP_SECRET")
    base_url = os.getenv("FEISHU_BASE_URL")
    batch_size = int(os.getenv("BATCH_SIZE", str(DEFAULT_BATCH_SIZE)))

    if not app_id or not app_secret or not base_url:
        raise RuntimeError("缺少 FEISHU_APP_ID / FEISHU_APP_SECRET / FEISHU_BASE_URL")

    started_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    app_token, table_id, _view_id = parse_base_url(base_url)
    log(f"导入任务开始: source=db table_id={table_id}")
    headers, rows = query_db_rows(DB_QUERY)
    log(f"读取数据库完成: rows={len(rows)}")

    try:
        token = get_tenant_access_token(app_id, app_secret)
        log("已获取 tenant_access_token")
        field_map, missing_headers = ensure_fields(app_token, table_id, token, headers)
        log(f"字段检查完成: field_count={len(field_map)} missing_count={len(missing_headers)}")
        existing_record_ids = list_all_record_ids(app_token, table_id, token)
        deleted = batch_delete_records(app_token, table_id, token, existing_record_ids, batch_size)
        records = build_records(rows, field_map, missing_headers)
        log(f"记录构建完成: source_rows={len(rows)} upload_records={len(records)}")
        created = batch_create_records(app_token, table_id, token, records, batch_size)
    except RuntimeError as exc:
        if "91403" in str(exc) or "Forbidden" in str(exc):
            print_permission_help(str(exc))
            raise RuntimeError("飞书多维表权限不足") from exc
        log(f"导入任务失败: error={exc}", level="ERROR")
        raise

    if missing_headers:
        log(f"存在未写入字段: missing_count={len(missing_headers)}", level="WARN")
    log(f"导入任务完成: deleted={deleted} created={created} source_rows={len(rows)}")
    return {
        "started_at": started_at,
        "finished_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "deleted": deleted,
        "created": created,
        "source_rows": len(rows),
        "missing_count": len(missing_headers),
    }


def main() -> int:
    try:
        summary = execute_sync()
        try:
            send_feishu_group_message(build_success_message(summary))
            log("飞书群通知发送成功")
        except Exception as notify_exc:
            log(f"飞书群通知发送失败: error={notify_exc}", level="ERROR")
        return 0
    except RuntimeError as exc:
        try:
            send_feishu_group_message(
                build_failure_message(str(exc), datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
            )
            log("失败通知已发送到飞书群", level="WARN")
        except Exception as notify_exc:
            log(f"失败通知发送失败: error={notify_exc}", level="ERROR")
        if str(exc) == "飞书多维表权限不足":
            return 2
        raise


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