调整查询SQL

import_logistics_to_feishu_bitable.py

@@ -9,10 +9,80 @@ 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_CSV_PATH = Path("物流.csv")
 DEFAULT_BATCH_SIZE = 500
+DEFAULT_SOURCE = "db"
+
+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
@@ -24,6 +94,12 @@ 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 load_env(path: Path) -> None:
     if not path.exists():
@@ -36,6 +112,27 @@ def load_env(path: Path) -> None:
         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]
@@ -84,6 +181,11 @@ def feishu_request(
     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(
@@ -107,7 +209,7 @@ def paginate_fields(app_token: str, table_id: str, token: str) -> List[dict]:
             f"/tables/{table_id}/fields?{parse.urlencode(query)}"
         )
         result = feishu_request("GET", url, token=token)
-        data = result.get("data", {})
+        data = get_response_data(result)
         fields.extend(data.get("items", []))
         if not data.get("has_more"):
             break
@@ -116,14 +218,18 @@ def paginate_fields(app_token: str, table_id: str, token: str) -> List[dict]:
     return fields
 
 
-def create_text_field(app_token: str, table_id: str, token: str, field_name: str) -> dict:
+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_TEXT,
+        "type": field_type,
     }
     result = feishu_request("POST", url, token=token, payload=payload)
-    return result["data"]["field"]
+    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 normalize_value(raw: Optional[str]) -> Optional[str]:
@@ -178,6 +284,43 @@ def read_csv_rows(csv_path: Path) -> Tuple[List[str], List[Dict[str, Optional[st
     return headers, rows
 
 
+def normalize_db_value(value) -> Optional[str]:
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        return value.strftime("%Y-%m-%d %H:%M:%S+08")
+    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,
@@ -191,7 +334,13 @@ def ensure_fields(
     for header in csv_headers:
         if header not in field_map:
             try:
-                created = create_text_field(app_token, table_id, token, header)
+                created = create_field(
+                    app_token,
+                    table_id,
+                    token,
+                    header,
+                    FIELD_TYPE_HINTS.get(header, FIELD_TYPE_TEXT),
+                )
                 field_map[header] = created
                 print(f"已创建字段: {header}")
             except RuntimeError as exc:
@@ -268,17 +417,25 @@ def main() -> int:
     app_id = os.getenv("FEISHU_APP_ID")
     app_secret = os.getenv("FEISHU_APP_SECRET")
     base_url = os.getenv("FEISHU_BASE_URL")
+    source = os.getenv("DATA_SOURCE", DEFAULT_SOURCE).strip().lower()
     csv_path = Path(os.getenv("CSV_PATH", str(DEFAULT_CSV_PATH)))
     batch_size = int(os.getenv("BATCH_SIZE", str(DEFAULT_BATCH_SIZE)))
 
     if not app_id or not app_secret or not base_url:
         print("缺少 FEISHU_APP_ID / FEISHU_APP_SECRET / FEISHU_BASE_URL", file=sys.stderr)
         return 1
-    if not csv_path.exists():
+    if source not in {"db", "csv"}:
+        print(f"不支持的数据源: {source}，仅支持 db / csv", file=sys.stderr)
+        return 1
+    if source == "csv" and not csv_path.exists():
         print(f"CSV 文件不存在: {csv_path}", file=sys.stderr)
         return 1
 
     app_token, table_id, _view_id = parse_base_url(base_url)
+    if source == "db":
+        headers, rows = query_db_rows(DB_QUERY)
+        print(f"读取数据库完成，共 {len(rows)} 行")
+    else:
         headers, rows = read_csv_rows(csv_path)
         print(f"读取 CSV 完成: {csv_path}，共 {len(rows)} 行")