From 2822c9b55849c1e50981955a1e5837b444863a90 Mon Sep 17 00:00:00 2001 From: BXY0414 <1079823339@qq.com> Date: Wed, 25 Mar 2026 18:01:45 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B0=83=E6=95=B4=E6=9F=A5=E8=AF=A2SQL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- import_logistics_to_feishu_bitable.py | 173 ++++++++++++++++++++++++-- 1 file changed, 165 insertions(+), 8 deletions(-) diff --git a/import_logistics_to_feishu_bitable.py b/import_logistics_to_feishu_bitable.py index 5a25468..aa27596 100644 --- a/import_logistics_to_feishu_bitable.py +++ b/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,19 +417,27 @@ 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) - headers, rows = read_csv_rows(csv_path) - print(f"读取 CSV 完成: {csv_path}，共 {len(rows)} 行") + 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)} 行") try: token = get_tenant_access_token(app_id, app_secret)