#!/usr/bin/env python3 import csv import json import os import sys import time from datetime import datetime from pathlib import Path from typing import Dict, Iterable, List, Optional, Tuple from urllib import error, parse, request ENV_PATH = Path(".env") DEFAULT_CSV_PATH = Path("物流.csv") DEFAULT_BATCH_SIZE = 500 # 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 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 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, ) -> 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") 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") except error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"Feishu API 请求失败: {exc.code} {detail}") from exc except error.URLError as exc: raise RuntimeError(f"无法连接飞书 API: {exc}") from exc result = json.loads(body) if result.get("code") != 0: raise RuntimeError( f"Feishu API 返回错误: code={result.get('code')} msg={result.get('msg')} body={body}" ) return result 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 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 = result.get("data", {}) fields.extend(data.get("items", [])) if not data.get("has_more"): break page_token = data.get("page_token") return fields def create_text_field(app_token: str, table_id: str, token: str, field_name: str) -> 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, } result = feishu_request("POST", url, token=token, payload=payload) return result["data"]["field"] def normalize_value(raw: Optional[str]) -> Optional[str]: if raw is None: return None value = raw.strip() if not value or value.upper() == "NULL": return None return value 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: # CSV 中是 `2026-03-15 14:20:30+08` 这种格式，转换成毫秒时间戳。 try: dt = datetime.strptime(value, "%Y-%m-%d %H:%M:%S+08") return int(dt.timestamp() * 1000) except ValueError: return value return value def read_csv_rows(csv_path: Path) -> Tuple[List[str], List[Dict[str, Optional[str]]]]: with csv_path.open("r", newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) headers = reader.fieldnames or [] rows: List[Dict[str, Optional[str]]] = [] for row in reader: rows.append({key: normalize_value(value) for key, value in row.items()}) return headers, rows 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_text_field(app_token, table_id, token, header) field_map[header] = created print(f"已创建字段: {header}") except RuntimeError as exc: missing_headers.append(header) print(f"跳过缺失字段: {header} ({exc})") 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): feishu_request("POST", url, token=token, payload={"records": batch}) created += len(batch) print(f"已写入批次 {batch_index}/{total_batches}，累计 {created} 条") time.sleep(0.15) return created def print_permission_help(message: str) -> None: print("飞书返回权限不足，当前应用还不能向这个多维表新增字段或记录。", file=sys.stderr) print(f"原始错误: {message}", file=sys.stderr) print( "请在飞书开放平台或多维表权限设置中，确认这个应用已被添加到该 Base，且具备多维表读取/写入权限。", file=sys.stderr, ) print( "如果你手动补齐缺失字段并授权写入，重新执行 `python3 import_logistics_to_feishu_bitable.py` 即可继续导入。", file=sys.stderr, ) def main() -> 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") 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(): 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)} 行") try: token = get_tenant_access_token(app_id, app_secret) field_map, missing_headers = ensure_fields(app_token, table_id, token, headers) records = build_records(rows, field_map, missing_headers) 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)) return 2 raise if missing_headers: print(f"以下字段未写入，请在多维表中手动创建后重跑: {', '.join(missing_headers)}") print(f"导入完成，共写入 {created} 条记录") return 0 if __name__ == "__main__": raise SystemExit(main())